#!/usr/bin/env python3
"""Local server for the L2 lyric review dashboard."""

from __future__ import annotations

import argparse
import csv
import json
import mimetypes
import os
import secrets
import subprocess
import sys
import threading
import time
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from urllib.parse import parse_qs, unquote, urlparse

import pymysql
import requests
from dbutils.pooled_db import PooledDB
from dotenv import load_dotenv


ROOT = Path(__file__).resolve().parent
REPORT_DIR = ROOT / "output" / "reports"
DASHBOARD = ROOT / "l2_review_dashboard.html"
REPORT_DIR.mkdir(parents=True, exist_ok=True)
GROUP_INDEX_CACHE: dict[tuple[str, int, int, str], list[dict[str, object]]] = {}
load_dotenv(ROOT / ".env")
REVIEW_CLAIM_TTL_SECONDS = max(60, int(os.getenv("REVIEW_CLAIM_TTL_SECONDS", "1800")))
REVIEW_ACCESS_CODE = os.getenv("REVIEW_ACCESS_CODE", "")
REVIEW_SESSION_TTL_SECONDS = max(300, int(os.getenv("REVIEW_SESSION_TTL_SECONDS", "43200")))
REVIEW_SESSIONS: dict[str, tuple[str, float]] = {}
REVIEW_SESSIONS_LOCK = threading.Lock()

TARGET_DB_CONFIG = {
    "host": os.getenv("TARGET_DB_HOST"),
    "port": int(os.getenv("TARGET_DB_PORT", 3306)),
    "user": os.getenv("TARGET_DB_USER"),
    "password": os.getenv("TARGET_DB_PASSWORD"),
    "database": os.getenv("TARGET_DB_NAME"),
    "charset": "utf8mb4",
    "cursorclass": pymysql.cursors.DictCursor,
}
TARGET_TABLE_NAME = os.getenv("TARGET_TABLE_NAME", "hk_songs")
TARGET_TABLE_NAME_TMP = os.getenv("TARGET_TABLE_NAME_TMP", f"{TARGET_TABLE_NAME}_import_staging")

# 连接池：mincached=2 保持 2 个空闲连接，maxcached=5 上限 5 个空闲，
# maxconnections=10 最大总数，ping=1 每次取连接前检测活性
_DB_POOL = PooledDB(
    creator=pymysql,
    mincached=2,
    maxcached=5,
    maxconnections=10,
    blocking=True,
    ping=1,
    **TARGET_DB_CONFIG,
)
STAGING_DB_PATH = "__staging_db__"
TARGET_COLUMNS = [
    "id", "name", "lyricist", "composer", "issue_status", "intro",
    "audio_url", "accompany_url", "lyrics_url", "lrc_url",
    "song_time", "song_start", "song_end",
    "creation_url", "opern_url", "cover_version", "issue_time",
    "cover_url", "animation_type", "bpm_class", "review_status",
    "in_status", "song_status", "commit_time", "review_time",
    "shelf_time", "review_remark", "create_time", "creator",
    "modify_time", "modifier", "deleted", "cooperate_type", "singer",
    "off_shelf_remark", "musician_id", "commit_id", "sheet_music",
    "commit_desc", "price", "source_table_name", "source_song_id",
    "lyric_archive_element_id", "melody_archive_element_id", "audio_fingerprint",
]
REVIEW_SCHEMA_COLUMNS = {
    "review_claimed_by": "varchar(64) DEFAULT NULL COMMENT '当前领取审核人'",
    "review_claim_token": "varchar(64) DEFAULT NULL COMMENT '领取会话凭证'",
    "review_claimed_at": "datetime DEFAULT NULL COMMENT '领取时间'",
    "review_claim_expires_at": "datetime DEFAULT NULL COMMENT '领取过期时间'",
    "review_version": "bigint(20) NOT NULL DEFAULT '0' COMMENT '审核乐观锁版本'",
}


class ReviewConflictError(Exception):
    """The staging row changed or is owned by another review session."""

    def __init__(self, message: str, current_record: dict[str, object] | None = None) -> None:
        super().__init__(message)
        self.current_record = current_record or {}


class ReviewAuthError(Exception):
    """The caller does not have a valid review editing session."""


def _review_login(
    reviewer: str,
    access_code: str = "",
    session_token: str = "",
) -> dict[str, object]:
    reviewer = reviewer.strip()
    if not reviewer:
        raise ReviewAuthError("请输入昵称")
    now = time.time()
    with REVIEW_SESSIONS_LOCK:
        expired = [token for token, (_name, expires_at) in REVIEW_SESSIONS.items() if expires_at <= now]
        for token in expired:
            REVIEW_SESSIONS.pop(token, None)
        existing = REVIEW_SESSIONS.get(session_token) if session_token else None
        if existing and existing[0] == reviewer and existing[1] > now:
            REVIEW_SESSIONS[session_token] = (reviewer, now + REVIEW_SESSION_TTL_SECONDS)
            return {"reviewer": reviewer, "session_token": session_token, "expires_in": REVIEW_SESSION_TTL_SECONDS}
    if not REVIEW_ACCESS_CODE:
        raise ReviewAuthError("服务端尚未配置 REVIEW_ACCESS_CODE，编辑功能已禁用")
    if not secrets.compare_digest(access_code.encode("utf-8"), REVIEW_ACCESS_CODE.encode("utf-8")):
        raise ReviewAuthError("暗号错误")
    token = secrets.token_urlsafe(32)
    with REVIEW_SESSIONS_LOCK:
        REVIEW_SESSIONS[token] = (reviewer, now + REVIEW_SESSION_TTL_SECONDS)
    return {"reviewer": reviewer, "session_token": token, "expires_in": REVIEW_SESSION_TTL_SECONDS}


def _require_review_session(payload: dict[str, object]) -> str:
    reviewer = str(payload.get("reviewer") or "").strip()
    session_token = str(payload.get("session_token") or "").strip()
    now = time.time()
    with REVIEW_SESSIONS_LOCK:
        session = REVIEW_SESSIONS.get(session_token)
        if not session or session[1] <= now:
            REVIEW_SESSIONS.pop(session_token, None)
            raise ReviewAuthError("编辑会话无效或已过期，请重新输入暗号")
        if session[0] != reviewer:
            raise ReviewAuthError("编辑会话与当前昵称不匹配")
        REVIEW_SESSIONS[session_token] = (reviewer, now + REVIEW_SESSION_TTL_SECONDS)
    return reviewer


def _conflict_response(handler: BaseHTTPRequestHandler, exc: ReviewConflictError) -> None:
    _json_response(
        handler,
        {
            "error": str(exc),
            "code": "review_conflict",
            "current_record": exc.current_record,
        },
        status=409,
    )


def _json_response(handler: BaseHTTPRequestHandler, payload: object, status: int = 200) -> None:
    body = json.dumps(payload, ensure_ascii=False, default=str).encode("utf-8")
    try:
        handler.send_response(status)
        handler.send_header("Content-Type", "application/json; charset=utf-8")
        handler.send_header("Cache-Control", "no-store")
        handler.send_header("Content-Length", str(len(body)))
        handler.end_headers()
        handler.wfile.write(body)
    except (BrokenPipeError, ConnectionResetError):
        return


def _error(handler: BaseHTTPRequestHandler, message: str, status: int = 400) -> None:
    _json_response(handler, {"error": message}, status=status)


def _safe_path(raw_path: str) -> Path:
    path = Path(unquote(raw_path)).expanduser()
    if not path.is_absolute():
        path = ROOT / path
    resolved = path.resolve()
    allowed_roots = [ROOT.resolve(), REPORT_DIR.resolve()]
    if not any(resolved == base or base in resolved.parents for base in allowed_roots):
        raise ValueError(f"path is outside workspace: {raw_path}")
    if not resolved.exists() or not resolved.is_file():
        raise FileNotFoundError(str(resolved))
    return resolved


def _read_csv(path: Path) -> list[dict[str, str]]:
    with open(path, "r", encoding="utf-8-sig", newline="") as f:
        return list(csv.DictReader(f))


def _iter_csv(path: Path):
    with open(path, "r", encoding="utf-8-sig", newline="") as f:
        yield from csv.DictReader(f)


def _target_conn():
    """从连接池获取连接，with 块结束时自动归还池中。"""
    return _DB_POOL.connection()


def _ensure_review_schema(apply_migration: bool = False) -> None:
    """Validate or add the columns required by collaborative reviewing."""
    with _target_conn() as conn, conn.cursor() as cursor:
        cursor.execute(f"SHOW COLUMNS FROM {TARGET_TABLE_NAME_TMP}")
        existing_columns = {str(row["Field"]) for row in cursor.fetchall()}
        missing = [name for name in REVIEW_SCHEMA_COLUMNS if name not in existing_columns]
        if missing and not apply_migration:
            names = ", ".join(missing)
            raise RuntimeError(
                f"暂存表缺少多人审核字段: {names}。"
                "请先使用 --migrate-review-schema 启动一次。"
            )
        for name in missing:
            cursor.execute(
                f"ALTER TABLE {TARGET_TABLE_NAME_TMP} "
                f"ADD COLUMN `{name}` {REVIEW_SCHEMA_COLUMNS[name]}"
            )
        cursor.execute(f"SHOW INDEX FROM {TARGET_TABLE_NAME_TMP} WHERE Key_name = %s", ("idx_staging_review_claim",))
        if not cursor.fetchone():
            if not apply_migration:
                raise RuntimeError(
                    "暂存表缺少索引 idx_staging_review_claim。"
                    "请先使用 --migrate-review-schema 启动一次。"
                )
            cursor.execute(
                f"ALTER TABLE {TARGET_TABLE_NAME_TMP} "
                "ADD KEY `idx_staging_review_claim` (`biz_review_status`, `review_claim_expires_at`)"
            )
        if apply_migration:
            conn.commit()


def _staging_summary() -> list[dict[str, str]]:
    sql = f"""
    SELECT
      COUNT(DISTINCT source_song_id) AS total_count,
      SUM(CASE WHEN dedup_action = 'new' THEN 1 ELSE 0 END) AS new_count,
      SUM(CASE WHEN dedup_action = 'merge' THEN 1 ELSE 0 END) AS merge_count,
      SUM(CASE WHEN dedup_action = 'review' THEN 1 ELSE 0 END) AS review_count,
      SUM(CASE WHEN dedup_action = 'skip' THEN 1 ELSE 0 END) AS skip_count,
      SUM(CASE WHEN dedup_action = 'review' AND (biz_review_status IS NULL OR biz_review_status IN ('pending', 'unsure')) THEN 1 ELSE 0 END) AS pending_review_count,
      SUM(CASE WHEN biz_review_status = 'approved_import' THEN 1 ELSE 0 END) AS approved_count,
      SUM(CASE WHEN biz_review_status = 'rejected_duplicate' THEN 1 ELSE 0 END) AS rejected_count,
      SUM(CASE WHEN biz_review_status = 'deleted' THEN 1 ELSE 0 END) AS deleted_count,
      SUM(CASE WHEN biz_review_status IN ('approved_import', 'rejected_duplicate', 'deleted') AND staging_status <> 'imported' THEN 1 ELSE 0 END) AS reviewed_count,
      SUM(CASE WHEN biz_review_status IN ('approved_import', 'rejected_duplicate', 'deleted') AND staging_status = 'imported' THEN 1 ELSE 0 END) AS submitted_count
    FROM (
      SELECT source_song_id, dedup_action, biz_review_status, staging_status
      FROM {TARGET_TABLE_NAME_TMP}
      WHERE (source_song_id, staging_id) IN (
        SELECT source_song_id, MAX(staging_id)
        FROM {TARGET_TABLE_NAME_TMP}
        GROUP BY source_song_id
      )
    ) latest
    """
    with _target_conn() as conn, conn.cursor() as cursor:
        cursor.execute(sql)
        row = cursor.fetchone() or {}
    return [{
        "top_k": "staging",
        "elapsed_seconds": "-",
        "throughput_per_second": "-",
        "avg_recalled_candidates": "-",
        "hit_count": str((row.get("merge_count") or 0) + (row.get("review_count") or 0)),
        "duplicate_count": str(row.get("merge_count") or 0),
        "review_count": str(row.get("review_count") or 0),
        "skip_count": str(row.get("skip_count") or 0),
        "new_count": str(row.get("new_count") or 0),
        "total_count": str(row.get("total_count") or 0),
        "approved_count": str(row.get("approved_count") or 0),
        "rejected_count": str(row.get("rejected_count") or 0),
        "deleted_count": str(row.get("deleted_count") or 0),
        "reviewed_count": str(row.get("reviewed_count") or 0),
        "submitted_count": str(row.get("submitted_count") or 0),
        "pending_review_count": str(row.get("pending_review_count") or 0),
    }]


def _staging_dashboard_row(row: dict[str, object]) -> dict[str, str]:
    lyrics_url = str(row.get("lyrics_url") or "")
    # 若 lyrics_url 不是有效 URL/路径，说明是原始歌词文本（skip_oss 模式），加前缀供前端直接使用
    if lyrics_url and not lyrics_url.startswith(("http://", "https://", "/")):
        lyrics_url = f"_raw:{lyrics_url}"
    return {
        "top_k": "staging",
        "rank": "1",
        "query_source_id": str(row.get("source_song_id") or ""),
        "query_name": str(row.get("name") or ""),
        "query_lyricist": str(row.get("lyricist") or ""),
        "query_composer": str(row.get("composer") or ""),
        "query_lyrics_path": lyrics_url,
        "candidate_id": str(row.get("matched_song_id") or ""),
        "candidate_name": "",
        "candidate_lyricist": "",
        "candidate_composer": "",
        "candidate_lyrics_path": "",
        "candidate_decision": str(row.get("dedup_action") or ""),
        "candidate_confidence": str(row.get("dedup_confidence") or ""),
        "candidate_reason": str(row.get("dedup_reason") or ""),
        "decision": str(row.get("dedup_action") or ""),
        "action": str(row.get("dedup_action") or ""),
        "source_id": str(row.get("source_song_id") or ""),
        "staging_id": str(row.get("staging_id") or ""),
        "review_status": str(row.get("biz_review_status") or ""),
        "review_note": str(row.get("biz_review_note") or ""),
        "reviewed_by": str(row.get("reviewed_by") or ""),
        "reviewed_at": str(row.get("reviewed_at") or ""),
        "review_version": str(row.get("review_version") or 0),
        "review_claimed_by": str(row.get("review_claimed_by") or ""),
        "review_claim_expires_at": str(row.get("review_claim_expires_at") or ""),
        "staging_status": str(row.get("staging_status") or ""),
        "l1_metadata_match": "1" if row.get("l1_matched_id") else "0",
        "l1_l2_conflict": "0",
        "audio_url": str(row.get("audio_url") or ""),
    }


def _staging_groups_response(
    mode: str,
    term: str,
    page: int,
    page_size: int,
    decision: str = "",
    client_token: str = "",
) -> dict[str, object]:
    where_clauses = []
    params: list[object] = []
    if term:
        where_clauses.append("(CAST(s.source_song_id AS CHAR) LIKE %s OR s.name LIKE %s OR s.lyricist LIKE %s OR s.composer LIKE %s)")
        like = f"%{term}%"
        params.extend([like, like, like, like])
    # hits 模式下只展示有命中的记录（merge/review）
    if mode == "hits":
        where_clauses.append("s.dedup_action IN ('merge', 'review')")
    # 决策筛选：l1_hit / l2_hit / conflict / new / skip
    if decision == 'l1_hit':
        where_clauses.append("s.l1_matched_id IS NOT NULL")
    elif decision == 'l2_duplicate':
        where_clauses.append("s.dedup_action = 'merge'")
    elif decision == 'l2_review':
        where_clauses.append("s.dedup_action = 'review'")
    elif decision == 'conflict':
        where_clauses.append("(s.l1_matched_id IS NOT NULL AND s.dedup_action = 'new') OR (s.l1_matched_id IS NULL AND s.dedup_action IN ('merge', 'review'))")
    elif decision == 'new':
        where_clauses.append("s.dedup_action = 'new' AND s.l1_matched_id IS NULL")
    elif decision == 'skip':
        where_clauses.append("s.dedup_action = 'skip'")
    elif decision == 'reviewed':
        where_clauses.append("s.biz_review_status IN ('approved_import', 'rejected_duplicate', 'deleted')")
        where_clauses.append("s.staging_status <> 'imported'")
    elif decision == 'submitted':
        where_clauses.append("s.biz_review_status IN ('approved_import', 'rejected_duplicate', 'deleted')")
        where_clauses.append("s.staging_status = 'imported'")
    elif decision == 'pending':
        where_clauses.append("(s.biz_review_status IS NULL OR s.biz_review_status IN ('pending', 'unsure'))")
        where_clauses.append("s.dedup_action = 'review'")
        where_clauses.append("s.staging_status <> 'imported'")
        where_clauses.append(
            "(s.review_claim_expires_at IS NULL OR s.review_claim_expires_at <= NOW() OR s.review_claim_token = %s)"
        )
        params.append(client_token)
    # 兼容旧的 dedup_action 筛选
    elif decision and decision in ('merge', 'review'):
        where_clauses.append("s.dedup_action = %s")
        params.append(decision)
    where = "WHERE " + " AND ".join(where_clauses) if where_clauses else ""
    # 同一首歌可能因多次导入批次产生多行，按 source_song_id 去重取最新一行
    latest_join = f"""
    FROM {TARGET_TABLE_NAME_TMP} s
    INNER JOIN (
        SELECT source_song_id, MAX(staging_id) AS max_staging_id
        FROM {TARGET_TABLE_NAME_TMP}
        GROUP BY source_song_id
    ) latest ON s.staging_id = latest.max_staging_id
    """
    count_sql = f"SELECT COUNT(*) AS n {latest_join} {where}"
    sql = f"""
    SELECT s.staging_id, s.source_song_id, s.name, s.lyricist, s.composer, s.dedup_action,
           s.biz_review_status, s.staging_status, s.l1_matched_id, s.review_version,
           CASE WHEN s.review_claim_expires_at > NOW() THEN s.review_claimed_by ELSE NULL END AS review_claimed_by,
           s.review_claim_expires_at,
           CASE WHEN s.review_claim_token = %s AND s.review_claim_expires_at > NOW() THEN 1 ELSE 0 END AS claimed_by_me
    {latest_join}
    {where}
    ORDER BY
      CASE s.dedup_action WHEN 'review' THEN 0 WHEN 'merge' THEN 1 ELSE 2 END,
      s.staging_create_time DESC
    LIMIT %s OFFSET %s
    """
    with _target_conn() as conn, conn.cursor() as cursor:
        cursor.execute(count_sql, params)
        total = (cursor.fetchone() or {}).get("n", 0)
        cursor.execute(sql, [client_token, *params, page_size, max(0, (page - 1) * page_size)])
        raw_rows = cursor.fetchall()
    # Python 层面再去重，防止 SQL 层面因数据库版本或数据异常未能完全去重
    seen_source_ids: set[str] = set()
    rows = []
    for row in raw_rows:
        sid = str(row.get("source_song_id") or "")
        if sid not in seen_source_ids:
            seen_source_ids.add(sid)
            rows.append(row)
    groups = [
        {
            "id": str(row.get("source_song_id") or ""),
            "query_source_id": str(row.get("source_song_id") or ""),
            "query_name": row.get("name") or "",
            "query_lyricist": row.get("lyricist") or "",
            "query_composer": row.get("composer") or "",
            "count": 1,
            "has_hit": row.get("dedup_action") in {"merge", "review"},
            "has_conflict": False,
            "has_l1_hint": bool(row.get("l1_matched_id")),
            "decision": row.get("dedup_action") or "",
            "reviewed": row.get("biz_review_status") in {"approved_import", "rejected_duplicate", "deleted"},
            "submitted": (
                row.get("biz_review_status") in {"approved_import", "rejected_duplicate", "deleted"}
                and row.get("staging_status") == "imported"
            ),
            "staging_id": str(row.get("staging_id") or ""),
            "review_version": int(row.get("review_version") or 0),
            "review_claimed_by": str(row.get("review_claimed_by") or ""),
            "review_claim_expires_at": str(row.get("review_claim_expires_at") or ""),
            "claimed_by_me": bool(row.get("claimed_by_me")),
        }
        for row in rows
    ]
    end = page * page_size
    return {"total": total, "page": page, "page_size": page_size, "has_more": end < total, "groups": groups}


def _staging_query_rows(query_id: str) -> dict[str, object]:
    sql = f"SELECT * FROM {TARGET_TABLE_NAME_TMP} WHERE source_song_id = %s ORDER BY staging_id DESC LIMIT 1"
    with _target_conn() as conn, conn.cursor() as cursor:
        cursor.execute(sql, (query_id,))
        row = cursor.fetchone()
        if not row:
            return {"rows": []}
        matched_song_id = row.get("matched_song_id")
        matched_row = None
        if matched_song_id:
            # 先尝试目标表查找（matched_id 是目标库 ID 的情况）
            cursor.execute(
                f"SELECT name, lyricist, composer, lyrics_url, audio_url FROM {TARGET_TABLE_NAME} WHERE id = %s",
                (matched_song_id,),
            )
            matched_row = cursor.fetchone()
            # 目标表找不到时，回退到暂存表查找（同批次内匹配，matched_id 是源库 source_song_id）
            if not matched_row:
                cursor.execute(
                    f"SELECT name, lyricist, composer, lyrics_url, audio_url FROM {TARGET_TABLE_NAME_TMP}"
                    f" WHERE source_song_id = %s ORDER BY staging_id DESC LIMIT 1",
                    (str(matched_song_id),),
                )
                matched_row = cursor.fetchone()
    dashboard_row = _staging_dashboard_row(row)
    has_main_match = bool(matched_song_id)
    # 主行 rank：仅 merge/review 有主匹配时显示 "1"，new 样本置空以便召回候选从 1 开始编号
    dashboard_row["rank"] = "1" if has_main_match else ""
    if matched_row:
        dashboard_row["candidate_name"] = str(matched_row.get("name") or "")
        dashboard_row["candidate_lyricist"] = str(matched_row.get("lyricist") or "")
        dashboard_row["candidate_composer"] = str(matched_row.get("composer") or "")
        cand_lyrics = str(matched_row.get("lyrics_url") or "")
        if cand_lyrics and not cand_lyrics.startswith(("http://", "https://", "/")):
            cand_lyrics = f"_raw:{cand_lyrics}"
        dashboard_row["candidate_lyrics_path"] = cand_lyrics
        dashboard_row["candidate_audio_url"] = str(matched_row.get("audio_url") or "")

    rows = [dashboard_row]

    # 解析 recalled_candidates JSON，为每个召回候选生成一张卡片
    recalled_json = row.get("recalled_candidates")
    if recalled_json:
        try:
            recalled_list = json.loads(recalled_json) if isinstance(recalled_json, str) else recalled_json
            main_candidate_id = str(matched_song_id or "")

            # 批量查找召回候选的歌词 URL（目标表 + 暂存表 fallback）
            recalled_ids_str = [
                str(c.get("id") or "")
                for c in recalled_list
                if c.get("id") and str(c.get("id")) != main_candidate_id
            ]
            # 转 int 避免 bigint/string 类型不匹配导致 IN 查询返空
            recalled_ids_int: list[int] = []
            for rid in recalled_ids_str:
                try:
                    recalled_ids_int.append(int(rid))
                except (ValueError, TypeError):
                    pass
            lyrics_map: dict[str, dict] = {}
            if recalled_ids_int:
                placeholders = ",".join(["%s"] * len(recalled_ids_int))
                with _target_conn() as tconn, tconn.cursor() as tcursor:
                    # 目标表查找
                    tcursor.execute(
                        f"SELECT id, name, lyricist, composer, lyrics_url, audio_url"
                        f" FROM {TARGET_TABLE_NAME} WHERE id IN ({placeholders})",
                        recalled_ids_int,
                    )
                    for mrow in tcursor.fetchall():
                        lyrics_map[str(mrow.get("id"))] = mrow
                    # 暂存表 fallback：目标表找不到的 ID 从暂存表补查
                    missing_ids = [rid for rid in recalled_ids_int if str(rid) not in lyrics_map]
                    if missing_ids:
                        ph2 = ",".join(["%s"] * len(missing_ids))
                        tcursor.execute(
                            f"SELECT source_song_id, name, lyricist, composer, lyrics_url, audio_url"
                            f" FROM {TARGET_TABLE_NAME_TMP}"
                            f" WHERE source_song_id IN ({ph2})"
                            f" ORDER BY staging_id DESC",
                            [str(mid) for mid in missing_ids],
                        )
                        for mrow in tcursor.fetchall():
                            sid = str(mrow.get("source_song_id"))
                            if sid not in lyrics_map:
                                lyrics_map[sid] = mrow

            rank = 2 if has_main_match else 1
            for cand in recalled_list:
                cand_id = str(cand.get("id") or "")
                # 跳过已经是主匹配候选的（避免重复显示）
                if cand_id and cand_id == main_candidate_id:
                    continue
                cand_row = dict(dashboard_row)
                cand_row["rank"] = str(rank)
                cand_row["candidate_id"] = cand_id
                # 优先用目标表/暂存表查到的元数据，回退到 JSON 中的信息
                target_info = lyrics_map.get(cand_id)
                if target_info:
                    cand_row["candidate_name"] = str(target_info.get("name") or cand.get("name") or "")
                    cand_row["candidate_lyricist"] = str(target_info.get("lyricist") or cand.get("lyricist") or "")
                    cand_row["candidate_composer"] = str(target_info.get("composer") or cand.get("composer") or "")
                    cand_lyrics = str(target_info.get("lyrics_url") or "")
                    if cand_lyrics and not cand_lyrics.startswith(("http://", "https://", "/")):
                        cand_lyrics = f"_raw:{cand_lyrics}"
                    cand_row["candidate_lyrics_path"] = cand_lyrics
                    cand_row["candidate_audio_url"] = str(target_info.get("audio_url") or "")
                else:
                    cand_row["candidate_name"] = str(cand.get("name") or "")
                    cand_row["candidate_lyricist"] = str(cand.get("lyricist") or "")
                    cand_row["candidate_composer"] = str(cand.get("composer") or "")
                    cand_row["candidate_lyrics_path"] = ""
                    cand_row["candidate_audio_url"] = ""
                cand_row["candidate_decision"] = str(cand.get("decision") or "new")
                cand_row["candidate_confidence"] = str(cand.get("confidence") or "")
                cand_row["candidate_jaccard"] = str(cand.get("jaccard") or "")
                cand_row["candidate_line_coverage"] = str(cand.get("line_coverage") or "")
                cand_row["candidate_reason"] = ""
                rows.append(cand_row)
                rank += 1
        except (json.JSONDecodeError, TypeError):
            pass

    return {"rows": rows}


def _review_snapshot(cursor, staging_id: int) -> dict[str, object]:
    cursor.execute(
        f"""
        SELECT staging_id, source_song_id, dedup_action, biz_review_status, biz_review_note,
               reviewed_by, reviewed_at, review_version, review_claimed_by,
               review_claim_expires_at, staging_status
        FROM {TARGET_TABLE_NAME_TMP}
        WHERE staging_id = %s
        """,
        (staging_id,),
    )
    return cursor.fetchone() or {}


def _claim_staging_review(staging_id: int, reviewer: str, client_token: str) -> dict[str, object]:
    if not reviewer or not client_token:
        raise ValueError("reviewer and client_token are required")
    with _target_conn() as conn, conn.cursor() as cursor:
        cursor.execute(
            f"""
            UPDATE {TARGET_TABLE_NAME_TMP}
            SET review_claimed_by = %s,
                review_claim_token = %s,
                review_claimed_at = NOW(),
                review_claim_expires_at = DATE_ADD(NOW(), INTERVAL %s SECOND),
                review_version = review_version + 1
            WHERE staging_id = %s
              AND staging_status <> 'imported'
              AND (
                biz_review_status IN ('pending', 'unsure')
                OR biz_review_status = 'not_required'
                OR biz_review_status IN ('approved_import', 'rejected_duplicate', 'deleted')
              )
              AND (
                review_claim_expires_at IS NULL
                OR review_claim_expires_at <= NOW()
                OR review_claim_token = %s
              )
            """,
            (reviewer, client_token, REVIEW_CLAIM_TTL_SECONDS, staging_id, client_token),
        )
        if cursor.rowcount != 1:
            current = _review_snapshot(cursor, staging_id)
            conn.rollback()
            if not current:
                message = f"找不到 staging_id={staging_id}，请刷新页面后重试"
            elif current.get("staging_status") == "imported":
                message = "该记录已经入库，只能浏览，不能再领取审核"
            elif current.get("review_claimed_by"):
                message = f"该记录已由 {current['review_claimed_by']} 领取"
            else:
                message = f"该记录当前状态为 {current.get('biz_review_status') or '-'}，不能领取"
            raise ReviewConflictError(message, current)
        current = _review_snapshot(cursor, staging_id)
        conn.commit()
    return current


def _heartbeat_staging_review(staging_id: int, client_token: str) -> dict[str, object]:
    with _target_conn() as conn, conn.cursor() as cursor:
        cursor.execute(
            f"""
            UPDATE {TARGET_TABLE_NAME_TMP}
            SET review_claim_expires_at = DATE_ADD(NOW(), INTERVAL %s SECOND)
            WHERE staging_id = %s
              AND review_claim_token = %s
              AND review_claim_expires_at > NOW()
            """,
            (REVIEW_CLAIM_TTL_SECONDS, staging_id, client_token),
        )
        if cursor.rowcount != 1:
            current = _review_snapshot(cursor, staging_id)
            conn.rollback()
            raise ReviewConflictError("审核领取已过期或已转给其他审核人", current)
        current = _review_snapshot(cursor, staging_id)
        conn.commit()
    return current


def _review_statuses(staging_ids: list[int], client_token: str = "") -> list[dict[str, object]]:
    if not staging_ids:
        return []
    placeholders = ",".join(["%s"] * len(staging_ids))
    with _target_conn() as conn, conn.cursor() as cursor:
        cursor.execute(
            f"""
            SELECT staging_id, source_song_id, biz_review_status, staging_status, reviewed_by, reviewed_at,
                   review_version,
                   CASE WHEN review_claim_expires_at > NOW() THEN review_claimed_by ELSE NULL END AS review_claimed_by,
                   review_claim_expires_at,
                   CASE WHEN review_claim_token = %s AND review_claim_expires_at > NOW() THEN 1 ELSE 0 END AS claimed_by_me
            FROM {TARGET_TABLE_NAME_TMP}
            WHERE staging_id IN ({placeholders})
            """,
            [client_token, *staging_ids],
        )
        return cursor.fetchall()


def _update_staging_review(
    staging_id: int,
    decision: str,
    note: str,
    reviewer: str,
    expected_version: int,
    client_token: str,
) -> dict[str, object]:
    """Persist one claimed review using optimistic concurrency control."""
    if decision == "pending":
        status_fields = """
            biz_review_status = 'pending', reviewed_by = NULL, reviewed_at = NULL,
            staging_status = CASE WHEN staging_status = 'deleted' THEN 'staged' ELSE staging_status END
        """
    elif decision == "deleted":
        status_fields = """
            biz_review_status = 'deleted', reviewed_by = %s, reviewed_at = NOW(),
            staging_status = 'deleted'
        """
    else:
        status_fields = "biz_review_status = %s, reviewed_by = %s, reviewed_at = NOW()"

    params: list[object]
    if decision == "pending":
        params = [note, staging_id, expected_version, client_token, reviewer]
    elif decision == "deleted":
        params = [reviewer, note, staging_id, expected_version, client_token, reviewer]
    else:
        params = [decision, reviewer, note, staging_id, expected_version, client_token, reviewer]
    with _target_conn() as conn, conn.cursor() as cursor:
        cursor.execute(
            f"""
            UPDATE {TARGET_TABLE_NAME_TMP}
            SET {status_fields},
                biz_review_note = NULLIF(%s, ''),
                review_version = review_version + 1,
                review_claimed_by = NULL,
                review_claim_token = NULL,
                review_claimed_at = NULL,
                review_claim_expires_at = NULL
            WHERE staging_id = %s
              AND review_version = %s
              AND review_claim_token = %s
              AND review_claimed_by = %s
              AND review_claim_expires_at > NOW()
            """,
            params,
        )
        if cursor.rowcount != 1:
            current = _review_snapshot(cursor, staging_id)
            conn.rollback()
            raise ReviewConflictError("记录已被其他人修改，当前提交未覆盖数据库", current)
        current = _review_snapshot(cursor, staging_id)
        conn.commit()
    return current


def _import_approved_staging(source_ids: list[str] | None = None) -> dict[str, object]:
    """Import approved rows while holding staging-row locks to prevent duplicate imports."""
    columns = ", ".join(f"`{column}`" for column in TARGET_COLUMNS)
    select_columns = ", ".join(f"s.`{column}`" for column in TARGET_COLUMNS)
    with _target_conn() as conn, conn.cursor() as cursor:
        source_filter = ""
        lock_params: list[object] = []
        if source_ids:
            source_placeholders = ",".join(["%s"] * len(source_ids))
            source_filter = f" AND s.source_song_id IN ({source_placeholders})"
            lock_params.extend(source_ids)
        cursor.execute(
            f"""
            SELECT s.staging_id, s.source_song_id
            FROM {TARGET_TABLE_NAME_TMP} s
            INNER JOIN (
                SELECT source_song_id, MAX(staging_id) AS max_staging_id
                FROM {TARGET_TABLE_NAME_TMP}
                GROUP BY source_song_id
            ) latest ON s.staging_id = latest.max_staging_id
            WHERE s.dedup_action = 'review'
              AND s.biz_review_status = 'approved_import'
              AND s.staging_status = 'staged'
              AND (s.review_claim_expires_at IS NULL OR s.review_claim_expires_at <= NOW())
              {source_filter}
            ORDER BY staging_id
            FOR UPDATE
            """,
            lock_params,
        )
        approved_rows = cursor.fetchall()
        if not approved_rows:
            conn.rollback()
            raise ValueError("没有可入库的审核通过样本")
        staging_ids = [row["staging_id"] for row in approved_rows]
        approved_source_ids = [str(row["source_song_id"]) for row in approved_rows]
        staging_placeholders = ",".join(["%s"] * len(staging_ids))
        cursor.execute(
            f"""
            INSERT INTO {TARGET_TABLE_NAME} ({columns})
            SELECT {select_columns}
            FROM {TARGET_TABLE_NAME_TMP} s
            WHERE s.staging_id IN ({staging_placeholders})
            """,
            staging_ids,
        )
        inserted_count = cursor.rowcount
        source_placeholders = ",".join(["%s"] * len(approved_source_ids))
        cursor.execute(
            f"""
            SELECT source_song_id, id AS target_id
            FROM {TARGET_TABLE_NAME}
            WHERE source_song_id IN ({source_placeholders})
            """,
            approved_source_ids,
        )
        id_map = {str(row["source_song_id"]): row["target_id"] for row in cursor.fetchall()}
        imported_updated = 0
        for approved_row in approved_rows:
            sid = str(approved_row["source_song_id"])
            target_id = id_map.get(sid)
            if target_id is None:
                raise RuntimeError(f"入库后未找到目标记录: source_song_id={sid}")
            cursor.execute(
                f"""
                UPDATE {TARGET_TABLE_NAME_TMP}
                SET staging_status = 'imported', imported_song_id = %s, error_message = NULL,
                    review_version = review_version + 1
                WHERE staging_id = %s AND staging_status = 'staged'
                """,
                (target_id, approved_row["staging_id"]),
            )
            imported_updated += cursor.rowcount
        conn.commit()
    return {
        "approved_count": len(approved_rows),
        "inserted_count": inserted_count,
        "imported_song_ids_updated": imported_updated,
        "source_ids": approved_source_ids,
    }


def _report_runs() -> list[dict[str, str]]:
    summaries = {
        path.name.replace("l2_topk_benchmark_summary_", "").removesuffix(".csv"): path
        for path in REPORT_DIR.glob("l2_topk_benchmark_summary_*.csv")
    }
    retrievals = {
        path.name.replace("l2_topk_retrieval_top10_", "").removesuffix(".csv"): path
        for path in REPORT_DIR.glob("l2_topk_retrieval_top10_*.csv")
    }
    hits = {
        path.name.replace("l2_topk_duplicate_hits_", "").removesuffix(".csv"): path
        for path in REPORT_DIR.glob("l2_topk_duplicate_hits_*.csv")
    }
    run_ids = sorted(set(summaries) & set(retrievals) & set(hits), reverse=True)
    benchmark_runs = [
        {
            "id": run_id,
            "type": "benchmark",
            "summary": str(summaries[run_id]),
            "retrieval": str(retrievals[run_id]),
            "hits": str(hits[run_id]),
        }
        for run_id in run_ids
    ]
    review_runs = [
        {
            "id": path.name.replace("review_decisions_", "").removesuffix(".csv"),
            "type": "import",
            "summary": str(path),
            "retrieval": str(path),
            "hits": str(path),
            "review": str(path),
        }
        for path in sorted(REPORT_DIR.glob("review_decisions_*.csv"), reverse=True)
    ]
    staging_run = {
        "id": "staging-db",
        "type": "staging",
        "summary": STAGING_DB_PATH,
        "retrieval": STAGING_DB_PATH,
        "hits": STAGING_DB_PATH,
        "review": STAGING_DB_PATH,
    }
    return [staging_run] + review_runs + benchmark_runs


def _row_decision(row: dict[str, str]) -> str:
    return row.get("action") or row.get("decision") or row.get("candidate_decision") or ""


def _is_import_review_row(row: dict[str, str]) -> bool:
    return "source_id" in row and "action" in row and "query_source_id" not in row


def _dashboard_row(row: dict[str, str]) -> dict[str, str]:
    if not _is_import_review_row(row):
        return row
    source_id = row.get("source_id", "")
    action = row.get("action", "")
    return {
        **row,
        "top_k": "import",
        "rank": "1",
        "query_source_id": source_id,
        "query_name": row.get("name", ""),
        "query_lyricist": row.get("lyricist", ""),
        "query_composer": row.get("composer", ""),
        "query_lyrics_path": row.get("query_lyrics_path", ""),
        "candidate_id": row.get("matched_id", ""),
        "candidate_name": row.get("matched_name", ""),
        "candidate_lyricist": row.get("matched_lyricist", ""),
        "candidate_composer": row.get("matched_composer", ""),
        "candidate_lyrics_path": row.get("matched_lyrics_path", ""),
        "candidate_decision": action,
        "candidate_confidence": row.get("confidence", ""),
        "candidate_reason": row.get("reason", ""),
        "decision": action,
        "l1_metadata_match": "1" if row.get("l1_matched_id") else "0",
        "l1_l2_conflict": "0",
    }


def _review_summary(path: Path) -> list[dict[str, str]]:
    counts = {"new": 0, "merge": 0, "review": 0}
    total = 0
    for row in _iter_csv(path):
        total += 1
        action = row.get("action", "")
        if action in counts:
            counts[action] += 1
    return [{
        "top_k": "import",
        "elapsed_seconds": "-",
        "throughput_per_second": "-",
        "avg_recalled_candidates": "-",
        "hit_count": str(counts["merge"] + counts["review"]),
        "duplicate_count": str(counts["merge"]),
        "review_count": str(counts["review"]),
        "new_count": str(counts["new"]),
        "total_count": str(total),
    }]


def _norm(value: str | None) -> str:
    return (value or "").strip().lower()


def _matches_term(row: dict[str, object], term: str) -> bool:
    if not term:
        return True
    haystack = " ".join(
        [
            row.get("query_source_id", ""),
            row.get("query_name", ""),
            row.get("query_lyricist", ""),
            row.get("query_composer", ""),
        ]
    ).lower()
    return term in haystack


def _group_index(path: Path, top_k: str) -> list[dict[str, object]]:
    stat = path.stat()
    cache_key = (str(path), stat.st_mtime_ns, stat.st_size, top_k)
    cached = GROUP_INDEX_CACHE.get(cache_key)
    if cached is not None:
        return cached

    groups_by_id: dict[str, dict[str, object]] = {}
    order = 0
    for raw_row in _iter_csv(path):
        row = _dashboard_row(raw_row)
        if top_k and str(row.get("top_k", "")) != top_k:
            continue
        query_id = row.get("query_source_id", "")
        if not query_id:
            continue

        group = groups_by_id.get(query_id)
        if group is None:
            group = {
                "id": query_id,
                "_order": order,
                "query_source_id": query_id,
                "query_name": row.get("query_name", ""),
                "query_lyricist": row.get("query_lyricist", ""),
                "query_composer": row.get("query_composer", ""),
                "count": 0,
                "has_hit": False,
                "has_conflict": False,
                "has_l1_hint": False,
                "decision": _row_decision(row),
            }
            groups_by_id[query_id] = group
            order += 1

        group["count"] = int(group["count"]) + 1
        decision = _row_decision(row)
        l1_match = row.get("l1_metadata_match") == "1"
        conflict = row.get("l1_l2_conflict") == "1"
        group["has_hit"] = bool(group["has_hit"]) or decision in {"duplicate", "review"}
        group["has_conflict"] = bool(group["has_conflict"]) or conflict
        group["has_l1_hint"] = bool(group["has_l1_hint"]) or (decision == "new" and l1_match)

    groups = sorted(
        groups_by_id.values(),
        key=lambda group: (
            0 if group["has_conflict"] else 1,
            0 if group["has_l1_hint"] else 1,
            0 if group["has_hit"] else 1,
            int(group["_order"]),
        ),
    )
    GROUP_INDEX_CACHE.clear()
    GROUP_INDEX_CACHE[cache_key] = groups
    return groups


def _groups_response(path: Path, top_k: str, mode: str, term: str, page: int, page_size: int, decision: str = '') -> dict[str, object]:
    groups = _group_index(path, top_k)
    # hits 模式下只展示有命中的记录（merge/review），排除纯 new
    if mode == "hits":
        groups = [g for g in groups if g.get("has_hit")]
    # 决策筛选
    if decision == 'l1_hit':
        groups = [g for g in groups if g.get("has_l1_hint")]
    elif decision == 'l2_duplicate':
        groups = [g for g in groups if g.get("decision") == "merge"]
    elif decision == 'l2_review':
        groups = [g for g in groups if g.get("decision") == "review"]
    elif decision == 'conflict':
        groups = [g for g in groups if g.get("has_conflict")]
    elif decision == 'new':
        groups = [g for g in groups if g.get("decision") == "new" and not g.get("has_l1_hint")]
    elif decision == 'skip':
        groups = [g for g in groups if g.get("decision") == "skip"]
    elif decision == 'reviewed':
        groups = [g for g in groups if g.get("reviewed")]
    elif decision == 'pending':
        groups = [g for g in groups if not g.get("reviewed")]
    elif decision and decision in ('merge', 'review', 'duplicate'):
        groups = [g for g in groups if g.get("decision") == decision]
    filtered = [group for group in groups if _matches_term(group, term)]
    start = max(0, (page - 1) * page_size)
    end = start + page_size
    page_groups = filtered[start:end]
    return {
        "total": len(filtered),
        "page": page,
        "page_size": page_size,
        "has_more": end < len(filtered),
        "groups": [{key: value for key, value in group.items() if not key.startswith("_")} for group in page_groups],
    }


def _query_rows_response(path: Path, top_k: str, query_id: str) -> dict[str, object]:
    rows = []
    found = False
    for raw_row in _iter_csv(path):
        row = _dashboard_row(raw_row)
        if str(row.get("top_k", "")) != top_k:
            continue
        current_query_id = row.get("query_source_id", "")
        if current_query_id == query_id:
            found = True
            rows.append(row)
            continue
        if found:
            break
    return {"rows": rows}


def _write_review_import_csv(rows: list[dict[str, str]]) -> Path:
    path = REPORT_DIR / f"review_import_approved_{time.strftime('%Y%m%d_%H%M%S')}.csv"
    with open(path, "w", encoding="utf-8", newline="") as f:
        writer = csv.DictWriter(f, fieldnames=["source_id", "review_decision", "review_note"])
        writer.writeheader()
        for row in rows:
            writer.writerow({
                "source_id": row.get("source_id", ""),
                "review_decision": row.get("review_decision", ""),
                "review_note": row.get("review_note", ""),
            })
    return path


def _run_import(review_csv: Path) -> dict[str, object]:
    cmd = [
        "uv",
        "run",
        "python",
        "import_hk_songs.py",
        "--review-result-csv",
        str(review_csv),
        "--load-existing-lyrics",
        "--skip-media-oss",
    ]
    result = subprocess.run(
        cmd,
        cwd=ROOT,
        text=True,
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
        check=False,
    )
    output_tail = result.stdout[-12000:]
    if result.returncode != 0:
        raise RuntimeError(
            f"入库脚本执行失败 (exit={result.returncode}):\n{output_tail}"
        )
    return {
        "command": " ".join(cmd),
        "returncode": result.returncode,
        "output": output_tail,
    }


class Handler(BaseHTTPRequestHandler):
    def log_message(self, fmt: str, *args: object) -> None:
        print(f"{self.address_string()} - {fmt % args}")

    def do_GET(self) -> None:
        parsed = urlparse(self.path)
        if parsed.path == "/":
            self._serve_file(DASHBOARD)
            return
        if parsed.path == "/api/reports":
            _json_response(self, {"runs": _report_runs()})
            return
        if parsed.path == "/api/csv":
            params = parse_qs(parsed.query)
            raw_path = params.get("path", [""])[0]
            try:
                if raw_path == STAGING_DB_PATH:
                    _json_response(self, {"path": raw_path, "rows": _staging_summary()})
                    return
                path = _safe_path(raw_path)
                if path.name.startswith("review_decisions_"):
                    _json_response(self, {"path": str(path), "rows": _review_summary(path)})
                else:
                    _json_response(self, {"path": str(path), "rows": _read_csv(path)})
            except Exception as exc:  # noqa: BLE001 - local diagnostic API
                print(f"[ERROR] /api/csv 异常: {exc}")
                _error(self, str(exc), status=404)
            return
        if parsed.path == "/api/groups":
            params = parse_qs(parsed.query)
            try:
                raw_path = params.get("path", [""])[0]
                top_k = params.get("top_k", [""])[0]
                mode = params.get("mode", ["retrieval"])[0]
                term = _norm(params.get("q", [""])[0])
                decision = _norm(params.get("decision", [""])[0])
                client_token = params.get("client_token", [""])[0].strip()
                page = max(1, int(params.get("page", ["1"])[0]))
                page_size = min(200, max(10, int(params.get("page_size", ["50"])[0])))
                if raw_path == STAGING_DB_PATH:
                    _json_response(
                        self,
                        _staging_groups_response(mode, term, page, page_size, decision, client_token),
                    )
                    return
                path = _safe_path(raw_path)
                _json_response(self, _groups_response(path, top_k, mode, term, page, page_size, decision))
            except Exception as exc:  # noqa: BLE001 - local diagnostic API
                _error(self, str(exc), status=404)
            return
        if parsed.path == "/api/review-statuses":
            params = parse_qs(parsed.query)
            try:
                raw_ids = params.get("staging_ids", [""])[0]
                staging_ids = [int(value) for value in raw_ids.split(",") if value.strip()]
                if len(staging_ids) > 200:
                    raise ValueError("最多同步 200 条记录")
                client_token = params.get("client_token", [""])[0].strip()
                _json_response(self, {"rows": _review_statuses(staging_ids, client_token)})
            except Exception as exc:  # noqa: BLE001
                _error(self, str(exc), status=400)
            return
        if parsed.path == "/api/query":
            params = parse_qs(parsed.query)
            try:
                raw_path = params.get("path", [""])[0]
                top_k = params.get("top_k", [""])[0]
                query_id = params.get("query_id", [""])[0]
                if raw_path == STAGING_DB_PATH:
                    _json_response(self, _staging_query_rows(query_id))
                    return
                path = _safe_path(raw_path)
                _json_response(self, _query_rows_response(path, top_k, query_id))
            except Exception as exc:  # noqa: BLE001 - local diagnostic API
                _error(self, str(exc), status=404)
            return
        if parsed.path == "/api/text":
            params = parse_qs(parsed.query)
            raw_path = params.get("path", [""])[0]
            try:
                if raw_path.startswith(("http://", "https://")):
                    resp = requests.get(raw_path, timeout=10)
                    resp.raise_for_status()
                    _json_response(self, {"path": raw_path, "text": resp.content.decode("utf-8", errors="replace")})
                    return
                path = _safe_path(raw_path)
                text = path.read_text(encoding="utf-8", errors="replace")
                _json_response(self, {"path": str(path), "text": text})
            except Exception as exc:  # noqa: BLE001 - local diagnostic API
                _error(self, str(exc), status=404)
            return
        if parsed.path.startswith("/static/"):
            self._serve_file(ROOT / parsed.path.lstrip("/"))
            return
        _error(self, "not found", status=404)

    def do_POST(self) -> None:
        parsed = urlparse(self.path)
        if parsed.path == "/api/review-login":
            try:
                length = int(self.headers.get("Content-Length", "0"))
                payload = json.loads(self.rfile.read(length).decode("utf-8") or "{}")
                result = _review_login(
                    str(payload.get("reviewer") or ""),
                    str(payload.get("access_code") or ""),
                    str(payload.get("session_token") or ""),
                )
                _json_response(self, result)
            except ReviewAuthError as exc:
                _error(self, str(exc), status=401)
            except Exception as exc:  # noqa: BLE001
                _error(self, str(exc), status=400)
            return
        if parsed.path in {"/api/review-claim", "/api/review-heartbeat"}:
            try:
                length = int(self.headers.get("Content-Length", "0"))
                payload = json.loads(self.rfile.read(length).decode("utf-8") or "{}")
                reviewer = _require_review_session(payload)
                staging_id = int(payload.get("staging_id") or 0)
                client_token = str(payload.get("client_token") or "").strip()
                if not staging_id:
                    raise ValueError("staging_id is required")
                if parsed.path == "/api/review-claim":
                    current = _claim_staging_review(staging_id, reviewer, client_token)
                else:
                    current = _heartbeat_staging_review(staging_id, client_token)
                _json_response(self, {"record": current})
            except ReviewConflictError as exc:
                _conflict_response(self, exc)
            except ReviewAuthError as exc:
                _error(self, str(exc), status=401)
            except Exception as exc:  # noqa: BLE001
                _error(self, str(exc), status=400)
            return
        if parsed.path == "/api/staging-review":
            try:
                length = int(self.headers.get("Content-Length", "0"))
                payload = json.loads(self.rfile.read(length).decode("utf-8") or "{}")
                reviewer = _require_review_session(payload)
                staging_id = int(payload.get("staging_id") or 0)
                decision = str(payload.get("decision", "")).strip()
                note = str(payload.get("note", "")).strip()
                client_token = str(payload.get("client_token") or "").strip()
                expected_version = int(payload.get("expected_version", -1))
                if not staging_id:
                    raise ValueError("staging_id is required")
                if not reviewer or not client_token or expected_version < 0:
                    raise ValueError("reviewer, client_token and expected_version are required")
                if decision not in ("approved_import", "rejected_duplicate", "unsure", "pending", "deleted"):
                    raise ValueError(f"invalid decision: {decision}")
                result = _update_staging_review(
                    staging_id, decision, note, reviewer, expected_version, client_token
                )
                _json_response(self, {"record": result})
            except ReviewConflictError as exc:
                _conflict_response(self, exc)
            except ReviewAuthError as exc:
                _error(self, str(exc), status=401)
            except Exception as exc:  # noqa: BLE001
                _error(self, str(exc), status=400)
            return
        if parsed.path != "/api/import-reviewed":
            _error(self, "not found", status=404)
            return
        try:
            length = int(self.headers.get("Content-Length", "0"))
            payload = json.loads(self.rfile.read(length).decode("utf-8") or "{}")
            _require_review_session(payload)
            source_ids = payload.get("source_ids")
            if source_ids is not None and not isinstance(source_ids, list):
                raise ValueError("source_ids must be a list")
            result = _import_approved_staging(
                [str(value).strip() for value in source_ids if str(value).strip()]
                if source_ids
                else None
            )
            approved = [
                {"source_id": source_id, "review_decision": "import", "review_note": ""}
                for source_id in result["source_ids"]
            ]
            review_csv = _write_review_import_csv(approved)
            _json_response(self, {"review_csv": str(review_csv), **result})
        except ReviewAuthError as exc:
            _error(self, str(exc), status=401)
        except Exception as exc:  # noqa: BLE001 - local diagnostic API
            _error(self, str(exc), status=400)

    def _serve_file(self, path: Path) -> None:
        if not path.exists() or not path.is_file():
            _error(self, "not found", status=404)
            return
        content_type = mimetypes.guess_type(path.name)[0] or "application/octet-stream"
        body = path.read_bytes()
        self.send_response(200)
        self.send_header("Content-Type", content_type)
        if path.suffix.lower() in {".html", ".js", ".css"}:
            self.send_header("Cache-Control", "no-cache")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)


def main() -> None:
    parser = argparse.ArgumentParser(description="Serve L2 lyric review dashboard")
    parser.add_argument("--host", default="127.0.0.1")
    parser.add_argument("--port", type=int, default=8765)
    parser.add_argument(
        "--migrate-review-schema",
        action="store_true",
        help="add the staging-table columns/index required for collaborative reviewing",
    )
    args = parser.parse_args()

    _ensure_review_schema(apply_migration=args.migrate_review_schema)
    server = ThreadingHTTPServer((args.host, args.port), Handler)
    print(f"L2 dashboard: http://{args.host}:{args.port}")
    server.serve_forever()


if __name__ == "__main__":
    try:
        main()
    except KeyboardInterrupt:
        sys.exit(0)