#!/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 subprocess
import sys
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 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")

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")
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",
]


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("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():
    return pymysql.connect(**TARGET_DB_CONFIG)


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
    FROM (
      SELECT source_song_id, dedup_action
      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),
        "new_count": str(row.get("new_count") or 0),
        "total_count": str(row.get("total_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 ""),
        "l1_metadata_match": "1" if row.get("l1_matched_id") else "0",
        "l1_l2_conflict": "0",
    }


def _staging_groups_response(mode: str, term: str, page: int, page_size: int) -> dict[str, object]:
    where_clauses = []
    params: list[object] = []
    if term:
        where_clauses.append("(CAST(source_song_id AS CHAR) LIKE %s OR name LIKE %s OR lyricist LIKE %s OR composer LIKE %s)")
        like = f"%{term}%"
        params.extend([like, like, like, like])
    # hits 模式下只展示有命中的记录（merge/review）
    if mode == "hits":
        where_clauses.append("dedup_action IN ('merge', 'review')")
    where = "WHERE " + " AND ".join(where_clauses) if where_clauses else ""
    # 同一首歌可能因多次导入批次产生多行，按 source_song_id 去重取最新一行
    count_sql = f"SELECT COUNT(DISTINCT source_song_id) AS n FROM {TARGET_TABLE_NAME_TMP} {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.l1_matched_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}
        {where}
        GROUP BY source_song_id
    ) latest ON s.staging_id = latest.max_staging_id
    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, [*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")),
        }
        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 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 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)
    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
    return {"rows": [dashboard_row]}


def _update_staging_review(source_song_id: str, decision: str, note: str, reviewer: str = "dashboard") -> dict[str, object]:
    """更新暂存表的人工审核状态。

    decision: 'approved_import' | 'rejected_duplicate' | 'unsure' | 'pending' (撤销)
    """
    if decision == "pending":
        # 撤销审核
        sql = f"""
        UPDATE {TARGET_TABLE_NAME_TMP}
        SET biz_review_status = 'pending',
            reviewed_by = NULL,
            reviewed_at = NULL,
            biz_review_note = NULLIF(%s, '')
        WHERE source_song_id = %s AND dedup_action = 'review'
        """
        params = [note, source_song_id]
    else:
        sql = f"""
        UPDATE {TARGET_TABLE_NAME_TMP}
        SET biz_review_status = %s,
            reviewed_by = %s,
            reviewed_at = NOW(),
            biz_review_note = NULLIF(%s, '')
        WHERE source_song_id = %s AND dedup_action = 'review'
        """
        params = [decision, reviewer, note, source_song_id]
    with _target_conn() as conn, conn.cursor() as cursor:
        cursor.execute(sql, params)
        affected = cursor.rowcount
        conn.commit()
    return {"source_song_id": source_song_id, "decision": decision, "affected": affected}


def _import_approved_staging(source_ids: list[str], reviewer: str = "dashboard") -> dict[str, object]:
    if not source_ids:
        raise ValueError("没有可入库的 source_id")
    placeholders = ",".join(["%s"] * len(source_ids))
    columns = ", ".join(f"`{column}`" for column in TARGET_COLUMNS)
    select_columns = ", ".join(f"s.`{column}`" for column in TARGET_COLUMNS)
    update_imported_sql = f"""
    UPDATE {TARGET_TABLE_NAME_TMP}
    SET biz_review_status = 'approved_import',
        reviewed_by = %s,
        reviewed_at = NOW()
    WHERE source_song_id IN ({placeholders})
      AND dedup_action = 'review'
      AND biz_review_status IN ('pending', 'unsure')
    """
    insert_sql = f"""
    INSERT INTO {TARGET_TABLE_NAME} ({columns})
    SELECT {select_columns}
    FROM {TARGET_TABLE_NAME_TMP} s
    WHERE s.source_song_id IN ({placeholders})
      AND s.dedup_action = 'review'
      AND s.biz_review_status = 'approved_import'
      AND s.staging_status <> 'imported'
    """
    # 入库后查询目标表实际 ID，正确回填 imported_song_id（不能用暂存表自己的 id）
    lookup_target_id_sql = f"""
    SELECT source_song_id, id AS target_id
    FROM {TARGET_TABLE_NAME}
    WHERE source_song_id IN ({placeholders})
    """
    mark_sql_template = f"""
    UPDATE {TARGET_TABLE_NAME_TMP}
    SET staging_status = 'imported',
        imported_song_id = %s,
        error_message = NULL
    WHERE source_song_id = %s
      AND dedup_action = 'review'
      AND biz_review_status = 'approved_import'
    """
    with _target_conn() as conn, conn.cursor() as cursor:
        cursor.execute(update_imported_sql, [reviewer, *source_ids])
        approved_count = cursor.rowcount
        cursor.execute(insert_sql, source_ids)
        inserted_count = cursor.rowcount
        # 从目标表查出实际生成的主键 ID，逐条回填 imported_song_id
        cursor.execute(lookup_target_id_sql, source_ids)
        id_map = {str(row["source_song_id"]): row["target_id"] for row in cursor.fetchall()}
        imported_updated = 0
        for sid, target_id in id_map.items():
            cursor.execute(mark_sql_template, [target_id, sid])
            imported_updated += cursor.rowcount
        conn.commit()
    return {
        "approved_count": approved_count,
        "inserted_count": inserted_count,
        "imported_song_ids_updated": imported_updated,
    }


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,
            }
            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) -> 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")]
    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,
    )
    return {
        "command": " ".join(cmd),
        "returncode": result.returncode,
        "output": result.stdout[-12000:],
    }


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])
                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))
                    return
                path = _safe_path(raw_path)
                _json_response(self, _groups_response(path, top_k, mode, term, page, page_size))
            except Exception as exc:  # noqa: BLE001 - local diagnostic API
                _error(self, str(exc), status=404)
            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/staging-review":
            try:
                length = int(self.headers.get("Content-Length", "0"))
                payload = json.loads(self.rfile.read(length).decode("utf-8") or "{}")
                source_id = str(payload.get("source_id", "")).strip()
                decision = str(payload.get("decision", "")).strip()
                note = str(payload.get("note", "")).strip()
                reviewer = str(payload.get("reviewer", "dashboard")).strip() or "dashboard"
                if not source_id:
                    raise ValueError("source_id is required")
                if decision not in ("approved_import", "rejected_duplicate", "unsure", "pending"):
                    raise ValueError(f"invalid decision: {decision}")
                result = _update_staging_review(source_id, decision, note, reviewer)
                _json_response(self, result)
            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 "{}")
            rows = payload.get("rows") or []
            if not isinstance(rows, list):
                raise ValueError("rows must be a list")
            approved = [
                {
                    "source_id": str(row.get("source_id", "")).strip(),
                    "review_decision": "import",
                    "review_note": str(row.get("review_note", "")).strip(),
                }
                for row in rows
                if str(row.get("source_id", "")).strip()
                and str(row.get("review_decision", "")).strip().lower() in {"import", "导入", "not_duplicate"}
            ]
            if not approved:
                raise ValueError("没有可入库的审核通过样本")
            review_csv = _write_review_import_csv(approved)
            result = _import_approved_staging([row["source_id"] for row in approved])
            _json_response(self, {"review_csv": str(review_csv), **result})
        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)
        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)
    args = parser.parse_args()

    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)