Data: profile, join locally, pull public feeds

Download .py · csv_profile.py

#!/usr/bin/env python3
"""BSV recipe: profile a CSV — column types, nulls, distinct counts, sample values.

Input : path to a CSV (default: writes and profiles a labeled synthetic sample).
Output: profile.csv + profile.md in the working directory.
Original BSV code, MIT. Synthetic sample rows are labeled examples, not customer or production data.
"""
import csv, os, sys
from collections import Counter

SAMPLE = """id,city,temp_c,note
1,Tokyo,22.1,example-row
2,Osaka,,example-row
3,Tokyo,19.4,example-row
4,Nagoya,21.0,example-row
5,Osaka,18.7,example-row
"""

def is_float(s):
    try:
        float(s); return True
    except Exception:
        return False

def main(path=None):
    if not path:
        path = "sample_labeled.csv"
        open(path, "w", encoding="utf-8").write(SAMPLE)
        print("wrote labeled synthetic sample:", path)
    rows = list(csv.DictReader(open(path, encoding="utf-8", newline="")))
    if not rows:
        raise SystemExit("empty csv")
    cols = list(rows[0].keys())
    out_rows = []
    lines = [f"# CSV profile for `{path}`", "", f"Rows: {len(rows)}  Columns: {len(cols)}", "",
             "Synthetic or practice files should stay labeled as such; do not treat sample numbers as live metrics.", ""]
    for c in cols:
        vals = [r.get(c, "") for r in rows]
        empty = sum(1 for v in vals if v is None or str(v).strip() == "")
        filled = [str(v).strip() for v in vals if v is not None and str(v).strip() != ""]
        numeric = all(is_float(v) for v in filled) if filled else False
        distinct = len(set(filled))
        top = Counter(filled).most_common(3)
        samples = ", ".join(repr(v) for v, _ in top) if top else ""
        out_rows.append({"column": c, "non_null": len(filled), "nulls": empty,
                         "distinct": distinct, "numeric": str(numeric).lower(), "top_values": samples})
        lines.append(f"## {c}")
        lines.append(f"- non-null: {len(filled)} / nulls: {empty} / distinct: {distinct} / numeric: {numeric}")
        lines.append(f"- top: {samples}")
        lines.append("")
    with open("profile.csv", "w", newline="", encoding="utf-8") as f:
        w = csv.DictWriter(f, fieldnames=["column", "non_null", "nulls", "distinct", "numeric", "top_values"])
        w.writeheader(); w.writerows(out_rows)
    open("profile.md", "w", encoding="utf-8").write("\n".join(lines) + "\n")
    print(f"profiled {len(rows)} rows × {len(cols)} cols -> profile.csv, profile.md")

if __name__ == "__main__":
    main(*sys.argv[1:])

Run on 2026-10-07 02:01 JST · Python 3.13.5

wrote labeled synthetic sample: sample_labeled.csv
profiled 5 rows × 4 cols -> profile.csv, profile.md

Download .py · jsonl_flatten.py

#!/usr/bin/env python3
"""BSV recipe: flatten a JSON Lines file into a CSV of top-level keys.

Input : path to .jsonl (default: writes a labeled synthetic sample).
Output: flat.csv
Original BSV code, MIT. Nested objects become JSON strings; arrays become '|'-joined text.
"""
import csv, json, os, sys

SAMPLE = [
    {"id": 1, "tag": "example-row", "metrics": {"a": 1, "b": 2}, "labels": ["x", "y"]},
    {"id": 2, "tag": "example-row", "metrics": {"a": 3}, "labels": ["z"]},
    {"id": 3, "tag": "example-row", "metrics": {}, "labels": []},
]

def cell(v):
    if isinstance(v, dict):
        return json.dumps(v, ensure_ascii=False, sort_keys=True)
    if isinstance(v, list):
        return "|".join(str(x) for x in v)
    return v

def main(path=None):
    if not path:
        path = "sample_labeled.jsonl"
        with open(path, "w", encoding="utf-8") as f:
            for row in SAMPLE:
                f.write(json.dumps(row, ensure_ascii=False) + "\n")
        print("wrote labeled synthetic sample:", path)
    rows = [json.loads(line) for line in open(path, encoding="utf-8") if line.strip()]
    keys = []
    for r in rows:
        for k in r:
            if k not in keys:
                keys.append(k)
    with open("flat.csv", "w", newline="", encoding="utf-8") as f:
        w = csv.DictWriter(f, fieldnames=keys)
        w.writeheader()
        for r in rows:
            w.writerow({k: cell(r.get(k)) for k in keys})
    print(f"flattened {len(rows)} lines → flat.csv ({len(keys)} columns)")

if __name__ == "__main__":
    main(*sys.argv[1:])

Run on 2026-10-07 02:01 JST · Python 3.13.5

wrote labeled synthetic sample: sample_labeled.jsonl
flattened 3 lines → flat.csv (4 columns)

Download .py · worldbank_series.py

#!/usr/bin/env python3
"""BSV recipe: download one World Bank indicator series for one country into CSV.

Input : country ISO2 (default JP), indicator id (default SP.POP.TOTL = population).
Output: wb_series.csv
Original BSV code, MIT. Data: World Bank Open Data API (CC BY 4.0 — cite World Bank).
"""
import csv, json, sys, urllib.parse, urllib.request

def main(country="JP", indicator="SP.POP.TOTL"):
    q = urllib.parse.urlencode({"format": "json", "per_page": "20000"})
    url = f"https://api.worldbank.org/v2/country/{country}/indicator/{indicator}?{q}"
    with urllib.request.urlopen(url, timeout=45) as r:
        payload = json.load(r)
    meta, rows = payload[0], payload[1] or []
    keep = [x for x in rows if x.get("value") is not None]
    keep.sort(key=lambda x: str(x.get("date") or ""))
    with open("wb_series.csv", "w", newline="", encoding="utf-8") as f:
        w = csv.writer(f)
        w.writerow(["country", "indicator", "year", "value"])
        for x in keep:
            w.writerow([country, indicator, x.get("date"), x.get("value")])
    print(f"{country} {indicator}: {len(keep)} yearly points -> wb_series.csv (source: World Bank Open Data)")
    if keep:
        print("latest:", keep[-1].get("date"), keep[-1].get("value"))

if __name__ == "__main__":
    main(*sys.argv[1:])

Run on 2026-10-07 02:01 JST · Python 3.13.5

JP SP.POP.TOTL: 66 yearly points -> wb_series.csv (source: World Bank Open Data)
latest: 2025 123366734

Download .py · open_meteo_daily.py

#!/usr/bin/env python3
"""BSV recipe: fetch recent daily weather for a lat/lon from Open-Meteo (no API key).

Input : latitude, longitude (default Tokyo 35.68,139.76), past_days (default 7).
Output: weather_daily.csv
Original BSV code, MIT. Data: Open-Meteo (CC BY 4.0 — cite Open-Meteo). Not a forecast product claim.
"""
import csv, json, sys, urllib.parse, urllib.request

def main(lat="35.68", lon="139.76", past_days="7"):
    q = urllib.parse.urlencode({
        "latitude": lat, "longitude": lon, "past_days": past_days,
        "daily": "temperature_2m_max,temperature_2m_min,precipitation_sum",
        "timezone": "auto",
    })
    url = "https://api.open-meteo.com/v1/forecast?" + q
    with urllib.request.urlopen(url, timeout=45) as r:
        data = json.load(r)
    daily = data.get("daily") or {}
    days = daily.get("time") or []
    with open("weather_daily.csv", "w", newline="", encoding="utf-8") as f:
        w = csv.writer(f)
        w.writerow(["date", "temp_max_c", "temp_min_c", "precip_mm", "lat", "lon"])
        for i, d in enumerate(days):
            w.writerow([
                d,
                (daily.get("temperature_2m_max") or [None])[i],
                (daily.get("temperature_2m_min") or [None])[i],
                (daily.get("precipitation_sum") or [None])[i],
                lat, lon,
            ])
    print(f"{len(days)} daily rows for {lat},{lon} -> weather_daily.csv (source: Open-Meteo)")

if __name__ == "__main__":
    main(*sys.argv[1:])

Run on 2026-10-07 02:01 JST · Python 3.13.5

14 daily rows for 35.68,139.76 -> weather_daily.csv (source: Open-Meteo)

Download .py · csv_schema_diff.py

#!/usr/bin/env python3
"""BSV recipe: diff two CSV schemas — shared, only-left, only-right columns + type guesses.

Input : two CSV paths (default: writes two labeled synthetic samples that intentionally differ).
Output: schema_diff.csv + schema_diff.md
Original BSV code, MIT. Synthetic rows are labeled examples, not customer data.
"""
import csv, os, sys
from collections import OrderedDict

A = """id,city,temp_c,note
1,Tokyo,22.1,example-row
2,Osaka,19.0,example-row
"""
B = """id,city,humidity_pct,note,sensor
1,Tokyo,55,example-row,demo
2,Kyoto,60,example-row,demo
"""

def guess(vals):
    filled = [v.strip() for v in vals if v is not None and str(v).strip() != ""]
    if not filled:
        return "empty"
    try:
        for v in filled:
            float(v)
        return "numeric"
    except Exception:
        return "text"

def cols(path):
    rows = list(csv.DictReader(open(path, encoding="utf-8", newline="")))
    if not rows:
        return OrderedDict()
    out = OrderedDict()
    for c in rows[0].keys():
        out[c] = guess([r.get(c, "") for r in rows])
    return out

def main(left=None, right=None):
    if not left or not right:
        left, right = "sample_a_labeled.csv", "sample_b_labeled.csv"
        open(left, "w", encoding="utf-8").write(A)
        open(right, "w", encoding="utf-8").write(B)
        print("wrote labeled synthetic pair:", left, right)
    la, rb = cols(left), cols(right)
    sa, sb = set(la), set(rb)
    rows = []
    for c in list(la) + [c for c in rb if c not in la]:
        side = "both" if c in sa and c in sb else ("left_only" if c in sa else "right_only")
        rows.append({"column": c, "side": side, "left_type": la.get(c, ""), "right_type": rb.get(c, ""),
                     "type_match": str(la.get(c) == rb.get(c)).lower() if side == "both" else ""})
    with open("schema_diff.csv", "w", newline="", encoding="utf-8") as f:
        w = csv.DictWriter(f, fieldnames=["column", "side", "left_type", "right_type", "type_match"])
        w.writeheader(); w.writerows(rows)
    lines = [f"# Schema diff `{left}` vs `{right}`", "",
             f"Shared: {len(sa & sb)}  Left-only: {len(sa - sb)}  Right-only: {len(sb - sa)}", "",
             "Synthetic practice files stay labeled; do not treat sample numbers as live metrics.", ""]
    for r in rows:
        lines.append(f"- `{r['column']}` · {r['side']} · L={r['left_type'] or '-'} R={r['right_type'] or '-'}")
    open("schema_diff.md", "w", encoding="utf-8").write("\n".join(lines) + "\n")
    print(f"diff {len(rows)} columns -> schema_diff.csv, schema_diff.md")

if __name__ == "__main__":
    main(*(sys.argv[1:3] or []))

Run on 2026-10-07 02:13 JST · Python 3.13.5

wrote labeled synthetic pair: sample_a_labeled.csv sample_b_labeled.csv
diff 6 columns -> schema_diff.csv, schema_diff.md

Download .py · sqlite_from_csv.py

#!/usr/bin/env python3
"""BSV recipe: load a CSV into SQLite and run a small GROUP BY demo.

Input : optional CSV path (default: labeled synthetic sample).
Output: practice.sqlite + groupby.csv
Original BSV code, MIT. Stdlib sqlite3 + csv only.
"""
import csv, os, sqlite3, sys

SAMPLE = """id,city,temp_c,note
1,Tokyo,22.1,example-row
2,Osaka,19.0,example-row
3,Tokyo,21.5,example-row
4,Nagoya,20.0,example-row
5,Osaka,18.2,example-row
"""

def main(path=None):
    if not path:
        path = "sample_labeled.csv"
        open(path, "w", encoding="utf-8").write(SAMPLE)
        print("wrote labeled synthetic sample:", path)
    rows = list(csv.DictReader(open(path, encoding="utf-8", newline="")))
    if not rows:
        raise SystemExit("empty csv")
    cols = list(rows[0].keys())
    db = "practice.sqlite"
    if os.path.exists(db):
        os.remove(db)
    con = sqlite3.connect(db)
    # quote identifiers safely for this demo (alphanumeric + underscore only)
    safe = [c for c in cols if c.replace("_", "").isalnum()]
    if len(safe) != len(cols):
        raise SystemExit("column names must be alphanumeric/underscore for this demo loader")
    con.execute(f"CREATE TABLE t ({', '.join(c + ' TEXT' for c in safe)})")
    con.executemany(f"INSERT INTO t VALUES ({','.join('?' for _ in safe)})",
                    [[r.get(c, "") for c in safe] for r in rows])
    con.commit()
    # Prefer a city-like column for the demo aggregate
    group_col = "city" if "city" in safe else safe[1 if len(safe) > 1 else 0]
    q = f"SELECT {group_col} AS grp, COUNT(*) AS n FROM t GROUP BY {group_col} ORDER BY n DESC, grp"
    out = list(con.execute(q))
    with open("groupby.csv", "w", newline="", encoding="utf-8") as f:
        w = csv.writer(f); w.writerow(["grp", "n"]); w.writerows(out)
    con.close()
    print(f"loaded {len(rows)} rows into {db}; groupby on {group_col} -> groupby.csv ({len(out)} groups)")

if __name__ == "__main__":
    main(*sys.argv[1:2])

Run on 2026-10-07 02:13 JST · Python 3.13.5

wrote labeled synthetic sample: sample_labeled.csv
loaded 5 rows into practice.sqlite; groupby on city -> groupby.csv (3 groups)

Download .py · frankfurter_fx.py

#!/usr/bin/env python3
"""BSV recipe: pull latest FX rates from the Frankfurter API (ECB reference rates, no key).

Input : base currency (default USD) and comma-separated quote list (default JPY,EUR,GBP).
Output: fx_latest.csv with base, quote, rate, as_of_date, source
Original BSV code, MIT. Cite Frankfurter / ECB if you republish numbers.
"""
import csv, json, sys, urllib.request

def main(base="USD", quotes="JPY,EUR,GBP"):
    base = base.upper().strip()
    qlist = [q.strip().upper() for q in quotes.split(",") if q.strip()]
    url = f"https://api.frankfurter.app/latest?from={base}&to={','.join(qlist)}"
    req = urllib.request.Request(url, headers={"User-Agent": "BSV-fields-recipe/1.0 (practice; +https://botshelfvampire.com)"})
    with urllib.request.urlopen(req, timeout=30) as r:
        data = json.loads(r.read().decode("utf-8"))
    as_of = data.get("date", "")
    rates = data.get("rates") or {}
    rows = [{"base": base, "quote": q, "rate": rates[q], "as_of_date": as_of, "source": "frankfurter.app / ECB"}
            for q in qlist if q in rates]
    if not rows:
        raise SystemExit(f"no rates returned for {base} -> {qlist}: {data}")
    with open("fx_latest.csv", "w", newline="", encoding="utf-8") as f:
        w = csv.DictWriter(f, fieldnames=["base", "quote", "rate", "as_of_date", "source"])
        w.writeheader(); w.writerows(rows)
    print(f"wrote {len(rows)} FX rows as_of {as_of} -> fx_latest.csv (source Frankfurter/ECB)")

if __name__ == "__main__":
    main(*sys.argv[1:3])

Run on 2026-10-07 02:13 JST · Python 3.13.5

wrote 3 FX rows as_of 2026-10-06 -> fx_latest.csv (source Frankfurter/ECB)

Download .py · usgs_quakes_day.py

#!/usr/bin/env python3
"""BSV recipe: flatten USGS past-day M2.5+ earthquakes GeoJSON into a CSV.

Input : optional feed URL (default: USGS 2.5_day summary).
Output: quakes_day.csv with time_utc, mag, place, lon, lat, depth_km, url
Original BSV code, MIT. USGS data are public domain; BSV does not alter magnitudes.
"""
import csv, json, sys, urllib.request
from datetime import datetime, timezone

DEFAULT = "https://earthquake.usgs.gov/earthquakes/feed/v1.0/summary/2.5_day.geojson"

def main(url=DEFAULT):
    with urllib.request.urlopen(url, timeout=45) as r:
        data = json.loads(r.read().decode("utf-8"))
    feats = data.get("features") or []
    rows = []
    for f in feats:
        p = f.get("properties") or {}
        g = (f.get("geometry") or {}).get("coordinates") or [None, None, None]
        lon, lat, depth = (g + [None, None, None])[:3]
        ms = p.get("time")
        t = datetime.fromtimestamp(ms / 1000, tz=timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") if ms else ""
        rows.append({"time_utc": t, "mag": p.get("mag"), "place": p.get("place"),
                     "lon": lon, "lat": lat, "depth_km": depth, "url": p.get("url")})
    rows.sort(key=lambda r: r["time_utc"] or "", reverse=True)
    with open("quakes_day.csv", "w", newline="", encoding="utf-8") as f:
        w = csv.DictWriter(f, fieldnames=["time_utc", "mag", "place", "lon", "lat", "depth_km", "url"])
        w.writeheader(); w.writerows(rows)
    print(f"wrote {len(rows)} events from USGS feed -> quakes_day.csv (public domain USGS)")

if __name__ == "__main__":
    main(*sys.argv[1:2])

Run on 2026-10-07 02:13 JST · Python 3.13.5

wrote 37 events from USGS feed -> quakes_day.csv (public domain USGS)

Download .py · nager_public_holidays.py

#!/usr/bin/env python3
"""BSV recipe: public holidays for a country/year (Nager.Date).

Practice / education only.
Input : country ISO (default JP), year (default 2026).
Output: public_holidays.csv.
Original BSV code, MIT. Data: date.nager.at.
"""
import csv, json, sys, urllib.request
UA = {"User-Agent": "BSV-fields-recipe/1.0 (education; +https://botshelfvampire.com)"}

def main(country="JP", year="2026"):
    country, year = country.strip().upper(), int(year)
    url = f"https://date.nager.at/api/v3/PublicHolidays/{year}/{country}"
    req = urllib.request.Request(url, headers=UA)
    with urllib.request.urlopen(req, timeout=30) as r:
        data = json.loads(r.read().decode())
    if not isinstance(data, list) or not data:
        raise SystemExit("empty holiday list")
    rows = [{"date": x.get("date"), "localName": x.get("localName"), "name": x.get("name"),
             "countryCode": x.get("countryCode"), "global": x.get("global")} for x in data]
    with open("public_holidays.csv", "w", newline="", encoding="utf-8") as f:
        w = csv.DictWriter(f, fieldnames=["date", "localName", "name", "countryCode", "global"]); w.writeheader(); w.writerows(rows)
    print(f"wrote {len(rows)} holidays {country} {year} -> public_holidays.csv")

if __name__ == "__main__":
    main(*sys.argv[1:3])

Run on 2026-10-07 03:20 JST · Python 3.13.5

wrote 16 holidays JP 2026 -> public_holidays.csv

Download .py · open_meteo_air_quality.py

#!/usr/bin/env python3
"""BSV recipe: Open-Meteo air-quality hourly sample for a lat/lon.

Practice / education only — not an official AQI product.
Input : lat lon (defaults 35.68 139.76), hours (default 24).
Output: air_quality.csv.
Original BSV code, MIT. Data: air-quality-api.open-meteo.com.
"""
import csv, json, sys, urllib.parse, urllib.request
UA = {"User-Agent": "BSV-fields-recipe/1.0 (education; +https://botshelfvampire.com)"}

def main(lat="35.68", lon="139.76", hours="24"):
    lat, lon, hours = float(lat), float(lon), int(hours)
    q = urllib.parse.urlencode({
        "latitude": lat, "longitude": lon,
        "hourly": "pm10,pm2_5,european_aqi",
        "forecast_days": 1,
    })
    url = f"https://air-quality-api.open-meteo.com/v1/air-quality?{q}"
    req = urllib.request.Request(url, headers=UA)
    with urllib.request.urlopen(req, timeout=45) as r:
        d = json.loads(r.read().decode())
    h = d.get("hourly") or {}
    times = h.get("time") or []
    rows = []
    for i, t in enumerate(times[:hours]):
        rows.append({"time": t, "pm10": (h.get("pm10") or [None])[i],
                     "pm2_5": (h.get("pm2_5") or [None])[i],
                     "european_aqi": (h.get("european_aqi") or [None])[i]})
    with open("air_quality.csv", "w", newline="", encoding="utf-8") as f:
        w = csv.DictWriter(f, fieldnames=["time", "pm10", "pm2_5", "european_aqi"]); w.writeheader(); w.writerows(rows)
    print(f"wrote {len(rows)} hourly AQ rows -> air_quality.csv")

if __name__ == "__main__":
    main(*sys.argv[1:4])

Run on 2026-10-07 03:20 JST · Python 3.13.5

wrote 24 hourly AQ rows -> air_quality.csv

Download .py · csv_value_counts.py

#!/usr/bin/env python3
"""BSV recipe: value_counts for one column of an embedded practice CSV.

Practice data hygiene only — browser Field Lab CSV profile is the interactive twin.
Input : column name (default city).
Output: value_counts.csv.
Original BSV code, MIT. Dependency: none.
"""
import csv, io, sys

SAMPLE = """id,city,temp_c
1,Tokyo,22.1
2,Osaka,18.7
3,Tokyo,19.4
4,Nagoya,21.0
5,Osaka,18.7
6,Tokyo,20.2
"""

def main(column="city"):
    rows = list(csv.DictReader(io.StringIO(SAMPLE)))
    if column not in rows[0]:
        raise SystemExit(f"column {column!r} missing")
    counts = {}
    for r in rows:
        v = r[column]
        counts[v] = counts.get(v, 0) + 1
    out = [{"value": k, "count": v, "column": column} for k, v in sorted(counts.items(), key=lambda kv: -kv[1])]
    with open("value_counts.csv", "w", newline="", encoding="utf-8") as f:
        w = csv.DictWriter(f, fieldnames=["column", "value", "count"]); w.writeheader(); w.writerows(out)
    print(f"wrote {len(out)} distinct values for {column!r} -> value_counts.csv")

if __name__ == "__main__":
    main(*sys.argv[1:2])

Run on 2026-10-07 03:20 JST · Python 3.13.5

wrote 3 distinct values for 'city' -> value_counts.csv

Download .py · sha256_lines_manifest.py

#!/usr/bin/env python3
"""BSV recipe: SHA-256 each non-empty line of stdin text embedded as practice lines.

Practice integrity hygiene only.
Input : none (uses embedded lines); optional label (default practice).
Output: sha256_manifest.csv.
Original BSV code, MIT. Dependency: hashlib.
"""
import csv, hashlib, sys

LINES = [
    "alpha-practice-row",
    "beta-practice-row",
    "gamma-practice-row",
]

def main(label="practice"):
    rows = []
    for i, line in enumerate(LINES):
        h = hashlib.sha256(line.encode()).hexdigest()
        rows.append({"label": label, "index": i, "sha256": h, "nbytes": len(line.encode())})
    with open("sha256_manifest.csv", "w", newline="", encoding="utf-8") as f:
        w = csv.DictWriter(f, fieldnames=["label", "index", "sha256", "nbytes"]); w.writeheader(); w.writerows(rows)
    print(f"wrote {len(rows)} line digests -> sha256_manifest.csv")

if __name__ == "__main__":
    main(*sys.argv[1:2])

Run on 2026-10-07 03:20 JST · Python 3.13.5

wrote 3 line digests -> sha256_manifest.csv