#!/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).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(*) AS total_count,
      SUM(dedup_action = 'new') AS new_count,
      SUM(dedup_action = 'merge') AS merge_count,
      SUM(dedup_action = 'review') AS review_count
    FROM {TARGET_TABLE_NAME_TMP}
    """
    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]:
    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": str(row.get("lyrics_url") or ""),
        "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 ""),
        "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 ""
    count_sql = f"SELECT COUNT(*) AS n FROM {TARGET_TABLE_NAME_TMP} {where}"
    sql = f"""
    SELECT staging_id, source_song_id, name, lyricist, composer, dedup_action,
           biz_review_status, l1_matched_id
    FROM {TARGET_TABLE_NAME_TMP}
    {where}
    ORDER BY
      CASE dedup_action WHEN 'review' THEN 0 WHEN 'merge' THEN 1 ELSE 2 END,
      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)])
        rows = cursor.fetchall()
    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()
    return {"rows": [_staging_dashboard_row(row)] if row else []}


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'
    """
    mark_sql = f"""
    UPDATE {TARGET_TABLE_NAME_TMP}
    SET staging_status = 'imported',
        imported_song_id = id,
        error_message = NULL
    WHERE source_song_id IN ({placeholders})
      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
        cursor.execute(mark_sql, source_ids)
        conn.commit()
    return {"approved_count": approved_count, "inserted_count": inserted_count}


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