"""Publish a new evidence-qualified revision; never repair parent geometry.

Only frozen source-derived ``faces`` constraints lacking semantic-yaw evidence
are omitted. Explicit user requirements survive. Below-floor target positions
are diagnostics, not clamped labels. This module intentionally uses stdlib only.
"""
from __future__ import annotations

import argparse
from collections import Counter
from copy import deepcopy
import hashlib
from itertools import zip_longest
import json
import math
import os
from pathlib import Path
import shutil
import tempfile

SPLITS = ("train", "validation", "test")
INPUT_FILES = tuple(f"{split}.jsonl" for split in SPLITS) + ("manifest.json", "rejections.jsonl")
OUTPUT_FILES = tuple(name for name in INPUT_FILES if name != "manifest.json")
POLICY = "source-derived-facing-evidence-v1"
BUILDER = "immutable-parent-review-v1"
FLOOR_TOLERANCE_M = 1e-4
LEGACY_CONSTRAINT_SOURCE = "frozen_sparse_legacy_request"
RAW_SOURCE_ROOT = Path("/Volumes/harddisk/3D_Room_Collections")
REVIEW_FIELDS = ("review_parent_scene_id", "review_policy", "review_source_qualification",
                "review_omitted_constraints", "review_geometry_diagnostics")


def _digest(path):
    value = hashlib.sha256()
    with Path(path).open("rb") as stream:
        for block in iter(lambda: stream.read(1024 * 1024), b""):
            value.update(block)
    return value.hexdigest()


def _reject_constant(value):
    raise ValueError(f"nonfinite JSON number: {value}")


def _read_json(value):
    return json.loads(value, parse_constant=_reject_constant)


def _parent_inputs(root):
    for name in INPUT_FILES:
        if not (root / name).is_file():
            raise FileNotFoundError(f"required parent input missing: {root / name}")
    raw = (root / "manifest.json").read_bytes()
    manifest = _read_json(raw)
    if not isinstance(manifest, dict) or manifest.get("schema_version") != "fastfill.v2":
        raise ValueError("parent manifest must identify fastfill.v2 data")
    hashes = {name: _digest(root / name) for name in INPUT_FILES if name != "manifest.json"}
    return manifest, {**hashes, "manifest.json": hashlib.sha256(raw).hexdigest()}


def _recheck(root, hashes):
    if any(_digest(root / name) != expected for name, expected in hashes.items()):
        raise ValueError("parent input or manifest changed during review")


def _new_output(parent_root, output):
    if Path(output).is_symlink():
        raise FileExistsError(f"immutable review output already exists as a symlink: {output}")
    parent, target = Path(parent_root).resolve(), Path(output).resolve()
    for protected in (parent, RAW_SOURCE_ROOT.resolve()):
        if target == protected or protected in target.parents or target in protected.parents:
            raise ValueError("output must be outside source data and its ancestors")
    if target.exists():
        raise FileExistsError(f"immutable review output already exists: {target}")
    return parent, target


def _complete(mask, dimensions):
    return mask is True or (isinstance(mask, list) and len(mask) == dimensions
                            and all(value is True for value in mask))


def _mask(sample, field, index, dimensions):
    values = sample["validity"].get(field, [])
    return index < len(values) and _complete(values[index], dimensions)


def _finite(value):
    return isinstance(value, (int, float)) and not isinstance(value, bool) and math.isfinite(value)


def _indices(sample):
    if not isinstance(sample, dict) or sample.get("schema_version") != "fastfill.v2":
        raise ValueError("parent rows must use fastfill.v2 schema")
    condition, target = sample.get("condition", {}), sample.get("target", {})
    if not isinstance(condition, dict) or not isinstance(target, dict):
        raise ValueError("condition and target must be objects")
    collections = (condition.get("objects"), target.get("objects"))
    ids = []
    for objects in collections:
        if not isinstance(objects, list) or any(not isinstance(obj, dict)
                or not isinstance(obj.get("id"), str) or not obj["id"] for obj in objects):
            raise ValueError("request and target objects require nonempty string IDs")
        names = [obj["id"] for obj in objects]
        if len(names) != len(set(names)):
            raise ValueError("duplicate request or target ID")
        ids.append(names)
    if set(ids[0]) != set(ids[1]):
        raise ValueError("target IDs must equal requested IDs exactly once")
    validity = sample.get("validity")
    if not isinstance(validity, dict) or any(not isinstance(validity.get(field, []), list)
            or len(validity.get(field, [])) > len(ids[1]) for field in ("position", "size", "yaw")):
        raise ValueError("invalid geometry validity rows")
    if not isinstance(condition.get("room"), dict) or not isinstance(condition.get("constraints", []), list):
        raise ValueError("room and constraints must be structured fields")
    provenance = sample.get("provenance")
    if not isinstance(provenance, dict) or any(not isinstance(provenance.get(key), str)
            or not provenance[key] for key in ("scene_id", "legacy_uid")):
        raise ValueError("parent provenance requires stable scene_id and legacy_uid")
    evidence = provenance.get("field_evidence", [])
    if (not isinstance(evidence, list) or len(evidence) > len(ids[1])
            or any(not isinstance(row, dict) for row in evidence)):
        raise ValueError("field evidence must follow target object indices")
    if any(key in provenance for key in REVIEW_FIELDS):
        raise ValueError("parent already contains review provenance")
    return {name: index for index, name in enumerate(ids[1])}


def _trusted_yaw(sample, index):
    """Facing needs a semantic front: a valid yaw whose symmetry order is 1 (axis-only mod-pi yaw is not evidence)."""
    orders = sample["validity"].get("yaw_symmetry_order", [])
    if not _mask(sample, "yaw", index, 1) or index < len(orders) and orders[index] != 1:
        return False
    angle = sample["target"]["objects"][index].get("yaw_rad")
    if not _finite(angle) or not -math.pi <= angle < math.pi:
        raise ValueError("valid yaw label must be finite and wrapped to [-pi, pi)")
    evidence = sample["provenance"].get("field_evidence", [])
    field = evidence[index] if index < len(evidence) else {}
    recorded = field.get("recorded_source_evidence", {})
    if not isinstance(recorded, dict):
        raise ValueError("recorded source evidence must be an object")
    return "front_unknown_reason" not in recorded


def _floor_diagnostics(sample):
    room = sample["condition"]["room"]
    known = room.get("floor_known") is True
    floor = room.get("floor_z_m") if known else None
    if known and not _finite(floor):
        raise ValueError("known floor requires a finite floor_z_m")
    conflicts = []
    for index, target in enumerate(sample["target"]["objects"]):
        if not known or not _mask(sample, "position", index, 3):
            continue
        position = target.get("bottom_center_m")
        if not isinstance(position, list) or len(position) != 3 or not all(map(_finite, position)):
            raise ValueError("valid position label must be finite XYZ")
        if position[2] < floor - FLOOR_TOLERANCE_M:
            conflicts.append({"object_id": target["id"], "bottom_minus_floor_m": position[2] - floor})
    return {"known_floor": known, "floor_z_m": floor, "tolerance_m": FLOOR_TOLERANCE_M,
            "below_known_floor": conflicts,
            "minimum_bottom_minus_floor_m": min((c["bottom_minus_floor_m"] for c in conflicts), default=None),
            "labels_modified": False}


def review_sample(parent):
    """Pure derivation: only omit unverified frozen legacy facing constraints."""
    indices = _indices(parent)
    legacy = parent["provenance"].get("constraint_source") == LEGACY_CONSTRAINT_SOURCE
    kept, omitted = [], []
    for constraint in parent["condition"].get("constraints", []):
        if not isinstance(constraint, dict):
            raise ValueError("constraint must be an object")
        if legacy and constraint.get("type") == "faces":
            object_id = constraint.get("object_id")
            if object_id not in indices:
                raise ValueError("source-derived facing refers to an unknown request ID")
            if not _trusted_yaw(parent, indices[object_id]):
                omitted.append({"constraint": deepcopy(constraint),
                                "reason": "unverified_source_derived_facing"})
                continue
        kept.append(deepcopy(constraint))
    result = deepcopy(parent)
    if "constraints" in parent["condition"]:
        result["condition"] = {**result["condition"], "constraints": kept}
    result["provenance"] = {**result["provenance"],
        "review_parent_scene_id": parent["provenance"]["scene_id"], "review_policy": POLICY,
        "review_source_qualification": "Frozen legacy constraints are source-derived annotations, not newly "
            "synthesized user requirements; unverified source-derived facing is omitted. "
            "Explicit/nonlegacy requirements and existing exchangeability groups are preserved.",
        "review_omitted_constraints": omitted, "review_geometry_diagnostics": _floor_diagnostics(parent)}
    return result


def _rows(root, split, seen):
    with (root / f"{split}.jsonl").open("rb") as stream:
        for number, raw in enumerate(stream, 1):
            if not raw.strip():
                continue
            try:
                sample = _read_json(raw)
                provenance = sample["provenance"]
                uid = provenance["legacy_uid"]
                if not isinstance(uid, str) or not uid or uid in seen or provenance.get("split") != split:
                    raise ValueError("duplicate/missing UID or provenance split disagrees with its file")
                seen.add(uid)
            except (ValueError, KeyError, TypeError) as exc:
                raise ValueError(f"invalid row {split}.jsonl:{number}: {exc}") from exc
            yield sample


def _empty_counts():
    return {key: Counter() for key in ("split_samples", "split_objects", "source_split_samples",
        "source_omitted_faces", "source_floor_conflict_objects", "totals")}


def _count(counts, sample, split):
    source = sample["provenance"].get("source", "unknown")
    omitted = len(sample["provenance"]["review_omitted_constraints"])
    conflicts = len(sample["provenance"]["review_geometry_diagnostics"]["below_known_floor"])
    counts["split_samples"][split] += 1
    counts["split_objects"][split] += len(sample["target"]["objects"])
    counts["source_split_samples"][f"{source}:{split}"] += 1
    for key, value in (("omitted_faces", omitted), ("rows_with_omitted_faces", bool(omitted)),
                       ("floor_conflict_objects", conflicts), ("rows_with_floor_conflicts", bool(conflicts))):
        counts["totals"][key] += value
    if omitted:
        counts["source_omitted_faces"][source] += omitted
    if conflicts:
        counts["source_floor_conflict_objects"][source] += conflicts


def _summary(counts):
    result = {key: dict(value) for key, value in counts.items() if key != "totals"}
    for key in ("split_samples", "split_objects"):
        result[key] = {split: counts[key][split] for split in SPLITS}
    return {**result, **counts["totals"], "samples_written": sum(counts["split_samples"].values()),
            "objects_written": sum(counts["split_objects"].values())}


def _write_records(parent, staging):
    seen, counts = set(), _empty_counts()
    for split in SPLITS:
        with (staging / f"{split}.jsonl").open("xb") as output:
            for record in _rows(parent, split, seen):
                child = review_sample(record)
                output.write((json.dumps(child, ensure_ascii=False, allow_nan=False,
                                         separators=(",", ":")) + "\n").encode())
                _count(counts, child, split)
    shutil.copyfile(parent / "rejections.jsonl", staging / "rejections.jsonl")
    return _summary(counts)


def build_reviewed(parent_root, output):
    """Stream all rows and atomically publish to a new immutable directory."""
    parent, target = _new_output(parent_root, output)
    manifest, hashes = _parent_inputs(parent)
    target.parent.mkdir(parents=True, exist_ok=True)
    with tempfile.TemporaryDirectory(prefix=f".{target.name}-", dir=target.parent) as temporary:
        staging = Path(temporary)
        result = {"schema_version": "fastfill.v2", "builder": BUILDER, "review_policy": POLICY,
            "parent_root": str(parent), "parent_builder": manifest.get("builder"),
            "parent_front_policy": manifest.get("front_policy"),
            "front_policy": manifest.get("front_policy"), "yaw_policy": manifest.get("yaw_policy"), "parent_sha256": hashes,
            "split_rule": "inherit every parent UID and split without re-splitting or filtering",
            "labels_modified": False, "source_data_modified": False,
            "implementation_sha256": _digest(__file__), **_write_records(parent, staging),
            "output_sha256": {name: _digest(staging / name) for name in OUTPUT_FILES}}
        (staging / "manifest.json").write_text(json.dumps(result, indent=2, allow_nan=False) + "\n")
        _recheck(parent, hashes)
        if target.exists() or target.is_symlink():
            raise FileExistsError("immutable review output appeared during derivation")
        os.rename(staging, target)
    return result


def verify_reviewed(data_root, parent_root):
    """Check fingerprints, every deterministic row, preserved identities and totals."""
    child, parent = Path(data_root).resolve(), Path(parent_root).resolve()
    _, parent_hashes = _parent_inputs(parent)
    manifest, child_hashes = _parent_inputs(child)
    if (manifest.get("builder") != BUILDER or manifest.get("review_policy") != POLICY
            or manifest.get("parent_sha256") != parent_hashes):
        raise ValueError("review manifest policy or parent fingerprints do not match")
    expected_hashes = {name: child_hashes[name] for name in OUTPUT_FILES}
    if manifest.get("output_sha256") != expected_hashes:
        raise ValueError("review output hashes do not match manifest")
    if child_hashes["rejections.jsonl"] != parent_hashes["rejections.jsonl"]:
        raise ValueError("parent rejections were changed")
    parent_seen, child_seen, counts = set(), set(), _empty_counts()
    for split in SPLITS:
        pairs = zip_longest(_rows(parent, split, parent_seen), _rows(child, split, child_seen))
        for number, (original, observed) in enumerate(pairs, 1):
            if original is None or observed != review_sample(original):
                raise ValueError(f"reviewed row differs from deterministic parent revision: {split}:{number}")
            _count(counts, observed, split)
    summary = _summary(counts)
    if any(manifest.get(key) != value for key, value in summary.items()):
        raise ValueError("review manifest counts do not match verified rows")
    if manifest.get("labels_modified") is not False or manifest.get("source_data_modified") is not False:
        raise ValueError("review manifest must declare unchanged labels and parent data")
    _recheck(parent, parent_hashes)
    _recheck(child, child_hashes)
    return {"verified": True, **summary, "manifest_sha256": child_hashes["manifest.json"]}


def main(argv=None):
    parser = argparse.ArgumentParser(description=__doc__)
    commands = parser.add_subparsers(dest="command", required=True)
    for command in ("build", "verify"):
        sub = commands.add_parser(command)
        sub.add_argument("--parent-root", type=Path, required=True)
        sub.add_argument("--output" if command == "build" else "--data-root", type=Path, required=True)
    args = vars(parser.parse_args(argv))
    command = args.pop("command")
    result = build_reviewed(**args) if command == "build" else verify_reviewed(**args)
    print(json.dumps(result, indent=2, allow_nan=False))


if __name__ == "__main__":
    main()