app/main.py

"""PoolCoach BE — FastAPI bọc engine ZonePlanner V2 (23/07/2026; một engine
duy nhất từ 31/07/2026 — BRIEF việc D, bản cũ oracle/hybrid ở nhánh v1-full).

Chạy từ gốc repo poolcoach-rl (hoặc dùng run_app.bat ở D:\\Khoa luan):
    python -m uvicorn app.main:app --port 8000

4 endpoint (contract §4 PoolCoach_App_Web_Design.md):
    GET  /api/table       hằng số bàn để FE dựng Canvas đúng tỉ lệ
    GET  /api/health      ok | warming_up | error (FE poll lúc mở trang)
    POST /api/recommend   thế bàn → 1 + N cú đã rank theo tiêu chí V2; CHỈ
                          cú rank 1 kèm trajectories (render lười, 30/07)
    POST /api/trajectory  thế bàn + (phi, V0, spin) → quỹ đạo MỘT cú. FE gọi
                          khi người dùng bấm sang một cú thay thế.
Thêm 04/08 (bàn giao 11, F3 MVP): GET /api/drills(+/{id}) đọc file
app/drills.json; POST/GET /api/drill-attempts ghi/đọc lượt tập tự khai
(ghi best-effort, đọc cần DB). GET /api/table-qr/{token} từ bàn giao 10.
Static FE mount ở "/" (app/static), Cache-Control: no-cache (31/07).
Warmup JIT Numba chạy nền lúc startup. Concurrency: 1 uvicorn worker +
asyncio.Lock quanh search (pooltool không cần thread-safe; MVP một người
dùng). V2 chạy serial 99–198 sim/cú — không ProcessPool, không net/torch.
Bàn giao 14 (04/08): set REDIS_URL → /api/recommend đi qua worker riêng
(scripts/engine_worker.py, transport app/jobqueue.py); không set → in-process
Y HỆT bản cũ. Engine hai mode là MỘT hàm: app/engine.py run_recommend.
"""

from __future__ import annotations

import asyncio
import functools
import sys
import threading
import time
from contextlib import asynccontextmanager
from datetime import timedelta, timezone
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
if str(ROOT / "src") not in sys.path:
    sys.path.insert(0, str(ROOT / "src"))

from fastapi import FastAPI, File, Form, HTTPException, UploadFile
from fastapi.staticfiles import StaticFiles

import sqlalchemy as sa

from . import analyzer_store as astore
from . import db, displays, drills, jobqueue, localcv, qr
from .engine import _pt, run_recommend
from .schemas import (AnalyzerCornersOut, AnalyzerDemoOut, AnalyzerShotItem,
                      AnalyzerShotNetBrief,
                      AnalyzerShotsOut, AnalyzerVideoCreateOut,
                      AnalyzerVideoInfo, DrillAttemptIn,
                      DrillAttemptItem, DrillAttemptOut, DrillAttemptsListOut,
                      DrillListItem, DrillOut, DrillsListOut, HealthOut,
                      RecommendRequest, ScanBall, ScanOut, TableInfo,
                      TablePocket, TableQrOut, TableQrTable,
                      TrajectoryRequest, TrajectoryResponse)

STATIC_DIR = Path(__file__).resolve().parent / "static"


# ------------------------------------------------------------------ state

class _State:
    env_h = None                    # PositionPlayEnv dùng chung mọi request
    jit_ready: bool = False
    warmup_s: float | None = None
    boot_error: str | None = None


state = _State()
_search_lock = asyncio.Lock()


def _boot() -> None:
    """Import pooltool + tạo env + kích JIT — chạy trong thread nền để
    uvicorn lên ngay, FE poll /api/health trong lúc chờ."""
    try:
        t0 = time.time()
        from poolcoach_rl.envs import PositionPlayEnv
        from poolcoach_rl.recommend import warmup

        state.env_h = PositionPlayEnv()
        print(f"[poolcoach] env san sang sau {time.time() - t0:.1f}s "
              f"-- warmup JIT Numba (~40s lan dau)...", flush=True)
        state.warmup_s = warmup(state.env_h)
        state.jit_ready = True
        print(f"[poolcoach] JIT xong sau {state.warmup_s:.1f}s -- API san sang",
              flush=True)
    except Exception as e:  # noqa: BLE001 — báo lỗi qua /api/health
        state.boot_error = f"{type(e).__name__}: {e}"
        print(f"[poolcoach] LOI khoi dong: {state.boot_error}", flush=True)


@asynccontextmanager
async def lifespan(_app: FastAPI):
    # DB bật/tắt theo DATABASE_URL — tắt là chạy degraded Y HỆT bản không DB
    # (BRIEF 04/08: Space HF không có Postgres, rebuild bất kỳ lúc nào).
    if db.setup():
        print("[poolcoach] DB bat (DATABASE_URL co)", flush=True)
    else:
        print("[poolcoach] DB tat -- khong co DATABASE_URL, "
              "table-qr se tra 503, recommend khong log", flush=True)
    if qr.secret_is_default():
        print("[poolcoach] WARNING: POOLCOACH_SECRET chua set -- token QR "
              "dang ky bang secret dev, KHONG dung cho prod", flush=True)
    # Queue bật/tắt theo REDIS_URL — cùng nếp DATABASE_URL ở trên: không set
    # là mode inprocess, chạy Y HỆT hôm nay (bàn giao 14 — Space không được vỡ).
    qmode = jobqueue.setup()
    if qmode == "queue":
        print("[poolcoach] queue bat (REDIS_URL co) -- /api/recommend di qua "
              "worker rieng, KHONG fallback in-process khi worker chet",
              flush=True)
    elif qmode == "local":
        # Một container (Space HF): CV worker là một LUỒNG trong chính tiến
        # trình này, engine vẫn in-process. Xem app/localcv.py về ba đường
        # đã cân và vì sao chọn đường này.
        print("[poolcoach] mode LOCAL (POOLCOACH_LOCAL_CV) -- CV worker chay "
              "bang luong nhung trong app; recommend van in-process",
              flush=True)
        localcv.start()
    else:
        print("[poolcoach] queue tat -- khong co REDIS_URL, recommend "
              "in-process nhu cu", flush=True)
    # Video mẫu đã phân tích sẵn (việc B): chép vào thư mục kết quả nếu chưa
    # có, để mở app KHÔNG GPU là thấy ngay danh sách cú (không phân tích lại).
    astore.seed_demo()
    threading.Thread(target=_boot, daemon=True, name="poolcoach-warmup").start()
    try:
        yield
    finally:
        if jobqueue.mode() == "local":
            localcv.stop()


app = FastAPI(title="PoolCoach", lifespan=lifespan)


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

@app.get("/api/health", response_model=HealthOut)
def api_health():
    """Trạng thái boot. `workers`/`hybrid_available` cũ gỡ 31/07 cùng pool +
    net (V2 chạy serial, không model) — FE chỉ đọc status/jit_ready/error.

    Bàn giao 14 THÊM mode/redis_ok/worker_alive — phản ánh đủ 3 trạng thái
    (inprocess / queue+alive / queue+worker chết) và KHÔNG được 500 khi Redis
    chết: redis_ok=False là câu trả lời, không phải lỗi. Bàn giao 19 THÊM
    cv_worker_alive (heartbeat CV worker — độc lập engine worker), cùng quy
    ước: inprocess → null "không áp dụng", Redis chết → False.
    """
    status = ("error" if state.boot_error
              else "ok" if state.jit_ready else "warming_up")
    qmode = jobqueue.mode()
    redis_ok = worker_alive = cv_worker_alive = None
    if qmode == "queue":
        redis_ok = jobqueue.ping()
        worker_alive = jobqueue.worker_alive() if redis_ok else False
        cv_worker_alive = (jobqueue.worker_alive(
            key=jobqueue.SCAN_HEARTBEAT_KEY) if redis_ok else False)
    elif qmode == "local":
        # `worker_alive` (engine) để NULL đúng nghĩa "không áp dụng": mode
        # local chạy engine in-process, không có engine worker để mà chết.
        redis_ok = jobqueue.ping()
        cv_worker_alive = (jobqueue.worker_alive(
            key=jobqueue.SCAN_HEARTBEAT_KEY) if redis_ok else False)
    return HealthOut(status=status, jit_ready=state.jit_ready,
                     error=state.boot_error, mode=qmode,
                     redis_ok=redis_ok, worker_alive=worker_alive,
                     cv_worker_alive=cv_worker_alive)


@app.get("/api/table", response_model=TableInfo)
def api_table():
    if state.boot_error:
        raise HTTPException(500, f"Server lỗi khởi động: {state.boot_error}")
    env_h = state.env_h
    if env_h is None:
        raise HTTPException(503, "Đang khởi động — thử lại sau")

    from poolcoach_rl.envs.position_env import BALL_R
    from poolcoach_rl.recommend import pocket_name

    # env._pockets và table.pockets cùng thứ tự (giả định đã dùng ở
    # render_png CLI — nghiệm thu smoke 23/07 pass)
    pocket_objs = list(env_h.table.pockets.values())
    pockets = [
        TablePocket(x=float(p[0]), y=float(p[1]),
                    r=float(getattr(pocket_objs[i], "radius", 2 * BALL_R)),
                    name=pocket_name(env_h, i))
        for i, p in enumerate(env_h._pockets)
    ]
    return TableInfo(w=float(env_h.w), l=float(env_h.l),
                     ball_r=float(BALL_R), pockets=pockets)


# ------------------------------------------------- /api/table-qr (04/08)

GUEST_SESSION_TTL = timedelta(hours=24)


@app.get("/api/table-qr/{token}", response_model=TableQrOut)
def api_table_qr(token: str):
    """Khách quét QR bàn → verify token → tạo guest session TTL 24h.

    Không cần engine/JIT — endpoint sống ngay từ giây đầu boot. Sync def
    (starlette chạy trong threadpool) vì call DB là blocking.
    """
    if not db.enabled():
        # Space HF chưa có Postgres — nói thẳng tính năng chưa bật, đừng 500.
        raise HTTPException(503, "Server chưa cấu hình cơ sở dữ liệu — "
                                 "tính năng QR bàn chưa bật trên bản này.")
    ids = qr.read_table_token(token)
    if ids is None:
        raise HTTPException(400, "Mã QR không hợp lệ — quét lại mã dán trên bàn.")
    tenant_id, table_id = ids

    with db.session() as s:
        tbl = s.get(db.Table, table_id)
        # Token ký đúng nhưng bàn đã xoá / gán nhầm tenant → cũng là "không
        # tồn tại", không lộ bàn của tenant khác.
        if tbl is None or tbl.tenant_id != tenant_id:
            raise HTTPException(404, "Bàn trong mã QR không còn tồn tại — "
                                     "gọi nhân viên đổi mã.")
        gs = db.GuestSession(tenant_id=tenant_id, table_id=table_id,
                             expires_at=db.utcnow() + GUEST_SESSION_TTL)
        s.add(gs)
        s.flush()          # áp default id/created_at trước khi ra khỏi session
        return TableQrOut(
            session_id=gs.id,
            table=TableQrTable(id=tbl.id, name=tbl.name),
            # DB lưu naive-UTC (app/db.py) — gắn lại tz để FE khỏi đoán múi giờ
            expires_at=gs.expires_at.replace(tzinfo=timezone.utc))


# --------------------------------- /api/drills + drill-attempts (F3, 04/08)

@app.get("/api/drills", response_model=DrillsListOut)
def api_drills():
    """Danh sách drill chuẩn hệ thống — đọc từ file repo (app/drills.json).

    Sống từ giây đầu boot: không cần engine/JIT, không cần DB — Space
    degraded vẫn demo được F3 (quyết định nền của BRIEF 04/08 bước 1).
    """
    return DrillsListOut(drills=[
        DrillListItem(id=d["id"], title=d["title"], tags=d["tags"],
                      reps=d["reps"])
        for d in drills.all_drills().values()])


@app.get("/api/drills/{drill_id}", response_model=DrillOut)
def api_drill_detail(drill_id: str):
    """Đủ layout/goal/scoring để FE dựng bàn + vẽ cue_zone."""
    d = drills.get_drill(drill_id)
    if d is None:
        raise HTTPException(404, f"Không có drill '{drill_id}' trong thư viện.")
    return DrillOut(**d)


@app.post("/api/drill-attempts", response_model=DrillAttemptOut)
def api_drill_attempt(req: DrillAttemptIn):
    """Ghi 1 lượt tập TỰ KHAI — best-effort như log recommendations (§5.3).

    Không DB (hoặc DB nổ giữa chừng) → vẫn 200 + ``persisted: false``:
    degraded vẫn tập được, chỉ mất lưu tiến bộ. ``drill_id`` rác thì 404 —
    đó là lỗi client, không liên quan DB nên không được nuốt.

    ``tenant_id``/``guest_session_id`` điền từ guest session nếu
    ``session_id`` trỏ tới một phiên CÒN HẠN — cùng check ``expires_at`` với
    log recommend (bàn giao 13, đồng bộ hai chỗ). Phiên THẬT nhưng hết hạn →
    row không gắn phiên (cả hai cột NULL): hết TTL là hết tư cách quy lượt
    tập về quán/phiên đó. ``session_id`` không có trong DB thì không phải
    phiên — giữ nguyên text làm dấu vết (không FK, như ``scan_id``), tenant
    NULL. Cả hai trường hợp đều KHÔNG chặn lượt tập — luật của bảng này là
    không bao giờ vứt một lượt tập vì chuyện lưu trữ.
    """
    if drills.get_drill(req.drill_id) is None:
        raise HTTPException(404, f"Không có drill '{req.drill_id}' trong thư viện.")
    if not db.enabled():
        return DrillAttemptOut(persisted=False)
    try:
        with db.session() as s:
            tenant_id, guest_session_id = None, req.session_id
            if req.session_id:
                gs = s.get(db.GuestSession, req.session_id)
                if gs is not None:
                    if gs.expires_at > db.utcnow():
                        tenant_id = gs.tenant_id
                    else:
                        guest_session_id = None
                        print(f"[poolcoach] WARNING: session het han trong "
                              f"drill-attempt -- bo qua, khong gan phien "
                              f"({req.session_id!r})", flush=True)
            row = db.DrillAttempt(tenant_id=tenant_id,
                                  guest_session_id=guest_session_id,
                                  drill_id=req.drill_id, result=req.result,
                                  detail=req.detail)
            s.add(row)
            s.flush()          # lấy id trước khi ra khỏi session
            return DrillAttemptOut(persisted=True, id=row.id)
    except Exception as e:  # noqa: BLE001 — best-effort đúng nghĩa đen
        print(f"[poolcoach] WARNING: khong ghi duoc drill attempt: "
              f"{type(e).__name__}: {e}", flush=True)
        return DrillAttemptOut(persisted=False)


@app.get("/api/drill-attempts", response_model=DrillAttemptsListOut)
def api_drill_attempts(session_id: str | None = None,
                       drill_id: str | None = None):
    """Lịch sử + tổng hợp lượt tập. KHÁC POST một cách có chủ đích: xem lịch
    sử thì PHẢI có DB — không có dữ liệu nào để best-effort, 503 nói thẳng."""
    if not db.enabled():
        raise HTTPException(503, "Server chưa cấu hình cơ sở dữ liệu — "
                                 "lịch sử lượt tập chưa xem được trên bản "
                                 "demo này (lượt tập không được lưu).")
    q = sa.select(db.DrillAttempt).order_by(db.DrillAttempt.id)
    if session_id:
        q = q.where(db.DrillAttempt.guest_session_id == session_id)
    if drill_id:
        q = q.where(db.DrillAttempt.drill_id == drill_id)
    with db.session() as s:
        rows = s.scalars(q).all()

    summary: dict[str, dict[str, int]] = {}
    for r in rows:
        c = summary.setdefault(r.drill_id, {"pass": 0, "fail": 0})
        c[r.result] = c.get(r.result, 0) + 1
    return DrillAttemptsListOut(
        attempts=[DrillAttemptItem(
            id=r.id, drill_id=r.drill_id, result=r.result,
            session_id=r.guest_session_id, detail=r.detail,
            # DB lưu naive-UTC — gắn lại tz như expires_at của table-qr
            created_at=r.created_at.replace(tzinfo=timezone.utc))
            for r in rows],
        summary=summary)


@app.post("/api/recommend")
async def api_recommend(req: RecommendRequest):
    if state.boot_error:
        raise HTTPException(500, f"Server lỗi khởi động: {state.boot_error}")
    if not state.jit_ready or state.env_h is None:
        raise HTTPException(503, "Đang khởi động vật lý (JIT ~40s) — thử lại sau")

    if req.balls is None:
        # Đường 3 bi cũ (cue/b1/b2) gỡ 31/07 cùng oracle/hybrid — client cũ
        # nhận 422 message rõ thay vì một shape response không ai đọc nữa.
        raise HTTPException(422, "Cần balls (thế bàn full rack) — đường 3 bi "
                                 "cũ đã gỡ 31/07/2026, xem nhánh v1-full")
    return await _recommend_full(req)


# ------------------------------------------------------- full rack (API v2)
# Toàn bộ phần DỰNG response (ShotFullOut/_full_response) chuyển sang
# app/engine.py (bàn giao 14) — worker cần nó mà không được kéo theo FastAPI
# app. Ở đây chỉ còn transport (inprocess/queue) + logging DB.

ENGINE_VER = "zoneplanner-v2"   # hằng số ghi vào log recommendations (04/08)


def _log_recommend(req: RecommendRequest, resp: dict,
                   latency_ms: int) -> None:
    """Ghi 1 row `recommendations` — BEST-EFFORT, không bao giờ ném.

    DB lỗi/tắt thì response vẫn đi ra bình thường (BRIEF 04/08: KHÔNG BAO GIỜ
    để DB làm chết `/api/recommend`). Vì thế nuốt MỌI exception, chỉ warning.

    `resp` là dict JSON-thuần của `run_recommend` — CÙNG shape ở cả hai mode
    (bàn giao 14), nên logging cũng là một code path duy nhất.

    `trajectories` không vào log, cố ý: hàng trăm điểm toạ độ mỗi cú chỉ để
    vẽ, còn phân tích thì tham số cú + outcome là đủ — sim tất định nên cần
    lại quỹ đạo cứ đưa 4 tham số cho `/api/trajectory`.
    """
    if not db.enabled():
        return
    try:
        top = resp["shots"][0] if resp["shots"] else None
        with db.session() as s:
            # Phiên bàn (bàn giao 12): session sống → row mang tenant_id của
            # quán. Session rác/hết hạn KHÔNG làm mất row — log nặc danh vẫn
            # có giá trị, chỉ warning để dev thấy client gửi phiên hỏng.
            tenant_id = None
            if req.session_id:
                gs = s.get(db.GuestSession, req.session_id)
                if gs is not None and gs.expires_at > db.utcnow():
                    tenant_id = gs.tenant_id
                else:
                    print(f"[poolcoach] WARNING: session_id la/het han trong "
                          f"recommend -- bo qua, khong gan tenant "
                          f"({req.session_id!r})", flush=True)
            s.add(db.Recommendation(
                tenant_id=tenant_id,
                scan_id=req.scan_id,
                balls={bid: {"x": p.x, "y": p.y}
                       for bid, p in req.balls.items()},
                edited=req.edited,
                engine_ver=ENGINE_VER,
                shot=({k: v for k, v in top.items() if k != "trajectories"}
                      if top is not None else None),
                alternatives=[{k: v for k, v in s2.items()
                               if k != "trajectories"}
                              for s2 in resp["shots"][1:]],
                latency_ms=latency_ms,
            ))
    except Exception as e:  # noqa: BLE001 — best-effort đúng nghĩa đen
        print(f"[poolcoach] WARNING: khong ghi duoc log recommend: "
              f"{type(e).__name__}: {e}", flush=True)


async def _submit_queue(loop, payload: dict) -> dict:
    """Một job qua Redis → dict response. BLPOP chạy trong executor để không
    block event loop (BRIEF 2.4 — nhất quán với cách engine được gọi).

    Queue lỗi (Redis chết / worker không trả lời) → 503 message tiếng Việt
    rõ + log, KHÔNG âm thầm fallback in-process — che chết worker tệ hơn lỗi
    rõ (BRIEF 1.4).

    Fast-fail (bàn giao 20, đối xứng /api/scan bàn giao 19): vắng heartbeat
    engine worker → 503 NGAY với message riêng, KHÔNG đẩy job rồi bắt client
    chờ trọn timeout 30s. Chỉ chặn ca "worker KHÔNG chạy"; worker chết SAU
    enqueue vẫn đi đường QueueTimeout như cũ (heartbeat TTL 15s — cửa sổ
    ngắn worker vừa chết vẫn timeout, chấp nhận, là thiết kế).
    `worker_alive_or_raise` ném QueueDown khi Redis chết — rơi đúng nhánh
    message "Redis" bên dưới, không đổ oan cho worker.
    """
    try:
        alive = await loop.run_in_executor(None, jobqueue.worker_alive_or_raise)
        if not alive:
            raise HTTPException(503, "Engine worker không chạy (không thấy "
                                "heartbeat) — mode queue cần engine worker "
                                "đang bật (scripts/engine_worker.py). Thử "
                                "lại sau khi bật worker.")
        reply = await loop.run_in_executor(
            None, functools.partial(jobqueue.submit, payload))
    except jobqueue.QueueTimeout as e:
        print(f"[poolcoach] LOI queue: het han cho ket qua ({e})", flush=True)
        raise HTTPException(503, f"Engine không trả lời trong "
                            f"{jobqueue.timeout_s():g}s — worker có thể đã "
                            f"chết hoặc đang kẹt. Thử lại sau; nếu lặp lại, "
                            f"kiểm tra process engine_worker.") from e
    except jobqueue.QueueDown as e:
        print(f"[poolcoach] LOI queue: Redis khong noi chuyen duoc ({e})",
              flush=True)
        raise HTTPException(503, "Không kết nối được hàng đợi engine (Redis) "
                            "— kiểm tra Redis và worker rồi thử lại.") from e

    if reply.get("ok"):
        return reply["result"]
    if reply.get("error") == "validation":
        # Cùng message validate_full nguyên văn — 422 hai mode trùng từng byte
        raise HTTPException(422, reply.get("message", ""))
    print(f"[poolcoach] LOI worker: {reply.get('message')}", flush=True)
    raise HTTPException(500, "Engine gặp lỗi khi tính cú: "
                        f"{reply.get('message', 'không rõ')}")


async def _recommend_full(req: RecommendRequest) -> dict:
    """Full rack — MỘT engine duy nhất từ 31/07/2026: ZonePlanner V2, từ bàn
    giao 14 qua hàm chung `run_recommend` (app/engine.py) để in-process và
    worker CÙNG code path; ở đây chỉ chọn transport.

    ``req.engine``/``req.topk`` BỎ QUA CÓ CHỦ ĐÍCH (G5.5): request cũ còn gửi
    "oracle"/"hybrid" vẫn 200 và nhận cú của zone V2 — app không còn engine
    nào khác để chiều theo, và 4xx hoá một client cũ chỉ vì nó gửi thừa một
    trường là làm phiền đúng người đang không làm gì sai. `search.engine`
    trong response khai "zone" — sự thật về đường đã chạy.

    Mode queue KHÔNG cầm `_search_lock`: env của process API không bị đụng
    tới (worker có env riêng và tự serial hoá bằng vòng BRPOP đơn).
    """
    payload = {"balls": {bid: {"x": p.x, "y": p.y}
                         for bid, p in req.balls.items()},
               "alternatives": req.alternatives}
    loop = asyncio.get_running_loop()

    t0 = time.perf_counter()
    if jobqueue.mode() == "queue":
        resp = await _submit_queue(loop, payload)
    else:
        call = functools.partial(run_recommend, payload, state.env_h)
        try:
            async with _search_lock:
                resp = await loop.run_in_executor(None, call)
        except ValueError as e:   # validate_full: ngoài bàn / chồng bi / id lạ
            raise HTTPException(422, str(e)) from e
    # round chứ không int-truncate, và kẹp >= 1: engine thật đo hàng nghìn ms,
    # nhưng engine mock trong test xong dưới 1 ms — 0 sẽ trông như "không đo".
    latency_ms = max(1, round((time.perf_counter() - t0) * 1000))

    # TV tại bàn (bàn giao 15, Bước 3): push best-effort xuống TV đã ghép.
    # Ở API LAYER sau khi có kết quả, KHÔNG trong run_recommend/worker — một
    # code path cho CẢ HAI transport (bất biến bàn giao 14, bẫy 27/07).
    # `push_recommend` tự nuốt mọi lỗi + chặn trần thời gian: không TV /
    # TV rớt thì response vẫn y nguyên, không chậm đi.
    if req.session_id:
        await displays.hub.push_recommend(req.session_id, payload["balls"],
                                          resp)

    # Ghi log qua executor: write DB là blocking, không để nó chặn event loop.
    # Best-effort — `_log_recommend` tự nuốt lỗi, response không phụ thuộc nó.
    await loop.run_in_executor(
        None, functools.partial(_log_recommend, req, resp, latency_ms))
    return resp


# ------------------------------------------------- /api/scan (05/08, F2)

# Ảnh FE ĐÃ resize (max dim ~1280) trước khi upload — payload bình thường
# vài trăm KB. Trần này chỉ chặn client lạ đẩy ảnh gốc chục MB làm nghẽn
# Redis (BRIEF "Nếu bí": payload >10MB là chuyện phải dừng lại hỏi — chặn
# từ cửa để nó không bao giờ thành chuyện).
SCAN_MAX_UPLOAD_B = 10 * 1024 * 1024


# exclude_unset: trường BallID (number/number_conf/wb — bàn giao 22) chỉ
# xuất hiện khi WORKER thật sự gửi. Worker cũ → balls giữ nguyên 4 key cũ,
# assertion test cũ sống nguyên; worker mới gửi number=null → null đi qua
# NGUYÊN VẸN (FE cần phân biệt "không gán được" để fallback).
@app.post("/api/scan", response_model=ScanOut, response_model_exclude_unset=True)
async def api_scan(image: UploadFile = File(...), corners: str = Form(...)):
    """Upload ảnh bàn + 4 góc pixel → CV worker detect → thế bi (m).

    CHỈ sống ở mode queue: CV chạy trên venv CUDA riêng (bàn giao 17), app
    không import torch nên không có đường in-process để fallback — degraded
    là 503 nói thẳng, cùng nếp "che chết worker tệ hơn lỗi rõ" (bàn giao 14).
    Từ bàn giao 19: vắng heartbeat CV worker → 503 NGAY trước khi enqueue
    (fast-fail), không bắt client chờ trọn timeout 30s.

    `corners` là JSON ``[[x,y]×4]`` pixel TRÊN ẢNH GỬI LÊN, thứ tự như
    ``poolcoach_cv.table_corners``: đi vòng quanh bàn, 2 góc đầu là một băng
    ngắn — (0,0) → (W,0) → (W,L) → (0,L). Route chỉ kiểm shape; nghĩa hình
    học (suy biến, thẳng hàng) do worker phán — message 422 nguyên văn.

    `scan_id` = uuid chuỗi mờ (như session_id): KHÔNG bảng scans, KHÔNG
    migration (chờ auth) — FE gửi lại trong /api/recommend để ghi log.
    """
    import base64
    import json as _json
    import uuid as _uuid

    if not jobqueue.cv_enabled():
        raise HTTPException(503, "Tính năng scan ảnh cần CV worker — server "
                                 "này chưa bật hàng đợi (REDIS_URL) nên chưa "
                                 "scan được. Thế bi vẫn đặt tay được như thường.")
    try:
        pts = _json.loads(corners)
        ok_shape = (isinstance(pts, list) and len(pts) == 4
                    and all(isinstance(p, list) and len(p) == 2
                            and all(isinstance(v, (int, float)) for v in p)
                            for p in pts))
    except ValueError:
        ok_shape = False
    if not ok_shape:
        raise HTTPException(422, "corners phải là JSON 4 điểm [x, y] pixel "
                                 "— chấm đủ 4 góc bàn theo thứ tự hướng dẫn.")
    raw = await image.read()
    if not raw:
        raise HTTPException(422, "Ảnh rỗng — chọn lại file ảnh bàn (JPEG/PNG).")
    if len(raw) > SCAN_MAX_UPLOAD_B:
        raise HTTPException(422, "Ảnh quá lớn (>10 MB) — app tự thu nhỏ ảnh "
                                 "trước khi gửi, tải lại trang rồi thử lại.")

    payload = {"image_b64": base64.b64encode(raw).decode("ascii"),
               "corners": pts}
    loop = asyncio.get_running_loop()
    t0 = time.perf_counter()
    try:
        # Fast-fail (bàn giao 19): vắng heartbeat CV worker → 503 NGAY với
        # message riêng, KHÔNG đẩy job rồi bắt client chờ trọn timeout 30s.
        # Chỉ chặn ca "worker KHÔNG chạy"; worker chết SAU enqueue vẫn đi
        # đường QueueTimeout như cũ (heartbeat TTL 15s — cửa sổ ngắn worker
        # vừa chết vẫn timeout, chấp nhận, là thiết kế). Check đặt SAU mọi
        # validate input: lỗi client là 422 bất kể worker sống chết.
        alive = await loop.run_in_executor(
            None, functools.partial(jobqueue.worker_alive_or_raise,
                                    key=jobqueue.SCAN_HEARTBEAT_KEY))
        if not alive:
            raise HTTPException(503, "CV worker không chạy (không thấy "
                                "heartbeat) — tính năng scan cần CV worker "
                                "đang bật (scripts/cv_worker.py). Thế bi vẫn "
                                "đặt tay được như thường.")
        reply = await loop.run_in_executor(
            None, functools.partial(jobqueue.submit, payload,
                                    jobs_key=jobqueue.SCAN_JOBS_KEY))
    except jobqueue.QueueTimeout as e:
        print(f"[poolcoach] LOI scan: het han cho CV worker ({e})", flush=True)
        raise HTTPException(503, f"CV worker không trả lời trong "
                            f"{jobqueue.timeout_s():g}s — tính năng scan cần "
                            f"CV worker đang chạy (scripts/cv_worker.py). "
                            f"Thử lại sau khi bật worker.") from e
    except jobqueue.QueueDown as e:
        print(f"[poolcoach] LOI scan: Redis khong noi chuyen duoc ({e})",
              flush=True)
        raise HTTPException(503, "Không kết nối được hàng đợi CV (Redis) — "
                            "tính năng scan cần Redis + CV worker sống. "
                            "Kiểm tra rồi thử lại.") from e

    if not reply.get("ok"):
        if reply.get("error") == "validation":
            raise HTTPException(422, reply.get("message", ""))
        print(f"[poolcoach] LOI cv_worker: {reply.get('message')}", flush=True)
        raise HTTPException(500, "CV worker gặp lỗi khi scan ảnh: "
                            f"{reply.get('message', 'không rõ')}")

    balls = reply["result"]["balls"]
    print(f"[poolcoach] scan xong sau "
          f"{(time.perf_counter() - t0) * 1000:.0f}ms -- {len(balls)} bi, "
          f"dropped={reply['result'].get('dropped')}", flush=True)
    return ScanOut(scan_id=_uuid.uuid4().hex,
                   balls=[ScanBall(**b) for b in balls])


# ------------------------------------- Broadcast Analyzer — helper chung
# (Endpoint 1-cú POST/GET /api/analyze của BG24 GỠ ở A2b — video 1 cú là
# trường hợp con của luồng đa cú /api/analyzer/*. Queue "analyze" + status
# key trong jobqueue GIỮ NGUYÊN: job per cú của luồng đa cú vẫn đi đường đó.)


def _probe_duration_s(path: Path) -> float | None:
    """Thời lượng video qua ffprobe (binary hệ thống — máy demo đã có ffmpeg
    9.0, BRIEF #8). Ba kết cục, cố ý tách bạch:

    - trả float: đo được — route enforce trần ngay tại POST (4xx tử tế).
    - trả None: máy KHÔNG có ffprobe — không đo được ở tầng API, worker vẫn
      enforce lần hai bằng cv2; POST cho qua, không 500.
    - ném ValueError: ffprobe CÓ mà không đọc được file — video hỏng, 422.
    """
    import shutil
    import subprocess

    exe = shutil.which("ffprobe")
    if exe is None:
        return None
    try:
        out = subprocess.run(
            [exe, "-v", "error", "-show_entries", "format=duration",
             "-of", "default=noprint_wrappers=1:nokey=1", str(path)],
            capture_output=True, text=True, timeout=15)
    except subprocess.TimeoutExpired as e:
        raise ValueError("ffprobe không trả lời trong 15s") from e
    raw = (out.stdout or "").strip()
    if out.returncode != 0 or not raw:
        raise ValueError((out.stderr or "không đọc được định dạng").strip()
                         .splitlines()[-1][:200])
    try:
        return float(raw)
    except ValueError as e:
        raise ValueError(f"ffprobe trả duration lạ: {raw!r}") from e


# ------------------------------------- /api/analyzer/* (13/08, lát A1)
# Analyzer đa cú: upload VIDEO (1..N cú, trần 20 phút) + 4 pocket pixel →
# job segment (cắt cú, queue pc:jobs:segment) → worker sinh N job analyze
# per cú NGUYÊN TRẠNG → JSON per cú ghi FILE ngay khi có (nếp BG31 sau vụ
# mất kết quả Redis TTL BG29b). GET đọc từ FILE làm nguồn sự thật, Redis
# chỉ bù tiến độ; DB ghi danh sách best-effort (nếp recommendations).


def _analyzer_video_dir(video_id: str):
    """Bọc astore.video_dir để test monkeypatch trỏ tmp_path (nếp
    _analyze_tmp_dir)."""
    return astore.video_dir(video_id)


def _parse_corners4(raw: str):
    """4 điểm [x, y] pixel từ JSON string — None khi hỏng (route trả 422).
    Cùng luật với /api/analyze; helper riêng để KHÔNG đụng code path cũ."""
    import json as _json

    try:
        pts = _json.loads(raw)
    except ValueError:
        return None
    ok = (isinstance(pts, list) and len(pts) == 4
          and all(isinstance(p, list) and len(p) == 2
                  and all(isinstance(v, (int, float)) for v in p)
                  for p in pts))
    return pts if ok else None


def _valid_video_id(video_id: str) -> bool:
    """uuid4 hex do chính route sinh — chặn mọi id lạ TRƯỚC khi ghép vào
    đường dẫn file (không có chuyện ../ đi lạc thư mục)."""
    return (len(video_id) == 32
            and all(c in "0123456789abcdef" for c in video_id))


@app.get("/api/analyzer/demo", response_model=AnalyzerDemoOut)
def api_analyzer_demo():
    """Id video mẫu ĐÃ PHÂN TÍCH SẴN đóng gói trong repo — FE mở tab là nạp.

    Không cần queue, không cần CV worker, không cần GPU: kết quả nằm sẵn
    trên đĩa và GET /shots đọc từ file. Đây là đường để "mở link thấy ngay
    danh sách cú" (BRIEF 14/08 việc B, mục tiêu demo gửi thầy).
    """
    ids = astore.demo_ids()
    return AnalyzerDemoOut(id=ids[0] if ids else None)


SUGGEST_MAX_UPLOAD_B = 10 * 1024 * 1024


@app.post("/api/analyzer/suggest-corners", response_model=AnalyzerCornersOut,
          response_model_exclude_none=True)
async def api_analyzer_suggest_corners(image: UploadFile = File(...)):
    """Frame đầu (JPEG/PNG do FE trích từ chính video) → 4 góc bàn ĐỀ XUẤT.

    Chạy IN-PROCESS kể cả mode queue — quyết định, không phải bỏ sót: đây là
    một lần dò màu thuần cv2/numpy vài chục ms trên MỘT ảnh, không đụng
    torch/YOLO, nên không có lý do đẩy qua CV worker (nếp /api/trajectory
    giữ in-process trong khi /api/recommend đi worker).

    LUÔN 200 khi ảnh đọc được: "không đề xuất" là một CÂU TRẢ LỜI hợp lệ
    (`ok=false` + lý do), không phải lỗi — FE quay về luồng chấm tay y hệt
    trước đây. Chỉ input rác (ảnh rỗng/quá to/không giải mã được) mới 422.
    Van sanity của đề xuất là van BG32 sẵn có trong camera.py, gọi nguyên
    qua ``autocorner.suggest_corners`` — không nới, không bản sao.
    """
    raw = await image.read()
    if not raw:
        raise HTTPException(422, "Ảnh rỗng — chọn lại video.")
    if len(raw) > SUGGEST_MAX_UPLOAD_B:
        raise HTTPException(422, "Ảnh frame quá lớn (>10 MB).")

    def _work() -> dict:
        import numpy as np

        try:
            import cv2
        except ImportError:
            # Bản app không cài cv2 (tầng requirements-cv) — nói thẳng, FE
            # về luồng chấm tay. KHÔNG 500: thiếu tính năng phụ ≠ hỏng app.
            return {"ok": False, "reason": "Server này chưa cài cv2 nên "
                                           "không gợi ý được góc bàn — chấm "
                                           "tay 4 pocket như thường."}
        from poolcoach_cv.autocorner import suggest_corners

        img = cv2.imdecode(np.frombuffer(raw, np.uint8), cv2.IMREAD_COLOR)
        if img is None:
            return None
        res = suggest_corners(img)
        return {"ok": res["ok"], "corners": res["corners"],
                "reason": res["reason"], "camera": res["camera"],
                "rails": res["rails"]}

    loop = asyncio.get_running_loop()
    out = await loop.run_in_executor(None, _work)
    if out is None:
        raise HTTPException(422, "Không giải mã được frame gửi lên — cần "
                                 "JPEG hoặc PNG.")
    return AnalyzerCornersOut(**out)


@app.post("/api/analyzer/videos", response_model=AnalyzerVideoCreateOut)
async def api_analyzer_upload(video: UploadFile = File(...),
                              corners: str = Form(...)):
    """Upload video 1..N cú + 4 pocket pixel (chấm MỘT lần trên frame đầu,
    mọi cú dùng chung homography — design §2) → job segment async → id.

    Nếp /api/analyze giữ nguyên từng bước: chỉ sống ở mode queue, validate
    input trước heartbeat check, fast-fail khi vắng CV worker, status
    "queued" ghi TRƯỚC enqueue (race đã bắt được bằng test ở BG24). KHÁC:
    video lưu vào thư mục kết quả data/analyzer/{id}/ (không phải temp) —
    kết quả per cú phải sống lâu hơn phiên, còn video gốc worker sẽ tự xoá
    sau khi scan xong (bản quyền, không đọng).
    """
    import uuid as _uuid

    if not jobqueue.cv_enabled():
        raise HTTPException(503, "Tính năng phân tích video cần CV worker — "
                                 "server này chưa bật hàng đợi (REDIS_URL) "
                                 "nên chưa phân tích được.")
    pts = _parse_corners4(corners)
    if pts is None:
        raise HTTPException(422, "corners phải là JSON 4 điểm [x, y] pixel "
                                 "— chấm đủ 4 pocket góc theo thứ tự hướng dẫn.")
    suffix = Path(video.filename or "").suffix.lower()
    if suffix not in astore.VIDEO_SUFFIXES:
        raise HTTPException(422, "Cần file MP4 hoặc MOV (video 1..nhiều cú, "
                                 f"≤ {astore.VIDEO_MAX_DUR_S / 60:g} phút) — "
                                 f"nhận {suffix or 'file không đuôi'}.")

    video_id = _uuid.uuid4().hex
    vdir = _analyzer_video_dir(video_id)
    vdir.mkdir(parents=True, exist_ok=True)
    path = vdir / f"video{suffix}"
    total = 0
    try:
        with open(path, "wb") as f:
            while chunk := await video.read(1 << 20):
                total += len(chunk)
                if total > astore.VIDEO_MAX_UPLOAD_B:
                    raise HTTPException(
                        422, "Video quá lớn (>1 GB) — cắt đoạn trận cần "
                             "phân tích (một rack ≤ 20 phút) rồi thử lại.")
                f.write(chunk)
        if total == 0:
            raise HTTPException(422, "Video rỗng — chọn lại file MP4/MOV.")
        try:
            dur = _probe_duration_s(path)
        except ValueError as e:
            raise HTTPException(422, f"Không đọc được video ({e}) — cần "
                                     f"MP4/MOV H.264 lành lặn.") from e
        if dur is not None and dur > astore.VIDEO_MAX_DUR_S + 0.5:
            raise HTTPException(
                422, f"Video dài {dur / 60:.1f} phút — giới hạn "
                     f"{astore.VIDEO_MAX_DUR_S / 60:g} phút cho một lần "
                     f"quét: cắt đoạn trận cần phân tích rồi thử lại.")

        loop = asyncio.get_running_loop()
        try:
            alive = await loop.run_in_executor(
                None, functools.partial(jobqueue.worker_alive_or_raise,
                                        key=jobqueue.SCAN_HEARTBEAT_KEY))
            if not alive:
                raise HTTPException(503, "CV worker không chạy (không thấy "
                                    "heartbeat) — tính năng phân tích video "
                                    "cần CV worker đang bật "
                                    "(scripts/cv_worker.py).")
            payload = {"video_id": video_id, "video_path": str(path),
                       "corners": pts, "filename": video.filename,
                       "out_dir": str(vdir)}
            await loop.run_in_executor(
                None, functools.partial(jobqueue.set_video_status, video_id,
                                        {"status": "queued",
                                         "progress": 0.0}))
            await loop.run_in_executor(
                None, functools.partial(jobqueue.enqueue, payload,
                                        jobs_key=jobqueue.SEGMENT_JOBS_KEY,
                                        job_id=video_id))
        except jobqueue.QueueDown as e:
            print(f"[poolcoach] LOI analyzer upload: Redis khong noi chuyen "
                  f"duoc ({e})", flush=True)
            raise HTTPException(503, "Không kết nối được hàng đợi CV (Redis) "
                                "— kiểm tra Redis + CV worker rồi thử lại.") from e
    except BaseException:
        # không thành job thì dọn cả thư mục — video bản quyền không đọng
        path.unlink(missing_ok=True)
        try:
            vdir.rmdir()
        except OSError:
            pass
        raise

    await asyncio.get_running_loop().run_in_executor(
        None, _log_analyzer_video, video_id, video.filename)
    print(f"[poolcoach] segment job {video_id} vao queue "
          f"({total / 1e6:.1f}MB, {video.filename})", flush=True)
    return AnalyzerVideoCreateOut(id=video_id)


def _log_analyzer_video(video_id: str, filename: str | None) -> None:
    """Row `analyzer_videos` — BEST-EFFORT như _log_recommend, không ném."""
    if not db.enabled():
        return
    try:
        with db.session() as s:
            s.add(db.AnalyzerVideo(id=video_id, filename=filename,
                                   status="queued"))
    except Exception as e:  # noqa: BLE001 — best-effort đúng nghĩa đen
        print(f"[poolcoach] WARNING: khong ghi duoc analyzer_videos: "
              f"{type(e).__name__}: {e}", flush=True)


def _shot_summary(item: dict, result: dict) -> None:
    """Điền tóm tắt tham số của MỘT cú từ JSON kết quả analyze vào dict
    hàng bảng — chỉ trích, không tính lại số nào."""
    m = result.get("metrics") or {}
    item.update(
        status="done",
        v0_mps=m.get("v0_mps"),
        phi_deg=m.get("phi_deg"),
        spin_class=result.get("spin_class"),
        spin_confidence=result.get("spin_confidence"),
        n_collisions=m.get("n_collisions"),
        n_warnings=len(result.get("warnings") or []),
        # van A2 phần 1 — chỉ trích từ JSON, không tính lại; kết quả trước
        # A2 không có key thì hàng không mọc cờ (exclude_none)
        suspect_not_shot=result.get("suspect_not_shot"),
    )
    # RMSE resim (A2 phần 2): cột bảng lấy bộ shotnet (model đang chạy app);
    # vắng shotnet thì fallback analytic NHƯNG dán nhãn qua rmse_set —
    # "ghi rõ bộ nào, đừng trộn" (BRIEF A2 bối cảnh 3)
    rsets = (result.get("resim") or {}).get("sets") or {}
    for set_name in ("shotnet", "analytic"):
        rmse = (rsets.get(set_name) or {}).get("rmse_mm")
        if rmse is not None:
            item["rmse_mm"] = rmse
            item["rmse_set"] = set_name
            break
    sn = result.get("shotnet")
    if sn:
        item["shotnet"] = AnalyzerShotNetBrief(
            v0_cue_mps=sn["v0_cue_mps"], phi_deg=sn["phi_deg"],
            a=sn["a"], b=sn["b"], spin_vert=sn["spin_vert"],
            spin_side=sn["spin_side"], confidence=sn["confidence"],
            confidence_raw=sn.get("confidence_raw"),
            det_density=sn.get("det_density"))


def _sync_analyzer_db(video_id: str, video_status: str, n_shots: int | None,
                      items: list[AnalyzerShotItem]) -> None:
    """Đồng bộ danh sách cú vào DB — BEST-EFFORT, chỉ ghi khi có THAY ĐỔI
    (GET bị FE poll mỗi giây, không được thành máy spam UPDATE). Nguồn sự
    thật vẫn là file; DB là sổ ghi danh sách để tra cứu sau (BRIEF A1)."""
    if not db.enabled():
        return
    try:
        with db.session() as s:
            v = s.get(db.AnalyzerVideo, video_id)
            if v is not None and (v.status != video_status
                                  or v.n_shots != n_shots):
                v.status = video_status
                v.n_shots = n_shots
                v.updated_at = db.utcnow()
            rows = {r.shot_idx: r for r in
                    s.query(db.AnalyzerShot).filter_by(video_id=video_id)}
            for it in items:
                brief = it.model_dump(exclude_none=True,
                                      exclude={"thumb_url", "result_url",
                                               "progress", "stage"})
                row = rows.get(it.idx)
                if row is None:
                    s.add(db.AnalyzerShot(
                        video_id=video_id, shot_idx=it.idx,
                        t_start_s=it.t_start_s, t_end_s=it.t_end_s,
                        status=it.status, reason=it.reason, result=brief))
                elif row.status != it.status:
                    row.status = it.status
                    row.reason = it.reason
                    row.result = brief
                    row.updated_at = db.utcnow()
    except Exception as e:  # noqa: BLE001 — best-effort đúng nghĩa đen
        print(f"[poolcoach] WARNING: khong sync duoc analyzer_shots: "
              f"{type(e).__name__}: {e}", flush=True)


def _collect_shots(video: str) -> AnalyzerShotsOut:
    """Dựng body danh sách cú của MỘT video — nguồn ghép theo thứ tự tin
    cậy: manifest + JSON per cú trên ĐĨA (bền, nguồn sự thật) → status key
    Redis (tiến độ job đang chạy) → "unknown" khi cả hai vắng (worker chết
    giữa chừng — nói thẳng, không đoán).

    Tách từ GET /shots (A2 phần 5) để export CSV/JSON dùng CHUNG — hai
    đường xuất và bảng phải cùng một số (gate A2.3/A2.5), không dựng lại.

    ĐỌC ĐĨA TRƯỚC (14/08, việc B): video ĐÃ có manifest trên đĩa thì phục vụ
    được kể cả khi không có hàng đợi nào — đúng điều module này vẫn tuyên bố
    ("Redis chỉ là kênh tiến độ, đĩa mới là nguồn sự thật", BG29b/BG31), và
    là thứ giữ cho video mẫu bundled mở được ngay cả khi tầng CV chết. Chỉ
    khi CHƯA có manifest mới cần hàng đợi để nói video đang ở đâu — không có
    thì 503 nguyên văn như cũ.
    """
    if not _valid_video_id(video):
        raise HTTPException(422, "video phải là id 32 ký tự hex do "
                                 "/api/analyzer/videos trả về.")
    vdir = _analyzer_video_dir(video)
    manifest = astore.read_json(astore.manifest_path(vdir))
    live = jobqueue.cv_enabled()
    if manifest is None and not live:
        raise HTTPException(503, "Tính năng phân tích video cần CV worker — "
                                 "server này chưa bật hàng đợi (REDIS_URL).")
    vst = None
    if live:
        try:
            vst = jobqueue.get_video_status(video)
        except jobqueue.QueueDown as e:
            if manifest is None:
                raise HTTPException(503, "Không kết nối được hàng đợi CV "
                                    "(Redis) — kiểm tra Redis rồi thử "
                                    "lại.") from e
            live = False        # có file rồi: Redis chết không cản đọc kết quả

    if manifest is None:
        # chưa có file: video đang xếp hàng/đang scan/lỗi — theo Redis
        if vst is None:
            if vdir.exists():
                info = AnalyzerVideoInfo(
                    id=video, status="error",
                    message="Mất dấu job quét video (không còn trạng thái "
                            "trong hàng đợi, chưa có danh sách cú trên đĩa) "
                            "— worker có chết giữa chừng không?")
            else:
                raise HTTPException(404, "Không thấy video này — id sai "
                                         "hoặc chưa từng upload.")
        else:
            info = AnalyzerVideoInfo(
                id=video, status=vst.get("status", "queued"),
                progress=vst.get("progress"), stage=vst.get("stage"),
                message=vst.get("message"))
        return AnalyzerShotsOut(video=info, shots=[], n_done=0, n_total=0)

    if manifest.get("status") == "error":
        info = AnalyzerVideoInfo(id=video, status="error",
                                 message=manifest.get("message"))
        _sync_analyzer_db(video, "error", None, [])
        return AnalyzerShotsOut(video=info, shots=[], n_done=0, n_total=0)

    shots_out: list[AnalyzerShotItem] = []
    n_done = 0
    for sh in manifest.get("shots", []):
        idx = int(sh["idx"])
        item = {
            "idx": idx,
            "t_start_s": sh["t_start_s"], "t_end_s": sh["t_end_s"],
            "t_onset_s": sh.get("t_onset_s"),
            "t_settle_s": sh.get("t_settle_s"),
            "status": "queued",
            "reason": sh.get("reason"),
            "thumb_url": (f"/api/analyzer/videos/{video}/shots/{idx}/thumb"
                          if sh.get("thumb") else None),
        }
        if sh.get("status") == "error":
            # cú segmentation đã kết án (mất cảnh, quá dài...) — không có
            # job analyze nào cho nó
            item["status"] = "error"
        else:
            item["result_url"] = (f"/api/analyzer/videos/{video}"
                                  f"/shots/{idx}/result")
            result = astore.read_json(astore.shot_json_path(vdir, idx))
            if result is None:
                # chưa có file — hỏi Redis tiến độ job per cú (chỉ khi còn
                # hàng đợi: đọc đĩa thuần thì không có ai để hỏi)
                rs = None
                if live:
                    try:
                        rs = jobqueue.get_analyze_status(
                            astore.shot_analyze_id(video, idx))
                    except jobqueue.QueueDown:
                        rs = None
                if rs is None:
                    item["status"] = "unknown"
                    item["reason"] = ("Không còn dấu vết job phân tích cú "
                                      "này (không file kết quả, hết hạn "
                                      "hàng đợi) — worker có chết giữa "
                                      "chừng không?")
                elif rs.get("status") == "running":
                    item["status"] = "running"
                    item["progress"] = rs.get("progress")
                    item["stage"] = rs.get("stage")
                elif rs.get("status") == "error":
                    item["status"] = "error"
                    item["reason"] = rs.get("message")
                elif rs.get("status") == "done" and rs.get("result"):
                    # file chưa kịp/không ghi được nhưng Redis còn — vẫn
                    # phục vụ số (đừng bắt người dùng đợi một file hỏng)
                    _shot_summary(item, rs["result"])
            elif result.get("error"):
                item["status"] = "error"
                item["reason"] = result.get("message")
            else:
                _shot_summary(item, result)
        # overlay per cú (A2 phần 3): file trên đĩa là nguồn sự thật —
        # có file là có nút phát/tải, kể cả cú bị van gắn cờ
        if astore.overlay_path(vdir, idx).exists():
            item["overlay_url"] = (f"/api/analyzer/videos/{video}"
                                   f"/shots/{idx}/overlay")
        if item["status"] == "done":
            n_done += 1
        shots_out.append(AnalyzerShotItem(**item))

    info = AnalyzerVideoInfo(
        id=video, status="done",
        n_shots=len(shots_out),
        duration_s=manifest.get("duration_s"),
        warnings=manifest.get("warnings") or None)
    _sync_analyzer_db(video, "done", len(shots_out), shots_out)
    return AnalyzerShotsOut(video=info, shots=shots_out,
                            n_done=n_done, n_total=len(shots_out))


@app.get("/api/analyzer/shots", response_model=AnalyzerShotsOut,
         response_model_exclude_none=True)
def api_analyzer_shots(video: str):
    """Danh sách cú của MỘT video — hiện DẦN theo tiến độ (FE poll).
    Sync def (threadpool) vì lệnh Redis blocking — nếp api_analyze_status.
    Toàn bộ logic ở _collect_shots (dùng chung với export A2 phần 5)."""
    return _collect_shots(video)


_ANV_STATUS_VN = {"done": "xong", "running": "đang phân tích",
                  "queued": "chờ phân tích", "unknown": "mất dấu",
                  "error": "không phân tích được"}


@app.get("/api/analyzer/videos/{video_id}/export.csv")
def api_analyzer_export_csv(video_id: str):
    """Xuất CSV toàn danh sách cú (A2 phần 5) — đủ cột của bảng + cờ van
    + RMSE, mỗi cú một hàng KỂ CẢ cú lỗi/cú bị van gắn cờ (không lọc).
    Ghi **UTF-8 BOM** (utf-8-sig) — tiếng Việt mở trong Excel không vỡ
    font (BRIEF A2 phần 5); CRLF cho Excel. Cùng nguồn _collect_shots với
    bảng — hai nơi không thể lệch nhau."""
    import csv
    import io

    from fastapi.responses import Response

    out = _collect_shots(video_id)
    buf = io.StringIO()
    w = csv.writer(buf, lineterminator="\r\n")
    w.writerow(["# cú", "t_onset_s", "t_start_s", "t_end_s", "trạng thái",
                "V0 gậy model (m/s)", "φ model (°)", "a", "b",
                "spin model", "tin cậy model",
                "V0 bi analytic (m/s)", "φ analytic (°)", "spin analytic",
                "tin cậy analytic", "RMSE resim (mm)", "bộ params RMSE",
                "nghi không phải cú đánh", "lý do van",
                "số va chạm", "số cảnh báo", "lý do lỗi"])
    for s in out.shots:
        sn = s.shotnet
        sus = s.suspect_not_shot
        w.writerow([
            s.idx, s.t_onset_s, s.t_start_s, s.t_end_s,
            _ANV_STATUS_VN.get(s.status, s.status),
            sn.v0_cue_mps if sn else None,
            sn.phi_deg if sn else None,
            sn.a if sn else None, sn.b if sn else None,
            f"{sn.spin_vert}+{sn.spin_side}" if sn else None,
            sn.confidence if sn else None,
            s.v0_mps, s.phi_deg, s.spin_class, s.spin_confidence,
            s.rmse_mm, s.rmse_set,
            (None if sus is None else ("có" if sus.flagged else "không")),
            "; ".join(sus.reasons) if sus and sus.reasons else None,
            s.n_collisions, s.n_warnings, s.reason,
        ])
    return Response(
        content=buf.getvalue().encode("utf-8-sig"),
        media_type="text/csv; charset=utf-8",
        headers={"Content-Disposition":
                 f'attachment; filename="analyzer_{video_id[:8]}_shots.csv"'})


@app.get("/api/analyzer/videos/{video_id}/export.json")
def api_analyzer_export_json(video_id: str):
    """Xuất JSON toàn danh sách cú (A2 phần 5) — đúng body của GET /shots
    (cùng _collect_shots), trả dạng attachment để lưu file."""
    from fastapi.responses import JSONResponse

    out = _collect_shots(video_id)
    return JSONResponse(
        out.model_dump(exclude_none=True),
        headers={"Content-Disposition":
                 f'attachment; filename="analyzer_{video_id[:8]}_shots.json"'})


@app.get("/api/analyzer/videos/{video_id}/shots/{idx}/result")
def api_analyzer_shot_result(video_id: str, idx: int):
    """JSON kết quả THÔ của một cú — trang chi tiết lát A1 (design §6:
    "click ra JSON thô là đủ"; trang đẹp là A2)."""
    from fastapi.responses import JSONResponse

    if not _valid_video_id(video_id):
        raise HTTPException(422, "video_id phải là id 32 ký tự hex.")
    data = astore.read_json(
        astore.shot_json_path(_analyzer_video_dir(video_id), idx))
    if data is None:
        raise HTTPException(404, "Chưa có kết quả cho cú này — cú đang chờ "
                                 "phân tích, hoặc id/cú sai.")
    return JSONResponse(data)


@app.get("/api/analyzer/videos/{video_id}/shots/{idx}/thumb")
def api_analyzer_shot_thumb(video_id: str, idx: int):
    """Thumbnail frame đầu cú (worker lưu lúc scan) — cột đầu bảng §3.1."""
    from fastapi.responses import FileResponse

    if not _valid_video_id(video_id):
        raise HTTPException(422, "video_id phải là id 32 ký tự hex.")
    p = astore.thumb_path(_analyzer_video_dir(video_id), idx)
    if not p.exists():
        raise HTTPException(404, "Không có thumbnail cho cú này.")
    return FileResponse(p, media_type="image/jpeg")


@app.get("/api/analyzer/videos/{video_id}/shots/{idx}/overlay")
def api_analyzer_shot_overlay(video_id: str, idx: int):
    """overlay.mp4 per cú (lát A2 phần 3) — worker render trong job
    analyze, trước khi xoá clip. Trang chi tiết phát bằng <video>; nút
    tải dùng chính URL này (thuộc tính download same-origin). Chứa frame
    broadcast — tính năng local/demo, file nằm data/ ngoài git."""
    from fastapi.responses import FileResponse

    if not _valid_video_id(video_id):
        raise HTTPException(422, "video_id phải là id 32 ký tự hex.")
    p = astore.overlay_path(_analyzer_video_dir(video_id), idx)
    if not p.exists():
        raise HTTPException(404, "Chưa có overlay cho cú này — cú chưa "
                                 "phân tích xong, hoặc kết quả thuộc lần "
                                 "chạy trước A2 (overlay không render hồi "
                                 "tố được vì clip đã xoá).")
    return FileResponse(p, media_type="video/mp4")


# --------------------------------------------------------- /api/trajectory

def _trajectory_sync(balls, phi, v0, side, vert):
    """Một `simulate_shot_multi(render=True)` + trích quỹ đạo. Chạy trong
    executor (pooltool nặng, và `env_h` dùng chung nên phải nằm trong
    `_search_lock` như mọi lần sim khác).

    `validate_full` gọi ở đây chứ không ở route: nó là CÙNG cửa vào mà
    `recommend_v2` dùng, nên input rác cho ra ĐÚNG message tiếng Việt của
    `/api/recommend`, không phải một bộ message thứ hai.

    `min_disp=TRAJ_MIN_DISP` — đúng tham số `recommend_v2` dùng khi render;
    thiếu nó thì quỹ đạo cú thay thế có thêm 5-7 bi đứng yên mà cú rank 1
    không có, và G6.2 sẽ đỏ vì một khác biệt không ai muốn.
    """
    from poolcoach_rl.recommend import (TRAJ_MIN_DISP, extract_trajectories,
                                        simulate_shot_multi, validate_full)

    validate_full(state.env_h, balls)
    rr = simulate_shot_multi(state.env_h, balls, phi, v0, side, vert,
                             render=True)
    if rr is None:
        return None
    return (extract_trajectories(rr["system"], min_disp=TRAJ_MIN_DISP),
            rr["balls_final"])


@app.post("/api/trajectory", response_model=TrajectoryResponse)
async def api_trajectory(req: TrajectoryRequest):
    """Quỹ đạo của MỘT cú đã biết tham số — phần lười của `/api/recommend`.

    Không nhận `engine`/`alternatives`: đây không phải một lần search, nó là
    một lần sim. Cú nào đáng vẽ đã do `/api/recommend` quyết.

    GIỮ IN-PROCESS kể cả khi mode queue bật — QUYẾT ĐỊNH (BRIEF 04/08 bàn
    giao 14, việc 1.5), không phải bỏ sót: mỗi request đúng 1 sim nhẹ, còn
    search 99–198 sim của recommend mới đáng đẩy qua worker.
    """
    import numpy as np

    if state.boot_error:
        raise HTTPException(500, f"Server lỗi khởi động: {state.boot_error}")
    if not state.jit_ready or state.env_h is None:
        raise HTTPException(503, "Đang khởi động vật lý (JIT ~40s) — thử lại sau")

    balls = {bid: np.array([p.x, p.y]) for bid, p in req.balls.items()}
    call = functools.partial(_trajectory_sync, balls, req.phi, req.v0,
                             req.side, req.vert)
    loop = asyncio.get_running_loop()
    try:
        async with _search_lock:
            out = await loop.run_in_executor(None, call)
    except ValueError as e:   # validate_full: ngoài bàn / chồng bi / id lạ
        raise HTTPException(422, str(e)) from e

    if out is None:
        # pooltool ném → `simulate_shot_multi` trả None. Nói THẲNG là sim
        # hỏng; trả quỹ đạo rỗng 200 thì FE vẽ bàn trống và không ai biết vì sao.
        raise HTTPException(422, "Không mô phỏng được cú này (pooltool ném "
                                 "lỗi) — kiểm lại tham số cú.")
    traj, balls_final = out
    return TrajectoryResponse(
        trajectories=traj,
        balls_final={bid: _pt(xy) for bid, xy in balls_final.items()})


class NoCacheStaticFiles(StaticFiles):
    """StaticFiles + ``Cache-Control: no-cache`` cho ``.js``/``.css``/``.html``.

    Nguyên nhân gốc bug tối 30/07: ``index.html`` nạp ``app.js`` không
    cache-busting và mount này không set header → trình duyệt giữ ``app.js``
    cũ hàng giờ sau khi Space đã deploy bản mới (alt không có quỹ đạo cho tới
    Ctrl+F5). ``no-cache`` chứ KHÔNG phải ``no-store``: trình duyệt vẫn được
    giữ bản sao và revalidate 304 bằng ETag/Last-Modified (starlette sẵn có),
    chỉ cấm dùng bản cũ mà không hỏi server.

    Subclass thay vì middleware toàn app, cố ý: header chỉ thuộc về tầng
    static, response ``/api/*`` không việc gì phải đi qua thêm một lớp. Set
    trên CẢ response 304 (``file_response`` của starlette trả
    ``NotModifiedResponse`` với bộ header đã chốt TRƯỚC khi mình kịp chèn —
    nên phải chèn sau khi ``super()`` trả về, không phải trên FileResponse).
    """

    def file_response(self, full_path, stat_result, scope, status_code=200):
        resp = super().file_response(full_path, stat_result, scope,
                                     status_code)
        if str(full_path).lower().endswith((".js", ".css", ".html")):
            resp.headers["Cache-Control"] = "no-cache"
        return resp


# TV tại bàn (bàn giao 15): WS /display/{id} + POST /api/displays/pair.
# Router phải vào TRƯỚC mount static — mount "/" nuốt mọi path còn lại.
app.include_router(displays.router)

# FE static — mount SAU các route /api để không nuốt chúng
app.mount("/", NoCacheStaticFiles(directory=str(STATIC_DIR), html=True),
          name="static")