#!/usr/bin/env python3
"""Self-test corrected XGD1 Redump-style media-profile routing.

v227 adds an offline, fast media-profile conformance test.  It verifies the
backing-store coordinate contract that the live bridge uses when an adjacent
.media.json is present, including locked video L0/L1 split routing and unlocked
GDR8050L host-LBA to reconstructed-ISO mapping.
"""
from __future__ import annotations

import argparse
import json
from pathlib import Path
from typing import Any

from gdr_media_model import GDRMediaModel

EXPECTED = {
    'sector_size': 2048,
    'reconstructed_output_sector_count': 3_820_880,
    'reconstructed_output_byte_count': 7_825_162_240,
    'layer_break_lba': 1_913_776,
    'game_output_start_lba': 198_144,
    'game_xiso_leadin_sector_count': 32,
    'xdfs_volume_lba_within_game_region': 32,
    'game_unlocked_source_start_lba': 32,
    'game_unlocked_source_sector_count': 3_431_232,
    'video_l0_sector_count': 6_832,
    'video_l1_start_lba': 3_820_720,
    'video_l1_sector_count': 160,
    'locked_visible_sector_count': 6_992,
    'unlocked_visible_sector_count': 3_431_264,
}


def load_jsonl(path: Path | None) -> list[dict[str, Any]]:
    if not path or not path.exists():
        return []
    rows: list[dict[str, Any]] = []
    with path.open('r', encoding='utf-8', errors='replace') as f:
        for line in f:
            line = line.strip()
            if not line:
                continue
            try:
                obj = json.loads(line)
            except Exception:
                continue
            if isinstance(obj, dict):
                rows.append(obj)
    return rows


def intish(v: Any, default: int | None = None) -> int | None:
    try:
        if v is None or v == '':
            return default
        return int(v)
    except Exception:
        return default


def truthy(v: Any) -> bool:
    if isinstance(v, bool):
        return v
    return str(v).lower() in ('1', 'true', 'yes', 'y')


def cdb_op(row: dict[str, Any]) -> str:
    h = str(row.get('cdb_hex') or '').replace(' ', '')
    return h[:2].upper() if len(h) >= 2 else ''


def resolve_iso(args: argparse.Namespace, rows: list[dict[str, Any]]) -> Path | None:
    if args.iso:
        return Path(args.iso)
    for row in rows:
        p = row.get('dynamic_media_path') or row.get('media_path')
        if p:
            return Path(str(p))
    if args.config and Path(args.config).exists():
        try:
            cfg = json.loads(Path(args.config).read_text(encoding='utf-8'))
            media = cfg.get('media') or {}
            p = media.get('iso_path') or media.get('path') or media.get('file')
            if p:
                return Path(str(p))
        except Exception:
            pass
    return None


def adjacent_sidecar(iso: Path | None, explicit: str | None) -> Path | None:
    if explicit:
        return Path(explicit)
    if not iso:
        return None
    for suffix in ('.xdvd.json', '.media.json'):
        p = iso.with_suffix(suffix)
        if p.exists():
            return p
    return iso.with_suffix('.xdvd.json')


def norm_segments(segs: list[dict[str, Any]]) -> list[dict[str, int | str]]:
    out = []
    for s in segs:
        out.append({
            'host_lba': int(s.get('host_lba', 0)),
            'iso_lba': int(s.get('iso_lba', 0)),
            'blocks': int(s.get('blocks', 0)),
            'segment_note': str(s.get('segment_note') or ''),
        })
    return out


def check_value(checks: list[dict[str, Any]], name: str, actual: Any, expected: Any) -> None:
    checks.append({
        'name': name,
        'actual': actual,
        'expected': expected,
        'ok': actual == expected,
    })


def check_route(checks: list[dict[str, Any]], model: GDRMediaModel, name: str, state: str, host_lba: int, blocks: int, expected: list[tuple[int, int, int, str]], *, read_map: str = 'firmware_window') -> None:
    segs, note = model.read_segments_for_host(state, host_lba, blocks, read_map=read_map)
    got = norm_segments(segs)
    exp = [{'host_lba': h, 'iso_lba': i, 'blocks': b, 'segment_note': n} for h, i, b, n in expected]
    checks.append({
        'name': name,
        'state': state,
        'host_lba': host_lba,
        'blocks': blocks,
        'note': note,
        'actual_segments': got,
        'expected_segments': exp,
        'ok': got == exp,
    })


def trace_observations(rows: list[dict[str, Any]]) -> dict[str, Any]:
    reads = [r for r in rows if cdb_op(r) in ('28', 'A8')]
    redump = [r for r in reads if truthy(r.get('media_is_redump_style'))]
    unlocked = [r for r in redump if str(r.get('dynamic_media_auth_stage') or '').lower() == 'unlocked']
    old_306080 = [r for r in unlocked if intish(r.get('effective_media_lba_delta'), None) == 306080]
    outside = [r for r in redump if 'outside' in str(r.get('read_map_note') or '').lower()]
    deltas: dict[int, int] = {}
    notes: dict[str, int] = {}
    for r in redump:
        d = intish(r.get('effective_media_lba_delta'), None)
        if d is not None:
            deltas[d] = deltas.get(d, 0) + 1
        note = str(r.get('read_map_note') or '')
        notes[note] = notes.get(note, 0) + 1
    return {
        'read_rows': len(reads),
        'redump_read_rows': len(redump),
        'redump_unlocked_read_rows': len(unlocked),
        'redump_unlocked_old_306080_rows': len(old_306080),
        'redump_outside_mapping_rows': len(outside),
        'redump_effective_delta_counts': dict(sorted(deltas.items())),
        'redump_read_map_note_counts': dict(sorted(notes.items())),
        'ok': (not redump) or (len(old_306080) == 0 and len(outside) == 0 and len(unlocked) > 0),
    }


def main() -> int:
    ap = argparse.ArgumentParser(description='v227 GDR/XGD1 Redump media-profile self-test')
    ap.add_argument('--config', type=Path)
    ap.add_argument('--iso', type=Path)
    ap.add_argument('--xdvd', type=Path, help='explicit .xdvd.json or .media.json')
    ap.add_argument('--jsonl', type=Path, help='optional live trace for observed mapping checks')
    ap.add_argument('--json', type=Path)
    ap.add_argument('--md', type=Path)
    args = ap.parse_args()

    rows = load_jsonl(args.jsonl)
    iso = resolve_iso(args, rows)
    sidecar = adjacent_sidecar(iso, str(args.xdvd) if args.xdvd else None)
    checks: list[dict[str, Any]] = []
    warnings: list[str] = []
    model_summary: dict[str, Any] = {}
    model_loaded = False
    route_probe_complete = False

    if not sidecar or not sidecar.exists():
        warnings.append(f'sidecar_missing:{sidecar}')
    else:
        try:
            model = GDRMediaModel.from_paths(sidecar, iso)
            model_loaded = True
            model_summary = model.summary()
            is_redump = bool(model.is_redump_style)
            check_value(checks, 'media_profile_loaded', bool(model.media_profile_loaded), True)
            check_value(checks, 'layout_type', model.layout_type, 'original_xbox_xgd1_redump_style_2048')
            check_value(checks, 'is_redump_style', is_redump, True)
            check_value(checks, 'sector_size', model.sector_size, EXPECTED['sector_size'])
            for key in (
                'reconstructed_output_sector_count', 'layer_break_lba', 'game_output_start_lba',
                'game_xiso_leadin_sector_count', 'xdfs_volume_lba_within_game_region',
                'game_unlocked_source_start_lba', 'game_unlocked_source_sector_count',
                'video_l0_sector_count', 'video_l1_start_lba', 'video_l1_sector_count',
                'locked_visible_sector_count', 'unlocked_visible_sector_count',
            ):
                check_value(checks, key, getattr(model, key), EXPECTED[key])
            if model.iso_byte_count:
                check_value(checks, 'iso_byte_count', model.iso_byte_count, EXPECTED['reconstructed_output_byte_count'])
                check_value(checks, 'iso_sector_count', model.iso_sector_count, EXPECTED['reconstructed_output_sector_count'])
            locked_cap = model.read_capacity_10_response('locked').hex().upper()
            unlocked_cap = model.read_capacity_10_response('unlocked').hex().upper()
            check_value(checks, 'locked_read_capacity_10_hex', locked_cap, '00001B4F00000800')
            check_value(checks, 'unlocked_read_capacity_10_hex', unlocked_cap, '00345B5F00000800')
            payloads = model.summary().get('payloads') or {}
            for key in ('ad_c0', 'pfi', 'dmi'):
                checks.append({'name': f'payload_{key}_present', 'actual': key in payloads, 'expected': True, 'ok': key in payloads})
            check_route(checks, model, 'locked_l0_start', 'locked', 0, 1, [(0, 0, 1, 'redump_locked_video_l0')])
            check_route(checks, model, 'locked_l0_end', 'locked', 6831, 1, [(6831, 6831, 1, 'redump_locked_video_l0')])
            check_route(checks, model, 'locked_l0_l1_cross_split', 'locked', 6831, 2, [
                (6831, 6831, 1, 'redump_locked_video_l0'),
                (6832, 3820720, 1, 'redump_locked_video_l1_tail'),
            ])
            check_route(checks, model, 'locked_l1_start', 'locked', 6832, 1, [(6832, 3820720, 1, 'redump_locked_video_l1_tail')])
            check_route(checks, model, 'locked_l1_end', 'locked', 6991, 1, [(6991, 3820879, 1, 'redump_locked_video_l1_tail')])
            check_route(checks, model, 'unlocked_start', 'unlocked', 0, 1, [(0, 198144, 1, 'redump_unlocked_game_xiso_region')])
            check_route(checks, model, 'unlocked_xiso_leadin_end', 'unlocked', 31, 1, [(31, 198175, 1, 'redump_unlocked_game_xiso_region')])
            check_route(checks, model, 'unlocked_xdfs_start', 'unlocked', 32, 1, [(32, 198176, 1, 'redump_unlocked_game_xiso_region')])
            check_route(checks, model, 'unlocked_last_visible_lba', 'unlocked', 3431263, 1, [(3431263, 3629407, 1, 'redump_unlocked_game_xiso_region')])
            check_route(checks, model, 'raw_redump_direct_last_lba', 'locked', 3820879, 1, [(3820879, 3820879, 1, 'direct')], read_map='raw_iso')
            route_probe_complete = True
            for w in model.summary().get('warnings') or []:
                warnings.append(str(w))
        except Exception as exc:
            warnings.append(f'model_load_failed:{exc}')

    trace = trace_observations(rows)
    failed = [c for c in checks if not c.get('ok')]
    # If no sidecar/model is available, keep the script non-fatal for legacy
    # traces.  Redump-profile live runs should have both model checks and trace
    # observations clean.
    if model_loaded:
        ok = len(failed) == 0 and (not rows or trace.get('ok', False))
        mode = 'redump_profile' if model_summary.get('is_redump_style') else 'legacy_or_no_media_profile'
    else:
        ok = bool(trace.get('ok', True))
        mode = 'trace_only_no_model'

    report = {
        'version': 'v227-media-profile-selftest',
        'config': str(args.config) if args.config else '',
        'jsonl': str(args.jsonl) if args.jsonl else '',
        'iso': str(iso) if iso else '',
        'sidecar': str(sidecar) if sidecar else '',
        'mode': mode,
        'model_loaded': model_loaded,
        'route_probe_complete': route_probe_complete,
        'check_count': len(checks),
        'failed_count': len(failed),
        'ok': ok,
        'checks': checks,
        'failures': failed,
        'warnings': list(dict.fromkeys(warnings)),
        'model_summary': model_summary,
        'trace_observations': trace,
    }

    if args.json:
        args.json.write_text(json.dumps(report, indent=2, sort_keys=True), encoding='utf-8')

    lines = [
        '# GDR/XGD1 v227 media-profile self-test', '',
        f'- mode: `{mode}`',
        f'- model_loaded: `{str(model_loaded).lower()}`',
        f'- route_probe_complete: `{str(route_probe_complete).lower()}`',
        f'- check_count: `{len(checks)}`',
        f'- failed_count: `{len(failed)}`',
        f'- ok: `{str(ok).lower()}`', '',
        '## Key paths', '',
        f'- iso: `{iso or ""}`',
        f'- sidecar: `{sidecar or ""}`', '',
        '## Trace observations', '',
        f'- read_rows: `{trace.get("read_rows")}`',
        f'- redump_read_rows: `{trace.get("redump_read_rows")}`',
        f'- redump_unlocked_read_rows: `{trace.get("redump_unlocked_read_rows")}`',
        f'- redump_unlocked_old_306080_rows: `{trace.get("redump_unlocked_old_306080_rows")}`',
        f'- redump_outside_mapping_rows: `{trace.get("redump_outside_mapping_rows")}`',
        f'- redump_effective_delta_counts: `{trace.get("redump_effective_delta_counts")}`', '',
        '## Failed checks', '',
    ]
    if failed:
        lines += ['| check | actual | expected |', '|---|---|---|']
        for c in failed:
            lines.append(f"| `{c.get('name')}` | `{c.get('actual', c.get('actual_segments'))}` | `{c.get('expected', c.get('expected_segments'))}` |")
    else:
        lines.append('None.')
    lines += ['', '## Route/capacity checks', '', '| check | ok |', '|---|---:|']
    for c in checks:
        lines.append(f"| `{c.get('name')}` | `{str(c.get('ok')).lower()}` |")
    if warnings:
        lines += ['', '## Warnings', '']
        for w in dict.fromkeys(warnings):
            lines.append(f'- `{w}`')
    if args.md:
        args.md.write_text('\n'.join(lines) + '\n', encoding='utf-8')

    print(f"[MEDIA-PROFILE-SELFTEST] mode={mode} model_loaded={str(model_loaded).lower()} route_probe_complete={str(route_probe_complete).lower()} checks={len(checks)} failed={len(failed)} ok={str(ok).lower()}")
    return 0 if ok else 2


if __name__ == '__main__':
    raise SystemExit(main())
