#!/usr/bin/env python3
"""Re-run the check behind "How to Share Photos Between iPhone and Android".

Four kinds of row in dataset.json:

  page     a vendor's own help page, fetched now; every quote must still appear on it
  code     a file in this repository; every quote must still be in it
  lab      a value this script measures again by running Viallo's own HEIC conversion
  derived  a table cell built from other rows; it holds when all of them hold

The lab needs macOS (sips), exiftool, Node 22 and `pnpm install` in the repository,
because it imports heic-convert and sharp from apps/api and the profile helpers from
apps/api/src/common/heic-convert.util.ts. The test HEIC comes from
$VIALLO_TEST_INPUTS/photo-48mp.heic, or is rebuilt from a NASA public-domain photo.

Exits 0 when every row holds, 1 otherwise. --recheck re-reads every page and skips
the lab when rebuilding its input would take longer than a minute.
"""

import csv
import hashlib
import html
import json
import os
import re
import shutil
import subprocess
import sys
import tempfile
import time
from pathlib import Path

HERE = Path(__file__).resolve().parent
REPO = HERE.parents[4]
API = REPO / "apps" / "api"
UA = "Mozilla/5.0 (Macintosh; Intel Mac OS X 14_0) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.0 Safari/605.1.15"
TABLE = ["reaches the other platform", "quality or size limit", "what the recipient needs"]
NASA = "https://images-assets.nasa.gov/image/KSC-20220119-PH-CSH01_0011/KSC-20220119-PH-CSH01_0011~orig.jpg"
P3 = "/System/Library/ColorSync/Profiles/Display P3.icc"
# sips' HEIC encoder differs between macOS releases, so a rebuilt input is compared
# on pixels and metadata, and byte counts only within this tolerance.
BYTE_TOLERANCE = 0.02
LAB_SECONDS = 60

PIPELINE = r"""
import { createRequire } from 'node:module';
import fs from 'node:fs';
const [,, apiDir, input, output] = process.argv;
const require = createRequire(apiDir + '/package.json');
const convert = require('heic-convert');
const sharp = require('sharp');
const { readHeifIccProfile, injectIccIntoJpeg } = await import(apiDir + '/src/common/heic-convert.util.ts');
const heic = fs.readFileSync(input);
// The same two steps as ImageProcessingService.convertHeicIfNeeded.
let jpeg = Buffer.from(await convert({ buffer: heic, format: 'JPEG', quality: 0.92 }));
const icc = readHeifIccProfile(heic);
if (icc) jpeg = await sharp(injectIccIntoJpeg(jpeg, icc), { failOn: 'none' }).withIccProfile('srgb').jpeg({ quality: 92 }).toBuffer();
fs.writeFileSync(output, jpeg);
"""


def norm(text):
    text = html.unescape(text)
    for src, dst in (("’", "'"), ("‘", "'"), ("“", '"'), ("”", '"'), (" ", " ")):
        text = text.replace(src, dst)
    text = re.sub(r"\s+", " ", text)
    return re.sub(r"\s+([.,;:!?])", r"\1", text).strip()


def page_text(url):
    proc = subprocess.run(
        ["curl", "-sL", "--compressed", "-m", "45", "-A", UA, "-H", "Accept-Language: en-US,en",
         "-w", "\n%{http_code}", url],
        capture_output=True,
    )
    body, _, code = proc.stdout.decode("utf-8", "replace").rpartition("\n")
    body = re.sub(r"(?is)<(script|style)[^>]*>.*?</\1>", " ", body)
    return code.strip(), norm(re.sub(r"<[^>]+>", " ", body))


def quotes(row):
    q = row.get("quote", "")
    return q if isinstance(q, list) else [q]


# ---------------------------------------------------------------- lab

def run(cmd, timeout=300):
    proc = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout)
    if proc.returncode != 0:
        raise RuntimeError("%s: %s" % (cmd[0], (proc.stderr or proc.stdout).strip()[:300]))
    return proc.stdout


def exif(path):
    out = run(["exiftool", "-j", "-n", "-ImageWidth", "-ImageHeight", "-FileSize", "-GPSLatitude",
               "-DateTimeOriginal", "-ProfileDescription", str(path)])
    return json.loads(out)[0]


def rebuild_input(tmp):
    """The steps that made photo-48mp.heic: a NASA photo, cropped, scaled, tagged as an iPhone."""
    src = tmp / "src.jpg"
    run(["curl", "-sf", "-o", str(src), NASA], timeout=120)
    run(["sips", "-c", "5120", "3840", str(src), "--out", str(tmp / "crop.jpg")])
    heic = tmp / "photo-48mp.heic"
    run(["sips", "-z", "8064", "6048", "-s", "format", "heic", str(tmp / "crop.jpg"), "--out", str(heic)])
    run(["exiftool", "-q", "-overwrite_original", "-all=", str(heic)])
    run(["exiftool", "-q", "-overwrite_original", "-Make=Apple", "-Model=iPhone 15 Pro", "-Software=18.0",
         "-DateTimeOriginal=2026:10:25 09:30:00", "-CreateDate=2026:10:25 09:30:00", "-OffsetTimeOriginal=-04:00",
         "-GPSLatitude=28.586", "-GPSLatitudeRef=N", "-GPSLongitude=80.651", "-GPSLongitudeRef=W",
         "-GPSAltitude=3", "-GPSAltitudeRef=0", "-Orientation#=1", str(heic)])
    return heic


def describe(meta):
    return {
        "pixels": "%dx%d" % (meta["ImageWidth"], meta["ImageHeight"]),
        "bytes": int(meta["FileSize"]),
        "gps": "present" if "GPSLatitude" in meta else "absent",
        "date": meta.get("DateTimeOriginal", "absent"),
        "icc": meta.get("ProfileDescription", "none"),
    }


def run_lab(tmp, recheck, expected_sha):
    if not (API / "node_modules" / "heic-convert").exists():
        raise RuntimeError("apps/api/node_modules has no heic-convert; run pnpm install first")
    given = os.environ.get("VIALLO_TEST_INPUTS")
    started = time.time()
    if given and (Path(given) / "photo-48mp.heic").exists():
        heic = Path(given) / "photo-48mp.heic"
    elif recheck:
        return None
    else:
        heic = rebuild_input(tmp)
    sha = hashlib.sha256(heic.read_bytes()).hexdigest()

    p3 = tmp / "p3.heic"
    run(["sips", "-s", "format", "heic", "--embedProfile", P3, str(heic), "--out", str(p3)])
    script = tmp / "pipeline.mjs"
    script.write_text(PIPELINE)
    plain_out, p3_out, cap = tmp / "plain.jpg", tmp / "p3.jpg", tmp / "cap.jpg"
    for src, dst in ((heic, plain_out), (p3, p3_out)):
        run(["node", "--experimental-strip-types", "--no-warnings", str(script), str(API), str(src), str(dst)])
    run(["sips", "-Z", "2048", str(plain_out), "--out", str(cap)])

    m_in, m_p3in = describe(exif(heic)), describe(exif(p3))
    m_plain, m_p3, m_cap = describe(exif(plain_out)), describe(exif(p3_out)), describe(exif(cap))
    w, h = map(int, m_in["pixels"].split("x"))
    cw, ch = map(int, m_cap["pixels"].split("x"))
    lab = {"input_sha256": sha, "seconds": round(time.time() - started, 1)}
    for prefix, m in (("in", m_in), ("p3_in", m_p3in), ("plain", m_plain), ("p3", m_p3), ("cap", m_cap)):
        for key, value in m.items():
            lab["%s_%s" % (prefix, key)] = value
    lab["in_mp"] = "%.2f" % (w * h / 1e6)
    lab["cap_mp"] = "%.2f" % (cw * ch / 1e6)
    lab["cap_share_pct"] = "%.2f" % (100.0 * cw * ch / (w * h))
    lab["same_input"] = sha == expected_sha
    return lab


def lab_holds(row, lab):
    key, want = row["lab_key"], row["value"]
    got = lab.get(key)
    if key.endswith("_bytes"):
        want_n, got_n = int(str(want).replace(",", "")), int(got)
        if got_n == want_n:
            return True, str(got_n)
        if not lab["same_input"] or key.startswith("p3"):
            return abs(got_n - want_n) <= want_n * BYTE_TOLERANCE, "%d (tolerance %.0f %%)" % (got_n, BYTE_TOLERANCE * 100)
        return False, str(got_n)
    return str(got) == str(want), str(got)


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

def csv_matches(data):
    table = {}
    for row in data["results"]:
        if row["property"] in TABLE:
            table.setdefault(row["option"], {})[row["property"]] = row["value"]
    with open(HERE / "dataset.csv", newline="") as handle:
        rows = list(csv.DictReader(handle))
    problems = []
    if [r["option"] for r in rows] != list(table):
        problems.append("options differ: csv %r, json %r" % ([r["option"] for r in rows], list(table)))
    for r in rows:
        for prop in TABLE:
            if table.get(r["option"], {}).get(prop) != r[prop]:
                problems.append("%s / %s: csv %r, json %r" % (r["option"], prop, r[prop], table.get(r["option"], {}).get(prop)))
    return problems


def main():
    recheck = "--recheck" in sys.argv[1:]
    data = json.loads((HERE / "dataset.json").read_text())
    failed = 0

    problems = csv_matches(data)
    for problem in problems:
        print("FAIL dataset.csv", problem)
    failed += len(problems)
    if not problems:
        print("ok   dataset.csv matches dataset.json")

    tmp = Path(tempfile.mkdtemp(prefix="iphone-android-"))
    lab, lab_error = None, None
    try:
        lab = run_lab(tmp, recheck, data["labInputSha256"])
        if lab is None:
            lab_error = "skipped under --recheck: set VIALLO_TEST_INPUTS to reuse the input instead of downloading it"
        else:
            print("ok   lab ran in %.1fs on input %s%s" % (lab["seconds"], lab["input_sha256"][:12],
                  "" if lab["same_input"] else " (rebuilt, differs from the recorded input)"))
    except Exception as error:  # noqa: BLE001
        lab_error = str(error)
        print("FAIL lab:", lab_error)
        failed += 1
    finally:
        shutil.rmtree(tmp, ignore_errors=True)

    pages, verdict = {}, {}
    for row in data["results"]:
        kind, label = row["kind"], "%-38s %s / %s" % (row["id"], row["option"], row["property"])
        if kind == "page":
            if row["url"] not in pages:
                pages[row["url"]] = page_text(row["url"])
            code, text = pages[row["url"]]
            missing = [q for q in quotes(row) if norm(q) not in text]
            ok = code == "200" and not missing
            detail = "" if ok else "HTTP %s, missing: %r" % (code, missing)
        elif kind == "code":
            path = REPO / row["url"].split(":", 1)[1]
            text = norm(path.read_text()) if path.exists() else ""
            missing = [q for q in quotes(row) if norm(q) not in text]
            ok = path.exists() and not missing
            detail = "" if ok else ("no file %s" % path if not path.exists() else "missing: %r" % missing)
        elif kind == "lab":
            if lab is None:
                print("SKIP %s (%s)" % (label, lab_error))
                verdict[row["id"]] = lab_error is not None and "skipped" in lab_error
                continue
            ok, got = lab_holds(row, lab)
            detail = "" if ok else "measured %s, dataset %s" % (got, row["value"])
        elif kind == "derived":
            ok = all(verdict.get(i, False) for i in row["from"])
            detail = "" if ok else "a source row failed: %r" % [i for i in row["from"] if not verdict.get(i)]
        else:
            print("SKIP %s (%s)" % (label, row["value"]))
            continue
        verdict[row["id"]] = ok
        print("%s %s %s" % ("ok  " if ok else "FAIL", label, detail))
        failed += 0 if ok else 1

    print()
    print("%s, dataset checked %s" % ("all rows hold" if not failed else "%d problem(s)" % failed, data["checkedAt"]))
    return 0 if not failed else 1


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