api/main.py

"""CarPrint API — see docs/API_CONTRACT.md."""
from __future__ import annotations

import hashlib
import hmac
import os
import re
import secrets
import sys
import threading
import time
import uuid
from dataclasses import dataclass, field
from datetime import datetime, timedelta, timezone
from pathlib import Path

import cv2
import numpy as np
from fastapi import BackgroundTasks, FastAPI, File, Form, HTTPException, UploadFile
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import FileResponse, JSONResponse, Response
from fastapi.staticfiles import StaticFiles
from pydantic import BaseModel

ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT))

from carprint.core.image_io import load_image  # noqa: E402
from carprint.core.quality import MESSAGES_RU  # noqa: E402
from carprint.damage import segmenter  # noqa: E402
from carprint.damage.estimate import SIDED_PARTS, DamageInput, estimate as cost_estimate, group_key, near_part_edge  # noqa: E402
from carprint.damage.scale import estimate_scale, measure, vehicle_spec  # noqa: E402
from carprint.identity.cohort import bank_paths  # noqa: E402
from carprint.identity.pipeline import make_frame  # noqa: E402
from carprint.identity.verify import Frame, Verdict, distinctiveness, verify  # noqa: E402
from carprint.identity.viz import match_lines, side_by_side  # noqa: E402
from carprint.integrity.hashing import DUPLICATE_MAX_DISTANCE, hamming, phash64, sha256_hex  # noqa: E402
from carprint.integrity.ledger import Ledger  # noqa: E402
from carprint.integrity.report import build_pdf  # noqa: E402
from carprint.timeline.diff import ZoneMap, compare as timeline_compare  # noqa: E402

DATA = Path(os.environ.get("CARPRINT_DATA", ROOT / "data" / "store"))
MODELS = Path(os.environ.get("CARPRINT_MODELS", ROOT / "models"))
# Operating point of the damage model, chosen on data (docs/EVALUATION.md): at 0.45 it flags 82% of 60 held-out damaged
# photos and 29% of 132 clean cohort photos (0.35: 95% and 50%). Findings below 0.6 are shown as "staff to confirm".
DAMAGE_SCORE = float(os.environ.get("CARPRINT_DAMAGE_SCORE", "0.45"))
COHORT = Path(os.environ.get("CARPRINT_COHORT", ROOT / "data" / "cohort"))
ARTIFACTS = DATA / "artifacts"
for d in (DATA, ARTIFACTS):
    d.mkdir(parents=True, exist_ok=True)
VEHICLE_SECRET = os.environ.get("CARPRINT_VEHICLE_SECRET", "dev-only-secret").encode()
COHORT_K = 8
REGISTRATION_NFA = -2.0  # a zone must be registered (a-contrario) before damage masks are carried through it
SHOT_PLAN = [
    ("front", "Перед целиком, 2–3 м"), ("front_left_45", "Диагональ слева спереди"), ("left", "Левый бок целиком"),
    ("rear_left_45", "Диагональ слева сзади"), ("rear", "Зад целиком"), ("rear_right_45", "Диагональ справа сзади"),
    ("right", "Правый бок целиком"), ("front_right_45", "Диагональ справа спереди"),
]
CHALLENGES = ["Покажите правое заднее колесо крупно", "Покажите левое переднее колесо крупно",
              "Покажите правое переднее колесо крупно", "Покажите левое заднее колесо крупно"]

app = FastAPI(title="CarPrint API", version="0.1.0")
WARM_MODELS = [m for m in os.environ.get("CARPRINT_WARM", "cobalt,matiz,nexia,gentra,spark").split(",") if m]


@app.on_event("startup")
def _warm_cohorts():
    """Compute the cohort banks of the common models in the background, so the jury's first check does not pay for

    them (on a CPU-only host this is ~40 s per model)."""
    def run():
        for m in WARM_MODELS:
            try:
                with compute_lock:
                    cohort_for(m)
                print(f"[carprint] cohort warmed: {m}")
            except Exception as e:
                print(f"[carprint] cohort warm-up failed for {m}: {e!r}")

    threading.Thread(target=run, daemon=True).start()
app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"])
ledger = Ledger(DATA / "passport.sqlite", DATA / "signing_key.pem")
compute_lock = threading.Lock()  # one heavy job at a time on a small GPU


@dataclass
class StoredFrame:
    frame_id: str
    inspection_id: str
    slot: str
    sha256: str
    phash: int
    capture_mode: str
    server_ts: str
    frame: Frame
    parts: list = field(default_factory=list)
    damages: list = field(default_factory=list)
    analyzed: bool = False
    overlay_url: str | None = None
    scale: object = None  # carprint.damage.scale.Scale
    sizes: list = field(default_factory=list)  # measure() per damage
    damage_inputs: list = field(default_factory=list)  # DamageInput per damage (cost model)
    response: dict | None = None  # analysis already rendered (overlay written once)


@dataclass
class Inspection:
    inspection_id: str
    vehicle_key: str
    role: str
    type: str
    nonce: str
    expires_at: datetime
    model_hint: str
    frames: list[StoredFrame] = field(default_factory=list)


INSPECTIONS: dict[str, Inspection] = {}
LAST_ZONES: dict[str, list] = {}  # verification_id -> ZoneEvidence (for the web studio's aligned view)
FRAMES: dict[str, StoredFrame] = {}
COHORT_CACHE: dict[str, list[Frame]] = {}
WEIGHTS: dict[str, np.ndarray] = {}


def now() -> datetime:
    return datetime.now(timezone.utc)


def err(status: int, code: str, message_ru: str):
    raise HTTPException(status_code=status, detail={"error": {"code": code, "message_ru": message_ru}})


def vehicle_key_for(identifier: str) -> str:
    return "veh_" + hmac.new(VEHICLE_SECRET, identifier.strip().upper().encode(), hashlib.sha256).hexdigest()[:24]


def cohort_for(model_hint: str) -> list[Frame]:
    key = model_hint or "generic"
    if key not in COHORT_CACHE:
        paths = bank_paths(COHORT, model_hint, COHORT_K) if COHORT.is_dir() else []
        COHORT_CACHE[key] = [make_frame(f"cohort_{i}", "extra", load_image(p.read_bytes()).rgb) for i, p in enumerate(paths)]
    return COHORT_CACHE[key]


def weights_for(sf: StoredFrame, model_hint: str) -> np.ndarray:
    if sf.frame_id not in WEIGHTS:
        WEIGHTS[sf.frame_id] = distinctiveness(sf.frame, cohort_for(model_hint))
    return WEIGHTS[sf.frame_id]


def blur_regions(rgb: np.ndarray, masks: list[np.ndarray]) -> np.ndarray:
    out = rgb.copy()
    if not masks:
        return out
    m = np.any(masks, axis=0)
    blurred = cv2.GaussianBlur(rgb, (0, 0), 25)
    out[m] = blurred[m]
    return out


def save_artifact(rgb: np.ndarray, prefix: str) -> str:
    name = f"{prefix}_{uuid.uuid4().hex[:12]}.jpg"
    cv2.imwrite(str(ARTIFACTS / name), cv2.cvtColor(rgb, cv2.COLOR_RGB2BGR), [cv2.IMWRITE_JPEG_QUALITY, 85])
    return f"/api/v1/artifacts/{name}"


def part_instance_at(sf: StoredFrame, inst):
    best, best_overlap = None, 0
    for p in sf.parts:
        if p.cls == "license_plate":
            continue
        ov = int(np.logical_and(p.mask, inst.mask).sum())
        if ov > best_overlap:
            best, best_overlap = p, ov
    return best


def part_at(sf: StoredFrame, inst) -> str | None:
    p = part_instance_at(sf, inst)
    return p.cls if p is not None else None


def _measure_damages(sf: StoredFrame):
    """Metric scale of the frame and size of every damage: the inputs of the cost model."""
    ins = INSPECTIONS.get(sf.inspection_id)
    sf.scale = estimate_scale(sf.parts, vehicle_spec(ins.model_hint if ins else ""), sf.frame.rgb.shape,
                              sf.frame.subject_box, sf.slot)
    sf.sizes, sf.damage_inputs = [], []
    for i, d in enumerate(sf.damages):
        part = part_instance_at(sf, d)
        size = measure(d.mask, part.cls if part else None, sf.scale)
        rel = float(d.mask.sum() / part.mask.sum()) if part is not None and part.mask.any() else None
        sf.sizes.append(size)
        # q = 1: the estimate is "what it costs if staff confirms the finding"; detector scores are not probabilities
        # (findings below AUTO_SCORE are shown as "staff to confirm"), so they must not zero the median
        sf.damage_inputs.append(DamageInput(d.cls, part.cls if part else None, 1.0, size["length_mm"],
                                            size.get("rel_unc_length"), rel,
                                            near_part_edge(d.mask, part.mask if part else None), f"dmg_{sf.frame_id}_{i}"))


def inspection_cost(frames: list[StoredFrame], model_hint: str, only: dict[str, set] | None = None) -> dict | None:
    """Cost of the damages of one inspection. A damage seen from several slots is costed once: every part is taken

    from the view where its damage is largest. `only` restricts to {frame_id: damage indices} (e.g. new damages)."""
    views: dict[str, list] = {}
    for sf in frames:
        per_group: dict[str, list] = {}
        for i, di in enumerate(sf.damage_inputs):
            if only is not None and i not in only.get(sf.frame_id, ()):
                continue
            key = group_key(di.part, sf.slot)
            if sf.slot in ("", "extra") and (di.part is None or di.part in SIDED_PARTS):
                key += f"@{sf.frame_id}"  # unknown side: a left and a right door must not collapse into one
            per_group.setdefault(key, []).append(di)
        for key, ds in per_group.items():
            views.setdefault(key, []).append((sum(d.rel_area or 0.0 for d in ds) or len(ds), ds))
    chosen = [(key, d) for key, v in views.items() for d in max(v, key=lambda t: t[0])[1]]
    if not chosen:
        return None
    est = cost_estimate(chosen, model=model_hint)
    est["note_ru"] = (est.get("note_ru", "") + " Диапазон для подтверждённых мастером находок.").strip()
    est["note_en"] = (est.get("note_en", "") + " Range for findings confirmed by staff.").strip()
    return est


def _size_dict(size: dict) -> dict:
    return {k: (round(v, 3) if isinstance(v, float) else v) for k, v in size.items() if k != "area_px"}


def analyze_frame(sf: StoredFrame) -> dict:
    if sf.response is not None:
        return sf.response
    if not sf.analyzed:
        try:
            parts_model, dmg_model = segmenter.load("parts", MODELS), segmenter.load("damage", MODELS)
            sf.parts = parts_model.predict(sf.frame.rgb, 0.4) if parts_model else []
            sf.damages = dmg_model.predict(sf.frame.rgb, DAMAGE_SCORE) if dmg_model else []
            _measure_damages(sf)
        except Exception as e:  # a damage-model failure must never cost the identity verdict
            print(f"[carprint] damage analysis failed for {sf.frame_id}: {e!r}")
            sf.parts, sf.damages, sf.sizes, sf.damage_inputs, sf.scale = [], [], [], [], None
        sf.analyzed = True
    plates = [p.mask for p in sf.parts if p.cls == "license_plate"]
    overlay = blur_regions(sf.frame.rgb, plates)
    signal = np.array([255, 91, 31])  # the site's accent: new damage reads the same everywhere
    for d in sf.damages:
        overlay[d.mask] = (0.55 * overlay[d.mask] + 0.45 * signal).astype(np.uint8)
        contours, _ = cv2.findContours(d.mask.astype(np.uint8), cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)
        cv2.drawContours(overlay, contours, -1, (255, 91, 31), 2, cv2.LINE_AA)
        x, y, w, h = d.bbox
        label = f"{d.cls} {d.score:.2f}"
        (tw, th), _ = cv2.getTextSize(label, cv2.FONT_HERSHEY_SIMPLEX, 0.55, 1)
        cv2.rectangle(overlay, (x, max(0, y - th - 10)), (x + tw + 10, max(th + 10, y)), (255, 91, 31), -1)
        cv2.putText(overlay, label, (x + 5, max(th + 5, y - 5)), cv2.FONT_HERSHEY_SIMPLEX, 0.55, (12, 13, 15), 1, cv2.LINE_AA)
    sf.overlay_url = save_artifact(overlay, "ovl")
    sf.response = {
        "frame_id": sf.frame_id, "slot": sf.slot, "quality": sf.frame.quality.to_dict(),
        "privacy": {"plates_blurred": len(plates), "persons_blurred": 0},
        "parts": [{"part": p.cls, "score": round(p.score, 3), "bbox": list(p.bbox)} for p in sf.parts if p.cls != "license_plate"],
        "damages": [{"damage_id": f"dmg_{sf.frame_id}_{i}", "part": part_at(sf, d), "class": d.cls, "score": round(d.score, 3),
                     "area_ratio": round(d.area_ratio, 4), "bbox": list(d.bbox), "size": _size_dict(sz),
                     "status": "auto" if d.score >= 0.6 else "needs_master"}
                    for i, (d, sz) in enumerate(zip(sf.damages, sf.sizes))],
        "scale": sf.scale.to_dict() if sf.scale is not None else None,
        "overlay_url": sf.overlay_url,
    }
    return sf.response


# ---------- endpoints ----------

class InspectionIn(BaseModel):
    vehicle_key: str | None = None
    vehicle_id_plain: str | None = None  # VIN or plate; hashed immediately, never stored
    vehicle_hint: dict | None = None
    role: str = "dealer"
    type: str = "checkin"


@app.post("/api/v1/inspections", status_code=201)
def create_inspection(body: InspectionIn):
    vk = body.vehicle_key or (vehicle_key_for(body.vehicle_id_plain) if body.vehicle_id_plain else f"veh_{uuid.uuid4().hex[:24]}")
    model_hint = ((body.vehicle_hint or {}).get("model") or "").strip().lower().replace(" ", "_")
    ins = Inspection(f"ins_{uuid.uuid4().hex[:16]}", vk, body.role, body.type, secrets.token_urlsafe(16),
                     now() + timedelta(hours=2), model_hint)
    INSPECTIONS[ins.inspection_id] = ins
    plan = [{"slot": s, "instruction": t, "required": True} for s, t in SHOT_PLAN]
    plan.append({"slot": "random_challenge", "instruction": secrets.choice(CHALLENGES), "required": True})
    return {"inspection_id": ins.inspection_id, "vehicle_key": vk, "nonce": ins.nonce,
            "expires_at": ins.expires_at.isoformat(), "shot_plan": plan}


def _get_inspection(inspection_id: str) -> Inspection:
    ins = INSPECTIONS.get(inspection_id)
    if ins is None:
        err(404, "NOT_FOUND", "Осмотр не найден")
    return ins


def _ingest(ins: Inspection, data: bytes, slot: str, capture_mode: str) -> tuple[StoredFrame, list[str]]:
    try:
        img = load_image(data)
    except Exception:
        err(400, "BAD_IMAGE", "Файл не является изображением")
    with compute_lock:
        frame = make_frame(f"frm_{uuid.uuid4().hex[:16]}", slot, img.rgb)
    ph = phash64(img.rgb)
    flags = []
    if capture_mode != "camera":
        flags.append("gallery_upload")
    for other in list(FRAMES.values()):
        if other.inspection_id != ins.inspection_id and hamming(ph, other.phash) <= DUPLICATE_MAX_DISTANCE:
            flags.append(f"duplicate_of:{other.frame_id}")
            break
    sf = StoredFrame(frame.frame_id, ins.inspection_id, slot, sha256_hex(data), ph, capture_mode,
                     now().isoformat(), frame)
    ins.frames.append(sf)
    FRAMES[sf.frame_id] = sf
    (DATA / ins.inspection_id).mkdir(exist_ok=True)
    (DATA / ins.inspection_id / f"{sf.frame_id}.bin").write_bytes(data)  # original evidence, never served
    return sf, flags


def _precompute_weights(sf: StoredFrame, model_hint: str):
    """Cohort distinctiveness of a frame is needed whenever it later serves as a reference: compute it off the

    request path so a verification does not pay K extra matches per reference frame."""
    try:
        with compute_lock:
            weights_for(sf, model_hint)
    except Exception as e:  # never fail an upload because of a cache warm-up
        print(f"[carprint] weight precompute failed for {sf.frame_id}: {e!r}")


@app.post("/api/v1/inspections/{inspection_id}/frames")
def upload_frame(inspection_id: str, background: BackgroundTasks, file: UploadFile = File(...),

                       slot: str = Form("extra"), nonce: str = Form(...), capture_mode: str = Form("gallery"),

                       client_ts: str | None = Form(None)):
    ins = _get_inspection(inspection_id)
    if not hmac.compare_digest(nonce, ins.nonce):
        err(409, "NONCE_MISMATCH", "Неверный код сессии съёмки")
    if now() > ins.expires_at:
        err(409, "NONCE_EXPIRED", "Сессия съёмки истекла — начните осмотр заново")
    sf, flags = _ingest(ins, file.file.read(), slot, capture_mode)
    q = sf.frame.quality
    if q.ok:
        background.add_task(_precompute_weights, sf, ins.model_hint)
    return {"frame_id": sf.frame_id, "sha256": sf.sha256, "server_ts": sf.server_ts, "quality": q.to_dict(),
            "integrity_flags": flags, "recapture_required": not q.ok,
            "message_ru": "Кадр принят" if q.ok else "; ".join(MESSAGES_RU[i] for i in q.issues)}


@app.post("/api/v1/inspections/{inspection_id}/analyze")
def analyze(inspection_id: str):
    ins = _get_inspection(inspection_id)
    t0 = time.perf_counter()
    with compute_lock:
        frames = [analyze_frame(sf) for sf in ins.frames]
        cost = inspection_cost(ins.frames, ins.model_hint)
    status = "PROCESSED" if any(sf.frame.quality.ok for sf in ins.frames) else "RECAPTURE_REQUIRED"
    return {"inspection_id": ins.inspection_id, "status": status, "frames": frames, "cost_estimate": cost,
            "latency_ms": round((time.perf_counter() - t0) * 1000)}


_TL_RANK = {"previously_recorded": 3, "newly_detected": 2, "previously_seen_now_absent": 1, "insufficient_evidence": 0}


def dedupe_timeline(items: list[dict]) -> list[dict]:
    """Every (reference, current) frame pair yields its own timeline; merge them so each damage appears once.

    A current damage keeps its strongest status (recorded > new > not visible); a reference damage that was recorded

    from any angle is not "gone", and "gone" yields to "not visible" when the views disagree."""
    recorded_refs = {t["ref_damage_id"] for t in items if t["status"] == "previously_recorded" and t.get("ref_damage_id")}
    best: dict[str, dict] = {}
    for t in items:
        if t.get("damage_id"):
            key = t["damage_id"]
            if key not in best or _TL_RANK[t["status"]] > _TL_RANK[best[key]["status"]]:
                best[key] = t
        else:
            key = t.get("ref_damage_id") or f"ref:{t['ref_frame']}:{t['ref_bbox']}"
            if key in recorded_refs:
                continue
            if key not in best or t["status"] == "insufficient_evidence":
                best[key] = t
    return list(best.values())


def _verify(ref: Inspection, cur: Inspection) -> dict:
    t0 = time.perf_counter()
    # A re-submitted reference photo would match itself perfectly: exclude it from evidence and flag it.
    recycled = {c.frame_id: r.frame_id for c in cur.frames for r in ref.frames
                if c.sha256 == r.sha256 or hamming(c.phash, r.phash) <= DUPLICATE_MAX_DISTANCE}
    fresh = [sf for sf in cur.frames if sf.frame_id not in recycled]
    with compute_lock:
        cohort = {sf.frame_id: cohort_for(ref.model_hint) for sf in ref.frames}
        if recycled and not fresh:
            v = Verdict("REVIEW", 0.0, [], ["Все фото текущего осмотра — копии эталонных снимков: нужен новый осмотр"])
        else:
            v = verify([sf.frame for sf in ref.frames], [sf.frame for sf in fresh], cohort)
        for sf in ref.frames + cur.frames:
            analyze_frame(sf)
        evidence, timeline = [], []
        by_id = {sf.frame_id: sf for sf in ref.frames + cur.frames}
        best_zone: dict[tuple[str, str], object] = {}
        registered: dict[tuple[str, str], list[ZoneMap]] = {}
        for z in v.zones:
            key = (z.ref_frame, z.cur_frame)
            if key not in best_zone or z.zone.n_inliers > best_zone[key].zone.n_inliers:
                best_zone[key] = z
            if z.zone.log10_nfa < REGISTRATION_NFA:
                registered.setdefault(key, []).append(ZoneMap(z.zone.H, z.zone.bbox0, z.zone.bbox1))
        for (rf, cf), z in sorted(best_zone.items(), key=lambda kv: kv[1].log10_nfa_distinct)[:6]:
            r, c = by_id[rf], by_id[cf]
            pair_zones = [e for e in v.zones if (e.ref_frame, e.cur_frame) == (rf, cf)]
            viz = side_by_side(r.frame, c.frame, pair_zones)
            rh, rw = r.frame.rgb.shape[:2]
            aligned = None
            if z.zone.log10_nfa < REGISTRATION_NFA:
                aligned = save_artifact(cv2.warpPerspective(c.frame.rgb, np.linalg.inv(z.zone.H), (rw, rh)), "al")
            evidence.append({**z.to_dict(), "slot": r.slot, "match_viz_url": save_artifact(viz, "m"), "aligned_url": aligned,
                             "matches": match_lines(pair_zones, r.frame.rgb.shape, c.frame.rgb.shape)})
        # The timeline only makes sense for the same car; each pair of frames contributes its registered panels.
        new_idx: dict[str, set] = {}
        old_idx: dict[str, set] = {}
        try:
            if v.decision != "DIFFERENT":
                for (rf, cf), zone_maps in registered.items():
                    r, c = by_id[rf], by_id[cf]
                    for item in timeline_compare(r.damages, c.damages, zone_maps, r.frame.rgb, c.frame.rgb.shape,
                                                 part_of=lambda inst, sf=c: part_at(sf, inst),
                                                 part_of_ref=lambda inst, sf=r: part_at(sf, inst)):
                        timeline.append({"status": item.status, "class": item.cls, "part": item.part, "slot": c.slot,
                                         "ref_frame": rf, "cur_frame": cf, "cur_bbox": item.cur_bbox, "ref_bbox": item.ref_bbox,
                                         "damage_id": f"dmg_{cf}_{item.cur_index}" if item.cur_index is not None else None,
                                         "ref_damage_id": f"dmg_{rf}_{item.ref_index}" if item.ref_index is not None else None,
                                         "score": round(item.score, 3) if item.score is not None else None,
                                         "first_seen_inspection_id": ref.inspection_id if item.status == "previously_recorded" else None})
                        if item.cur_index is not None and item.status in ("newly_detected", "previously_recorded"):
                            target = new_idx if item.status == "newly_detected" else old_idx
                            target.setdefault(cf, set()).add(item.cur_index)
            timeline = dedupe_timeline(timeline)
            # New = new in some registered view and recorded in none (a damage seen as old from another angle is old).
            only_new = {f: ids - old_idx.get(f, set()) for f, ids in new_idx.items()}
            model = ref.model_hint or cur.model_hint
            cost_new = inspection_cost(cur.frames, model, only_new) if any(only_new.values()) else None
            cost_all = inspection_cost(cur.frames, model)
        except Exception as e:  # timeline and cost are additions to the verdict, never a reason to lose it
            print(f"[carprint] timeline/cost failed: {e!r}")
            timeline, cost_new, cost_all = [], None, None
        vid = f"ver_{uuid.uuid4().hex[:12]}"
        LAST_ZONES[vid] = v.zones
        while len(LAST_ZONES) > 256:
            LAST_ZONES.pop(next(iter(LAST_ZONES)), None)
    flags = sorted({"gallery_upload" for sf in cur.frames if sf.capture_mode != "camera"})
    flags += [f"recycled_reference_photo:{cid}->{rid}" for cid, rid in recycled.items()]
    level = "red" if recycled else ("yellow" if flags else "green")
    if recycled:
        v.reasons.insert(0, f"{len(recycled)} фото повторяют снимки эталонного осмотра и исключены из доказательств")
    return {"verification_id": vid, "decision": v.decision, "score_raw": v.score_raw,
            "score_note": "Число совпавших независимых ячеек с индивидуальными признаками; не вероятность",
            "reasons_ru": v.reasons, "reasons_en": [reason_en(r) for r in v.reasons], "cohort_size": v.cohort_size,
            "evidence": evidence, "damage_timeline": timeline,
            "cost_estimate_new": cost_new, "cost_estimate_current": cost_all,
            "integrity": {"flags": flags, "level": level}, "latency_ms": round((time.perf_counter() - t0) * 1000)}


REASONS_EN = [
    (r"Совпали индивидуальные признаки в (\d+) независимых ячейках \((\d+) зон\)",
     "Individual marks agree in {0} independent cells ({1} zones)"),
    (r"Когорта машин той же модели мала \((\d+) < (\d+)\): автоматический SAME запрещён",
     "Same-model cohort too small ({0} < {1}): automatic SAME is disabled"),
    (r"Ракурсы совмещаются по дизайну модели, но индивидуальные метки эталона не найдены",
     "The views align on the model's design, but the reference's individual marks are absent"),
    (r"Не найдено сопоставимых участков: снимите те же ракурсы, что в эталоне",
     "No comparable areas found: shoot the same angles as the reference"),
    (r"Доказательств недостаточно: (\d+) совпавших ячеек из (\d+) нужных",
     "Not enough evidence: {0} matching cells of {1} needed"),
    (r"Нет кадров достаточного качества в одном из осмотров", "One of the inspections has no frame of sufficient quality"),
    (r"(\d+) фото повторяют снимки эталонного осмотра и исключены из доказательств",
     "{0} photo(s) repeat the reference inspection and are excluded from evidence"),
    (r"Все фото текущего осмотра — копии эталонных снимков: нужен новый осмотр",
     "Every current photo is a copy of a reference photo: a new inspection is needed"),
]


def reason_en(text: str) -> str:
    for pattern, template in REASONS_EN:
        m = re.fullmatch(pattern, text)
        if m:
            return template.format(*m.groups())
    return text


class VerificationIn(BaseModel):
    reference_inspection_id: str
    current_inspection_id: str


@app.post("/api/v1/verifications")
def verifications(body: VerificationIn):
    return _verify(_get_inspection(body.reference_inspection_id), _get_inspection(body.current_inspection_id))


@app.post("/api/v1/compare")
def compare(images_a: list[UploadFile] = File(...), images_b: list[UploadFile] = File(...),

                  model_hint: str = Form("")):
    if not (1 <= len(images_a) <= 16 and 1 <= len(images_b) <= 16):
        err(422, "BAD_REQUEST", "Нужно от 1 до 16 фото в каждой группе")
    insp = []
    for group in (images_a, images_b):
        ins = Inspection(f"ins_{uuid.uuid4().hex[:16]}", f"veh_{uuid.uuid4().hex[:24]}", "owner", "compare",
                         "", now() + timedelta(hours=2), model_hint.strip().lower())
        INSPECTIONS[ins.inspection_id] = ins
        for f in group:
            _ingest(ins, f.file.read(), "extra", "gallery")
        insp.append(ins)
    result = _verify(insp[0], insp[1])
    result["inspections"] = {"reference": insp[0].inspection_id, "current": insp[1].inspection_id}
    with compute_lock:
        result["analysis_a"] = [analyze_frame(sf) for sf in insp[0].frames]
        result["analysis_b"] = [analyze_frame(sf) for sf in insp[1].frames]
    return result


class Finding(BaseModel):
    damage_id: str
    action: str
    comment: str = ""


class ConfirmIn(BaseModel):
    operator_id: str
    findings: list[Finding] = []
    customer_signed: bool = False


@app.post("/api/v1/inspections/{inspection_id}/confirm")
def confirm(inspection_id: str, body: ConfirmIn):
    ins = _get_inspection(inspection_id)
    payload = {"inspection_id": ins.inspection_id, "type": ins.type, "role": ins.role, "server_ts": now().isoformat(),
               "operator_id": body.operator_id, "customer_signed": body.customer_signed,
               "frames": [{"frame_id": f.frame_id, "slot": f.slot, "sha256": f.sha256, "capture_mode": f.capture_mode}
                          for f in ins.frames],
               "findings": [f.model_dump() for f in body.findings]}
    rec = ledger.append(ins.vehicle_key, f"rec_{uuid.uuid4().hex[:12]}", payload)
    return {**rec, "public_key_url": "/api/v1/passport/public-key"}


@app.get("/api/v1/inspections/{inspection_id}/report.pdf")
def report(inspection_id: str):
    ins = _get_inspection(inspection_id)
    with compute_lock:
        for sf in ins.frames:
            if sf.overlay_url is None:
                analyze_frame(sf)
        cost = inspection_cost(ins.frames, ins.model_hint)
    record = next((r for r in ledger.timeline(ins.vehicle_key) if r["payload"].get("inspection_id") == ins.inspection_id), None)
    damages = [f"{d.cls} ({part_at(sf, d) or 'деталь не определена'}, {d.score:.2f})" for sf in ins.frames for d in sf.damages]
    rows = [("Осмотр", ins.inspection_id), ("Автомобиль (ключ)", ins.vehicle_key), ("Тип / роль", f"{ins.type} / {ins.role}"),
            ("Кадров", str(len(ins.frames))), ("Находки модели", "; ".join(damages) or "не обнаружено"),
            ("Оценка ремонта (ориентир)", f"P10–P90 {cost['p10']:,}–{cost['p90']:,} сум, медиана {cost['p50']:,} сум"
             .replace(",", " ") if cost else "нет находок"),
            ("Запись паспорта", record["record_hash"] if record else "не подтверждено приёмщиком"),
            ("Предыдущая запись", record["prev_hash"] if record else "—")]
    rows += [(f"Кадр {sf.slot}", f"SHA-256 {sf.sha256[:32]}… {sf.capture_mode} {sf.server_ts}") for sf in ins.frames]
    overlays = [(f"{sf.slot} — {sf.sha256[:16]}", ARTIFACTS / sf.overlay_url.rsplit("/", 1)[-1]) for sf in ins.frames if sf.overlay_url]
    pdf = build_pdf("CarPrint — акт осмотра", rows, overlays,
                    "Находки модели не являются решением: каждая подтверждается сотрудником. Номера размыты. "
                    "Целостность записи проверяется по цепочке хешей и подписи Ed25519 (/api/v1/passport/public-key).")
    return Response(content=pdf, media_type="application/pdf",
                    headers={"Content-Disposition": f'inline; filename="{ins.inspection_id}.pdf"'})


@app.get("/api/v1/vehicles/{vehicle_key}/timeline")
def timeline(vehicle_key: str):
    recs = ledger.timeline(vehicle_key)
    return {"vehicle_key": vehicle_key, "records": list(reversed(recs)), "chain_valid": ledger.verify_chain(vehicle_key)}


@app.get("/api/v1/passport/public-key")
def public_key():
    return {"algorithm": "Ed25519", "pem": ledger.public_key_pem}


@app.get("/api/v1/artifacts/{name}")
def artifact(name: str):
    path = (ARTIFACTS / name).resolve()
    if path.parent != ARTIFACTS.resolve() or not path.exists():
        err(404, "NOT_FOUND", "Файл не найден")
    return FileResponse(path, media_type="image/jpeg")


@app.get("/api/v1/metrics")
def metrics():
    path = ROOT / "eval" / "results.json"
    return JSONResponse(content=__import__("json").loads(path.read_text()) if path.exists() else {"status": "not_evaluated"})


@app.get("/api/v1/health")
def health():
    import torch

    return {"status": "ok", "device": "cuda" if torch.cuda.is_available() else "cpu",
            "models": {"parts": getattr(segmenter.load("parts", MODELS), "name", None),
                       "damage": getattr(segmenter.load("damage", MODELS), "name", None),
                       "matcher": "aliked+lightglue"}}


# Web studio: compatibility endpoints, then the built frontend (mounted last so API routes win).
from api.compat import build_router  # noqa: E402

app.include_router(build_router(sys.modules[__name__]))
WEB_DIST = Path(os.environ.get("CARPRINT_WEB_DIST", ROOT / "web" / "dist"))
if WEB_DIST.exists():
    app.mount("/assets", StaticFiles(directory=WEB_DIST / "assets"), name="assets")

    @app.get("/{path:path}", include_in_schema=False)
    def spa(path: str):
        target = (WEB_DIST / path).resolve()
        if path and target.is_file() and WEB_DIST.resolve() in target.parents:
            return FileResponse(target)
        if path.startswith("api/"):
            err(404, "NOT_FOUND", "Нет такого метода API")
        return FileResponse(WEB_DIST / "index.html", headers={"Cache-Control": "no-cache"})