#!/usr/bin/env python3
"""Re-runs the evidence check for the Viallo post full-resolution-photo-sharing.

Four kinds of row, each verified against the thing it was read from:

  docs     the cited page is fetched again and the recorded sentence must still be on it
  code     the Viallo source line a product claim rests on must still be in the repository
  lab      the two test files are rebuilt from a public-domain NASA photograph and the
           two conversions the post describes are re-measured
  derived  a figure computed from a documented rule and a measured pixel count, recomputed

Exit 0 when every row holds, 1 otherwise.

    python3 check.py             full run, rebuilds the test files and measures them
    python3 check.py --recheck   re-reads the pages, skips a lab step that takes over a minute

The lab needs macOS `sips` (HEIC decode and encode) and Pillow with numpy; `exiftool` is used
to write the synthetic camera tags when it is present, and the EXIF rows are skipped when it is
not. Nothing is uploaded anywhere: both conversions are run locally with the parameters the
Viallo code sets.
"""

import html
import io
import json
import math
import os
import re
import shutil
import subprocess
import sys
import tempfile
import time
import urllib.request

HERE = os.path.dirname(os.path.abspath(__file__))
DATASET = os.path.join(HERE, 'dataset.json')

# The script ships at apps/web/public/blog/<slug>/ inside the Viallo repository.
VIALLO_REPO = os.environ.get(
    'VIALLO_REPO', os.path.abspath(os.path.join(HERE, '..', '..', '..', '..', '..')))

BROWSER_UA = (
    'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) '
    'AppleWebKit/537.36 (KHTML, like Gecko) Chrome/127.0.0.0 Safari/537.36'
)
# help.instagram.com and faq.whatsapp.com answer a plain client with a shell that carries no
# article body; the crawler UA gets the same page with the text in it.
CRAWLER_UA = 'Googlebot/2.1 (+http://www.google.com/bot.html)'
CRAWLER_HOSTS = ('help.instagram.com', 'faq.whatsapp.com')

# Public-domain NASA photograph the two test files are cut from; a 12 MP JPEG and a 48.8 MP
# HEIC, so the conversions run on the sizes the post talks about.
NASA_SOURCE = (
    'https://images-assets.nasa.gov/image/KSC-20220119-PH-CSH01_0011/'
    'KSC-20220119-PH-CSH01_0011~orig.jpg'
)

# sips rebuilds agree to well under these; see the tolerance note in dataset.json.
BYTES_TOLERANCE = 0.01
PSNR_TOLERANCE = 0.5
RECHECK_LAB_BUDGET = 60


def user_agent_for(url):
    for host in CRAWLER_HOSTS:
        if host in url:
            return CRAWLER_UA
    return BROWSER_UA


def fetch(url, timeout=60):
    req = urllib.request.Request(url, headers={'User-Agent': user_agent_for(url)})
    with urllib.request.urlopen(req, timeout=timeout) as resp:
        return resp.read()


def normalise(text):
    text = text.replace('’', "'").replace('‘', "'")
    text = text.replace('“', '"').replace('”', '"')
    text = text.replace('–', '-').replace('—', '-')
    text = text.replace(' ', ' ')
    text = re.sub(r'\s+', ' ', text).strip()
    # Dropping a tag leaves a space where a link or a bold run ended, so a page that renders
    # "storage space." reads as "storage space ." here. Both are the same sentence.
    return re.sub(r'\s+([.,;:!?)\]])', r'\1', text)


def page_text(raw):
    """The visible words of a page, including text parked inside its JSON payload.

    Both Meta help centres, and Google's support pages, ship part of the article as escaped
    JSON inside a script tag, so the escapes are resolved before the tags are dropped.
    """
    text = raw.decode('utf-8', 'replace')
    text = html.unescape(text)
    for _ in range(3):
        text = re.sub(r'\\u([0-9a-fA-F]{4})', lambda m: chr(int(m.group(1), 16)), text)
        text = text.replace('\\"', '"').replace("\\'", "'").replace('\\/', '/')
        text = text.replace('\\n', ' ')
    text = re.sub(r'<[^>]+>', ' ', text)
    return normalise(text)


# ------------------------------------------------------------------- repository rows

def read_repo_file(path):
    if not os.path.isfile(os.path.join(VIALLO_REPO, 'packages', 'shared', 'package.json')):
        return None
    with open(os.path.join(VIALLO_REPO, path), encoding='utf-8') as fh:
        return fh.read()


# ------------------------------------------------------------------- lab: the test files

def run(cmd):
    return subprocess.run(cmd, capture_output=True, text=True)


def build_inputs(workdir):
    """Rebuild the two test files, the way the post's inputs were made.

    Both come from one public-domain NASA photograph cropped to 5120x3840, then resized: a
    4032x3024 JPEG at quality 90 and an 8064x6048 HEIC. A fixed set of camera tags is written
    into both so the EXIF columns have something to lose.
    """
    src = os.path.join(workdir, 'src.jpg')
    with open(src, 'wb') as fh:
        fh.write(fetch(NASA_SOURCE))
    crop = os.path.join(workdir, 'crop.jpg')
    jpeg = os.path.join(workdir, 'photo-12mp.jpg')
    heic = os.path.join(workdir, 'photo-48mp.heic')
    run(['sips', '-c', '5120', '3840', src, '--out', crop])
    run(['sips', '-z', '4032', '3024', '-s', 'formatOptions', '90', crop, '--out', jpeg])
    run(['sips', '-z', '8064', '6048', '-s', 'format', 'heic', crop, '--out', heic])
    os.unlink(crop)

    have_exiftool = shutil.which('exiftool') is not None
    if have_exiftool:
        tags = [
            '-Make=Apple', '-Model=iPhone 15 Pro', '-Software=18.0',
            '-LensModel=iPhone 15 Pro back triple camera 6.765mm f/1.78',
            '-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',
            '-ExposureTime=1/120', '-FNumber=1.78', '-ISO=80', '-Orientation#=1',
        ]
        for path in (jpeg, heic):
            run(['exiftool', '-q', '-overwrite_original', '-all=', path])
            run(['exiftool', '-q', '-overwrite_original'] + tags + [path])
    for path in (jpeg, heic):
        if not os.path.getsize(path):
            raise RuntimeError('sips wrote an empty file at %s' % path)
    return jpeg, heic, have_exiftool


def psnr(a, b):
    import numpy as np
    left = np.asarray(a, dtype='float64')
    right = np.asarray(b, dtype='float64')
    mse = ((left - right) ** 2).mean()
    return 10 * math.log10(255.0 * 255.0 / mse) if mse else 99.0


def lab_display_rendition(jpeg_path):
    """The rendition the lightbox opens: 2048 px long edge, JPEG quality 82."""
    from PIL import Image
    original = Image.open(jpeg_path)
    original.load()
    original_bytes = os.path.getsize(jpeg_path)
    rgb = original.convert('RGB')
    width, height = rgb.size
    scale = 2048 / max(width, height)
    target = (round(width * scale), round(height * scale)) if scale < 1 else (width, height)
    reference = rgb.resize(target, Image.LANCZOS)
    buffer = io.BytesIO()
    reference.save(buffer, 'JPEG', quality=82)
    data = buffer.getvalue()
    out = Image.open(io.BytesIO(data))
    out.load()
    return {
        'dimensions': list(out.size),
        'source_dimensions': [width, height],
        'bytes': len(data),
        'source_bytes': original_bytes,
        'share_of_source_pct': round(100.0 * len(data) / original_bytes, 2),
        'psnr_db': round(psnr(reference, out.convert('RGB')), 2),
        'exif_fields': len(dict(out.getexif())),
        'icc_profile': out.info.get('icc_profile') is not None,
    }


def lab_heic_conversion(heic_path):
    """HEIC decoded and re-encoded as JPEG at quality 92, full pixel dimensions."""
    from PIL import Image
    source_bytes = os.path.getsize(heic_path)
    handle, work = tempfile.mkstemp(suffix='.png')
    os.close(handle)
    try:
        run(['sips', '-s', 'format', 'png', heic_path, '--out', work])
        decoded = Image.open(work)
        decoded.load()
        rgb = decoded.convert('RGB')
        buffer = io.BytesIO()
        rgb.save(buffer, 'JPEG', quality=92)
        data = buffer.getvalue()
    finally:
        os.unlink(work)
    out = Image.open(io.BytesIO(data))
    out.load()
    return {
        'dimensions': list(out.size),
        'source_dimensions': list(rgb.size),
        'source_bytes': source_bytes,
        'bytes': len(data),
        'share_of_source_pct': round(100.0 * len(data) / source_bytes, 2),
        'psnr_db': round(psnr(rgb, out.convert('RGB')), 2),
        'exif_fields': len(dict(out.getexif())),
        'icc_profile': out.info.get('icc_profile') is not None,
    }


LABS = {
    'display-rendition-12mp': lab_display_rendition,
    'heic-to-jpeg-92': lab_heic_conversion,
}


# ------------------------------------------------------------------- derived rows

def derived_value(row, lab):
    """Recompute a figure from a documented rule applied to a measured file."""
    kind = row['derived']['kind']
    if kind == 'pixels-kept':
        width, height = lab[row['derived']['lab']]['source_dimensions']
        long_edge = row['derived']['long_edge']
        if width > long_edge:
            kept = long_edge, round(height * long_edge / width)
        else:
            kept = width, height
        target = row['derived'].get('dims')
        if target and list(target) != list(kept):
            return 'dims %dx%d, not the recorded %dx%d' % (kept[0], kept[1], target[0], target[1])
        return round(100.0 * kept[0] * kept[1] / (width * height), 2)
    if kind == 'megapixel-cap-pct':
        cap = row['derived']['megapixels'] * 1_000_000
        total = lab[row['derived']['lab']]['source_pixels']
        return round(100.0 * min(cap, total) / total, 2)
    if kind == 'sum-bytes':
        return sum(row['derived']['parts'])
    raise RuntimeError('unknown derived kind %r' % kind)


# ------------------------------------------------------------------- comparison

def close_enough(expected, measured, key):
    if isinstance(expected, list):
        return list(measured) == list(expected)
    if isinstance(expected, bool):
        return measured is expected
    if key == 'psnr_db':
        return abs(expected - measured) <= PSNR_TOLERANCE
    if isinstance(expected, int):
        if expected == measured:
            return True
        return abs(expected - measured) <= max(1, expected * BYTES_TOLERANCE)
    if isinstance(expected, float):
        return abs(expected - measured) <= max(0.05, abs(expected) * BYTES_TOLERANCE)
    return expected == measured


def measure_all(rows):
    """Rebuild the inputs and run every lab once, returning the measurements."""
    workdir = tempfile.mkdtemp(prefix='viallo-fr-check-')
    labs = {}
    try:
        jpeg_path, heic_path, have_exiftool = build_inputs(workdir)
        paths = {'display-rendition-12mp': jpeg_path, 'heic-to-jpeg-92': heic_path}
        for name in sorted({r['lab'] for r in rows if r.get('type') == 'lab'}):
            started = time.time()
            labs[name] = LABS[name](paths[name])
            labs[name]['seconds'] = round(time.time() - started, 1)
            width, height = labs[name]['source_dimensions']
            labs[name]['source_pixels'] = width * height
        return labs, have_exiftool
    finally:
        shutil.rmtree(workdir, ignore_errors=True)


def main(argv):
    recheck = '--recheck' in argv
    with open(DATASET, encoding='utf-8') as fh:
        dataset = json.load(fh)
    rows = dataset['results']

    failures = []
    skipped = []
    pages = {}

    for row in rows:
        label = '%s | %s' % (row['option'], row['property'])
        kind = row.get('type')

        if kind == 'docs':
            url = row['url']
            if url not in pages:
                try:
                    pages[url] = page_text(fetch(url))
                except Exception as error:                     # noqa: BLE001
                    pages[url] = None
                    failures.append('%s: page not reachable (%s) %s' % (label, error, url))
            text = pages[url]
            if text is None:
                continue
            if normalise(row['quote']) not in text:
                failures.append('%s: quote gone from %s' % (label, url))
            continue

        if kind == 'code':
            try:
                content = read_repo_file(row['path'])
            except Exception as error:                         # noqa: BLE001
                failures.append('%s: %s' % (label, error))
                continue
            if content is None:
                skipped.append('%s: no Viallo checkout at %s' % (label, VIALLO_REPO))
                continue
            if row['quote'] not in content:
                failures.append('%s: line gone from %s' % (label, row['path']))
            continue

        if kind in ('lab', 'derived'):
            continue                                             # measured once, below

        failures.append('%s: unknown row type %r' % (label, kind))

    lab_seconds = dataset.get('labSeconds') or 0
    if recheck and lab_seconds > RECHECK_LAB_BUDGET:
        skipped.append('lab: the full run took %ss, over the %ss recheck budget'
                       % (lab_seconds, RECHECK_LAB_BUDGET))
        labs = None
    else:
        labs, have_exiftool = measure_all(rows)
        if not have_exiftool:
            print('note: exiftool is missing, the test files carry no camera tags')

    if labs is not None:
        for row in rows:
            label = '%s | %s' % (row['option'], row['property'])
            if row.get('type') == 'lab':
                measured = labs[row['lab']]
                bad = [key for key, expected in row['expect'].items()
                       if not close_enough(expected, measured.get(key), key)]
                if bad:
                    failures.append('%s: %s recorded %s, measured %s' % (
                        label, bad, {k: row['expect'][k] for k in bad},
                        {k: measured.get(k) for k in bad}))
            elif row.get('type') == 'derived':
                got = derived_value(row, labs)
                if row.get('expect_number') is not None:
                    if not close_enough(float(row['expect_number']), float(got), 'pct'):
                        failures.append('%s: recorded %s, the rule gives %s'
                                        % (label, row['expect_number'], got))
                elif got != row['value']:
                    failures.append('%s: recorded %r, the rule gives %r'
                                    % (label, row['value'], got))
        slowest = max((labs[name]['seconds'] for name in labs), default=0)
        print('lab: %d file(s) rebuilt and measured, slowest step %.0fs' % (len(labs), slowest))
    else:
        for row in rows:
            if row.get('type') == 'derived':
                skipped.append('%s | %s: computed from the lab files, which this run skipped'
                               % (row['option'], row['property']))

    for line in skipped:
        print('SKIP  ' + line)
    for line in failures:
        print('FAIL  ' + line)
    print('%d rows checked, %d skipped, %d failed'
          % (len(rows) - len(skipped), len(skipped), len(failures)))
    return 0 if not failures else 1


if __name__ == '__main__':
    sys.exit(main(sys.argv[1:]))