#!/usr/bin/env python3 """Normalize raw NPCI JSON fetches into Projects/payments-stat-hub/data/payments.duckdb.""" import json, re, calendar from pathlib import Path import duckdb RAW = Path("/home/workspace/Projects/payments-stat-hub/raw/npci") DB = Path("/home/workspace/Projects/payments-stat-hub/data/payments.duckdb") MONTHS = {m.lower(): i for i, m in enumerate(calendar.month_name) if m} def num(v): if v is None: return None s = str(v).replace(",", "").strip() if s in ("", "-", "NA", "N/A"): return None try: return float(s) except ValueError: return None def month_label_to_date(label): m = re.match(r"^([A-Za-z]+)-(\d{4})$", str(label).strip()) if not m or m.group(1).lower() not in MONTHS: return None return f"{m.group(2)}-{MONTHS[m.group(1).lower()]:02d}-01" def npci_day_to_date(label): m = re.match(r"^([A-Za-z]+) (\d{1,2}), (\d{4})$", str(label).strip()) if not m or m.group(1).lower() not in MONTHS: return None return f"{m.group(3)}-{MONTHS[m.group(1).lower()]:02d}-{int(m.group(2)):02d}" def file_key_to_date(key): m = re.match(r"^(\d{4})-([A-Za-z]+)$", key) if not m or m.group(2).lower() not in MONTHS: return None return f"{m.group(1)}-{MONTHS[m.group(2).lower()]:02d}-01" META = {"id", "createdAt", "updatedAt", "publishedAt", "locale", "product_name", "tab_name", "row_number", "month", "year"} def load_rows(path): try: d = json.loads(path.read_text()) except Exception: return [] return (d.get("data") or {}).get("results") or [] def pick(row, pattern): for k, v in row.items(): if re.search(pattern, k, re.I) and k not in META: return v return None def main(): DB.parent.mkdir(parents=True, exist_ok=True) if DB.exists(): DB.unlink() con = duckdb.connect(str(DB)) con.execute(""" CREATE TABLE npci_product_monthly ( product TEXT, month DATE, volume_mn DOUBLE, value_cr DOUBLE, banks INT, payload JSON, src_file TEXT)""") con.execute(""" CREATE TABLE npci_trended_monthly ( product TEXT, month DATE, total_volume_mn DOUBLE, total_value_cr DOUBLE, avg_volume DOUBLE, avg_value DOUBLE, payload JSON, src_file TEXT)""") con.execute(""" CREATE TABLE npci_daily ( product TEXT, day DATE, volume_mn DOUBLE, value_cr DOUBLE, payload JSON, src_file TEXT)""") con.execute(""" CREATE TABLE npci_other ( product TEXT, stream TEXT, period DATE, n_rows INT, rows JSON, src_file TEXT)""") con.execute(""" CREATE TABLE fetch_log ( ts TEXT, url TEXT, file TEXT, status INT, total INT, rows INT)""") counts = {"product_monthly": 0, "trended_monthly": 0, "daily": 0, "other": 0, "files": 0} for path in sorted(RAW.rglob("*.json")): if path.name == "tabs-catalog.json": continue product = path.relative_to(RAW).parts[0].lower() stream = path.parent.name.lower() src = str(path.relative_to(RAW)) rows = load_rows(path) if not rows: continue counts["files"] += 1 r0 = rows[0] keys = set(r0.keys()) if "npci_day" in keys: fam = "daily" elif "total_volume" in keys: fam = "trended" elif any(re.match(r"volume", k, re.I) for k in keys - META): fam = "product_monthly" else: fam = "other" if fam == "daily": for r in rows: day = npci_day_to_date(r.get("npci_day")) if not day: continue con.execute("INSERT INTO npci_daily VALUES (?,?,?,?,?,?)", [product, day, num(pick(r, r"^upi_volume|^volume")), num(pick(r, r"^upi_value|^value")), json.dumps(r, ensure_ascii=False), src]) counts["daily"] += 1 elif fam == "trended": for r in rows: mth = month_label_to_date(r.get("month")) if not mth: continue con.execute("INSERT INTO npci_trended_monthly VALUES (?,?,?,?,?,?,?,?)", [product, mth, num(r.get("total_volume")), num(r.get("total_value")), num(r.get("avg_volume")), num(r.get("avg_value")), json.dumps(r, ensure_ascii=False), src]) counts["trended_monthly"] += 1 elif fam == "product_monthly": for r in rows: mth = month_label_to_date(r.get("month")) if not mth: continue banks = None for k, v in r.items(): if "banks" in k.lower(): banks = int(num(v)) if num(v) is not None else None break con.execute("INSERT INTO npci_product_monthly VALUES (?,?,?,?,?,?,?)", [product, mth, num(pick(r, r"volume")), num(pick(r, r"value")), banks, json.dumps(r, ensure_ascii=False), src]) counts["product_monthly"] += 1 else: period = file_key_to_date(path.stem) con.execute("INSERT INTO npci_other VALUES (?,?,?,?,?,?)", [product, stream, period, len(rows), json.dumps(rows, ensure_ascii=False), src]) counts["other"] += len(rows) for line in open(RAW / "manifest.jsonl"): try: r = json.loads(line) con.execute("INSERT INTO fetch_log VALUES (?,?,?,?,?,?)", [r.get("ts"), r.get("url"), r.get("file"), r.get("status"), r.get("totalCount"), r.get("rows")]) except Exception: pass con.execute("""CREATE VIEW rail_monthly AS SELECT product, month, volume_mn, value_cr, 'product_stats' AS src FROM npci_product_monthly UNION ALL SELECT product, month, total_volume_mn, total_value_cr, 'trended' AS src FROM npci_trended_monthly""") con.execute("""CREATE OR REPLACE TABLE npci_product_monthly_dedup AS SELECT * EXCLUDE (rn) FROM ( SELECT *, ROW_NUMBER() OVER ( PARTITION BY product, month ORDER BY (src_file LIKE '%/product-monthly/%') DESC) AS rn FROM npci_product_monthly) WHERE rn = 1""") con.execute("""CREATE VIEW upi_monthly AS SELECT month, banks, volume_mn, value_cr FROM npci_product_monthly_dedup WHERE product='upi' ORDER BY month""") print(counts) for t in ("npci_product_monthly", "npci_trended_monthly", "npci_daily", "npci_other", "fetch_log"): n = con.execute(f"SELECT COUNT(*) FROM {t}").fetchone()[0] print(f"{t}: {n:,}") print("UPI monthly range:", con.execute( "SELECT MIN(month), MAX(month), COUNT(*) FROM npci_product_monthly_dedup WHERE product='upi'").fetchone()) print("UPI daily range:", con.execute( "SELECT MIN(day), MAX(day), COUNT(*) FROM npci_daily WHERE product='upi'").fetchone()) con.close() if __name__ == "__main__": main()