from __future__ import annotations

import copy
from functools import lru_cache
import json
import math
import re
import sys
from pathlib import Path
from typing import Any

from jsonschema import Draft202012Validator
from jsonschema.exceptions import SchemaError

from vendor.cdsl_preview_runtime import step_to_glb
from app.settings import Settings


def load_engine(settings: Settings) -> Any:
    parent = str(settings.engine_root.parent)
    if parent not in sys.path:
        sys.path.insert(0, parent)
    import cdsl_engine

    return cdsl_engine


def _walk(value: Any) -> list[tuple[str, Any]]:
    result: list[tuple[str, Any]] = []
    if isinstance(value, dict):
        for key, child in value.items():
            result.append((str(key), child))
            result.extend(_walk(child))
    elif isinstance(value, list):
        for child in value:
            result.extend(_walk(child))
    return result


@lru_cache(maxsize=16)
def _read_schema_document(path_value: str, modified_ns: int) -> dict[str, Any]:
    """Load an immutable runtime schema once per on-disk version."""
    del modified_ns  # The mtime is intentionally part of the cache key.
    schema_path = Path(path_value)
    try:
        schema = json.loads(schema_path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as error:
        raise RuntimeError("The local engine schema document is unavailable or invalid") from error
    if not isinstance(schema, dict) or not isinstance(schema.get("operation_contracts"), dict):
        raise RuntimeError("The local engine schema has no operation contract registry")
    return schema


def _engine_schema(engine: Any) -> dict[str, Any]:
    schema_path = Path(str(engine.__file__)).with_name("profile_schema.json")
    try:
        modified_ns = schema_path.stat().st_mtime_ns
    except OSError as error:
        raise RuntimeError("The local engine schema document is unavailable or invalid") from error
    return _read_schema_document(str(schema_path), modified_ns)


@lru_cache(maxsize=16)
def _cdsl_validator(path_value: str, modified_ns: int) -> Draft202012Validator:
    """Compile the JSON Schema once per runtime schema revision."""
    del modified_ns
    schema_path = Path(path_value)
    try:
        schema = json.loads(schema_path.read_text(encoding="utf-8"))
        Draft202012Validator.check_schema(schema)
    except (OSError, json.JSONDecodeError, SchemaError) as error:
        raise RuntimeError("The local CDSL JSON Schema is unavailable or invalid") from error
    return Draft202012Validator(schema)


def _cdsl_schema_path(engine: Any) -> Path:
    document = _engine_schema(engine)
    schema_name = str(document.get("cdsl_json_schema_file") or "")
    if not schema_name or Path(schema_name).name != schema_name:
        raise RuntimeError("The local engine schema has an invalid CDSL JSON Schema path")
    return Path(str(engine.__file__)).with_name(schema_name)


def load_cdsl_json_schema(engine: Any) -> dict[str, Any]:
    schema_path = _cdsl_schema_path(engine)
    try:
        schema = json.loads(schema_path.read_text(encoding="utf-8"))
        Draft202012Validator.check_schema(schema)
    except (OSError, json.JSONDecodeError, SchemaError) as error:
        raise RuntimeError("The local CDSL JSON Schema is unavailable or invalid") from error
    return schema


def _validate_cdsl_json_schema(cdsl: dict[str, Any], engine: Any) -> None:
    schema_path = _cdsl_schema_path(engine)
    try:
        validator = _cdsl_validator(str(schema_path), schema_path.stat().st_mtime_ns)
    except OSError as error:
        raise RuntimeError("The local CDSL JSON Schema is unavailable or invalid") from error
    errors = sorted(validator.iter_errors(cdsl), key=lambda error: (list(error.absolute_path), error.message))
    if not errors:
        return
    error = errors[0]
    location = "$" + "".join(
        f"[{item}]" if isinstance(item, int) else f".{item}"
        for item in error.absolute_path
    )
    raise ValueError(f"CDSL schema violation at {location}: {error.message}")


def validate_cdsl_shape(cdsl: dict[str, Any], engine: Any) -> None:
    """Validate the static Runtime CDSL contract without resolving topology."""
    if not isinstance(cdsl, dict):
        raise ValueError("CDSL must be a JSON object")
    if cdsl.get("schema") not in {"cad.cdsl.llm.v1", "cad.runtime.v1"}:
        raise ValueError("Unsupported CDSL schema")
    _validate_cdsl_json_schema(cdsl, engine)
    part_id = str(cdsl.get("part_id") or "")
    if not re.fullmatch(r"[a-zA-Z0-9_-]{3,80}", part_id):
        raise ValueError("part_id must use letters, numbers, underscores, or hyphens")
    forbidden = {"compiler_context", "unknown_shape", "complex_arc_shape", "contour_edges_mm", "contour_regions_mm", "entities"}
    for key, value in _walk(cdsl):
        if key in forbidden or (isinstance(value, str) and value in {"unknown_shape", "complex_arc_shape"}):
            raise ValueError(f"Training-unsafe CDSL field: {key}")
    features = cdsl.get("features")
    sketches = cdsl.get("geometry", {}).get("sketches")
    if not isinstance(features, list) or not features or not isinstance(sketches, list):
        raise ValueError("CDSL requires a feature list and a geometry.sketches array")
    if cdsl.get("schema") == "cad.runtime.v1":
        bodies = cdsl.get("bodies")
        if not isinstance(bodies, list) or not bodies:
            raise ValueError("Runtime CDSL requires server-assigned bodies")
        body_ids = [str(body.get("id") or "") for body in bodies if isinstance(body, dict)]
        body_names = [str(body.get("name") or "") for body in bodies if isinstance(body, dict)]
        if len(body_ids) != len(bodies) or len(body_ids) != len(set(body_ids)) or len(body_names) != len(set(body_names)):
            raise ValueError("Runtime CDSL bodies must have unique server IDs and local names")
    sketch_ids = {str(sketch.get("id")) for sketch in sketches}
    semantic_contract = _engine_schema(engine)
    try:
        atomic_contracts = engine.materialized_feature_contracts(semantic_contract)
    except (AttributeError, ValueError) as error:
        raise RuntimeError("The local engine operation contract registry is invalid") from error
    declared_atomic_ids = {str(atomic_id) for atomic_id in atomic_contracts}
    registered_atomic_ids = {str(atomic_id) for atomic_id in getattr(engine, "SUPPORTED_ATOMIC_IDS", ())}
    supported_atomic_ids = sorted(declared_atomic_ids & registered_atomic_ids)
    feature_ids: set[str] = set()
    for feature in features:
        fid = str(feature.get("id") or "")
        if not fid or fid in feature_ids:
            raise ValueError("Feature ids must be unique")
        feature_ids.add(fid)
        atomic_id = str(feature.get("atomic_id") or "")
        if not atomic_id:
            raise ValueError(f"Feature {fid} has no atomic_id")
        contract = atomic_contracts.get(atomic_id)
        if atomic_id not in supported_atomic_ids or not isinstance(contract, dict):
            if feature.get("execution_status") == "deferred":
                raise ValueError(f"Feature {fid} is deferred and cannot be rebuilt by the current engine")
            raise ValueError(
                f"Unsupported CDSL atomic_id: {atomic_id}. Supported: {', '.join(supported_atomic_ids)}"
            )
        params = feature.get("params")
        if not isinstance(params, dict):
            raise ValueError(f"Feature {fid} params must be an object")
        for parameter_name in contract.get("required_params") or []:
            if params.get(parameter_name) is None:
                raise ValueError(f"Feature {fid} ({atomic_id}) is missing required parameter: {parameter_name}")
        # 距离检查只对"以 distance_mm 驱动"的 extrude 契约生效：extrude_cut_through
        # 等贯通型契约的作者参数不含 distance_mm（以 end_condition 表达贯通），
        # 不能按盲切假设要求距离。用契约声明判定，避免硬编码原子名单漂移。
        author_props = (contract.get("author_params_schema") or {}).get("properties") or {}
        if (
            atomic_id.startswith("extrude_")
            and "distance_mm" in author_props
            and float(params.get("distance_mm") or 0) <= 0
        ):
            raise ValueError(f"Feature {fid} ({atomic_id}) requires distance_mm > 0 for runtime rebuild")
        if atomic_id.startswith("revolve_") and float(params.get("angle_deg") or 0) <= 0:
            raise ValueError(f"Feature {fid} ({atomic_id}) requires angle_deg > 0 for runtime rebuild")
        if contract.get("requires_sketch") and str(feature.get("sketch_id") or "") not in sketch_ids:
            raise ValueError(f"Feature {fid} ({atomic_id}) requires a valid sketch_id")
        for dependency in feature.get("depends_on") or []:
            if dependency not in feature_ids:
                raise ValueError(f"Feature {fid} has a forward or missing dependency")
    for sketch in sketches:
        profile = sketch.get("profile")
        if sketch.get("profile_from"):
            continue
        if not isinstance(profile, dict):
            raise ValueError(f"Sketch {sketch.get('id')} has no self-contained profile")
        profile_type = str(profile.get("type") or "")
        if profile_type == "polygon":
            if not profile.get("vertices"):
                raise ValueError("Polygon profiles require vertices")
        elif profile_type not in engine.SHAPE_GENERATORS:
            raise ValueError(f"Unsupported CDSL profile: {profile_type}")


def validate_cdsl(cdsl: dict[str, Any], engine: Any) -> None:
    """Validate static CDSL and the current executable topology semantics."""
    validate_cdsl_shape(cdsl, engine)
    try:
        analysis = engine.analyze_cdsl(copy.deepcopy(cdsl))
    except Exception as error:
        raise ValueError(f"CDSL engine runtime preflight failed: {error}") from error
    if not analysis.runtime_eligible:
        first = next((result for result in analysis.feature_results if not result.executable), None)
        if first is None:
            raise ValueError(f"CDSL engine runtime preflight failed: {analysis.document_blockers[0].code}")
        blockers = ", ".join(blocker.code for blocker in first.blockers)
        raise ValueError(f"CDSL engine runtime preflight failed: feature {first.feature_id}: {blockers}")


def _parameter_id(path: list[str]) -> str:
    return "param_" + "_".join(re.sub(r"[^a-zA-Z0-9]+", "_", item).strip("_") for item in path)


def _parameter_label(path: list[str]) -> str:
    return " / ".join(path[-2:]).replace("_mm", " (mm)").replace("_", " ")


def _derived_parameters(cdsl: dict[str, Any]) -> list[dict[str, Any]]:
    parameters: list[dict[str, Any]] = []

    def add(path: list[str], value: Any, group: str) -> None:
        if isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite(float(value)):
            return
        number = float(value)
        magnitude = max(abs(number), 1.0)
        parameters.append({
            "id": _parameter_id(path),
            "name": ".".join(path),
            "display_name": _parameter_label(path),
            "path": path,
            "value": number,
            "default_value": number,
            "minimum": 0.01 if number >= 0 else -magnitude * 10,
            "maximum": magnitude * 10,
            "step": 0.1 if abs(number) < 100 else 1.0,
            "precision": 2,
            "unit": "mm" if path[-1].endswith("_mm") else "",
            "group": group,
            "editable": True,
        })

    for feature_index, feature in enumerate(cdsl.get("features") or []):
        for key, value in (feature.get("params") or {}).items():
            add(["features", str(feature_index), "params", str(key)], value, "Features")
    for sketch_index, sketch in enumerate(cdsl.get("geometry", {}).get("sketches") or []):
        profile = sketch.get("profile") or {}

        def walk_profile(value: Any, path: list[str]) -> None:
            if isinstance(value, dict):
                for key, child in value.items():
                    walk_profile(child, [*path, str(key)])
            elif isinstance(value, list):
                # Coordinates are topology anchors, not user-facing dimensions.
                return
            else:
                add(path, value, "Sketches")

        walk_profile(profile, ["geometry", "sketches", str(sketch_index), "profile"])
    return parameters


def parameter_contract(cdsl: dict[str, Any]) -> dict[str, Any]:
    declared = cdsl.get("meta", {}).get("editable_parameters")
    if isinstance(declared, list) and declared:
        values = [item for item in declared if isinstance(item, dict) and isinstance(item.get("path"), list)]
        if values:
            return {"schema_version": "1.0", "parameters": values, "source": "declared"}
    return {"schema_version": "1.0", "parameters": _derived_parameters(cdsl), "source": "derived"}


def topology_snapshot(
    engine_result: dict[str, Any],
    *,
    task_id: str = "",
    revision_id: str = "",
    preview: dict[str, Any] | None = None,
) -> dict[str, Any]:
    raw_records = [
        raw for raw in engine_result.get("topology_records") or ()
        if isinstance(raw, dict) and raw.get("record_id") and raw.get("kind")
    ]
    active_body_id = next(
        (
            str(result.get("body_id"))
            for result in reversed(engine_result.get("feature_results") or ())
            if isinstance(result, dict) and result.get("body_id")
        ),
        "",
    )
    if not active_body_id:
        active_body_id = next(
            (
                str(raw.get("body_id"))
                for raw in reversed(raw_records)
                if raw.get("kind") == "body" and raw.get("body_id")
            ),
            "",
        )

    # The runtime retains historical B-rep records for provenance, but only
    # the final body can resolve face, edge, vertex, and body selectors.
    active_records = [
        raw for raw in raw_records
        if not active_body_id
        or raw.get("kind") in {"plane", "axis"}
        or str(raw.get("body_id") or "") == active_body_id
    ]
    records: list[dict[str, Any]] = []
    for raw in active_records:
        kind = str(raw.get("kind"))
        records.append({
            "record_id": str(raw["record_id"]),
            "kind": kind,
            "feature_id": str(raw.get("feature_id") or ""),
            "body_id": str(raw.get("body_id") or "") or None,
            "owner_feature_ids": [str(item) for item in raw.get("owner_feature_ids") or () if str(item)],
            "geometry": copy.deepcopy(raw.get("geometry") or {}),
            "executable": kind in {"body", "face", "edge", "vertex", "plane", "axis"},
            "synthetic": False,
        })
    # Preview/B-rep fallback faces are useful for visual explanation only.
    # Keep them in the unified audit snapshot, but never expose them as
    # executable selector candidates.
    if not any(item.get("kind") == "face" for item in records):
        for index, raw in enumerate((preview or {}).get("topology_faces") or ()):
            if not isinstance(raw, dict):
                continue
            record_id = str(raw.get("id") or f"synthetic:face:{index}")
            center = raw.get("center")
            normal = raw.get("normal")
            raw_bbox = raw.get("bbox")
            if isinstance(raw_bbox, dict) and isinstance(raw_bbox.get("min"), list) and isinstance(raw_bbox.get("max"), list):
                raw_bbox = [*raw_bbox["min"], *raw_bbox["max"]]
            geometry = {
                "surface_type": str(raw.get("surface_type") or "unknown"),
                "center_mm": copy.deepcopy(center) if isinstance(center, list) else None,
                "normal": copy.deepcopy(normal) if isinstance(normal, list) else None,
                "bbox_mm": copy.deepcopy(raw_bbox or {}),
            }
            records.append({
                "record_id": record_id,
                "kind": "face",
                "feature_id": "",
                "body_id": None,
                "owner_feature_ids": [],
                "geometry": geometry,
                "executable": False,
                "synthetic": True,
            })
    return {
        "schema_version": "cad.topology.v1",
        "task_id": task_id,
        "revision_id": revision_id,
        "snapshot_id": f"{task_id}/{revision_id}" if task_id and revision_id else "",
        "body_id": active_body_id,
        "records": records,
    }


def topology_sidecars(
    engine_result: dict[str, Any],
    preview: dict[str, Any] | None = None,
    *,
    snapshot: dict[str, Any] | None = None,
) -> tuple[dict[str, Any], dict[str, Any]]:
    runtime_records = (snapshot or topology_snapshot(engine_result)).get("records") or []
    runtime_faces = [item for item in runtime_records if item.get("kind") == "face" and item.get("executable", True)]
    runtime_edges = [item for item in runtime_records if item.get("kind") == "edge" and item.get("executable", True)]
    if runtime_faces or runtime_edges:
        references = []
        for record in runtime_faces:
            geometry = record.get("geometry") or {}
            references.append({
                "id": str(record.get("record_id")),
                "selectorType": "face",
                "label": str(geometry.get("surface_type") or "face"),
                "center": geometry.get("center_mm"),
                "normal": geometry.get("normal"),
                "frame": {
                    "origin_mm": geometry.get("center_mm"),
                    "normal": geometry.get("normal"),
                    "x_dir": [1, 0, 0],
                    "y_dir": [0, 1, 0],
                },
                "bbox": geometry.get("bbox_mm") or {},
                "surface_type": str(geometry.get("surface_type") or "unknown"),
                "owner_feature_ids": record.get("owner_feature_ids") or [],
                "source": "runtime_snapshot",
                "snapshot_id": (snapshot or {}).get("snapshot_id") or "",
                "executable": True,
            })
        edge_records = [
            {
                **record,
                "selectorType": "edge",
                "source": "runtime_snapshot",
                "snapshot_id": (snapshot or {}).get("snapshot_id") or "",
            }
            for record in runtime_edges
        ]
        return (
            {"schema_version": "cad.topology.v1", "references": references, "edges": edge_records},
            {"schema_version": "cad.topology.v1", "edges": edge_records},
        )
    topology_faces = (preview or {}).get("topology_faces")
    if isinstance(topology_faces, list) and topology_faces:
        references = []
        for face in topology_faces:
            if not isinstance(face, dict):
                continue
            frame = face.get("frame")
            center = face.get("center")
            normal = face.get("normal")
            if not isinstance(frame, dict) or not isinstance(center, list) or not isinstance(normal, list):
                continue
            references.append({
                "id": str(face.get("id") or f"face_{len(references):03d}"),
                "selectorType": "face",
                "label": str(face.get("surface_type") or "face"),
                "center": center,
                "normal": normal,
                "frame": frame,
                "bbox": face.get("bbox") or {},
                "surface_type": str(face.get("surface_type") or "unknown"),
                "triangle_start": int(face.get("triangle_start") or 0),
                "triangle_count": int(face.get("triangle_count") or 0),
                "source": "preview",
                "synthetic": True,
                "executable": False,
            })
        if references:
            return ({"schema_version": "1.1", "references": references}, {"schema_version": "1.0", "edges": []})

    bbox = engine_result.get("bbox_mm") or {}
    minimum = [float(value) for value in bbox.get("min") or [0, 0, 0]]
    maximum = [float(value) for value in bbox.get("max") or [0, 0, 0]]
    if len(minimum) != 3 or len(maximum) != 3:
        raise ValueError("Engine result is missing a valid bounding box")
    center = [(minimum[index] + maximum[index]) / 2 for index in range(3)]
    definitions = [
        ("top", [center[0], center[1], maximum[2]], [0, 0, 1], [1, 0, 0], [0, 1, 0]),
        ("bottom", [center[0], center[1], minimum[2]], [0, 0, -1], [1, 0, 0], [0, -1, 0]),
        ("right", [maximum[0], center[1], center[2]], [1, 0, 0], [0, 1, 0], [0, 0, 1]),
        ("left", [minimum[0], center[1], center[2]], [-1, 0, 0], [0, 1, 0], [0, 0, -1]),
        ("front", [center[0], maximum[1], center[2]], [0, 1, 0], [1, 0, 0], [0, 0, -1]),
        ("back", [center[0], minimum[1], center[2]], [0, -1, 0], [1, 0, 0], [0, 0, 1]),
    ]
    references = [
        {
            "id": f"face_{name}", "selectorType": "face", "label": name,
            "center": point, "normal": normal,
            "frame": {"origin_mm": point, "normal": normal, "x_dir": x_dir, "y_dir": y_dir},
            "bbox": {"min": minimum, "max": maximum},
            "source": "bbox_fallback",
            "synthetic": True,
            "executable": False,
        }
        for name, point, normal, x_dir, y_dir in definitions
    ]
    return ({"schema_version": "1.0", "references": references}, {"schema_version": "1.0", "edges": []})