#!/usr/bin/env python3
"""Re-run the evidence check behind the Phayan post thai-tones-with-audio.

Three checks live here:

  lab            the five tone files the page plays for กา are downloaded from
                 the public API, converted with ffmpeg and their pitch contours
                 measured with numpy. Every number in dataset.json must come
                 back within tolerance.
  primary-docs   every quote recorded against a third party's own page must
                 still appear on that page (whitespace, quotes, dashes and HTML
                 entities normalised before comparing).
  product-data   production cannot be reached from here, so the dataset's own
                 counts are checked instead: every cell built from fewer than 10
                 learners must be suppressed, and the ten contour distances
                 must be present.

Exit 0 when everything holds, 1 otherwise.

  python3 check.py            both checks
  python3 check.py --recheck  re-reads the pages; skips the lab step if it
                              would run longer than a minute
  python3 check.py --record   rewrite the lab rows and pairwise distances in
                              dataset.json from a live measurement (used once,
                              when the package was built)
"""

import argparse
import hashlib
import html
import json
import re
import shutil
import subprocess
import sys
import tempfile
import time
import urllib.request
import wave
from pathlib import Path

import numpy as np

ROOT = Path(__file__).resolve().parent
DATASET = ROOT / "dataset.json"
CSV = ROOT / "dataset.csv"
UA = "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126 Safari/537.36"

# Frozen analysis parameters. Changing any of these changes the numbers, so they
# are part of the method, not tuning knobs.
WIN_S = 0.040
HOP_S = 0.010
F0_MIN_HZ = 60.0
F0_MAX_HZ = 400.0
RMS_GATE = 0.02
ACF_FLOOR = 0.35
ONSET_SKIP_S = 0.060
TOL = 0.06


# ---------------------------------------------------------------- text checks

def fetch(url, timeout=40):
    req = urllib.request.Request(url, headers={"User-Agent": UA})
    with urllib.request.urlopen(req, timeout=timeout) as r:
        return r.read()


def html_to_text(raw):
    raw = re.sub(r"(?is)<(script|style|noscript|svg)[^>]*>.*?</\1>", " ", raw)
    raw = re.sub(r"(?s)<!--.*?-->", " ", raw)
    raw = re.sub(r"(?i)</(p|div|li|h[1-6]|tr|section|br)>", "\n", raw)
    raw = re.sub(r"(?s)<[^>]+>", " ", raw)
    return html.unescape(raw)


def normalise(text):
    text = html.unescape(text)
    text = text.replace("‘", "'").replace("’", "'")
    text = text.replace("“", '"').replace("”", '"')
    text = re.sub(r"[‐-―−]", "-", text)
    text = text.replace(" ", " ")
    return re.sub(r"\s+", " ", text).strip().lower()


def quote_present(page_text, quote):
    return normalise(quote) in normalise(page_text)


# ------------------------------------------------------------- audio / pitch

def decode_mp3(data):
    if not shutil.which("ffmpeg"):
        raise RuntimeError("ffmpeg not on PATH")
    with tempfile.TemporaryDirectory() as td:
        src = Path(td) / "in.mp3"
        dst = Path(td) / "out.wav"
        src.write_bytes(data)
        subprocess.run(
            ["ffmpeg", "-nostdin", "-v", "error", "-y", "-i", str(src),
             "-ac", "1", "-ar", "16000", "-f", "wav", str(dst)],
            check=True, capture_output=True,
        )
        with wave.open(str(dst)) as w:
            sr = w.getframerate()
            x = np.frombuffer(w.readframes(w.getnframes()), dtype=np.int16)
    return x.astype(np.float64) / 32768.0, sr


def f0_track(x, sr):
    W = int(sr * WIN_S)
    H = int(sr * HOP_S)
    lo, hi = int(sr / F0_MAX_HZ), int(sr / F0_MIN_HZ)
    times, freqs = [], []
    for s in range(0, len(x) - W + 1, H):
        frame = x[s:s + W]
        frame = frame - frame.mean()
        if np.sqrt((frame ** 2).mean()) < RMS_GATE:
            continue
        ac = np.correlate(frame, frame, "full")[W - 1:]
        if ac[0] <= 0:
            continue
        ac = ac / ac[0]
        seg = ac[lo:hi]
        k = int(np.argmax(seg)) + lo
        if 0 < k < len(ac) - 1:
            a, b, c = ac[k - 1], ac[k], ac[k + 1]
            denom = a - 2 * b + c
            if denom != 0:
                k = k + float(np.clip(0.5 * (a - c) / denom, -0.5, 0.5))
        if ac[int(round(k))] < ACF_FLOOR:
            continue
        times.append(s / sr)
        freqs.append(sr / k)
    t = np.array(times)
    f = np.array(freqs)
    if len(t) == 0:
        raise RuntimeError("no voiced frames")
    keep = t >= t[0] + ONSET_SKIP_S
    return t[keep], f[keep]


def measure(data):
    x, sr = decode_mp3(data)
    t, f = f0_track(x, sr)
    st = 12 * np.log2(f / np.median(f))
    pos = np.linspace(0.0, 1.0, len(f))
    return {
        "voiced_frames": int(len(f)),
        "voiced_duration_s": round(len(f) * HOP_S, 2),
        "median_f0_hz": round(float(np.median(f)), 1),
        "min_f0_hz": int(round(float(f.min()))),
        "max_f0_hz": int(round(float(f.max()))),
        "f0_range_semitones": round(float(np.percentile(st, 95) - np.percentile(st, 5)), 2),
        "start_semitones_vs_median": round(float(st[0]), 2),
        "peak_semitones_vs_median": round(float(st.max()), 2),
        "trough_semitones_vs_median": round(float(st.min()), 2),
        "end_semitones_vs_median": round(float(st[-1]), 2),
        "peak_position": round(float(pos[int(np.argmax(f))]), 2),
        "trough_position": round(float(pos[int(np.argmin(f))]), 2),
        "_shape": np.interp(np.linspace(0, 1, 101), pos, st),
    }


property_order = [
    "voiced_frames", "voiced_duration_s", "median_f0_hz", "min_f0_hz", "max_f0_hz",
    "f0_range_semitones", "start_semitones_vs_median", "peak_semitones_vs_median",
    "trough_semitones_vs_median", "end_semitones_vs_median",
    "peak_position", "trough_position",
]


def same(expected, got):
    if isinstance(expected, int):
        return int(got) == expected
    return abs(float(got) - float(expected)) <= TOL


# ---------------------------------------------------------------------- main

def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--recheck", action="store_true",
                    help="re-read the pages; skip the lab step if it would exceed a minute")
    ap.add_argument("--record", action="store_true",
                    help="rewrite the lab rows in dataset.json from a live measurement")
    ap.add_argument("--csv", action="store_true",
                    help="rewrite dataset.csv from dataset.json")
    args = ap.parse_args()

    ds = json.loads(DATASET.read_text(encoding="utf-8"))
    failures = 0
    rows_checked = 0

    # ---- lab: the five tone files
    measured, hashes = {}, {}
    audio_skip = False
    tones = ds["lab"]["tones"]
    for i, tone in enumerate(tones):
        n = tone["tone"]
        t0 = time.time()
        data = fetch(tone["url"])
        hashes[n] = hashlib.sha256(data).hexdigest()
        measured[n] = measure(data)
        if args.recheck and i == 0 and (time.time() - t0) * len(tones) > 60:
            audio_skip = True
            measured.clear()
            hashes.clear()
            break

    if args.record:
        shapes = {}
        lab_rows = []
        for tone in ds["lab"]["tones"]:
            n = tone["tone"]
            m = measured[n]
            shapes[n] = m["_shape"]
            for prop in property_order:
                lab_rows.append({
                    "group": "tone_audio",
                    "item": tone["item"],
                    "property": prop,
                    "value": m[prop],
                    "unit": tone.get("units", {}).get(prop, ""),
                    "quote": "",
                    "url": tone["url"],
                    "status": "measured",
                })
        pairs = []
        keys = sorted(shapes)
        for i, a in enumerate(keys):
            for b in keys[i + 1:]:
                d = float(np.sqrt(np.mean((shapes[a] - shapes[b]) ** 2)))
                pairs.append({"pair": "%d-%d" % (a, b), "distance_semitones_rms": round(d, 2)})
        pairs.sort(key=lambda p: p["distance_semitones_rms"])
        ds["lab"]["sha256"] = hashes
        ds["lab"]["shape_distance"] = pairs
        ds["results"] = [r for r in ds["results"] if r["group"] != "tone_audio"] + lab_rows
        DATASET.write_text(json.dumps(ds, indent=2, ensure_ascii=False) + "\n", encoding="utf-8")
        print("recorded %d lab rows and %d pairwise distances" % (len(lab_rows), len(pairs)))

    if audio_skip:
        print("SKIP lab (--recheck, over a minute)")
    else:
        for row in ds["results"]:
            if row["group"] != "tone_audio":
                continue
            tone = next(t for t in ds["lab"]["tones"] if t["item"] == row["item"])
            n = tone["tone"]
            got = measured[n][row["property"]]
            rows_checked += 1
            if same(row["value"], got):
                print("ok    %s / %s = %s" % (row["item"], row["property"], got))
            else:
                failures += 1
                print("FAIL  %s / %s expected %s got %s" % (row["item"], row["property"], row["value"], got))
        for n, want in ds["lab"].get("sha256", {}).items():
            got = hashes.get(int(n))
            if got != want:
                failures += 1
                print("FAIL  tone %s audio bytes changed (%s != %s)" % (n, got, want))

    # ---- primary-docs: every recorded quote still on its page
    pages = {}
    for url in {r["url"] for r in ds["results"] if r.get("url") and r.get("quote")}:
        try:
            pages[url] = html_to_text(fetch(url).decode("utf-8", "replace"))
        except Exception as exc:  # noqa: BLE001 - a page we cannot read is a failure
            pages[url] = None
            print("ERR   could not fetch %s (%s)" % (url, exc))

    for row in ds["results"]:
        if row["group"] == "tone_audio":
            continue
        if not row.get("quote"):
            print("skip  %s / %s recorded as %s, no supporting sentence" % (row["item"], row["property"], row["status"]))
            continue
        rows_checked += 1
        text = pages.get(row["url"])
        if text is None:
            failures += 1
            print("FAIL  %s / %s page not readable" % (row["item"], row["property"]))
        elif quote_present(text, row["quote"]):
            print("ok    %s / %s" % (row["item"], row["property"]))
        else:
            failures += 1
            print("FAIL  %s / %s quote gone from %s" % (row["item"], row["property"], row["url"]))

    if args.csv:
        import csv as _csv
        with CSV.open("w", newline="", encoding="utf-8") as fh:
            w = _csv.writer(fh)
            w.writerow(["group", "item", "property", "value", "unit", "quote", "url", "status"])
            for row in ds["results"]:
                w.writerow([row["group"], row["item"], row["property"], row["value"],
                            row.get("unit", ""), row.get("quote", ""), row.get("url", ""), row.get("status", "")])
        print("wrote %d rows to dataset.csv" % len(ds["results"]))

    # ---- product-data: suppression and internal counts
    pdata = ds.get("productData")
    if pdata:
        floor = pdata["minLearnersPerCell"]
        for cell in pdata["results"]:
            rows_checked += 1
            learners = cell["learners"]
            small = learners == "<%d" % floor or (isinstance(learners, int) and learners < floor)
            leaked = [k for k in ("attempts", "correct") if cell[k] != "suppressed"]
            if small and leaked:
                failures += 1
                print("FAIL  productData / %s publishes %s from fewer than %d learners" % (cell["tag"], ",".join(leaked), floor))
            elif not small and any(not isinstance(cell[k], int) or cell[k] < 0 for k in ("attempts", "correct")):
                failures += 1
                print("FAIL  productData / %s has a malformed count" % cell["tag"])
            elif not small and cell["correct"] > cell["attempts"]:
                failures += 1
                print("FAIL  productData / %s correct exceeds attempts" % cell["tag"])
            else:
                print("ok    productData / %s (learners %s)" % (cell["tag"], learners))
        if pdata.get("approval") != "owner, 26 October 2026, counts only":
            failures += 1
            print("FAIL  productData approval missing")
    pairs = ds["lab"].get("shape_distance", [])
    rows_checked += 1
    if len({p["pair"] for p in pairs}) == 10:
        print("ok    lab / ten pairwise contour distances")
    else:
        failures += 1
        print("FAIL  lab / expected 10 pairwise distances, found %d" % len(pairs))

    # ---- csv must agree with the json
    if CSV.exists():
        want = len({(r["group"], r["item"], r["property"]) for r in ds["results"]})
        got = max(0, len(CSV.read_text(encoding="utf-8").strip().splitlines()) - 1)
        if want != got:
            failures += 1
            print("FAIL  dataset.csv has %d rows, dataset.json has %d" % (got, want))

    print("\n%d checks, %d failed" % (rows_checked, failures))
    return 0 if failures == 0 else 1


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