#!/usr/bin/env python3
"""
Clean the raw FARS 2020-2022 CSVs into a set of graph-shaped CSVs, and generate the Neo4j / TigerGraph import scripts from them.

It does exactly two things:
  1. Raw tables -> csv/nodes/<Label>.tsv and csv/edges/<Src>_<REL>_<Dst>.tsv (one file per label / per edge kind)
  2. From the inferred column types, generate neo4j/*.cypher, tigergraph/*.gsql, expected-counts.tsv, schema.json
The data itself is not loaded through Python: Neo4j uses LOAD CSV, TigerGraph uses a loading job, HDFS uses a plain put.

Usage:
    python3 build_csv.py            # reads ../<year>/National/*.csv, writes into this directory

Cleaning rules (see README.md for details):
  * Encoding: each file is decoded as UTF-8 first, falling back to cp1252 (2020-2022 contain both). Output is always UTF-8 with LF.
  * Text: NBSP -> space, \\x1a / en and em dashes -> '-', curly quotes -> straight quotes, TAB/newline -> space, strip leading/trailing whitespace.
  * Output format: TAB-separated, unquoted (see write_tsv). TigerGraph's loader does not understand RFC 4180 "" escaping,
    and the data has many inch marks ", so comma + quote CSV is not used.
  * Column names: all upper case (vpicdecode 2020 is camelCase). Columns are unioned across years; columns missing in a year are left empty.
    UNDERIDE (2020) and UNDEROVERRIDE (2021+) use different coding schemes, so they are kept as two columns, not merged.
  * Columns that clash with GSQL reserved words are renamed: ORDER -> RACE_ORDER, TRIM -> TRIM1.
  * IDs: accident_id=<year>_<ST_CASE>, vehicle_id=<accident_id>_<VEH_NO>,
        person_id=<vehicle_id>_<PER_NO>, event_id=<accident_id>_<EVENTNUM> (same as the old database).
  * 1:1 side tables are folded into their main node: pbtype / safetyeq / miper -> Person, miacc / midrvacc -> Accident.
  * parkwork (parked/working vehicles) is folded into Vehicle: column names are mapped back to vehicle column names using the old fars database's rules
    (PARKWORK_COLUMN_MAP; all others drop the P prefix), and source_table records the origin (vehicle / parkwork).
    VEH_NO in both tables shares one numbering space with no overlap; pvehiclesf is folded into vehiclesf the same way.
  * Edge rows whose source or target does not exist are dropped, and duplicate edges for the same (src, dst) are removed; both counts go into build-report.txt.
"""

from __future__ import annotations

import csv
import io
import json
import re
import sys
import time
from collections import Counter, defaultdict
from pathlib import Path

import pandas as pd

YEARS = (2020, 2021, 2022)
HERE = Path(__file__).resolve().parent
RAW = HERE.parent                       # /datasets/prop_datasets/fars
CSV_DIR = HERE / "csv"

NEO4J_DB = "fars20-22"
NEO4J_IMPORT_ROOT = Path("/datasets/prop_datasets")   # neo4j.conf: server.directories.import
TG_GRAPH = "fars20_22"

COLUMN_RENAMES = {"ORDER": "RACE_ORDER", "ORDERNAME": "RACE_ORDERNAME", "TRIM": "TRIM1"}

# parkwork column -> vehicle column (same as vehicle_import_all_columns.py in the old fars database); columns not listed drop the leading P
PARKWORK_COLUMN_MAP = {
    "PTYPE": "UNITTYPE", "PTYPENAME": "UNITTYPENAME", "PBODYTYP": "BODY_TYP", "PBODYTYPNAME": "BODY_TYPNAME",
    "PMODYEAR": "MOD_YEAR", "PMODYEARNAME": "MOD_YEARNAME", "PFUECODE": "FUELCODE",
    "PTRAILER": "TOW_VEH", "PTRAILERNAME": "TOW_VEHNAME", "PCARGTYP": "CARGO_BT", "PCARGTYPNAME": "CARGO_BTNAME",
    "PHAZPLAC": "HAZ_PLAC", "PHAZPLACNAME": "HAZ_PLACNAME", "PSP_USE": "SPEC_USE", "PSP_USENAME": "SPEC_USENAME",
    "PEM_USE": "EMER_USE", "PEM_USENAME": "EMER_USENAME", "PVEH_SEV": "DEFORMED", "PVEH_SEVNAME": "DEFORMEDNAME",
    "PFIRE": "FIRE_EXP", "PFIRENAME": "FIRE_EXPNAME",
}

# GSQL reserved words that could clash with column names (not exhaustive, but enough for FARS); a clash fails immediately instead of halfway through a load
GSQL_RESERVED = set("""
ACCUM ADD ALL ALLOCATE ALTER AND ANY AS ASC AVG BAG BATCH BETWEEN BIGINT BLOB BOOL BOOLEAN BOTH BREAK BY
CALL CASCADE CASE CATCH CHAR CHARACTER CHECK CLOB COALESCE COMPRESS CONST CONSTRAINT CONTINUE COST COUNT
CREATE CURRENT_DATE CURRENT_TIME CURRENT_TIMESTAMP CURSOR DATETIME DATETIME_ADD DATETIME_SUB DECIMAL
DECLARE DEFAULT DELETE DESC DISTRIBUTED DO DOUBLE DROP EDGE ELSE ELSEIF END ESCAPE EXCEPTION EXISTS FALSE
FILE FILTER FIXED_BINARY FLOAT FOR FOREACH FROM GLOBAL GRANTS GRAPH GROUP GROUPBY HAVING HEADER IF IN INDEX
INPUT_LINE_FILTER INSERT INT INT16 INT32 INT32_T INT64_T INT8 INTEGER INTERPRET INTERSECT INTERVAL INTO IS
ISEMPTY JOB JOIN JSONARRAY JSONOBJECT KAFKA KEY LEADING LIKE LIMIT LIST LOAD LOADACCUM LOG LONG MAP MINUS
NOBODY NOT NOW NULL OFFSET ON OR ORDER PINNED POST_ACCUM PRIMARY PRIMARY_ID PRINT PROXY QUERY QUIT RAISE
RANGE REDUCE REPLACE RETURN RETURNS S3 SAMPLE SELECT SELECTVERTEX SET STATIC STRING SUM TARGET TEMP_TABLE
THEN TO TO_CSV TO_DATETIME TOKEN TOKEN_LEN TOKENBANK TRAILING TRIM TRUE TRY TUPLE TYPE TYPEDEF UINT UINT16
UINT32 UINT32_T UINT64_T UINT8 UINT8_T UNION UPDATE UPSERT USING VALUES VERTEX WHEN WHERE WHILE WITH
""".split())

TEXT_TRANS = str.maketrans({
    " ": " ", "\x1a": "-", "–": "-", "—": "-",
    "‘": "'", "’": "'", "“": '"', "”": '"',
    "\r": " ", "\n": " ", "\t": " ",
})

report_lines: list[str] = []


def log(msg: str = "") -> None:
    print(msg, flush=True)
    report_lines.append(msg)


# ----------------------------------------------------------------------------- read raw tables

def read_raw(table: str, year: int) -> pd.DataFrame:
    path = RAW / str(year) / "National" / f"{table}.csv"
    data = path.read_bytes()
    try:
        text = data.decode("utf-8")
    except UnicodeDecodeError:
        text = data.decode("cp1252", errors="replace")
    text = text.lstrip("")
    df = pd.read_csv(io.StringIO(text), dtype=str, keep_default_na=False)
    df.columns = [COLUMN_RENAMES.get(c.strip().upper(), c.strip().upper()) for c in df.columns]
    df = df.apply(lambda s: s.str.translate(TEXT_TRANS).str.strip())
    df.insert(0, "DATA_YEAR", str(year))
    return df


def read_table(table: str) -> pd.DataFrame:
    """Merge the three years; columns are unioned, and columns missing in a year are filled with empty strings."""
    frames = [read_raw(table, y) for y in YEARS]
    cols: list[str] = []
    for f in frames:
        cols += [c for c in f.columns if c not in cols]
    df = pd.concat(frames, ignore_index=True)[cols].fillna("")
    missing = {c: [y for y, f in zip(YEARS, frames) if c not in f.columns] for c in cols}
    missing = {c: ys for c, ys in missing.items() if ys}
    log(f"  {table:<18} rows={len(df):>8,}  cols={len(cols):>3}" + (f"  columns missing in some years: {missing}" if missing else ""))
    return df


def intstr(s: pd.Series) -> pd.Series:
    """Normalize keys like '001' / '1.0' to '1'."""
    return pd.to_numeric(s, errors="raise").astype("int64").astype(str)


def add_keys(df: pd.DataFrame) -> pd.DataFrame:
    df = df.copy()
    df["ST_CASE"] = intstr(df["ST_CASE"])
    df["accident_id"] = df["DATA_YEAR"] + "_" + df["ST_CASE"]
    if "VEH_NO" in df:
        df["VEH_NO"] = intstr(df["VEH_NO"])
        df["vehicle_id"] = df["accident_id"] + "_" + df["VEH_NO"]
    if "PER_NO" in df:
        df["PER_NO"] = intstr(df["PER_NO"])
        df["person_id"] = df["vehicle_id"] + "_" + df["PER_NO"]
    if "EVENTNUM" in df:
        df["EVENTNUM"] = intstr(df["EVENTNUM"])
        df["event_id"] = df["accident_id"] + "_" + df["EVENTNUM"]
    return df


def front(df: pd.DataFrame, first: list[str]) -> pd.DataFrame:
    return df[first + [c for c in df.columns if c not in first]]


def fold_one_to_one(base: pd.DataFrame, extra: pd.DataFrame, key: str, name: str,
                    drop: list[str], prefix: str = "") -> pd.DataFrame:
    """Fold the columns of a 1:1 side table into the main table (left join). Side-table keys must be unique and all present in the main table."""
    extra = extra.drop(columns=[c for c in drop if c in extra.columns and c != key])
    if prefix:
        extra = extra.rename(columns={c: prefix + c for c in extra.columns if c != key})
    clash = (set(extra.columns) & set(base.columns)) - {key}
    assert not clash, f"{name} has column names that clash with the main table: {clash}"
    assert extra[key].is_unique, f"{key} in {name} is not unique"
    orphan = (~extra[key].isin(base[key])).sum()
    assert orphan == 0, f"{name} has {orphan} rows not found in the main table"
    log(f"  folded in {name}: {len(extra):,} rows, {len(extra.columns) - 1} columns")
    return base.merge(extra, on=key, how="left").fillna("")


# ----------------------------------------------------------------------------- type inference

INT_RE = r"-?(?:0|[1-9]\d{0,17})"
FLOAT_RE = r"-?\d+\.\d+(?:[eE][-+]?\d+)?|-?\d+[eE][-+]?\d+"


def infer_type(s: pd.Series) -> str:
    v = s[s != ""]
    if v.empty:
        return "STRING"
    if v.str.fullmatch(INT_RE).all():
        return "INT"
    if v.str.fullmatch(f"{INT_RE}|{FLOAT_RE}").all():
        return "DOUBLE"
    return "STRING"


# ----------------------------------------------------------------------------- graph model

nodes: dict[str, dict] = {}     # label -> {key, key_type, df}
edges: list[dict] = []          # {type, src, dst, df(src,dst,props...), source}


def add_node(label: str, key: str, df: pd.DataFrame, key_type: str = "STRING") -> None:
    df = front(df, [key])
    dup = df[key].duplicated().sum()
    assert dup == 0, f"{label}.{key} has {dup} duplicates"
    nodes[label] = {"key": key, "key_type": key_type, "df": df}


def add_edge(rel: str, src: str, dst: str, df: pd.DataFrame, source: str) -> None:
    """The first two columns of df are src / dst; the rest are edge properties."""
    df = df.copy()
    df.columns = ["src", "dst"] + list(df.columns[2:])
    n0 = len(df)
    ok = df["src"].isin(nodes[src]["df"][nodes[src]["key"]]) & df["dst"].isin(nodes[dst]["df"][nodes[dst]["key"]])
    dangling = int((~ok).sum())
    df = df[ok]
    dup = int(df.duplicated(["src", "dst"]).sum())
    df = df.drop_duplicates(["src", "dst"])
    note = []
    if dangling:
        note.append(f"dropped missing endpoint {dangling:,}")
    if dup:
        note.append(f"deduplicated {dup:,}")
    log(f"  ({src})-[:{rel}]->({dst}) <- {source}: {n0:,} -> {len(df):,}" + (f"  [{', '.join(note)}]" if note else ""))
    edges.append({"type": rel, "src": src, "dst": dst, "df": df.reset_index(drop=True), "source": source})


def lookup_rows(df: pd.DataFrame, code_col: str) -> pd.DataFrame:
    """code -> name; when a code's name differs across years, the latest year's spelling wins."""
    t = df[[code_col, code_col + "NAME", "DATA_YEAR"]].copy()
    t.columns = ["id", "name", "DATA_YEAR"]
    t["id"] = intstr(t["id"])
    t = t.sort_values("DATA_YEAR").drop_duplicates("id", keep="last")
    return t[["id", "name"]]


def build() -> None:
    log("== Reading raw tables (2020-2022 merged)")
    T = {t: add_keys(read_table(t)) for t in [
        "accident", "miacc", "midrvacc", "vehicle", "parkwork", "person", "pbtype", "safetyeq", "miper",
        "cevent", "vevent", "vsoe", "vpicdecode", "vpictrailerdecode", "drugs",
        "weather", "crashrf", "damage", "distract", "drimpair", "driverrf", "factor", "maneuver",
        "vehiclesf", "pvehiclesf", "violatn", "vision", "race", "personrf",
        "nmcrash", "nmdistract", "nmimpair", "nmprior",
    ]}
    keycols = ["DATA_YEAR", "STATE", "STATENAME", "ST_CASE", "VEH_NO", "PER_NO",
               "accident_id", "vehicle_id", "person_id"]

    log("\n== Nodes")
    acc = T["accident"]
    acc = fold_one_to_one(acc, T["miacc"], "accident_id", "miacc", keycols, "MIACC_")
    acc = fold_one_to_one(acc, T["midrvacc"], "accident_id", "midrvacc", keycols, "MIDRVACC_")
    add_node("Accident", "accident_id", acc)

    veh = T["vehicle"].assign(source_table="vehicle")
    pw = T["parkwork"]
    pw = pw.rename(columns={c: PARKWORK_COLUMN_MAP.get(c, c[1:]) for c in pw.columns
                            if c not in keycols and c.startswith("P")}).assign(source_table="parkwork")
    unmapped = [c for c in pw.columns if c not in veh.columns]
    assert not unmapped, f"parkwork columns with no vehicle mapping: {unmapped}"
    assert not set(veh["vehicle_id"]) & set(pw["vehicle_id"]), "VEH_NO overlaps between vehicle and parkwork"
    log(f"  folded in parkwork: {len(pw):,} rows, {sum(c not in keycols for c in pw.columns) - 1} columns mapped to vehicle column names")
    veh = pd.concat([veh, pw], ignore_index=True)[list(veh.columns)].fillna("")
    add_node("Vehicle", "vehicle_id", front(veh, ["vehicle_id", "accident_id", "source_table"]))

    per = T["person"].copy()
    # Non-motorists have VEH_NO=0 and belong to no vehicle
    per["vehicle_id"] = per["vehicle_id"].where(per["VEH_NO"] != "0", "")
    per = fold_one_to_one(per, T["pbtype"], "person_id", "pbtype", keycols)
    per = fold_one_to_one(per, T["safetyeq"], "person_id", "safetyeq", keycols)
    per = fold_one_to_one(per, T["miper"], "person_id", "miper", keycols, "MIPER_")
    add_node("Person", "person_id", front(per, ["person_id", "accident_id", "vehicle_id"]))

    add_node("Event", "event_id", front(T["cevent"], ["event_id", "accident_id"]))

    vp = T["vpicdecode"].copy()
    vp["vpic_id"] = vp["vehicle_id"]
    add_node("VPICDecode", "vpic_id", front(vp, ["vpic_id", "vehicle_id", "accident_id"]))

    tr = T["vpictrailerdecode"].copy()
    tr["TRAILER_NO"] = intstr(tr["TRAILER_NO"])
    tr["trailer_id"] = tr["vehicle_id"] + "_" + tr["TRAILER_NO"]
    add_node("Trailer", "trailer_id", front(tr, ["trailer_id", "vehicle_id", "accident_id"]))

    dr = T["drugs"].copy()
    dr["vehicle_id"] = dr["vehicle_id"].where(dr["VEH_NO"] != "0", "")
    dr["seq"] = (dr.groupby("person_id").cumcount() + 1).astype(str)
    dr["drug_event_id"] = dr["person_id"] + "_" + dr["seq"]
    add_node("DrugEvent", "drug_event_id", front(dr.drop(columns="seq"), ["drug_event_id", "person_id", "accident_id"]))

    # Lookup-table nodes {id, name}
    LOOKUPS = [  # (table, code column, label, relationship, entity label, direction: 'to_entity' = code->entity, 'from_entity' = entity->code)
        ("weather", "WEATHER", "Weather", "OCCURRED_IN", "Accident", "from_entity"),
        ("crashrf", "CRASHRF", "CrashRF", "CRASH_FACTOR_OF", "Accident", "to_entity"),
        ("damage", "DAMAGE", "Damage", "POSITION_OF", "Vehicle", "to_entity"),
        ("distract", "DRDISTRACT", "Distract", "DISTRACTION_OF", "Vehicle", "to_entity"),
        ("drimpair", "DRIMPAIR", "DrImpair", "IMPAIRMENT_OF", "Vehicle", "to_entity"),
        ("driverrf", "DRIVERRF", "DriverRF", "DRIVER_RF_OF", "Vehicle", "to_entity"),
        ("factor", "VEHICLECC", "Factor", "FACTOR_OF", "Vehicle", "to_entity"),
        ("maneuver", "MANEUVER", "Maneuver", "MANEUVER_OF", "Vehicle", "to_entity"),
        ("vehiclesf", "VEHICLESF", "VehicleSF", "VEHICLE_SF_OF", "Vehicle", "to_entity"),
        ("violatn", "VIOLATION", "Violatn", "COMMITTED_BY", "Vehicle", "to_entity"),
        ("vision", "VISION", "Vision", "VISION_OF", "Vehicle", "to_entity"),
        ("race", "RACE", "Race", "RACE_OF", "Person", "to_entity"),
        ("personrf", "PERSONRF", "PersonRF", "PERSON_RF_OF", "Person", "to_entity"),
        ("nmcrash", "NMCC", "NMCrash", "NMCRASH_OF", "Person", "to_entity"),
        ("nmdistract", "NMDISTRACT", "NMDistract", "NMDISTRACT_OF", "Person", "to_entity"),
        ("nmimpair", "NMIMPAIR", "NMImpair", "NMIMPAIR_OF", "Person", "to_entity"),
        ("nmprior", "NMACTION", "NMPrior", "NMPRIOR_OF", "Person", "to_entity"),
    ]
    lk_frames: dict[str, list[pd.DataFrame]] = defaultdict(list)
    for table, code, label, *_ in LOOKUPS:
        lk_frames[label].append(lookup_rows(T[table], code))
    lk_frames["VehicleSF"].append(lookup_rows(T["pvehiclesf"], "PVEHICLESF"))
    lk_frames["DrugTest"].append(lookup_rows(T["drugs"], "DRUGSPEC"))
    lk_frames["Drug"].append(lookup_rows(T["drugs"], "DRUGRES"))
    for label, frames in lk_frames.items():
        lk = pd.concat(frames).drop_duplicates("id", keep="first")
        lk = lk.sort_values("id", key=lambda s: s.astype(int))
        add_node(label, "id", lk, key_type="INT")
    for label in lk_frames:
        log(f"  {label:<15} {len(nodes[label]['df']):>4} codes")
    for label in ("Accident", "Vehicle", "Person", "Event", "VPICDecode", "Trailer", "DrugEvent"):
        d = nodes[label]["df"]
        log(f"  {label:<15} {len(d):>9,} rows  {len(d.columns):>3} cols")

    log("\n== Edges")
    add_edge("INVOLVES", "Accident", "Vehicle", veh[["accident_id", "vehicle_id"]], "vehicle + parkwork")
    add_edge("HAS_VICTIM", "Accident", "Person", T["person"][["accident_id", "person_id"]], "person")
    add_edge("RESULTED_IN", "Accident", "Event", T["cevent"][["accident_id", "event_id"]], "cevent")

    occ = per[per["vehicle_id"] != ""]
    veh_ids = set(nodes["Vehicle"]["df"]["vehicle_id"])
    unknown = (~occ["vehicle_id"].isin(veh_ids)).sum()
    log(f"  person.VEH_NO>0 but vehicle missing: {unknown}")
    add_edge("HAS_OCCUPANT", "Vehicle", "Person", occ[["vehicle_id", "person_id"]], "person")

    ce = T["cevent"]
    s = ce.assign(v=ce["accident_id"] + "_" + intstr(ce["VNUMBER1"]))
    add_edge("SUBJECT_OF", "Vehicle", "Event", s[["v", "event_id", "AOI1", "AOI1NAME"]].rename(columns={"AOI1": "AOI", "AOI1NAME": "AOINAME"}), "cevent.VNUMBER1")
    o = ce.assign(v=ce["accident_id"] + "_" + intstr(ce["VNUMBER2"]))
    o = o[~o["VNUMBER2"].isin(["5555", "9999"])][["v", "event_id", "AOI2", "AOI2NAME"]].rename(columns={"AOI2": "AOI", "AOI2NAME": "AOINAME"})
    add_edge("OBJECT_OF", "Vehicle", "Event", o, "cevent.VNUMBER2")

    # vevent: each vehicle's own event sequence. AOI is taken from this vehicle's point of view (AOI1 if it is VNUMBER1, else AOI2).
    # That is exactly what vsoe contains, so vsoe is not imported separately; first check below that the two agree.
    ve = T["vevent"].copy()
    ve["AOI"] = ve["AOI1"].where(ve["VNUMBER1"] == ve["VEH_NO"], ve["AOI2"])
    ve["AOINAME"] = ve["AOI1NAME"].where(ve["VNUMBER1"] == ve["VEH_NO"], ve["AOI2NAME"])
    vs = T["vsoe"]
    k = ["vehicle_id", "VEVENTNUM", "SOE", "AOI"]
    same = Counter(map(tuple, ve[k].values.tolist())) == Counter(map(tuple, vs[k].values.tolist()))
    log(f"  (vehicle, VEVENTNUM, SOE, AOI) derived from vevent matches vsoe exactly: {same}")
    assert same, "vsoe cannot be derived from vevent and must be imported separately"
    add_edge("HAS_VEVENT", "Vehicle", "Event", ve[["vehicle_id", "event_id", "VEVENTNUM", "AOI", "AOINAME"]], "vevent")

    for label, rel, table, frame, props in (("VPICDecode", "HAS_VPIC", "vpicdecode", vp, []),
                                            ("Trailer", "HAS_TRAILER", "vpictrailerdecode", tr, ["TRAILER_NO"])):
        cols = ["vehicle_id", nodes[label]["key"]] + props
        add_edge(rel, "Vehicle", label, frame[cols], table)

    add_edge("DRUG_EVENT_OF", "DrugEvent", "Person", dr[["drug_event_id", "person_id"]], "drugs")
    add_edge("DRUG_TEST_OF", "DrugTest", "DrugEvent", dr[["DRUGSPEC", "drug_event_id"]].assign(DRUGSPEC=intstr(dr["DRUGSPEC"])), "drugs.DRUGSPEC")
    add_edge("RESULT_OF", "Drug", "DrugEvent", dr[["DRUGRES", "drug_event_id"]].assign(DRUGRES=intstr(dr["DRUGRES"])), "drugs.DRUGRES")

    entity_key = {"Accident": "accident_id", "Vehicle": "vehicle_id", "Person": "person_id"}
    for table, code, label, rel, entity, direction in LOOKUPS:
        d = T[table].copy()
        if table == "vehiclesf":
            pv = T["pvehiclesf"].rename(columns={"PVEHICLESF": "VEHICLESF", "PVEHICLESFNAME": "VEHICLESFNAME"})
            d = pd.concat([d, pv], ignore_index=True)
            table = "vehiclesf + pvehiclesf"
        d[code] = intstr(d[code])
        props = ["RACE_ORDER", "MULTRACE"] if table == "race" else []
        cols = [entity_key[entity], code] if direction == "from_entity" else [code, entity_key[entity]]
        if direction == "from_entity":
            add_edge(rel, entity, label, d[cols + props], table)
        else:
            add_edge(rel, label, entity, d[cols + props], table)


# ----------------------------------------------------------------------------- write CSVs

def write_tsv(df: pd.DataFrame, path: Path) -> None:
    """TAB-separated, unquoted, LF. TAB/newline were already replaced with spaces during cleaning, so no field contains a delimiter.
    Neo4j's CSV parser only enters quoted mode when a field starts with ", and reads a " in mid-field literally;
    so the only case to handle is values starting with " (just 1 in the whole dataset: VPIC SERIES '"Authentic" w/ ABS'),
    where " is replaced with '."""
    df = df.copy()
    for c in df.columns:
        lead = df[c].str.startswith('"')
        if lead.any():
            log(f"  {path.name}.{c}: {int(lead.sum())} values start with \", replacing \" with '")
            df.loc[lead, c] = df.loc[lead, c].str.replace('"', "'", regex=False)
        assert not df[c].str.contains("[\t\n\r]", regex=True).any(), f"{path.name}.{c} contains TAB/newline"
    with open(path, "w", encoding="utf-8", newline="\n") as fh:
        fh.write("\t".join(df.columns) + "\n")
        for row in df.itertuples(index=False, name=None):
            fh.write("\t".join(row) + "\n")


def write_csvs() -> dict:
    (CSV_DIR / "nodes").mkdir(parents=True, exist_ok=True)
    (CSV_DIR / "edges").mkdir(parents=True, exist_ok=True)
    schema: dict = {"years": list(YEARS), "nodes": {}, "edges": []}
    log("\n== Writing CSVs and column types")
    for label, n in nodes.items():
        df = n["df"]
        for c in df.columns:
            assert re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]*", c), f"{label}.{c} is not a valid identifier"
            assert c.upper() not in GSQL_RESERVED, f"{label}.{c} is a GSQL reserved word; add it to COLUMN_RENAMES"
        cols = [{"name": c, "type": n["key_type"] if c == n["key"] else infer_type(df[c]),
                 "nullable": bool((df[c] == "").any())} for c in df.columns]
        rel = f"csv/nodes/{label}.tsv"
        write_tsv(df, HERE / rel)
        schema["nodes"][label] = {"file": rel, "key": n["key"], "rows": len(df), "columns": cols}
        types = Counter(c["type"] for c in cols)
        log(f"  {rel:<45} {len(df):>9,} rows  {dict(types)}")
    names = [f"{e['src']}_{e['type']}_{e['dst']}" for e in edges]
    assert len(names) == len(set(names)), "The same (src, type, dst) appears twice; edge files would overwrite each other"
    for e in edges:
        df = e["df"]
        rel = f"csv/edges/{e['src']}_{e['type']}_{e['dst']}.tsv"
        write_tsv(df, HERE / rel)
        props = [{"name": c, "type": infer_type(df[c]), "nullable": bool((df[c] == "").any())} for c in df.columns[2:]]
        schema["edges"].append({"type": e["type"], "src": e["src"], "dst": e["dst"], "file": rel,
                                "source": e["source"], "rows": len(df), "props": props})
        log(f"  {rel:<45} {len(df):>9,} rows")
    # All (src,dst) pairs of one edge type must have the same properties (a TigerGraph edge type has a single property set)
    by_type: dict[str, list] = defaultdict(list)
    for e in schema["edges"]:
        by_type[e["type"]].append(e)
    for t, es in by_type.items():
        names = {tuple(p["name"] for p in e["props"]) for e in es}
        assert len(names) == 1, f"Edge {t} has inconsistent properties across files: {names}"
        # Use the "widest" type across files
        for i, p in enumerate(es[0]["props"]):
            ts = {e["props"][i]["type"] for e in es}
            wide = "STRING" if "STRING" in ts else "DOUBLE" if "DOUBLE" in ts else "INT"
            for e in es:
                e["props"][i]["type"] = wide
    (HERE / "schema.json").write_text(json.dumps(schema, indent=1, ensure_ascii=False))
    return schema


# ----------------------------------------------------------------------------- generate Cypher

def cy_name(c: str) -> str:
    return c if re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]*", c) else f"`{c}`"


def cy_value(c: str, t: str) -> str:
    v = f"row.{cy_name(c)}"
    return {"INT": f"toInteger({v})", "DOUBLE": f"toFloat({v})"}.get(t, v)


def gen_neo4j(schema: dict) -> None:
    d = HERE / "neo4j"
    d.mkdir(exist_ok=True)
    url = lambda rel: "file:///" + str((HERE / rel).relative_to(NEO4J_IMPORT_ROOT))
    hdr = f"// Generated by build_csv.py; do not edit by hand. Database `{NEO4J_DB}`; CSV paths are relative to server.directories.import in neo4j.conf.\n"

    s = [hdr, "// Unique constraints (which also provide the indexes MATCH uses). Run before the import.\n"]
    for label, n in schema["nodes"].items():
        s.append(f"CREATE CONSTRAINT {label.lower()}_{n['key']} IF NOT EXISTS FOR (n:{label}) REQUIRE n.{n['key']} IS UNIQUE;")
    # Plain indexes for querying by year / foreign key
    for label, prop in (("Vehicle", "accident_id"), ("Vehicle", "source_table"), ("Person", "accident_id"),
                        ("Person", "vehicle_id"), ("Event", "accident_id"), ("Accident", "DATA_YEAR")):
        s.append(f"CREATE INDEX {label.lower()}_{prop.lower()} IF NOT EXISTS FOR (n:{label}) ON (n.{prop});")
    s.append("CALL db.awaitIndexes(600);")
    (d / "schema.cypher").write_text("\n".join(s) + "\n")

    out = [hdr, "// Each section starts with '// ---- step:'; import_neo4j.sh runs and times the sections one by one using this marker.",
           "// In LOAD CSV an empty field is null, and toInteger(null) is still null, so missing values are not written as properties.\n"]
    for label, n in schema["nodes"].items():
        props = ",\n    ".join(f"{cy_name(c['name'])}: {cy_value(c['name'], c['type'])}" for c in n["columns"])
        out.append(f"// ---- step: node {label}  ({n['rows']:,} rows)\n"
                   f"LOAD CSV WITH HEADERS FROM '{url(n['file'])}' AS row FIELDTERMINATOR '\\t'\n"
                   f"CALL (row) {{\n  CREATE (:{label} {{\n    {props}\n  }})\n}} IN TRANSACTIONS OF 10000 ROWS;\n")
    (d / "nodes.cypher").write_text("\n".join(out))

    out = [hdr, "// Each section starts with '// ---- step:'; import_neo4j.sh runs and times the sections one by one using this marker.",
           "// Cleaning already guarantees both endpoints exist and (src,dst) is unique in edge files, so plain MATCH + CREATE is used.\n"]
    for e in schema["edges"]:
        sk = schema["nodes"][e["src"]]["key"]
        dk = schema["nodes"][e["dst"]]["key"]
        sv = "toInteger(row.src)" if schema["nodes"][e["src"]]["columns"][0]["type"] == "INT" else "row.src"
        dv = "toInteger(row.dst)" if schema["nodes"][e["dst"]]["columns"][0]["type"] == "INT" else "row.dst"
        props = ", ".join(f"{cy_name(p['name'])}: {cy_value(p['name'], p['type'])}" for p in e["props"])
        props = f" {{{props}}}" if props else ""
        out.append(f"// ---- step: edge {e['src']}-{e['type']}->{e['dst']}  ({e['rows']:,} rows)\n"
                   f"LOAD CSV WITH HEADERS FROM '{url(e['file'])}' AS row FIELDTERMINATOR '\\t'\n"
                   f"CALL (row) {{\n"
                   f"  MATCH (s:{e['src']} {{{sk}: {sv}}})\n"
                   f"  MATCH (d:{e['dst']} {{{dk}: {dv}}})\n"
                   f"  CREATE (s)-[:{e['type']}{props}]->(d)\n"
                   f"}} IN TRANSACTIONS OF 10000 ROWS;\n")
    (d / "edges.cypher").write_text("\n".join(out))


# ----------------------------------------------------------------------------- generate GSQL

def gen_tigergraph(schema: dict) -> None:
    d = HERE / "tigergraph"
    d.mkdir(exist_ok=True)
    g = TG_GRAPH
    hdr = f"// Generated by build_csv.py; do not edit by hand. Graph {g}; uses the same CSVs as Neo4j.\n"

    s = [hdr,
         "// All types are graph-local (graph-level schema change job), so DROP GRAPH removes only these types.",
         "// Every primary key sets primary_id_as_attribute=\"true\", so queries can use accident_id etc. directly as attributes.",
         "// TigerGraph has no null: empty CSV fields load as 0 for INT/DOUBLE and as an empty string for STRING.\n",
         f"CREATE GRAPH {g}()\n", f"USE GRAPH {g}", f"CREATE SCHEMA_CHANGE JOB {g}_schema FOR GRAPH {g} {{"]
    for label, n in schema["nodes"].items():
        cols = n["columns"]
        attrs = ",\n    ".join([f"PRIMARY_ID {cols[0]['name']} {cols[0]['type']}"] +
                               [f"{c['name']} {c['type']}" for c in cols[1:]])
        s.append(f"  ADD VERTEX {label} (\n    {attrs}\n  ) WITH primary_id_as_attribute=\"true\";")
    by_type: dict[str, list] = defaultdict(list)
    for e in schema["edges"]:
        by_type[e["type"]].append(e)
    for t, es in by_type.items():
        pairs = " | ".join(f"FROM {e['src']}, TO {e['dst']}" for e in es)
        props = "".join(f", {p['name']} {p['type']}" for p in es[0]["props"])
        s.append(f"  ADD DIRECTED EDGE {t} ({pairs}{props}) WITH REVERSE_EDGE=\"reverse_{t}\";")
    s += ["}", f"RUN SCHEMA_CHANGE JOB {g}_schema", f"DROP JOB {g}_schema"]
    (d / "schema.gsql").write_text("\n".join(s) + "\n")

    path = lambda rel: f"m1:{HERE / rel}"
    using = 'USING HEADER="true", SEPARATOR="\\t", EOL="\\n"'
    j = [hdr,
         "// Two jobs: load all vertices first, then all edges. Edges use VERTEX_MUST_EXIST, so rows with a missing endpoint are rejected and counted",
         "// instead of creating empty vertices with only a primary key (cleaning already guarantees endpoints exist, so there should be 0 rejections).",
         "// m1: means the shared disk /datasets is read on node m1 and then distributed across the cluster.\n",
         f"USE GRAPH {g}\n", f"CREATE LOADING JOB load_vertices FOR GRAPH {g} {{"]
    for label, n in schema["nodes"].items():
        j.append(f"  DEFINE FILENAME f_{label} = \"{path(n['file'])}\";")
    for label, n in schema["nodes"].items():
        vals = ", ".join(f'$"{c["name"]}"' for c in n["columns"])
        j.append(f"  LOAD f_{label} TO VERTEX {label} VALUES ({vals})\n    {using};")
    j += ["}\n", f"CREATE LOADING JOB load_edges FOR GRAPH {g} {{"]
    for e in schema["edges"]:
        j.append(f"  DEFINE FILENAME f_{e['src']}_{e['type']}_{e['dst']} = \"{path(e['file'])}\";")
    for e in schema["edges"]:
        multi = len(by_type[e["type"]]) > 1
        # When an edge type has several (FROM, TO) pairs, VALUES must name this row's endpoint types after each id
        src = f'$"src" {e["src"]}' if multi else '$"src"'
        dst = f'$"dst" {e["dst"]}' if multi else '$"dst"'
        vals = ", ".join([src, dst] + [f'$"{p["name"]}"' for p in e["props"]])
        j.append(f"  LOAD f_{e['src']}_{e['type']}_{e['dst']} TO EDGE {e['type']} VALUES ({vals})\n"
                 f"    {using}, VERTEX_MUST_EXIST=\"true\";")
    j.append("}")
    (d / "load_jobs.gsql").write_text("\n".join(j) + "\n")


def write_expected(schema: dict) -> None:
    rows = ["kind\ttype\tcount"]
    for label, n in schema["nodes"].items():
        rows.append(f"vertex\t{label}\t{n['rows']}")
    tot: Counter = Counter()
    for e in schema["edges"]:
        tot[e["type"]] += e["rows"]
    for t, c in tot.items():
        rows.append(f"edge\t{t}\t{c}")
    (HERE / "expected-counts.tsv").write_text("\n".join(rows) + "\n")
    log(f"\n== Total: {sum(n['rows'] for n in schema['nodes'].values()):,} vertices, {sum(tot.values()):,} edges, "
        f"{len(schema['nodes'])} vertex types, {len(tot)} edge types")


def main() -> None:
    t0 = time.time()
    build()
    schema = write_csvs()
    gen_neo4j(schema)
    gen_tigergraph(schema)
    write_expected(schema)
    log(f"\nDone in {time.time() - t0:.1f}s")
    (HERE / "build-report.txt").write_text("\n".join(report_lines) + "\n")


if __name__ == "__main__":
    sys.exit(main())