#!/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,
      SUM(CASE WHEN dedup_action = 'skip' THEN 1 ELSE 0 END) AS skip_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),
        "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),
    }]


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",
        "audio_url": str(row.get("audio_url") or ""),
    }


def _staging_groups_response(mode: str, term: str, page: int, page_size: int, decision: str = '') -> 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')")
    # 决策筛选：l1_hit / l2_hit / conflict / new / skip
    if decision == 'l1_hit':
        where_clauses.append("l1_matched_id IS NOT NULL")
    elif decision == 'l2_duplicate':
        where_clauses.append("dedup_action = 'merge'")
    elif decision == 'l2_review':
        where_clauses.append("dedup_action = 'review'")
    elif decision == 'conflict':
        where_clauses.append("(l1_matched_id IS NOT NULL AND dedup_action = 'new') OR (l1_matched_id IS NULL AND dedup_action IN ('merge', 'review'))")
    elif decision == 'new':
        where_clauses.append("dedup_action = 'new' AND l1_matched_id IS NULL")
    elif decision == 'skip':
        where_clauses.append("dedup_action = 'skip'")
    # 兼容旧的 dedup_action 筛选
    elif decision and decision in ('merge', 'review'):
        where_clauses.append("dedup_action = %s")
        params.append(decision)
    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")),
            "decision": row.get("dedup_action") or "",
        }
        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 _update_staging_review(source_song_id: str, decision: str, note: str, reviewer: str = "dashboard") -> dict[str, object]:
    """更新暂存表的人工审核状态。

    decision: 'approved_import' | 'rejected_duplicate' | 'unsure' | 'pending' (撤销) | 'deleted'
    """
    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]
    elif decision == "deleted":
        # 删除：不限定 dedup_action，对 review/skip 等各种 action 均可执行
        sql = f"""
        UPDATE {TARGET_TABLE_NAME_TMP}
        SET staging_status = 'deleted',
            biz_review_status = 'deleted',
            reviewed_by = %s,
            reviewed_at = NOW(),
            biz_review_note = NULLIF(%s, '')
        WHERE source_song_id = %s
        """
        params = [reviewer, 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,
                "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 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,
    )
    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])
                decision = _norm(params.get("decision", [""])[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, decision))
                    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/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", "deleted"):
                    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)