"""The tables of the catalog dataset nyc-subway-crime-and-ridership (catalog/<id>.yaml), and their check.

    from catalog_subway import tables, validate
    t = tables()                 # {file name: DataFrame}, in the registry's column order
    problems = validate(t)       # [] when every row of the article's data.csv is reproduced

Built from the same loaders as the article "Is the subway safer per ride?" and from its own functions
for the rides (``paid_rides``, ``month_pattern``, ``fiscal_months``), imported rather than copied, so
the dataset and the published article cannot drift apart: ``validate`` holds the monthly panel, summed,
to every row of the article's data.csv and to the figures its page prints from them.
"""
from __future__ import annotations

import sys, pathlib; sys.path.insert(0, str(pathlib.Path(__file__).resolve().parents[1]))  # src/: _paths and the library

import json

import pandas as pd

import _paths  # noqa: F401  (puts the shared library on sys.path)
from article_kit import r
from articles.is_the_subway_safer_per_ride import analysis as art
from nycdata import city, crime, mta, popest, weather

ID = "nyc-subway-crime-and-ridership"
FIRST_RIDE_MONTH = (1997, 7)                     # the first month of fiscal 1998, the long view's first year
#: The widenings of felony assault since 2019, by the day each came into force (catalog events).
WIDENINGS = (pd.Timestamp("2022-09-25"), pd.Timestamp("2024-07-19"), pd.Timestamp("2024-10-17"))
#: The offences the monthly panel breaks out, as the complaint files name them (the six major by prefix,
#: grand larceny of a motor vehicle left out, as crime.is_subway_major does).
SIX = {"murder": "MURDER", "rape": "RAPE", "robbery": "ROBBERY", "felony_assault": "FELONY ASSAULT",
       "burglary": "BURGLARY", "grand_larceny": "GRAND LARCENY"}
FA_CODES = {"felony_assault_ordinary": "ordinary", "felony_assault_strangulation": "strangulation",
            "felony_assault_police_or_peace_officer": "police or peace officer",
            "felony_assault_other_public_service_employee": "other public service employee"}
LAW = {"all_felonies": "FELONY", "all_misdemeanors": "MISDEMEANOR", "all_violations": "VIOLATION"}
COUNTS = (["six_major", *SIX, *FA_CODES, "theft_of_services", *LAW, "all_complaints"])
RATED = ("six_major", "felony_assault", "robbery", "grand_larceny")
FY_NOTES = {1998: "The fiscal 2001 edition marks this figure as based on preliminary data."}
#: The counts the finer tables (day, hour, station) carry: the rated offences and the totals.
FINE = ("six_major", "felony_assault", "robbery", "grand_larceny", "all_felonies", "all_complaints")
#: Report delay bands in days (occurrence-start day to report day), each [low, high].
LAG_BANDS = (("same day", 0, 0), ("1 day", 1, 1), ("2-7 days", 2, 7), ("8-30 days", 8, 30), ("31-90 days", 31, 90),
             ("91-365 days", 91, 365), ("over a year", 366, None))
DAYS = ("Monday", "Tuesday", "Wednesday", "Thursday", "Friday", "Saturday", "Sunday")
BUS = "BUS (NYC TRANSIT)"


def _method(year: int, month: int) -> str:
    if (year, month) >= (2020, 3):
        return "daily_counts"
    return "annual_total_less_daily" if year == 2020 else "annual_total_x_2019_pattern"


def _fy(year: int, month: int) -> int:
    return year + 1 if month >= 7 else year


def _ym(t: pd.Timestamp) -> str:
    return f"{t.year:04d}-{t.month:02d}"


def _int(x) -> object:
    return pd.NA if x is None or pd.isna(x) else int(round(float(x)))


def _rate(n, exposure, nd: int = 4) -> object:
    return pd.NA if exposure is None or pd.isna(exposure) or not exposure else round(1e6 * n / float(exposure), nd)


def _offence_column(name: str, major: bool) -> str | None:
    if not major:
        return None
    if name == "FELONY ASSAULT":
        return "felony_assault"
    for col, prefix in SIX.items():
        if name.startswith(prefix):
            return col
    return None


def inputs() -> dict:
    """Everything the tables are made from, read once: complaints, codes, cutoffs, rides and evasion."""
    d, late = crime.subway_complaints()
    codes, codes_late = crime.subway_complaints_by_code()
    ride = mta.long()
    sub = ride[ride["Mode"] == "Subway"]
    daily = sub.groupby([sub["Date"].dt.year, sub["Date"].dt.month])["Count"].sum()
    annual = mta.subway_annual().set_index("year")["trips"]
    pattern = art.month_pattern(mta.implied_2019_by_month("Subway", 2024, skip_leap_day=True))
    paid = art.paid_rides(annual, daily, pattern)
    last_day = sub["Date"].max()
    last_full = last_day if last_day.is_month_end else last_day - pd.offsets.MonthEnd(1)
    detail, _ = crime.subway_complaint_detail()
    bus = mta.daily()["Bus"]
    rain = weather.daily_rain()
    return {"d": d, "late": late, "codes": codes, "cutoff": crime.subway_report_cutoffs(), "annual": annual,
            "paid": paid, "last_ride_month": (int(last_full.year), int(last_full.month)),
            "entries": mta.entries_from_paid(paid), "entries_lo": mta.entries_from_paid(paid, shift=-0.01),
            "entries_hi": mta.entries_from_paid(paid, shift=0.01), "evasion": mta.fare_evasion(),
            "mmr": crime.transit_major_felonies(),
            # the finer tables
            "detail": detail, "places": crime.subway_complaint_places(), "stations": mta.station_monthly(),
            "station_list": mta.stations(), "records": crime.subway_complaint_records(),
            "subway_daily": sub[sub["Date"] <= last_day].set_index("Date")["Count"], "bus_daily": bus,
            "weather": weather.daily().merge(rain, on="date", how="outer").set_index("date"),
            "city": crime.city_complaints()[0], "bus_crime": crime.bus_complaints()[0],
            "arrests": crime.transit_arrests(), "shelter": city.shelter_daily().set_index("date")["Total Individuals in Shelter"],
            "population": popest.july_population().groupby("year")["pop"].sum()}


def _month_counts(x: dict) -> pd.DataFrame:
    """Complaints by month (index Timestamp) in the panel's count columns."""
    d = x["d"].copy()
    d["col"] = [_offence_column(o, m) for o, m in zip(d["ofns_desc"], d["major"])]
    six = d[d["major"]].groupby("m")["n"].sum().rename("six_major")
    each = d.dropna(subset=["col"]).pivot_table(index="m", columns="col", values="n", aggfunc="sum")
    law = d.pivot_table(index="m", columns="law_cat_cd", values="n", aggfunc="sum").rename(
        columns={v: k for k, v in LAW.items()})
    fa = x["codes"][x["codes"]["ky_cd"].eq(106)].pivot_table(index="m", columns="kind", values="n", aggfunc="sum")
    fa = fa.rename(columns={v: k for k, v in FA_CODES.items()})
    tos = x["codes"][x["codes"]["pd_cd"].eq(art.THEFT_OF_SERVICES)].groupby("m")["n"].sum().rename("theft_of_services")
    out = pd.concat([six, each, fa, tos, law], axis=1)
    months = pd.date_range(d["m"].min(), d["m"].max(), freq="MS")
    out = out.reindex(months).reindex(columns=[c for c in COUNTS if c != "all_complaints"]).fillna(0).astype(int)
    out["all_complaints"] = out[list(LAW)].sum(axis=1)
    return out


def _indicators(d: pd.DataFrame) -> pd.DataFrame:
    """The detail rows with one count column per tally the finer tables carry (each row's ``n`` in
    the columns it belongs to, 0 elsewhere), so any grouping sums them."""
    x = d.copy()
    col = pd.Series([_offence_column(o, m) for o, m in zip(x["ofns_desc"], x["major"])], index=x.index)
    x["six_major"] = x["n"].where(x["major"], 0)
    for k in ("felony_assault", "robbery", "grand_larceny"):
        x[k] = x["n"].where(col.eq(k), 0)
    x["offence_group"] = col.where(col.notna(), x["law_cat_cd"].map(
        {"FELONY": "other_felonies", "MISDEMEANOR": "misdemeanors", "VIOLATION": "violations"}))
    for k, v in LAW.items():
        x[k] = x["n"].where(x["law_cat_cd"].eq(v), 0)
    x["all_complaints"] = x["n"]
    x["theft_of_services"] = x["n"].where((x["ky_cd"].eq(343) & x["pd_cd"].eq(art.THEFT_OF_SERVICES)).fillna(False), 0)
    x["on_the_hour"] = x["n"].where(x["time"].str.endswith(":00:00").fillna(False), 0)
    x["day_type"] = x["date"].dt.dayofweek.map(lambda i: "weekend" if i >= 5 else "weekday")
    return x


def _daily(ind: pd.DataFrame, monthly: pd.DataFrame, x: dict) -> pd.DataFrame:
    """One row per day of the panel's months: complaints, the subway's paid rides that day (from
    March 2020) and Central Park's weather."""
    first, last = ind["m"].min(), ind["m"].max() + pd.offsets.MonthEnd(0)
    days = pd.date_range(first, last, freq="D")
    cols = ["six_major", "felony_assault", "robbery", "grand_larceny", *LAW, "all_complaints", "theft_of_services"]
    g = ind.groupby("date")[cols].sum().reindex(days, fill_value=0)
    w = x["weather"].reindex(days)
    complete = monthly.set_index("month")["complete"]
    rides = x["subway_daily"]
    return pd.DataFrame({
        "date": days.strftime("%Y-%m-%d"), "year": days.year, "month": days.strftime("%Y-%m"),
        "day_of_week": [DAYS[i] for i in days.dayofweek], "weekend": days.dayofweek >= 5,
        **{c: g[c].astype(int).to_numpy() for c in cols},
        "paid_rides": pd.array([_int(rides.get(t)) for t in days], dtype="Int64"),
        "tmax_f": pd.array([_int(v) for v in w["tmax_f"]], dtype="Int64"),
        "tmin_f": pd.array([_int(v) for v in w["tmin_f"]], dtype="Int64"),
        "precip_in": w["prcp_in"].round(2).to_numpy(), "snow_in": w["snow_in"].round(1).to_numpy(),
        "felony_assault_widenings_in_force": [sum(wd <= t for wd in WIDENINGS) for t in days],
        "complete": [bool(complete[t.strftime("%Y-%m")]) for t in days]})


def _hourly(ind: pd.DataFrame, monthly: pd.DataFrame) -> pd.DataFrame:
    """Complaints by month, weekday or weekend, and the hour the offence began (0-23), every cell
    of the grid present."""
    months = pd.date_range(ind["m"].min(), ind["m"].max(), freq="MS")
    grid = pd.MultiIndex.from_product([months, ["weekday", "weekend"], range(24)], names=["m", "day_type", "hour"])
    g = ind.groupby(["m", "day_type", "hour"])[[*FINE, "on_the_hour"]].sum().reindex(grid, fill_value=0).reset_index()
    out = pd.DataFrame({"month": g["m"].map(_ym), "day_type": g["day_type"], "hour": g["hour"].astype(int)})
    for c in FINE:
        out[c] = g[c].astype(int)
    out["all_complaints_on_the_hour"] = g["on_the_hour"].astype(int)
    out["complete"] = out["month"].map(monthly.set_index("month")["complete"]).astype(bool)
    return out


def _incomplete(monthly: pd.DataFrame) -> pd.Series:
    """How many of each year's months the panel flags as not yet complete."""
    return (~monthly["complete"]).groupby(monthly["year"]).sum().astype(int)


def _stations(ind: pd.DataFrame, x: dict, last_month: pd.Timestamp, monthly: pd.DataFrame) -> tuple[pd.DataFrame, pd.DataFrame]:
    """Complaints and entries by station complex and calendar year, and the crosswalk from each
    place the NYPD records to its complex."""
    keys = ["station_name", "boro_nm", "latitude", "longitude"]
    pl = x["places"]
    det = ind.fillna({"station_name": "(null)", "boro_nm": "(null)"}).merge(
        pl[keys + ["complex_id", "match"]], on=keys, how="left", validate="many_to_one")
    det["by_name"] = det["n"].where(det["match"].eq("name"), 0)
    c = det.groupby(["complex_id", "year"])[[*FINE, "by_name"]].sum()
    st = x["stations"]
    meta = st.sort_values("month").groupby("station_complex_id").last()[["station_complex", "borough", "latitude", "longitude"]]
    # a complex with complaints but no ridership row (a Staten Island Railway station) keeps its
    # complaints, named and placed from the MTA's station list
    extra = sorted(set(det["complex_id"].dropna().astype(int)) - set(meta.index))
    if extra:
        sl = x["station_list"]
        sl = sl[sl["Complex ID"].isin(extra)].groupby("Complex ID").first()
        boro = {"Bk": "Brooklyn", "M": "Manhattan", "Q": "Queens", "Bx": "Bronx", "SI": "Staten Island"}
        meta = pd.concat([meta, pd.DataFrame({"station_complex": sl["Stop Name"] + " (" + sl["Daytime Routes"].str.replace(" ", ",") + ")",
                                              "borough": sl["Borough"].map(boro), "latitude": sl["GTFS Latitude"],
                                              "longitude": sl["GTFS Longitude"]}, index=sl.index)]).sort_index()
    years = range(int(ind["year"].min()), int(ind["year"].max()) + 1)
    rows = []
    for cid, m in meta.iterrows():
        for y in years:
            months = 12 if y < last_month.year else last_month.month
            s = st[(st["station_complex_id"] == cid) & (st["month"].dt.year == y) & (st["month"].dt.month <= months)]
            rides, ride_months = (int(s["ridership"].sum()), int(len(s))) if len(s) else (pd.NA, 0)
            k = c.loc[(cid, y)] if (cid, y) in c.index else pd.Series(0, index=[*FINE, "by_name"])
            row = {"complex_id": int(cid), "station_complex": m["station_complex"], "borough": m["borough"],
                   "latitude": round(float(m["latitude"]), 6), "longitude": round(float(m["longitude"]), 6),
                   "year": y, "months": months, "months_incomplete": int(_incomplete(monthly).get(y, 0)),
                   **{f: int(k[f]) for f in FINE},
                   "all_complaints_matched_by_name": int(k["by_name"]), "rides": rides, "ride_months": ride_months}
            full = ride_months == months
            for f in ("six_major", "felony_assault", "robbery", "all_complaints"):
                row[f"{f}_per_m_rides"] = _rate(row[f], rides if full else None)
            rows.append(row)
    out = pd.DataFrame(rows)
    out["rides"] = out["rides"].astype("Int64")
    walk = pd.DataFrame({"station_name": pl["station_name"], "borough": pl["boro_nm"], "latitude": pl["latitude"],
                         "longitude": pl["longitude"], "complaints": pl["n"].astype(int), "complex_id": pl["complex_id"],
                         "matched_station": pl["stop_name"], "distance_m": pl["distance_m"], "match": pl["match"]})
    return out, walk


#: The record file's columns made from a source field, and the field (blank or "(null)" there is empty here).
RECORD_SOURCE = {"complaint_time": "cmplnt_fr_tm", "end_date": "cmplnt_to_dt", "end_time": "cmplnt_to_tm",
                 "law_category": "law_cat_cd", "offence": "ofns_desc", "pd_code": "pd_cd", "pd_description": "pd_desc",
                 "attempted_or_completed": "crm_atpt_cptd_cd", "jurisdiction": "juris_desc", "precinct": "addr_pct_cd",
                 "borough": "boro_nm", "patrol_borough": "patrol_boro", "transit_district": "transit_district",
                 "station_name": "station_name", "latitude": "latitude", "longitude": "longitude",
                 "victim_age_group": "vic_age_group", "victim_sex": "vic_sex", "suspect_age_group": "susp_age_group",
                 "suspect_sex": "susp_sex"}
SOURCE_BLANK = ("", "(null)")


def _records(x: dict, split_year: int) -> pd.DataFrame:
    """One row per complaint, the 109 late reports included and flagged: the NYPD's descriptive
    fields with a few derived ones. A cell is empty only where the NYPD's file is blank or says
    "(null)"; "UNKNOWN" and the like are values the NYPD recorded and stay as they are."""
    r, late = x["records"]
    a = pd.concat([r.assign(late_report=False), late.assign(late_report=True)], ignore_index=True)
    a = a.loc[a.index.repeat(a["n"])].reset_index(drop=True)

    def src(field: str) -> pd.Series:
        return a[field].where(~a[field].isin(SOURCE_BLANK), None)

    began = pd.to_datetime(a["cmplnt_fr_dt"])
    reported = pd.to_datetime(a["rpt_dt"])
    keys = ["station_name", "boro_nm", "latitude", "longitude"]
    pl = x["places"][keys + ["complex_id", "match"]]
    at = pd.DataFrame({"station_name": a["station_name"].replace("", "(null)"), "boro_nm": a["boro_nm"].replace("", "(null)"),
                       "latitude": pd.to_numeric(a["latitude"]), "longitude": pd.to_numeric(a["longitude"])})
    matched = at.merge(pl, on=keys, how="left", validate="many_to_one")
    hour = pd.to_numeric(src("cmplnt_fr_tm").str[:2], errors="coerce").astype("Int64")
    out = pd.DataFrame({
        "complaint_date": began.dt.strftime("%Y-%m-%d"), "complaint_time": src("cmplnt_fr_tm"),
        "end_date": src("cmplnt_to_dt").str[:10], "end_time": src("cmplnt_to_tm"),
        "report_date": reported.dt.strftime("%Y-%m-%d"), "year": began.dt.year, "month": began.dt.strftime("%Y-%m"),
        "day_of_week": [DAYS[i] for i in began.dt.dayofweek], "hour": hour,
        "report_delay_days": (reported - began).dt.days,
        "law_category": src("law_cat_cd"), "offence": src("ofns_desc"),
        "key_code": pd.to_numeric(a["ky_cd"]).astype("Int64"), "pd_code": pd.to_numeric(src("pd_cd")).astype("Int64"),
        "pd_description": src("pd_desc"), "is_major_six": a["ofns_desc"].map(crime.is_subway_major),
        "attempted_or_completed": src("crm_atpt_cptd_cd"), "jurisdiction": src("juris_desc"),
        "precinct": pd.to_numeric(src("addr_pct_cd")).astype("Int64"), "borough": src("boro_nm"),
        "patrol_borough": src("patrol_boro"), "transit_district": pd.to_numeric(src("transit_district")).astype("Int64"),
        # the coordinates as the NYPD's file writes them: parsed and written again, a few would move in
        # their seventeenth digit
        "station_name": src("station_name"), "latitude": src("latitude"), "longitude": src("longitude"),
        "complex_id": matched["complex_id"].astype("Int64").to_numpy(), "station_match": matched["match"].to_numpy(),
        "victim_age_group": src("vic_age_group"), "victim_sex": src("vic_sex"),
        "suspect_age_group": src("susp_age_group"), "suspect_sex": src("susp_sex"),
        "complaint_file": ["year_to_date" if lr or y >= split_year else "historic" for lr, y in zip(a["late_report"], began.dt.year)],
        "late_report": a["late_report"]})
    cut = {"historic": x["cutoff"]["historic"].strftime("%Y-%m-%d"), "year_to_date": x["cutoff"]["ytd"].strftime("%Y-%m-%d")}
    out.insert(out.columns.get_loc("late_report"), "report_cutoff", out["complaint_file"].map(cut))
    order = ["complaint_date", "complaint_time", "report_date", "law_category", "key_code", "pd_code", "station_name",
             "latitude", "longitude", "end_date", "end_time", "victim_age_group", "victim_sex", "suspect_age_group",
             "suspect_sex", "precinct", "late_report"]
    return out.sort_values(order, na_position="first", kind="mergesort").reset_index(drop=True)


def _victims(ind: pd.DataFrame, monthly: pd.DataFrame) -> pd.DataFrame:
    g = ind.groupby(["year", "law_cat_cd", "ofns_desc", "vic_age_group", "vic_sex"], dropna=False)["n"].sum().reset_index()
    return pd.DataFrame({"year": g["year"].astype(int), "law_category": g["law_cat_cd"], "offence": g["ofns_desc"],
                         "victim_age_group": g["vic_age_group"], "victim_sex": g["vic_sex"],
                         "complaints": g["n"].astype(int),
                         "months_incomplete": g["year"].map(_incomplete(monthly)).astype(int)}).sort_values(
        ["year", "law_category", "offence", "victim_age_group", "victim_sex"]).reset_index(drop=True)


def _report_lag(ind: pd.DataFrame, cutoff: dict, split_year: int, monthly: pd.DataFrame) -> pd.DataFrame:
    lag = (ind["report_date"] - ind["date"]).dt.days
    band = pd.Series(pd.NA, index=ind.index, dtype="object")
    for name, lo, hi in LAG_BANDS:
        band[(lag >= lo) & ((lag <= hi) if hi is not None else True)] = name
    g = ind.assign(lag_band=band).groupby(["year", "offence_group", "lag_band"])["n"].sum()
    groups = ["rape", "robbery", "felony_assault", "burglary", "grand_larceny", "other_felonies", "misdemeanors", "violations"]
    years = range(int(ind["year"].min()), int(ind["year"].max()) + 1)
    rows = [{"year": y, "offence_group": o, "lag_band": b, "lag_days_min": lo, "lag_days_max": hi,
             "complaints": int(g.get((y, o, b), 0)),
             "report_cutoff": cutoff["historic" if y < split_year else "ytd"].strftime("%Y-%m-%d"),
             "months_incomplete": int(_incomplete(monthly).get(y, 0))}
            for y in years for o in groups for b, lo, hi in LAG_BANDS]
    out = pd.DataFrame(rows)
    out["lag_days_max"] = out["lag_days_max"].astype("Int64")
    return out


def _arrests(a: pd.DataFrame) -> pd.DataFrame:
    g = a.groupby(["m", "law_cat_cd", "ky_cd", "ofns_desc", "pd_cd", "pd_desc"], dropna=False)["n"].sum().reset_index()
    return pd.DataFrame({"month": g["m"].map(_ym), "law_category": g["law_cat_cd"], "key_code": g["ky_cd"].astype("Int64"),
                         "offence": g["ofns_desc"], "pd_code": g["pd_cd"].astype("Int64"), "pd_description": g["pd_desc"],
                         "arrests": g["n"].astype(int)}).sort_values(
        ["month", "law_category", "key_code", "pd_code", "offence", "pd_description"], na_position="first").reset_index(drop=True)


def _context(monthly: pd.DataFrame, x: dict) -> pd.DataFrame:
    """The city around the subway, month by month: crime in the whole city, the population, crime
    and rides on buses, transit arrests and the shelter census."""
    months = pd.to_datetime(monthly["month"])

    def tally(d: pd.DataFrame) -> pd.DataFrame:
        col = pd.Series([_offence_column(o, m) for o, m in zip(d["ofns_desc"], d["major"])], index=d.index)
        t = pd.DataFrame({"m": d["m"], "six_major": d["n"].where(d["major"], 0),
                          "felony_assault": d["n"].where(col.eq("felony_assault"), 0),
                          "robbery": d["n"].where(col.eq("robbery"), 0),
                          "grand_larceny": d["n"].where(col.eq("grand_larceny"), 0), "all_complaints": d["n"]})
        return t.groupby("m").sum().reindex(months, fill_value=0)

    c = tally(x["city"])
    b = x["bus_crime"]
    on_bus = tally(b[b["prem_typ_desc"].eq(BUS)])
    at_stop = tally(b[~b["prem_typ_desc"].eq(BUS)])
    a = x["arrests"]
    arr = pd.DataFrame({"m": a["m"], "all": a["n"], "felony": a["n"].where(a["law_cat_cd"].eq("F"), 0),
                        "tos": a["n"].where((a["ky_cd"].eq(343) & a["pd_cd"].eq(art.THEFT_OF_SERVICES)).fillna(False), 0)}).groupby("m").sum().reindex(months)
    bus = x["bus_daily"]
    bm = bus.groupby(bus.index.to_period("M")).agg(["sum", "count", "size"])
    sh = x["shelter"]
    shm = sh.groupby(sh.index.to_period("M")).agg(["mean", "count"])
    rows = []
    for t, (_, mrow) in zip(months, monthly.iterrows()):
        p = t.to_period("M")
        days = p.days_in_month
        rides = int(bm.loc[p, "sum"]) if p in bm.index and bm.loc[p, "count"] == days else pd.NA
        rows.append({"month": _ym(t), "city_six_major": int(c.loc[t, "six_major"]),
                     "city_felony_assault": int(c.loc[t, "felony_assault"]), "city_robbery": int(c.loc[t, "robbery"]),
                     "city_grand_larceny": int(c.loc[t, "grand_larceny"]), "city_all_complaints": int(c.loc[t, "all_complaints"]),
                     "subway_share_of_city_six_major": round(mrow["six_major"] / c.loc[t, "six_major"], 4),
                     "nyc_population": _int(x["population"].get(t.year)),
                     "bus_six_major": int(on_bus.loc[t, "six_major"]), "bus_felony_assault": int(on_bus.loc[t, "felony_assault"]),
                     "bus_all_complaints": int(on_bus.loc[t, "all_complaints"]),
                     "bus_stop_and_terminal_complaints": int(at_stop.loc[t, "all_complaints"]), "bus_rides": rides,
                     "transit_arrests": _int(arr.loc[t, "all"]), "transit_arrests_felony": _int(arr.loc[t, "felony"]),
                     "transit_arrests_theft_of_services": _int(arr.loc[t, "tos"]),
                     "shelter_individuals": _int(shm.loc[p, "mean"]) if p in shm.index else pd.NA,
                     "shelter_census_days": int(shm.loc[p, "count"]) if p in shm.index else 0,
                     "complete": bool(mrow["complete"])})
    out = pd.DataFrame(rows)
    for k in ("nyc_population", "bus_rides", "transit_arrests", "transit_arrests_felony",
              "transit_arrests_theft_of_services", "shelter_individuals"):
        out[k] = out[k].astype("Int64")
    return out


def tables(x: dict | None = None) -> dict[str, pd.DataFrame]:
    """The product's data files as DataFrames, keyed by file name."""
    x = x or inputs()
    paid, entries, ev = x["paid"], x["entries"], x["evasion"].set_index(["year", "quarter"])
    split_year = int(x["cutoff"]["historic"].year) + 1    # the first year the year-to-date file holds
    counts = _month_counts(x)
    late = x["late"].copy()
    late["six"] = late["ofns_desc"].map(crime.is_subway_major)
    late_all = late.groupby("m")["n"].sum()
    late_six = late[late["six"]].groupby("m")["n"].sum()

    def share(year: int, month: int, col: str):
        v = ev["share" if col == "share" else "moe"].get((year, (month - 1) // 3 + 1))
        return pd.NA if v is None or pd.isna(v) else round(float(v), 4)

    # ---- the monthly panel
    rows = []
    for t, c in counts.iterrows():
        y, mo = int(t.year), int(t.month)
        f = "historic" if y < split_year else "year_to_date"
        cut = x["cutoff"]["historic" if f == "historic" else "ytd"]
        p, e = paid.get((y, mo)), entries.get((y, mo))
        row = {"month": _ym(t), "year": y, "fiscal_year": _fy(y, mo), **{k: int(c[k]) for k in COUNTS},
               "paid_rides": _int(p), "paid_rides_method": _method(y, mo), "unpaid_share": share(y, mo, "share"),
               "unpaid_share_moe": share(y, mo, "moe"), "entries_est": _int(e)}
        row.update({f"{k}_per_m_paid": _rate(c[k], p) for k in RATED})
        row.update({f"{k}_per_m_entries": _rate(c[k], e) for k in RATED})
        month_end = t + pd.offsets.MonthEnd(0)
        row["felony_assault_widenings_in_force"] = sum(w <= month_end for w in WIDENINGS)
        row["complaint_file"] = f
        row["report_cutoff"] = cut.strftime("%Y-%m-%d")
        # late reports exist only for months of the historic file; for the year-to-date file's months
        # they are already in the count, so the cell is left empty (not applicable), never zero
        row["late_reports_all"] = int(late_all.get(t, 0)) if f == "historic" else pd.NA
        row["late_reports_six_major"] = int(late_six.get(t, 0)) if f == "historic" else pd.NA
        row["complete"] = bool(cut >= t + pd.offsets.MonthEnd(2))
        rows.append(row)
    monthly = pd.DataFrame(rows)
    monthly[["late_reports_all", "late_reports_six_major"]] = monthly[["late_reports_all", "late_reports_six_major"]].astype("Int64")

    # ---- calendar years: whole years of complaints only
    years = [y for y in sorted(monthly["year"].unique()) if (monthly["year"] == y).sum() == 12]
    arows = []
    for y in years:
        keys = art.months_of(y)
        p = sum(paid[k] for k in keys)
        es = [entries.get(k) for k in keys]
        e = sum(es) if all(v is not None and pd.notna(v) for v in es) else None
        c = counts[counts.index.year == y].sum()
        mo = monthly[monthly["year"] == y]
        row = {"year": y, **{k: int(c[k]) for k in COUNTS}, "paid_rides": _int(p),
               "paid_rides_method": "official_total" if y <= 2020 else "daily_counts",
               "official_annual_total": _int(x["annual"].get(y)),
               "unpaid_share": round(1 - p / e, 4) if e else pd.NA, "entries_est": _int(e)}
        row.update({f"{k}_per_m_paid": _rate(c[k], p) for k in RATED})
        row.update({f"{k}_per_m_entries": _rate(c[k], e) for k in RATED})
        row["felony_assault_widenings_in_force"] = int(mo["felony_assault_widenings_in_force"].max())
        row["months_incomplete"] = int((~mo["complete"]).sum())
        arows.append(row)
    annual = pd.DataFrame(arows)

    # ---- fiscal years: the Mayor's Management Report against the rides of the same months
    prov = json.loads((_paths.REFERENCE / "source_provenance.json").read_text(encoding="utf-8"))
    edition_id = {pathlib.Path(v["dest"]).name: k for k, v in prov.items() if v.get("dest")}
    frows = []
    for _, m in x["mmr"].iterrows():
        fy = int(m["fiscal_year"])
        keys = art.fiscal_months(fy)
        p = sum(paid[k] for k in keys)
        es = [entries.get(k) for k in keys]
        e = sum(es) if all(v is not None and pd.notna(v) for v in es) else None
        methods = sorted({_method(*k) for k in keys}, key=["annual_total_x_2019_pattern", "annual_total_less_daily", "daily_counts"].index)
        frows.append({"fiscal_year": fy, "months": f"{fy - 1}-07 to {fy}-06", "major_felonies": int(m["n"]),
                      "crime_scope": "transit system (NYPD Transit Bureau)", "rides_scope": "subway paid entries",
                      "mmr_source_id": edition_id[str(m["edition"])], "note": FY_NOTES.get(fy, ""), "paid_rides": _int(p),
                      "paid_rides_method": " + ".join(methods), "major_felonies_per_m_paid": _rate(m["n"], p),
                      "unpaid_share": round(1 - p / e, 4) if e else pd.NA, "entries_est": _int(e),
                      "major_felonies_per_m_entries": _rate(m["n"], e)})
    fiscal = pd.DataFrame(frows)

    # ---- rides by month, from the first month of fiscal 1998 to the last whole month of the daily file
    sm = x["stations"].groupby("month")["ridership"].sum()
    station_sum = {(t.year, t.month): v for t, v in sm.items()}
    rrows = []
    for (y, mo), p in paid.items():
        if (y, mo) < FIRST_RIDE_MONTH or (y, mo) > x["last_ride_month"]:
            continue
        rrows.append({"month": f"{y:04d}-{mo:02d}", "year": y, "quarter": (mo - 1) // 3 + 1, "fiscal_year": _fy(y, mo),
                      "paid_rides": _int(p), "paid_rides_method": _method(y, mo),
                      "official_annual_total": _int(x["annual"].get(y)), "unpaid_share": share(y, mo, "share"),
                      "unpaid_share_moe": share(y, mo, "moe"), "entries_est": _int(entries.get((y, mo))),
                      "entries_est_unpaid_minus_1pt": _int(x["entries_lo"].get((y, mo))),
                      "entries_est_unpaid_plus_1pt": _int(x["entries_hi"].get((y, mo))),
                      "station_sum_rides": _int(station_sum.get((y, mo)))})
    rides = pd.DataFrame(rrows)
    rides["station_sum_rides"] = rides["station_sum_rides"].astype("Int64")

    evasion = x["evasion"].assign(period=lambda v: v["year"].astype(str) + "-Q" + v["quarter"].astype(str))
    evasion = pd.DataFrame({"period": evasion["period"], "year": evasion["year"], "quarter": evasion["quarter"],
                            "unpaid_share": evasion["share"].round(4), "margin_of_error": evasion["moe"].round(4)})

    d = x["d"]
    offences = pd.DataFrame({"month": d["m"].map(_ym), "law_category": d["law_cat_cd"], "offence": d["ofns_desc"],
                             "is_major_six": d["major"].astype(bool), "complaints": d["n"].astype(int),
                             "complaint_file": ["historic" if y < split_year else "year_to_date" for y in d["year"]]})
    offences = offences.sort_values(["month", "law_category", "offence"]).reset_index(drop=True)
    flags = monthly.set_index("month")[["report_cutoff", "complete"]]
    offences = offences.join(flags, on="month")

    # every key code by internal code, from the detail files (they equal the by-code files where both count)
    k = x["detail"].groupby(["m", "ky_cd", "ofns_desc", "pd_cd", "pd_desc"])["n"].sum().reset_index()
    codes = pd.DataFrame({"month": k["m"].map(_ym), "key_code": k["ky_cd"].astype(int), "offence": k["ofns_desc"],
                          "pd_code": k["pd_cd"].astype(int), "pd_description": k["pd_desc"],
                          "complaints": k["n"].astype(int),
                          "complaint_file": ["historic" if t.year < split_year else "year_to_date" for t in k["m"]]})
    codes = codes.sort_values(["month", "key_code", "pd_code", "offence", "pd_description"]).reset_index(drop=True).join(flags, on="month")

    ind = _indicators(x["detail"])
    stations, walk = _stations(ind, x, counts.index.max(), monthly)
    return {"subway_crime_monthly.csv": monthly, "subway_crime_annual.csv": annual,
            "transit_major_felonies_fiscal.csv": fiscal, "subway_ridership_monthly.csv": rides,
            "fare_evasion_quarterly.csv": evasion, "subway_complaints_by_offence_monthly.csv": offences,
            "subway_complaints_by_code_monthly.csv": codes,
            "subway_crime_daily.csv": _daily(ind, monthly, x), "subway_crime_by_hour.csv": _hourly(ind, monthly),
            "subway_stations_annual.csv": stations, "subway_station_crosswalk.csv": walk,
            "subway_victims_annual.csv": _victims(ind, monthly),
            "subway_report_delay_annual.csv": _report_lag(ind, x["cutoff"], split_year, monthly),
            "transit_arrests_monthly.csv": _arrests(x["arrests"]), "city_context_monthly.csv": _context(monthly, x),
            "subway_complaint_records.csv": _records(x, split_year)}


def _validate_finer(t: dict[str, pd.DataFrame]) -> list[str]:
    """The finer tables count the panel's complaints, no more and no fewer: each summed back to the
    month or year equals the panel; and the figures the card quotes from them."""
    bad = []
    m = t["subway_crime_monthly.csv"].set_index("month")
    a = t["subway_crime_annual.csv"].set_index("year")
    for name, keys in (("subway_crime_daily.csv", "month"), ("subway_crime_by_hour.csv", "month")):
        g = t[name].groupby(keys)[list(FINE)].sum()
        if not g.reindex(m.index).fillna(-1).astype(int).equals(m[list(FINE)].astype(int)):
            bad.append(f"{name} does not sum to the panel in every month")
    d = t["subway_crime_daily.csv"]
    counted = m[m["paid_rides_method"].eq("daily_counts")]
    dm = d.groupby("month")["paid_rides"].sum().reindex(counted.index)
    if not ((dm - counted["paid_rides"]).abs() <= 31).all():
        bad.append("the daily paid rides do not add up to the monthly paid rides of the counted months")
    if not d.groupby("month")["theft_of_services"].sum().reindex(m.index).eq(m["theft_of_services"]).all():
        bad.append("the daily theft of services does not sum to the panel's")
    if d["date"].duplicated().any() or len(d) != (pd.Timestamp(d["date"].iloc[-1]) - pd.Timestamp(d["date"].iloc[0])).days + 1:
        bad.append("the daily file skips or repeats a day")
    if d.loc[d["date"] >= "2020-03-01", "paid_rides"].isna().any() or d.loc[d["date"] < "2020-03-01", "paid_rides"].notna().any():
        bad.append("daily paid rides must be present from 1 March 2020 and only then")
    by_year = m.groupby("year")[list(FINE)].sum()
    s = t["subway_stations_annual.csv"].groupby("year")[list(FINE)].sum()
    if not s.equals(by_year.astype(s.dtypes.iloc[0])):
        bad.append("the station table does not sum to the panel in every year (a complaint left without a complex)")
    w = t["subway_station_crosswalk.csv"]
    if int(w["complaints"].sum()) != int(m["all_complaints"].sum()) + 109:
        bad.append("the crosswalk does not hold every complaint place (the panel's plus the 109 late reports)")
    share = w.groupby("match")["complaints"].sum() / w["complaints"].sum()
    if not (0.92 <= share.get("name", 0) < 0.93 and share.get("uncertain", 0) < 0.01):
        bad.append(f"station matches: {share.round(4).to_dict()}; the card says 92% by name within 300 m, under 1% uncertain")
    st = t["subway_stations_annual.csv"]
    if st.duplicated(["complex_id", "year"]).any() or st["complex_id"].nunique() < 420:
        bad.append("the station table's key (complex_id, year) is not unique, or complexes are missing")
    for name, col in (("subway_victims_annual.csv", "complaints"), ("subway_report_delay_annual.csv", "complaints")):
        g = t[name].groupby("year")[col].sum()
        if not g.equals(by_year["all_complaints"].astype(g.dtype)):
            bad.append(f"{name} does not sum to the panel's complaints in every year")
    ar = t["transit_arrests_monthly.csv"]
    tos = ar[ar["key_code"].eq(343) & ar["pd_code"].eq(art.THEFT_OF_SERVICES)].groupby(ar["month"].str[:4])["arrests"].sum()
    if (int(tos.get("2014", -1)), int(tos.get("2019", -1)), int(tos.get("2025", -1))) != (22776, 2712, 12475):
        bad.append(f"theft-of-services arrests: {tos.to_dict()}; the card says 22,776 in 2014, 2,712 in 2019, 12,475 in 2025")
    c = t["city_context_monthly.csv"].set_index("month")
    if not (c["city_six_major"] >= m["six_major"]).all() or c["nyc_population"].loc["2014-01":"2025-12"].isna().any():
        bad.append("the city's six major felonies must be at least the subway's, and the population present 2014-2025")
    if not c["transit_arrests_theft_of_services"].groupby(c.index.str[:4]).sum().reindex(tos.index).eq(tos).all():
        bad.append("the context file's theft-of-services arrests differ from the arrest table's")
    r = t["subway_ridership_monthly.csv"].set_index("month")
    ratio = (r["station_sum_rides"] / r["paid_rides"]).dropna()
    pre = ratio.loc[:"2020-02"]
    nov = pre[pre.index.str.endswith("-11")]
    w5 = [f"2019-{i:02d}" for i in range(1, 6)]
    jm = float(r.loc[w5, "station_sum_rides"].sum() / r.loc[w5, "paid_rides"].sum())
    if not (r["station_sum_rides"].first_valid_index() == "2017-02" and float((pre - 1).abs().max()) <= 0.06
            and (nov.between(1.03, 1.045)).all() and abs(jm - 1) <= 0.004):
        bad.append(f"station sums against modelled rides: {pre.min():.3f}-{pre.max():.3f}, Novembers {nov.round(3).to_dict()}, "
                   f"January-May 2019 {jm:.4f}; the card says counted within 6% of modelled from February 2017 to February 2020, "
                   "Novembers 3-4.5% above, "
                   "January-May 2019 within 0.4%")
    return bad


#: Phrases the card and events.csv quote from saved documents, each checked in its saved copy.
QUOTED = {"governor_ny_pause_2020.html": ["8PM on Sunday, March 22"],
          "mta_24_hour_service_restored_2021.html": ["May 6, 2020", "the closure covered 1 a.m. to 5 a.m.", "on February 15",
                                                     "from 2 a.m. to 4 a.m.", "2 a.m. on Monday, May 17"],
          "mta_fares_aug_2023.html": ["new fares take effect Sunday, Aug. 20", "will rise to $2.90, up from the current fare of $2.75"],
          "mta_fares_jan_2026.html": ["from $2.90 to $3", "take effect in January 2026"],
          "mta_crz_toll_phasing_2024.html": ["Sunday Morning, January 5, 2025"]}


def _validate_quotes(t: dict[str, pd.DataFrame]) -> list[str]:
    """Every figure the card's story, questions and cautions quote, against the files (and, for the
    phrases events.csv rests on, against the saved documents)."""
    import html as _html
    import re

    bad = []
    m = t["subway_crime_monthly.csv"]
    a = t["subway_crime_annual.csv"].set_index("year")
    pct = lambda v: int(round(100 * float(v)))
    if (round(a.loc[2019, "paid_rides"] / 1e9, 1), round(a.loc[2020, "paid_rides"] / 1e7) * 10) != (1.7, 640):
        bad.append("the card says 1.7 billion paid rides in 2019 and 640 million in 2020")
    cols = ["paid_rides", "felony_assault", "robbery", "grand_larceny"]
    q = a.loc[2020, cols].astype(float) / a.loc[2019, cols].astype(float)
    if (pct(q["paid_rides"]), pct(q["felony_assault"]), pct(q["robbery"])) != (38, 96, 105) or not 0.45 <= q["grand_larceny"] <= 0.55:
        bad.append("the card says 2020 had 38% of 2019's rides, 96% of its felony assaults, 105% of its robberies and half its grand larcenies")
    w = m[m["month"].str[5:].isin(["01", "02", "03", "04", "05"])].groupby("year")[["paid_rides", "entries_est"]].sum()
    if (pct(1 - w.loc[2019, "paid_rides"] / w.loc[2019, "entries_est"]), pct(1 - w.loc[2025, "paid_rides"] / w.loc[2025, "entries_est"])) != (4, 10):
        bad.append("the card says about 4% unpaid in early (January-May) 2019 and 10% in early 2025")
    s = t["subway_stations_annual.csv"]
    b = s[s["year"].eq(2025)].groupby("borough")[["six_major", "rides"]].sum()
    ratio = (b.loc["Bronx", "six_major"] / b.loc["Bronx", "rides"]) / (b.loc["Manhattan", "six_major"] / b.loc["Manhattan", "rides"])
    if not 2.7 <= ratio < 3.0:
        bad.append(f"Bronx against Manhattan per ride in 2025: {ratio:.2f}; the card says almost three times")
    d = t["subway_crime_daily.csv"]
    dd = d[d["year"].between(2023, 2025)].groupby("weekend")[["six_major", "paid_rides"]].sum()
    wk = (dd.loc[True, "six_major"] / dd.loc[True, "paid_rides"]) / (dd.loc[False, "six_major"] / dd.loc[False, "paid_rides"])
    if not 1.85 <= wk <= 2.15:
        bad.append(f"weekend against weekday per ride, 2023-2025: {wk:.2f}; the card says about twice")
    h = t["subway_crime_by_hour.csv"]
    small = lambda x: x.loc[x["hour"].between(1, 4), "all_complaints"].sum() / x["all_complaints"].sum()
    shares = (pct(small(h[h["month"].str[:4].eq("2019")])), pct(small(h[h["month"].between("2020-06", "2021-01")])),
              pct(small(h[h["month"].str[:4].eq("2022")])))
    if shares != (11, 4, 12):
        bad.append(f"complaints beginning 1-5 a.m.: {shares}%; the card says 11% in 2019, 4% during the closure, 12% in 2022")
    f = t["transit_major_felonies_fiscal.csv"].set_index("fiscal_year")["major_felonies_per_m_paid"]
    if (round(f.loc[1998], 1), round(f.loc[2026], 1)) != (5.0, 1.7):
        bad.append("the card says 5.0 transit felonies per million rides in fiscal 1998 and 1.7 in fiscal 2026")
    c = t["city_context_monthly.csv"]
    cy = c.groupby(c["month"].str[:4])["city_six_major"].sum()
    sy = m.groupby(m["month"].str[:4])["six_major"].sum()
    if (round(100 * sy["2019"] / cy["2019"], 1), round(100 * sy["2025"] / cy["2025"], 1)) != (2.8, 2.0):
        bad.append("the card says the subway held 2.8% of the city's six major felonies in 2019 and 2.0% in 2025")
    sh = c.set_index("month")["shelter_individuals"]
    if not 1.9 <= sh["2024-01"] / sh["2022-01"] < 2.0:
        bad.append("the card says the shelter census nearly doubled from January 2022 to January 2024")
    det, late = crime.subway_complaint_detail()
    tm = det.groupby("time")["n"].sum()
    if not tm.get("00:01:00", 0) > 15 * tm.get("00:02:00", 1):
        bad.append("the card says 00:01 is more than fifteen times as common as 00:02")
    dec = det[det["m"].eq("2025-12-01") & det["law_cat_cd"].eq("FELONY")]["n"].sum()
    dec_late = late[late["m"].eq("2025-12-01") & late["law_cat_cd"].eq("FELONY")]["n"].sum()
    if pct(dec_late / (dec + dec_late)) != 5:
        bad.append(f"December 2025 felonies filed in 2026: {dec_late} of {dec + dec_late}; the card says about 5%")
    r = t["subway_ridership_monthly.csv"].set_index("month").loc["2024-01":]
    above = float(r["station_sum_rides"].sum() / r.loc[r["station_sum_rides"].notna(), "paid_rides"].sum())
    if not 1.005 <= above <= 1.015:
        bad.append(f"station sums against daily counts since 2024: {above:.4f}; the card says about 1% above")
    for name, phrases in QUOTED.items():
        text = (_paths.REFERENCE / "reports" / name).read_text(encoding="utf-8", errors="replace")
        text = re.sub(r"\s+", " ", _html.unescape(re.sub(r"<[^>]+>", " ", text))).replace("\xa0", " ")
        bad += [f"{name} no longer says {p!r}" for p in phrases if p not in text]
    return bad


def _validate_records(t: dict[str, pd.DataFrame]) -> list[str]:
    """The record file holds the panel's complaints and the 109 late reports, one row each; a cell is
    empty exactly where the NYPD's file is blank or "(null)", and every other value is the NYPD's
    (so nothing was blanked or dropped here); and the figures the card quotes from it."""
    bad = []
    rec = t["subway_complaint_records.csv"]
    m = t["subway_crime_monthly.csv"]
    r, late = crime.subway_complaint_records()
    raw = pd.concat([r, late], ignore_index=True)
    if len(rec) != int(m["all_complaints"].sum()) + 109 or int(rec["late_report"].sum()) != 109 or len(rec) != int(raw["n"].sum()):
        bad.append(f"the record file has {len(rec):,} rows ({int(rec['late_report'].sum())} late); the panel and the late reports make "
                   f"{int(m['all_complaints'].sum()) + 109:,}")
    on = rec[~rec["late_report"]]
    g = on.fillna({"offence": "(null)"}).groupby(["month", "law_category", "offence"]).size()
    o = t["subway_complaints_by_offence_monthly.csv"].groupby(["month", "law_category", "offence"])["complaints"].sum()
    if not g.sort_index().equals(o.sort_index().astype(g.dtype)):
        bad.append("the record file, counted by month and offence, is not the offence table")
    for col, field in RECORD_SOURCE.items():
        blank = int(raw.loc[raw[field].isin(SOURCE_BLANK), "n"].sum())
        if int(rec[col].isna().sum()) != blank:
            bad.append(f"{col}: {int(rec[col].isna().sum())} empty cells, but the NYPD's {field} is blank in {blank}")
        want = raw.loc[~raw[field].isin(SOURCE_BLANK)].groupby(field)["n"].sum()
        got = rec[col].dropna().astype("string").value_counts()
        if col in ("pd_code", "precinct", "transit_district"):
            want.index = [str(int(float(v))) for v in want.index]
            want = want.groupby(level=0).sum()
        elif col == "end_date":
            want.index = [v[:10] for v in want.index]
            want = want.groupby(level=0).sum()
        if not got.sort_index().equals(want.sort_index().astype(got.dtype)):
            bad.append(f"{col}: its values are not the NYPD's {field}, value for value")
    # whole records: the multiset of normalised source tuples equals the file's, so no value moved between rows
    fields = list(RECORD_SOURCE.items()) + [("key_code", "ky_cd")]
    src = pd.DataFrame({col: raw[f].where(~raw[f].isin(SOURCE_BLANK), "") for col, f in fields})
    src["complaint_date"], src["report_date"] = raw["cmplnt_fr_dt"].str[:10], raw["rpt_dt"].str[:10]
    src["end_date"] = src["end_date"].str[:10]
    for col in ("pd_code", "precinct", "transit_district", "key_code"):
        src[col] = src[col].map(lambda v: str(int(float(v))) if v else "")
    want = src.assign(n=raw["n"]).groupby(list(src.columns))["n"].sum()
    out = rec[list(src.columns)].astype("string").fillna("")
    got = out.groupby(list(out.columns)).size()
    if not got.sort_index().equals(want.sort_index().astype(got.dtype)):
        bad.append("the record file's rows, field by field together, are not the NYPD extract's")
    quoted = {"rows": len(rec), "end_date": int(rec["end_date"].isna().sum()),
              "suspect_age_group": int(rec["suspect_age_group"].isna().sum()), "suspect_sex": int(rec["suspect_sex"].isna().sum()),
              "suspect_unknown": int(rec["suspect_age_group"].eq("UNKNOWN").sum())}
    if quoted != {"rows": 187739, "end_date": 16297, "suspect_age_group": 33830, "suspect_sex": 31482, "suspect_unknown": 29911}:
        bad.append(f"the record file: {quoted}; the card quotes 187,739 rows, 16,297 without an end date, 33,830 and 31,482 "
                   "without the suspect's age group and sex, 29,911 suspect age groups recorded UNKNOWN")
    return bad


def validate(t: dict[str, pd.DataFrame], article_csv: pathlib.Path | None = None) -> list[str]:
    """Problems, if any: the panel's own sums, and every row of the article's published data.csv
    reproduced from the panel (counts exactly, rides to the article's rounding)."""
    bad = []
    m, a, f = t["subway_crime_monthly.csv"], t["subway_crime_annual.csv"], t["transit_major_felonies_fiscal.csv"]
    if not (m["six_major"] == m[list(SIX)].sum(axis=1)).all():
        bad.append("six_major is not the sum of the six offences in every month")
    if not (m["felony_assault"] == m[list(FA_CODES)].sum(axis=1)).all():
        bad.append("the felony-assault codes do not add up to felony_assault in every month")
    if int(m["murder"].sum()) != 0:
        bad.append("the card says the subway premises records hold no murder")
    o, k = t["subway_complaints_by_offence_monthly.csv"], t["subway_complaints_by_code_monthly.csv"]
    if o.duplicated(["month", "law_category", "offence", "complaint_file"]).any():
        bad.append("the long table's key (month, law_category, offence, complaint_file) is not unique")
    if k.duplicated(["month", "key_code", "offence", "pd_code", "complaint_file"]).any():
        bad.append("the code table's key is not unique")
    if set(m["month"]) != set(o["month"]):
        bad.append("a month of the panel has no complaint record at all (a gap in the source, not a zero)")
    fa = k[k["key_code"].eq(106)].groupby("month")["complaints"].sum()
    if not (fa.reindex(m["month"]).fillna(0).astype(int).values == m["felony_assault"].values).all():
        bad.append("the code extract's felony assaults differ from the offence file's in some month")
    tos = k[k["key_code"].eq(343) & k["pd_code"].eq(art.THEFT_OF_SERVICES)].groupby("month")["complaints"].sum()
    if not (tos.reindex(m["month"]).fillna(0).astype(int).values == m["theft_of_services"].values).all():
        bad.append("theft_of_services is not the code table's PD_CD 478 in some month")
    if m.loc[m["complaint_file"].eq("year_to_date"), "late_reports_all"].notna().any():
        bad.append("late reports must be empty (not applicable) for the year-to-date file's months")
    if int(m["late_reports_all"].sum()) != 107:
        bad.append(f"late reports: {int(m['late_reports_all'].sum())}, the card says 107 from 2014 (of the article's 109)")
    tos = a.set_index("year")["theft_of_services"]
    if (int(tos.get(2019, -1)), int(tos.get(2025, -1))) != (54, 13009):
        bad.append("theft of services: the card says 54 in 2019 and 13,009 in 2025")
    if int(f.set_index("fiscal_year").loc[2026, "major_felonies"]) != 2176:
        bad.append("fiscal 2026's major felonies are not the 2,176 the card quotes")
    fy26 = int(m.loc[m["fiscal_year"].eq(2026), "six_major"].sum())
    if fy26 != 2093:
        bad.append(f"the six major felonies in the records for fiscal 2026 are {fy26}, the card says 2,093")
    if list(m.loc[~m["complete"], "month"]) != ["2025-12", "2026-06"]:
        bad.append(f"incomplete months are {list(m.loc[~m['complete'], 'month'])}, the card says December 2025 and June 2026")
    bad += _validate_finer(t)
    bad += _validate_quotes(t)
    bad += _validate_records(t)

    src = article_csv or (_paths.PROJECT / "site" / "public" / "articles" / art.SLUG / "data.csv")
    pub = pd.read_csv(src)
    key = {"six major": "six_major", "felony assault": "felony_assault", "robbery": "robbery",
           "grand larceny": "grand_larceny", "all felonies": "all_felonies",
           "theft of services (NYPD code 478)": "theft_of_services",
           **{f"felony assault: {v}": k for k, v in FA_CODES.items()}}
    ay = a.set_index("year")
    for _, row in pub.iterrows():
        series, period, col = row["series"], str(row["period"]), key.get(row["offence"])
        if series.endswith("calendar year"):
            got_n, got_p = int(ay.loc[int(period), col]), r(ay.loc[int(period), "paid_rides"] / 1e6, 1)
            got_e = None
        elif series.endswith("January to May"):
            w = m[(m["year"] == int(period)) & m["month"].str[-2:].astype(int).isin(art.WINDOW)]
            got_n, got_p, got_e = int(w[col].sum()), r(w["paid_rides"].sum() / 1e6, 1), r(w["entries_est"].sum() / 1e6, 1)
        else:
            fy = f.set_index("fiscal_year").loc[int(period[2:])]
            got_n, got_p, got_e = int(fy["major_felonies"]), r(fy["paid_rides"] / 1e6, 0), None
        want_p = float(row["paid_rides_m"])
        if got_n != int(row["count"]) or abs(got_p - want_p) > 0.051:
            bad.append(f"{series} {period} {row['offence']}: {got_n}, {got_p} against the article's {row['count']}, {want_p}")
        if got_e is not None and pd.notna(row["entries_m"]) and abs(got_e - float(row["entries_m"])) > 0.051:
            bad.append(f"{series} {period} entries: {got_e} against the article's {row['entries_m']}")
    return bad