hot100.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Random LeetCode Hot 100 picker with progress and wrong-answer review tracking."""

from __future__ import annotations

import argparse
import http.server
import json
import mimetypes
import os
import random
import re
import shutil
import socketserver
import subprocess
import sys
import threading
import urllib.error
import urllib.parse
import urllib.request
import uuid
import webbrowser
from datetime import datetime, timezone
from pathlib import Path
from typing import Any


ROOT = Path(__file__).resolve().parent
DATA_PATH = ROOT / "hot100.json"
STATE_PATH = ROOT / "progress.json"
WEB_ROOT = ROOT / "web"
REVIEW_JOBS_PATH = ROOT / "review_jobs"

PLAN_SLUG = "top-100-liked"
LEETCODE_GRAPHQL = "https://leetcode.com/graphql/"
CN_STUDY_PLAN_URL = "https://leetcode.cn/studyplan/top-100-liked/"
CN_PROBLEM_URL = "https://leetcode.cn/problems/{slug}/"
LEETCODE_SLUG_RE = re.compile(r"leetcode\.cn/problems/([^/?#)\s]+)")
CODEX_EXEC_TIMEOUT_SECONDS = 1800

STATUSES = ("todo", "picked", "done", "skipped")
DIFFICULTY_TEXT = {
    "EASY": "Easy",
    "MEDIUM": "Medium",
    "HARD": "Hard",
}


QUERY = """
query studyPlanV2Detail($slug: String!) {
  studyPlanV2Detail(planSlug: $slug) {
    slug
    name
    questionNum
    planSubGroups {
      slug
      name
      questions {
        titleSlug
        title
        translatedTitle
        questionFrontendId
        difficulty
      }
    }
  }
}
""".strip()


class Hot100Error(RuntimeError):
    """Expected user-facing error."""


def now_iso() -> str:
    return datetime.now(timezone.utc).astimezone().isoformat(timespec="seconds")


def read_json(path: Path) -> dict[str, Any]:
    try:
        with path.open("r", encoding="utf-8") as f:
            return json.load(f)
    except FileNotFoundError:
        raise Hot100Error(f"找不到文件：{path}")
    except json.JSONDecodeError as exc:
        raise Hot100Error(f"JSON 文件损坏：{path} ({exc})")


def write_json(path: Path, payload: dict[str, Any]) -> None:
    tmp_path = path.with_suffix(path.suffix + ".tmp")
    with tmp_path.open("w", encoding="utf-8") as f:
        json.dump(payload, f, ensure_ascii=False, indent=2)
        f.write("\n")
    tmp_path.replace(path)


def study_root() -> Path | None:
    # Standalone layout: hot100.py lives beside 知识点/ and 练习/.
    if (ROOT / "练习").is_dir() and (ROOT / "知识点").is_dir():
        return ROOT
    if ROOT.parent.name == "练习" and ROOT.parent.parent.exists():
        return ROOT.parent.parent
    for parent in ROOT.parents:
        if (parent / "练习").is_dir() and (parent / "知识点").is_dir():
            return parent
    return None


def local_note_dirs() -> dict[str, Path]:
    root = study_root()
    if not root:
        return {}
    candidates = {
        "练习": root / "练习",
        "知识点": root / "知识点",
    }
    return {name: path for name, path in candidates.items() if path.is_dir()}


def build_related_notes_by_slug() -> dict[str, list[dict[str, str]]]:
    related: dict[str, list[dict[str, str]]] = {}
    for kind, directory in local_note_dirs().items():
        for path in sorted(directory.glob("*.md")):
            try:
                text = path.read_text(encoding="utf-8")
            except OSError:
                continue
            for slug in sorted(set(LEETCODE_SLUG_RE.findall(text))):
                related.setdefault(slug, []).append(
                    {
                        "kind": kind,
                        "title": path.stem,
                        "url": f"/notes/{urllib.parse.quote(kind)}/{urllib.parse.quote(path.name)}",
                        "path": str(path),
                    }
                )
    return related


def attach_related_notes(data: dict[str, Any]) -> dict[str, Any]:
    related = build_related_notes_by_slug()
    for q in data.get("questions", []):
        q["related_notes"] = related.get(q.get("slug"), [])
    return data


def project_root() -> Path:
    return study_root() or ROOT


def codex_executable() -> str:
    candidates = [
        os.environ.get("HOT100_CODEX_BIN"),
        shutil.which("codex"),
        "/Applications/Codex.app/Contents/Resources/codex",
    ]
    for candidate in candidates:
        if candidate and Path(candidate).exists():
            return candidate
    raise Hot100Error("找不到 Codex 可执行文件。可设置 HOT100_CODEX_BIN 指向 codex。")


def safe_job_id(slug: str) -> str:
    timestamp = datetime.now().strftime("%Y%m%d-%H%M%S")
    slug_part = re.sub(r"[^a-zA-Z0-9-]+", "-", slug).strip("-")[:48] or "problem"
    return f"{timestamp}-{slug_part}-{uuid.uuid4().hex[:8]}"


def job_dir(job_id: str) -> Path:
    if not re.fullmatch(r"[0-9A-Za-z_.-]+", job_id):
        raise Hot100Error("任务 ID 无效")
    return REVIEW_JOBS_PATH / job_id


def read_text_if_exists(path: Path, limit: int | None = None) -> str:
    if not path.exists():
        return ""
    text = path.read_text(encoding="utf-8", errors="replace")
    if limit is not None and len(text) > limit:
        return text[-limit:]
    return text


def write_job_status(path: Path, payload: dict[str, Any]) -> None:
    previous = {}
    status_path = path / "status.json"
    if status_path.exists():
        try:
            previous = read_json(status_path)
        except Hot100Error:
            previous = {}
    previous.update(payload)
    previous["updated_at"] = now_iso()
    write_json(status_path, previous)


def read_job_status(job_id: str) -> dict[str, Any]:
    path = job_dir(job_id)
    status_path = path / "status.json"
    if not status_path.exists():
        raise Hot100Error(f"找不到复盘任务：{job_id}")
    payload = read_json(status_path)
    payload["job_id"] = job_id
    return payload


def read_job(job_id: str) -> dict[str, Any]:
    path = job_dir(job_id)
    payload = read_job_status(job_id)
    payload["log_tail"] = read_text_if_exists(path / "codex.log", limit=12000)
    payload["result"] = read_text_if_exists(path / "result.md", limit=24000)
    return payload


def review_jobs_for_slug(slug: str, limit: int = 5) -> list[dict[str, Any]]:
    if not REVIEW_JOBS_PATH.exists():
        return []
    jobs = []
    for path in sorted(REVIEW_JOBS_PATH.iterdir(), reverse=True):
        if not path.is_dir():
            continue
        try:
            job = read_job_status(path.name)
        except Hot100Error:
            continue
        if job.get("question", {}).get("slug") != slug:
            continue
        jobs.append(job)
        if len(jobs) >= limit:
            break
    return jobs


def review_jobs_by_slug(limit: int = 5) -> dict[str, list[dict[str, Any]]]:
    if not REVIEW_JOBS_PATH.exists():
        return {}

    grouped: dict[str, list[dict[str, Any]]] = {}
    for path in sorted(REVIEW_JOBS_PATH.iterdir(), reverse=True):
        if not path.is_dir():
            continue
        try:
            job = read_job_status(path.name)
        except Hot100Error:
            continue
        slug = job.get("question", {}).get("slug")
        if not slug:
            continue
        jobs = grouped.setdefault(slug, [])
        if len(jobs) < limit:
            jobs.append(job)
    return grouped


def review_history_for_question(q: dict[str, Any], jobs: list[dict[str, Any]] | None = None) -> dict[str, Any]:
    notes = q.get("related_notes") or []
    jobs = jobs if jobs is not None else review_jobs_for_slug(q.get("slug", ""))
    succeeded_jobs = [job for job in jobs if job.get("status") == "succeeded"]
    return {
        "mode": "incremental" if notes or succeeded_jobs else "new",
        "has_related_notes": bool(notes),
        "has_successful_review": bool(succeeded_jobs),
        "related_notes_count": len(notes),
        "jobs": [
            {
                "job_id": job.get("job_id"),
                "status": job.get("status"),
                "created_at": job.get("created_at"),
                "input_path": job.get("input_path"),
                "result_path": job.get("result_path"),
            }
            for job in jobs
        ],
    }


def list_review_jobs(limit: int = 8) -> list[dict[str, Any]]:
    if not REVIEW_JOBS_PATH.exists():
        return []
    jobs = []
    for path in sorted(REVIEW_JOBS_PATH.iterdir(), reverse=True):
        if not path.is_dir():
            continue
        try:
            jobs.append(read_job(path.name))
        except Hot100Error:
            continue
        if len(jobs) >= limit:
            break
    return jobs


def related_note_lines(q: dict[str, Any]) -> str:
    notes = q.get("related_notes") or []
    if not notes:
        return "- 暂无已关联本地资料。请创建一对新的 `知识点/主题名.md` 与 `练习/主题名.md`。"
    return "\n".join(f"- {note['kind']}：`{note['path']}`" for note in notes)


def review_job_lines(history: dict[str, Any]) -> str:
    jobs = history.get("jobs") or []
    if not jobs:
        return "- 暂无历史复盘任务。"
    return "\n".join(
        "- {job_id} / {status} / input: `{input_path}` / result: `{result_path}`".format(
            job_id=job.get("job_id"),
            status=job.get("status"),
            input_path=job.get("input_path"),
            result_path=job.get("result_path"),
        )
        for job in jobs
    )


def build_review_prompt(q: dict[str, Any], input_path: Path, history: dict[str, Any]) -> str:
    root = project_root()
    mode_text = "增量分析" if history.get("mode") == "incremental" else "首次复盘"
    return f"""
你是 Codex，正在维护一个中文刷题复习资料库。请基于用户刚完成的一道 LeetCode Hot 100 题的材料，生成或更新长期复习文档。

工作目录：`{root}`

必须先阅读并遵守：
- `AGENTS.md`
- `模板/知识点模板.md`
- `模板/练习模板.md`

题目信息：
- 题号：{q.get('id')}
- 标题：{q.get('title')}
- 英文 slug：{q.get('slug')}
- 难度：{q.get('difficulty')}
- Hot100 分组：{q.get('topic')}
- 原题链接：{q.get('url')}

用户提交的复盘材料保存在：
- `{input_path}`

重要前提：
- 用户输入默认就是针对上面这道“当前题目”的材料，通常只会包含通过代码、踩坑、疑问和补充说明。
- 不要要求用户再粘贴题面；请使用上面的题号、标题、slug、链接和 Hot100 分组作为题目上下文。
- 不要把用户代码误判成独立题目；它一定优先归属于当前题。

复盘模式：{mode_text}

当前已关联的本地资料：
{related_note_lines(q)}

当前题的历史复盘任务：
{review_job_lines(history)}

你的任务：
1. 先检查这道题是否已经复盘过：
   - 如果上方已有本地资料或成功的历史复盘任务，本次必须按“增量分析”处理。
   - 如果没有任何资料或历史成功任务，本次才按“首次复盘”处理。
2. 阅读用户材料，识别其中的通过代码、踩坑、疑问、复杂度、可改进点。
3. 首次复盘时：创建一对新的 `知识点/主题名.md` 与 `练习/主题名.md`。
4. 增量分析时：先阅读已有资料和历史复盘输入/输出，只把新代码、新坑点、新改进建议合并进去；不要无脑整篇重写，不要覆盖掉已经有价值的内容。
5. 文件风格要和已有复习资料一致：中文、短结论、面试高频、结构清晰，不堆完整原题。
6. 知识点文件至少覆盖：
   - 这题为什么属于某个核心模式
   - 用户代码是否正确，复杂度是什么
   - 可改进空间和更标准写法
   - 易错点、判断口诀、最小复习清单
   - 原题链接和题意摘要
7. 练习文件至少覆盖：
   - 围绕本题核心模式设计 5 到 8 个短练习
   - 每题使用 `<details><summary>查看答案</summary> ... </details>`
   - 综合例题放当前题，并说明为什么当前主题能解决这题
8. 如果新建主题，更新：
   - `README.md`
   - `知识点/README.md`
   - `练习/README.md`
9. 不要改无关文件，不要删除用户已有内容；必要时做小范围整理。

完成后，在最终回答里先说明本次是“首次复盘”还是“增量分析”，再列出你创建或修改的文件，并用 3 到 6 条总结本次代码复盘结论。
""".strip()


def create_review_input(q: dict[str, Any], material: str, path: Path) -> None:
    content = f"""# Hot100 复盘输入：{q.get('id')}. {q.get('title')}

- 题目：{q.get('url')}
- slug：{q.get('slug')}
- 难度：{q.get('difficulty')}
- Hot100 分组：{q.get('topic')}
- 创建时间：{now_iso()}
- 说明：下面的用户输入默认针对这道当前题，用户通常不会再次粘贴题面。

## 用户输入

{material.strip()}
"""
    path.write_text(content, encoding="utf-8")


def run_codex_review(path: Path) -> None:
    prompt_path = path / "prompt.md"
    result_path = path / "result.md"
    log_path = path / "codex.log"
    write_job_status(
        path,
        {
            "status": "running",
            "started_at": now_iso(),
            "finished_at": None,
            "return_code": None,
            "error": None,
        },
    )

    if os.environ.get("HOT100_REVIEW_DRY_RUN") == "1":
        result_path.write_text("DRY RUN：已生成复盘任务，但未真正调用 Codex。\n", encoding="utf-8")
        log_path.write_text("dry run\n", encoding="utf-8")
        write_job_status(path, {"status": "succeeded", "finished_at": now_iso(), "return_code": 0})
        return

    try:
        cmd = [
            codex_executable(),
            "--ask-for-approval",
            "never",
            "exec",
            "-C",
            str(project_root()),
            "--sandbox",
            "workspace-write",
            "--skip-git-repo-check",
            "-o",
            str(result_path),
            "-",
        ]
    except Hot100Error as exc:
        write_job_status(path, {"status": "failed", "finished_at": now_iso(), "error": str(exc)})
        return

    prompt = prompt_path.read_text(encoding="utf-8")
    with log_path.open("w", encoding="utf-8") as log:
        log.write(f"$ {' '.join(cmd)}\n\n")
        log.flush()
        try:
            completed = subprocess.run(
                cmd,
                input=prompt,
                text=True,
                cwd=str(project_root()),
                stdout=log,
                stderr=subprocess.STDOUT,
                timeout=CODEX_EXEC_TIMEOUT_SECONDS,
            )
        except subprocess.TimeoutExpired:
            write_job_status(path, {"status": "failed", "finished_at": now_iso(), "error": "Codex 复盘超时"})
            return
        except OSError as exc:
            write_job_status(path, {"status": "failed", "finished_at": now_iso(), "error": str(exc)})
            return

    status = "succeeded" if completed.returncode == 0 else "failed"
    payload: dict[str, Any] = {
        "status": status,
        "finished_at": now_iso(),
        "return_code": completed.returncode,
    }
    if completed.returncode != 0:
        payload["error"] = f"Codex 返回非 0 状态码：{completed.returncode}"
    write_job_status(path, payload)


def start_review_job(q: dict[str, Any], material: str) -> dict[str, Any]:
    material = material.strip()
    if not material:
        raise Hot100Error("复盘材料不能为空")

    REVIEW_JOBS_PATH.mkdir(parents=True, exist_ok=True)
    review_job_id = safe_job_id(q["slug"])
    path = job_dir(review_job_id)
    path.mkdir()

    input_path = path / "input.md"
    prompt_path = path / "prompt.md"
    create_review_input(q, material, input_path)
    history = review_history_for_question(q)
    prompt_path.write_text(build_review_prompt(q, input_path, history), encoding="utf-8")
    write_job_status(
        path,
        {
            "job_id": review_job_id,
            "status": "queued",
            "created_at": now_iso(),
            "review_mode": history.get("mode"),
            "has_previous_review": history.get("mode") == "incremental",
            "question": {
                "id": q.get("id"),
                "title": q.get("title"),
                "slug": q.get("slug"),
                "url": q.get("url"),
            },
            "input_path": str(input_path),
            "prompt_path": str(prompt_path),
            "result_path": str(path / "result.md"),
            "log_path": str(path / "codex.log"),
        },
    )

    thread = threading.Thread(target=run_codex_review, args=(path,), daemon=True)
    thread.start()
    return read_job(review_job_id)


def retry_review_job(review_job_id: str) -> dict[str, Any]:
    path = job_dir(review_job_id)
    status_path = path / "status.json"
    prompt_path = path / "prompt.md"
    if not status_path.exists():
        raise Hot100Error(f"找不到复盘任务：{review_job_id}")
    if not prompt_path.exists():
        raise Hot100Error(f"找不到复盘提示词，无法重试：{review_job_id}")

    job = read_job(review_job_id)
    status = job.get("status")
    if status in ("queued", "running"):
        raise Hot100Error("这个复盘任务还在执行中，不需要重试")
    if status != "failed":
        raise Hot100Error("只有失败的复盘任务可以重试")

    retry_count = int(job.get("retry_count") or 0) + 1
    write_job_status(
        path,
        {
            "status": "queued",
            "retried_at": now_iso(),
            "retry_count": retry_count,
            "finished_at": None,
            "return_code": None,
            "error": None,
        },
    )

    thread = threading.Thread(target=run_codex_review, args=(path,), daemon=True)
    thread.start()
    return read_job(review_job_id)


def fetch_hot100() -> dict[str, Any]:
    body = json.dumps({"query": QUERY, "variables": {"slug": PLAN_SLUG}}).encode("utf-8")
    request = urllib.request.Request(
        LEETCODE_GRAPHQL,
        data=body,
        headers={
            "content-type": "application/json",
            "user-agent": "hot100-random/1.0",
        },
        method="POST",
    )
    try:
        with urllib.request.urlopen(request, timeout=20) as response:
            payload = json.loads(response.read().decode("utf-8"))
    except urllib.error.URLError as exc:
        raise Hot100Error(f"刷新题单失败：无法访问 LeetCode ({exc})")
    except json.JSONDecodeError as exc:
        raise Hot100Error(f"刷新题单失败：LeetCode 返回了无法解析的数据 ({exc})")

    if payload.get("errors"):
        raise Hot100Error(f"刷新题单失败：{payload['errors']}")

    plan = payload.get("data", {}).get("studyPlanV2Detail")
    if not plan:
        raise Hot100Error("刷新题单失败：没有拿到 studyPlanV2Detail 数据")

    questions: list[dict[str, Any]] = []
    seen: set[str] = set()
    for group in plan.get("planSubGroups") or []:
        topic = group.get("name") or "Unknown"
        for q in group.get("questions") or []:
            slug = q.get("titleSlug")
            if not slug or slug in seen:
                continue
            seen.add(slug)
            questions.append(
                {
                    "id": str(q.get("questionFrontendId") or ""),
                    "slug": slug,
                    "title": q.get("translatedTitle") or q.get("title") or slug,
                    "english_title": q.get("title") or slug,
                    "difficulty": DIFFICULTY_TEXT.get(q.get("difficulty"), q.get("difficulty") or ""),
                    "topic": topic,
                    "url": CN_PROBLEM_URL.format(slug=slug),
                }
            )

    if len(questions) != 100:
        raise Hot100Error(f"刷新题单失败：预期 100 题，实际拿到 {len(questions)} 题")

    return {
        "plan_slug": PLAN_SLUG,
        "plan_name": plan.get("name") or "Top 100 Liked",
        "source": "https://leetcode.com/studyplan/top-100-liked/",
        "cn_study_plan": CN_STUDY_PLAN_URL,
        "fetched_at": now_iso(),
        "questions": questions,
    }


def extra_questions_from_data(data: dict[str, Any]) -> list[dict[str, Any]]:
    extra_questions = data.get("extra_questions", [])
    if extra_questions is None:
        return []
    if not isinstance(extra_questions, list):
        raise Hot100Error("题单文件无效：extra_questions 必须是数组")
    for q in extra_questions:
        if not isinstance(q, dict):
            raise Hot100Error("题单文件无效：extra_questions 里必须都是题目对象")
    return extra_questions


def merge_question_pools(data: dict[str, Any]) -> dict[str, Any]:
    questions = data.get("questions")
    if not isinstance(questions, list) or not questions:
        raise Hot100Error(f"题单文件无效：{DATA_PATH}")

    merged: list[dict[str, Any]] = []
    seen: set[str] = set()
    for q in [*questions, *extra_questions_from_data(data)]:
        if not isinstance(q, dict):
            raise Hot100Error("题单文件无效：questions 里必须都是题目对象")
        slug = str(q.get("slug") or "").strip()
        if not slug:
            raise Hot100Error("题单文件无效：题目缺少 slug")
        if slug in seen:
            continue
        seen.add(slug)
        merged.append(q)

    return {**data, "questions": merged}


def refresh_data(path: Path) -> dict[str, Any]:
    previous_data = read_json(path) if path.exists() else {}
    extra_questions = extra_questions_from_data(previous_data)
    aliases_by_slug = {
        str(q.get("slug")): q.get("aliases")
        for q in previous_data.get("questions", [])
        if isinstance(q, dict) and q.get("slug") and isinstance(q.get("aliases"), list)
    }
    data = fetch_hot100()
    for question in data["questions"]:
        aliases = aliases_by_slug.get(question.get("slug"))
        if aliases:
            question["aliases"] = aliases
    if extra_questions:
        data["extra_questions"] = extra_questions
    write_json(path, data)
    return merge_question_pools(data)


def load_data(path: Path = DATA_PATH) -> dict[str, Any]:
    if not path.exists():
        print("未找到本地题单，正在首次刷新...")
        data = fetch_hot100()
        write_json(path, data)
        return attach_related_notes(merge_question_pools(data))
    data = read_json(path)
    return attach_related_notes(merge_question_pools(data))


def empty_state() -> dict[str, Any]:
    timestamp = now_iso()
    return {
        "version": 1,
        "plan_slug": PLAN_SLUG,
        "created_at": timestamp,
        "updated_at": timestamp,
        "last_selected": None,
        "wrong_review_cycle": 1,
        "items": {},
    }


def load_state(path: Path = STATE_PATH) -> dict[str, Any]:
    if not path.exists():
        return empty_state()
    state = read_json(path)
    if state.get("plan_slug") != PLAN_SLUG or not isinstance(state.get("items"), dict):
        raise Hot100Error(f"状态文件无效或不属于 {PLAN_SLUG}：{path}")
    return state


def save_state(state: dict[str, Any], path: Path = STATE_PATH) -> None:
    state["updated_at"] = now_iso()
    write_json(path, state)


def questions_by_slug(data: dict[str, Any]) -> dict[str, dict[str, Any]]:
    return {q["slug"]: q for q in data["questions"]}


def question_status(state: dict[str, Any], slug: str) -> str:
    return state.get("items", {}).get(slug, {}).get("status", "todo")


def question_is_wrong(state: dict[str, Any], slug: str) -> bool:
    return state.get("items", {}).get(slug, {}).get("wrong") is True


def find_question(data: dict[str, Any], token: str) -> dict[str, Any]:
    token = token.strip()
    if not token:
        raise Hot100Error("题目标识不能为空")

    needle = token.lower()
    matches = []
    for q in data["questions"]:
        aliases = q.get("aliases") or []
        if not isinstance(aliases, list):
            aliases = []
        values = [
            q.get("slug"),
            q.get("id"),
            q.get("title"),
            q.get("english_title"),
            *aliases,
        ]
        terms = [str(value).strip().lower() for value in values if str(value or "").strip()]
        if needle in terms:
            return q
        if any(needle in term for term in terms):
            matches.append(q)

    if len(matches) == 1:
        return matches[0]
    if matches:
        preview = ", ".join(format_label(q) for q in matches[:8])
        raise Hot100Error(f"匹配到多道题，请更精确一点：{preview}")
    raise Hot100Error(f"找不到题目：{token}")


def latest_selected_question(data: dict[str, Any], state: dict[str, Any]) -> dict[str, Any]:
    slug = state.get("last_selected")
    if not slug:
        raise Hot100Error("还没有随机抽过题。先运行：python3 hot100.py next")
    by_slug = questions_by_slug(data)
    if slug not in by_slug:
        raise Hot100Error(f"上次抽到的题不在当前题单中：{slug}")
    return by_slug[slug]


def ensure_item(state: dict[str, Any], slug: str) -> dict[str, Any]:
    items = state.setdefault("items", {})
    if slug not in items:
        items[slug] = {
            "status": "todo",
            "picked_at": None,
            "done_at": None,
            "skipped_at": None,
            "attempts": 0,
            "note": "",
            "wrong": False,
            "wrong_at": None,
            "wrong_cleared_at": None,
            "wrong_reviewed_at": None,
            "wrong_review_count": 0,
            "wrong_review_cycle": 0,
        }
    return items[slug]


def set_status(state: dict[str, Any], slug: str, status: str, note: str | None = None) -> None:
    if status not in STATUSES:
        raise Hot100Error(f"未知状态：{status}")

    item = ensure_item(state, slug)
    previous = item.get("status", "todo")
    item["status"] = status

    if status == "picked" and previous != "picked":
        item["picked_at"] = now_iso()
        item["attempts"] = int(item.get("attempts") or 0) + 1
        item["done_at"] = None
        item["skipped_at"] = None
    elif status == "done":
        item["done_at"] = now_iso()
        item["skipped_at"] = None
    elif status == "skipped":
        item["skipped_at"] = now_iso()
    elif status == "todo":
        item["done_at"] = None
        item["skipped_at"] = None

    if note is not None:
        item["note"] = note


def set_wrong(state: dict[str, Any], slug: str, wrong: bool, note: str | None = None) -> None:
    item = ensure_item(state, slug)
    was_wrong = item.get("wrong") is True
    item["wrong"] = wrong

    if wrong and not was_wrong:
        item["wrong_at"] = now_iso()
        item["wrong_cleared_at"] = None
        item["wrong_review_cycle"] = 0
    elif not wrong and was_wrong:
        item["wrong_cleared_at"] = now_iso()

    if note is not None:
        item["note"] = note


def format_label(q: dict[str, Any]) -> str:
    return f"{q['id']}. {q['title']} ({q['difficulty']})"


def print_question(
    q: dict[str, Any],
    status: str,
    show_topic: bool = False,
    wrong: bool = False,
) -> None:
    print(format_label(q))
    wrong_text = " | 错题" if wrong else ""
    print(f"状态：{status}{wrong_text}")
    if show_topic:
        print(f"分组：{q['topic']}")
    print(f"链接：{q['url']}")


def summarize(data: dict[str, Any], state: dict[str, Any]) -> dict[str, int]:
    counts = {status: 0 for status in STATUSES}
    for q in data["questions"]:
        counts[question_status(state, q["slug"])] += 1
    counts["total"] = len(data["questions"])
    counts["wrong"] = sum(1 for q in data["questions"] if question_is_wrong(state, q["slug"]))
    counts["wrong_review_remaining"] = wrong_review_remaining(data, state)
    return counts


def public_question(
    q: dict[str, Any],
    state: dict[str, Any],
    review_jobs: list[dict[str, Any]] | None = None,
) -> dict[str, Any]:
    item = state.get("items", {}).get(q["slug"], {})
    return {
        **q,
        "status": question_status(state, q["slug"]),
        "note": item.get("note", ""),
        "attempts": item.get("attempts", 0),
        "picked_at": item.get("picked_at"),
        "done_at": item.get("done_at"),
        "skipped_at": item.get("skipped_at"),
        "wrong": item.get("wrong") is True,
        "wrong_at": item.get("wrong_at"),
        "wrong_cleared_at": item.get("wrong_cleared_at"),
        "wrong_reviewed_at": item.get("wrong_reviewed_at"),
        "wrong_review_count": int(item.get("wrong_review_count") or 0),
        "review_history": review_history_for_question(q, review_jobs),
    }


def current_question(data: dict[str, Any], state: dict[str, Any]) -> dict[str, Any] | None:
    slug = state.get("last_selected")
    if not slug:
        return None
    return questions_by_slug(data).get(slug)


def public_snapshot(data: dict[str, Any], state: dict[str, Any]) -> dict[str, Any]:
    attach_related_notes(data)
    current = current_question(data, state)
    review_index = review_jobs_by_slug()
    return {
        "plan": {
            "name": data.get("plan_name", "Top 100 Liked"),
            "source": data.get("source"),
            "cn_study_plan": data.get("cn_study_plan", CN_STUDY_PLAN_URL),
            "fetched_at": data.get("fetched_at"),
        },
        "counts": summarize(data, state),
        "current": public_question(current, state, review_index.get(current["slug"], [])) if current else None,
        "questions": [public_question(q, state, review_index.get(q["slug"], [])) for q in data["questions"]],
    }


def pick_next_question(data: dict[str, Any], state: dict[str, Any], include_skipped: bool = False) -> dict[str, Any] | None:
    allowed = {"todo"}
    if include_skipped:
        allowed.add("skipped")

    candidates = [q for q in data["questions"] if question_status(state, q["slug"]) in allowed]
    if not candidates:
        return None

    q = random.choice(candidates)
    set_status(state, q["slug"], "picked")
    state["last_selected"] = q["slug"]
    return q


def wrong_review_remaining(data: dict[str, Any], state: dict[str, Any]) -> int:
    cycle = int(state.get("wrong_review_cycle") or 1)
    return sum(
        1
        for q in data["questions"]
        if question_is_wrong(state, q["slug"])
        and int(state.get("items", {}).get(q["slug"], {}).get("wrong_review_cycle") or 0) != cycle
    )


def pick_wrong_question(
    data: dict[str, Any],
    state: dict[str, Any],
) -> tuple[dict[str, Any] | None, bool]:
    wrong_questions = [q for q in data["questions"] if question_is_wrong(state, q["slug"])]
    if not wrong_questions:
        return None, False

    cycle = int(state.get("wrong_review_cycle") or 1)
    candidates = [
        q
        for q in wrong_questions
        if int(state.get("items", {}).get(q["slug"], {}).get("wrong_review_cycle") or 0) != cycle
    ]
    started_new_cycle = False
    if not candidates:
        cycle += 1
        state["wrong_review_cycle"] = cycle
        candidates = wrong_questions
        started_new_cycle = True

    q = random.choice(candidates)
    item = ensure_item(state, q["slug"])
    item["wrong_review_cycle"] = cycle
    item["wrong_reviewed_at"] = now_iso()
    item["wrong_review_count"] = int(item.get("wrong_review_count") or 0) + 1
    state["last_selected"] = q["slug"]
    return q, started_new_cycle


def export_tsv(data: dict[str, Any], state: dict[str, Any]) -> str:
    rows = []
    for q in data["questions"]:
        status = question_status(state, q["slug"])
        item = state.get("items", {}).get(q["slug"], {})
        note = (item.get("note") or "").replace("\n", " ")
        wrong = "yes" if item.get("wrong") is True else "no"
        wrong_review_count = str(int(item.get("wrong_review_count") or 0))
        rows.append(
            "\t".join(
                [
                    q["id"],
                    q["title"],
                    q["difficulty"],
                    q["topic"],
                    status,
                    wrong,
                    wrong_review_count,
                    q["url"],
                    note,
                ]
            )
        )
    return "\n".join(rows) + "\n"


def cmd_refresh(args: argparse.Namespace) -> int:
    data = refresh_data(args.data)
    extra_count = len(extra_questions_from_data(read_json(args.data)))
    print(f"已刷新题单：{len(data['questions'])} 题")
    if extra_count:
        print(f"其中扩展题：{extra_count} 题")
    print(f"保存到：{args.data}")
    print(f"来源：{data['source']}")
    return 0


def cmd_next(args: argparse.Namespace) -> int:
    data = load_data(args.data)
    state = load_state(args.state)

    q = pick_next_question(data, state, include_skipped=args.include_skipped)
    if not q:
        counts = summarize(data, state)
        print("没有可抽取的新题了。")
        print_counts(counts)
        if counts["skipped"]:
            print("提示：想把跳过的题也纳入随机池，可以运行：python3 hot100.py next --include-skipped")
        print("重新开始可以运行：python3 hot100.py reset --confirm")
        return 0

    save_state(state, args.state)

    print_question(q, "picked", show_topic=args.show_topic, wrong=question_is_wrong(state, q["slug"]))
    if args.open:
        webbrowser.open(q["url"])
    return 0


def cmd_review_wrong(args: argparse.Namespace) -> int:
    data = load_data(args.data)
    state = load_state(args.state)
    q, started_new_cycle = pick_wrong_question(data, state)
    if not q:
        print("错题本还是空的。先标记一道错题：python3 hot100.py wrong")
        return 0

    save_state(state, args.state)
    if started_new_cycle:
        print("上一轮错题已全部复习，已自动开始新一轮。")
    print_question(q, question_status(state, q["slug"]), show_topic=args.show_topic, wrong=True)
    print(f"本轮剩余：{wrong_review_remaining(data, state)} 题")
    if args.open:
        webbrowser.open(q["url"])
    return 0


def cmd_current(args: argparse.Namespace) -> int:
    data = load_data(args.data)
    state = load_state(args.state)
    q = latest_selected_question(data, state)
    print_question(
        q,
        question_status(state, q["slug"]),
        show_topic=args.show_topic,
        wrong=question_is_wrong(state, q["slug"]),
    )
    return 0


def cmd_mark(args: argparse.Namespace, status: str) -> int:
    data = load_data(args.data)
    state = load_state(args.state)
    q = find_question(data, args.problem) if args.problem else latest_selected_question(data, state)
    set_status(state, q["slug"], status, note=args.note)
    state["last_selected"] = q["slug"]
    save_state(state, args.state)

    action = {"done": "已完成", "skipped": "已跳过", "todo": "已恢复为 todo"}[status]
    print(f"{action}：{format_label(q)}")
    print(f"链接：{q['url']}")
    return 0


def cmd_wrong(args: argparse.Namespace, wrong: bool) -> int:
    data = load_data(args.data)
    state = load_state(args.state)
    q = find_question(data, args.problem) if args.problem else latest_selected_question(data, state)
    set_wrong(state, q["slug"], wrong, note=args.note)
    state["last_selected"] = q["slug"]
    save_state(state, args.state)

    action = "已加入错题本" if wrong else "已移出错题本"
    print(f"{action}：{format_label(q)}")
    print(f"链接：{q['url']}")
    return 0


def cmd_list(args: argparse.Namespace) -> int:
    data = load_data(args.data)
    state = load_state(args.state)
    rows = []
    for q in data["questions"]:
        status = question_status(state, q["slug"])
        if args.status != "all" and status != args.status:
            continue
        if args.wrong_only and not question_is_wrong(state, q["slug"]):
            continue
        rows.append((status, q, question_is_wrong(state, q["slug"])))

    if not rows:
        print("没有符合条件的题。")
        return 0

    for status, q, wrong in rows:
        topic = f" | {q['topic']}" if args.show_topic else ""
        wrong_text = " [错题]" if wrong else ""
        print(f"[{status}]{wrong_text} {format_label(q)}{topic}")
        if args.with_url:
            print(f"    {q['url']}")
    return 0


def print_counts(counts: dict[str, int]) -> None:
    total = counts["total"]
    print(
        "进度："
        f"done {counts['done']}/{total}, "
        f"picked {counts['picked']}, "
        f"skipped {counts['skipped']}, "
        f"todo {counts['todo']}, "
        f"wrong {counts['wrong']}（本轮剩余 {counts['wrong_review_remaining']}）"
    )


def cmd_stats(args: argparse.Namespace) -> int:
    data = load_data(args.data)
    state = load_state(args.state)
    print_counts(summarize(data, state))
    print(f"题单更新时间：{data.get('fetched_at', 'unknown')}")
    print(f"状态文件：{args.state}")
    return 0


def cmd_reset(args: argparse.Namespace) -> int:
    if not args.confirm:
        raise Hot100Error("这会清空所有进度和错题记录。确认要重置请运行：python3 hot100.py reset --confirm")
    state = empty_state()
    save_state(state, args.state)
    print(f"已重置进度：{args.state}")
    return 0


def cmd_export(args: argparse.Namespace) -> int:
    data = load_data(args.data)
    state = load_state(args.state)
    print(export_tsv(data, state), end="")
    return 0


class ThreadingHTTPServer(socketserver.ThreadingMixIn, http.server.HTTPServer):
    daemon_threads = True
    allow_reuse_address = True


class Hot100WebHandler(http.server.BaseHTTPRequestHandler):
    data_path = DATA_PATH
    state_path = STATE_PATH
    web_root = WEB_ROOT

    server_version = "Hot100Random/1.0"

    def log_message(self, format: str, *args: Any) -> None:
        return

    def send_payload(self, status: int, body: bytes, content_type: str) -> None:
        self.send_response(status)
        self.send_header("content-type", content_type)
        self.send_header("content-length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def send_json(self, payload: dict[str, Any], status: int = 200) -> None:
        body = json.dumps(payload, ensure_ascii=False).encode("utf-8")
        self.send_payload(status, body, "application/json; charset=utf-8")

    def send_error_json(self, message: str, status: int = 400) -> None:
        self.send_json({"ok": False, "error": message}, status=status)

    def read_json_body(self) -> dict[str, Any]:
        length = int(self.headers.get("content-length", "0") or 0)
        if length == 0:
            return {}
        raw = self.rfile.read(length)
        try:
            payload = json.loads(raw.decode("utf-8"))
        except json.JSONDecodeError as exc:
            raise Hot100Error(f"请求 JSON 无法解析：{exc}")
        if not isinstance(payload, dict):
            raise Hot100Error("请求体必须是 JSON object")
        return payload

    def load_pair(self) -> tuple[dict[str, Any], dict[str, Any]]:
        return load_data(self.data_path), load_state(self.state_path)

    def do_GET(self) -> None:
        parsed = urllib.parse.urlparse(self.path)
        path = parsed.path
        try:
            if path == "/api/state":
                data, state = self.load_pair()
                self.send_json({"ok": True, **public_snapshot(data, state)})
                return
            if path == "/api/reviews":
                self.send_json({"ok": True, "jobs": list_review_jobs()})
                return
            if path.startswith("/api/review/"):
                review_job_id = path.rsplit("/", 1)[-1]
                self.send_json({"ok": True, "job": read_job(review_job_id)})
                return
            if path == "/api/export":
                data, state = self.load_pair()
                body = export_tsv(data, state).encode("utf-8")
                self.send_response(200)
                self.send_header("content-type", "text/tab-separated-values; charset=utf-8")
                self.send_header("content-disposition", 'attachment; filename="hot100-progress.tsv"')
                self.send_header("content-length", str(len(body)))
                self.end_headers()
                self.wfile.write(body)
                return
            if path.startswith("/notes/"):
                self.serve_note(path)
                return
            if path.startswith("/api/"):
                self.send_error_json("接口不存在", status=404)
                return
            self.serve_static(path)
        except Hot100Error as exc:
            self.send_error_json(str(exc), status=400)
        except OSError as exc:
            self.send_error_json(f"服务错误：{exc}", status=500)

    def do_POST(self) -> None:
        parsed = urllib.parse.urlparse(self.path)
        path = parsed.path
        try:
            body = self.read_json_body()
            if path == "/api/next":
                data, state = self.load_pair()
                q = pick_next_question(data, state, include_skipped=bool(body.get("includeSkipped")))
                if not q:
                    self.send_json(
                        {
                            "ok": False,
                            "error": "没有可抽取的新题了",
                            **public_snapshot(data, state),
                        },
                        status=409,
                    )
                    return
                save_state(state, self.state_path)
                self.send_json({"ok": True, "selected": public_question(q, state), **public_snapshot(data, state)})
                return

            if path == "/api/next-wrong":
                data, state = self.load_pair()
                q, started_new_cycle = pick_wrong_question(data, state)
                if not q:
                    self.send_json(
                        {
                            "ok": False,
                            "error": "错题本还是空的，先标记一道错题",
                            **public_snapshot(data, state),
                        },
                        status=409,
                    )
                    return
                save_state(state, self.state_path)
                self.send_json(
                    {
                        "ok": True,
                        "selected": public_question(q, state),
                        "started_new_cycle": started_new_cycle,
                        **public_snapshot(data, state),
                    }
                )
                return

            if path == "/api/status":
                slug = str(body.get("slug") or "").strip()
                status = str(body.get("status") or "").strip()
                note = body.get("note")
                if not slug:
                    raise Hot100Error("缺少题目 slug")
                if status not in STATUSES:
                    raise Hot100Error(f"未知状态：{status}")
                if note is not None and not isinstance(note, str):
                    raise Hot100Error("备注必须是字符串")
                data, state = self.load_pair()
                q = find_question(data, slug)
                set_status(state, q["slug"], status, note=note)
                state["last_selected"] = q["slug"]
                save_state(state, self.state_path)
                self.send_json({"ok": True, "question": public_question(q, state), **public_snapshot(data, state)})
                return

            if path == "/api/wrong":
                slug = str(body.get("slug") or "").strip()
                wrong = body.get("wrong")
                note = body.get("note")
                if not slug:
                    raise Hot100Error("缺少题目 slug")
                if not isinstance(wrong, bool):
                    raise Hot100Error("wrong 必须是布尔值")
                if note is not None and not isinstance(note, str):
                    raise Hot100Error("备注必须是字符串")
                data, state = self.load_pair()
                q = find_question(data, slug)
                set_wrong(state, q["slug"], wrong, note=note)
                state["last_selected"] = q["slug"]
                save_state(state, self.state_path)
                self.send_json({"ok": True, "question": public_question(q, state), **public_snapshot(data, state)})
                return

            if path == "/api/refresh":
                data = refresh_data(self.data_path)
                state = load_state(self.state_path)
                self.send_json({"ok": True, **public_snapshot(data, state)})
                return

            if path == "/api/reset":
                if body.get("confirm") is not True:
                    raise Hot100Error("重置进度需要 confirm=true")
                data = load_data(self.data_path)
                state = empty_state()
                save_state(state, self.state_path)
                self.send_json({"ok": True, **public_snapshot(data, state)})
                return

            if path.startswith("/api/review/") and path.endswith("/retry"):
                parts = path.strip("/").split("/")
                if len(parts) != 4 or parts[0] != "api" or parts[1] != "review" or parts[3] != "retry":
                    self.send_error_json("接口不存在", status=404)
                    return
                review_job_id = urllib.parse.unquote(parts[2])
                job = retry_review_job(review_job_id)
                self.send_json({"ok": True, "job": job})
                return

            if path == "/api/review":
                slug = str(body.get("slug") or "").strip()
                material = body.get("material")
                if not slug:
                    raise Hot100Error("缺少题目 slug")
                if not isinstance(material, str):
                    raise Hot100Error("复盘材料必须是字符串")
                data, _state = self.load_pair()
                q = find_question(data, slug)
                attach_related_notes(data)
                job = start_review_job(q, material)
                self.send_json({"ok": True, "job": job})
                return

            self.send_error_json("接口不存在", status=404)
        except Hot100Error as exc:
            self.send_error_json(str(exc), status=400)
        except OSError as exc:
            self.send_error_json(f"服务错误：{exc}", status=500)

    def serve_note(self, request_path: str) -> None:
        parts = request_path.strip("/").split("/", 2)
        if len(parts) != 3:
            self.send_error(404)
            return

        kind = urllib.parse.unquote(parts[1])
        filename = urllib.parse.unquote(parts[2])
        directory = local_note_dirs().get(kind)
        if not directory:
            self.send_error(404)
            return

        candidate = (directory / Path(filename).name).resolve()
        root = directory.resolve()
        if candidate.parent != root:
            self.send_error(403)
            return
        if not candidate.is_file():
            self.send_error(404)
            return
        self.send_payload(200, candidate.read_bytes(), "text/markdown; charset=utf-8")

    def serve_static(self, request_path: str) -> None:
        relative = "index.html" if request_path in ("", "/") else request_path.lstrip("/")
        candidate = (self.web_root / relative).resolve()
        root = self.web_root.resolve()
        if root != candidate and root not in candidate.parents:
            self.send_error(403)
            return
        if not candidate.is_file():
            candidate = root / "index.html"
        content_type = mimetypes.guess_type(candidate.name)[0] or "application/octet-stream"
        self.send_payload(200, candidate.read_bytes(), content_type)


def make_handler(data_path: Path, state_path: Path, web_root: Path) -> type[Hot100WebHandler]:
    class ConfiguredHot100WebHandler(Hot100WebHandler):
        pass

    ConfiguredHot100WebHandler.data_path = data_path
    ConfiguredHot100WebHandler.state_path = state_path
    ConfiguredHot100WebHandler.web_root = web_root
    return ConfiguredHot100WebHandler


def start_server(host: str, port: int, handler: type[Hot100WebHandler]) -> ThreadingHTTPServer:
    candidates = [0] if port == 0 else list(range(port, port + 21))
    last_error: OSError | None = None
    for candidate in candidates:
        try:
            return ThreadingHTTPServer((host, candidate), handler)
        except OSError as exc:
            last_error = exc
    raise Hot100Error(f"无法启动本地服务：{last_error}")


def cmd_web(args: argparse.Namespace) -> int:
    load_data(args.data)
    if not args.web_root.exists():
        raise Hot100Error(f"找不到前端目录：{args.web_root}")

    handler = make_handler(args.data, args.state, args.web_root)
    server = start_server(args.host, args.port, handler)
    host, port = server.server_address
    display_host = "127.0.0.1" if host in ("", "0.0.0.0") else host
    url = f"http://{display_host}:{port}/"

    print(f"Hot100 可视化界面已启动：{url}", flush=True)
    print("按 Ctrl+C 停止服务", flush=True)
    if args.open:
        webbrowser.open(url)
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        print("\n已停止服务")
    finally:
        server.server_close()
    return 0


def add_common_args(parser: argparse.ArgumentParser) -> None:
    parser.add_argument("--data", type=Path, default=DATA_PATH, help=f"题单 JSON，默认 {DATA_PATH}")
    parser.add_argument("--state", type=Path, default=STATE_PATH, help=f"进度 JSON，默认 {STATE_PATH}")


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        description="不重复随机抽 LeetCode Hot 100 和扩展题，并记录完成状态和错题复习进度。",
    )
    add_common_args(parser)
    subparsers = parser.add_subparsers(dest="command", required=True)

    refresh = subparsers.add_parser("refresh", help="从 LeetCode 刷新 Hot 100 题单，并保留扩展题")
    refresh.set_defaults(func=cmd_refresh)

    next_cmd = subparsers.add_parser("next", help="随机抽一道还没抽过的题")
    next_cmd.add_argument("--show-topic", action="store_true", help="显示题型分组")
    next_cmd.add_argument("--include-skipped", action="store_true", help="把 skipped 题也放回随机池")
    next_cmd.add_argument("--open", action="store_true", help="用默认浏览器打开题目")
    next_cmd.set_defaults(func=cmd_next)

    review_wrong = subparsers.add_parser("review-wrong", help="从错题本随机抽一道复习；一轮内不重复")
    review_wrong.add_argument("--show-topic", action="store_true", help="显示题型分组")
    review_wrong.add_argument("--open", action="store_true", help="用默认浏览器打开题目")
    review_wrong.set_defaults(func=cmd_review_wrong)

    current = subparsers.add_parser("current", help="查看上一次抽到的题")
    current.add_argument("--show-topic", action="store_true", help="显示题型分组")
    current.set_defaults(func=cmd_current)

    done = subparsers.add_parser("done", help="标记完成；不传题目时默认标记上一次抽到的题")
    done.add_argument("problem", nargs="?", help="题号、slug 或标题")
    done.add_argument("--note", help="记录一句复盘备注")
    done.set_defaults(func=lambda args: cmd_mark(args, "done"))

    skip = subparsers.add_parser("skip", help="标记跳过；默认不再随机到，除非 next --include-skipped")
    skip.add_argument("problem", nargs="?", help="题号、slug 或标题")
    skip.add_argument("--note", help="记录一句跳过原因")
    skip.set_defaults(func=lambda args: cmd_mark(args, "skipped"))

    wrong = subparsers.add_parser("wrong", help="加入错题本；不传题目时默认使用当前题")
    wrong.add_argument("problem", nargs="?", help="题号、slug 或标题")
    wrong.add_argument("--note", help="记录一句错因")
    wrong.set_defaults(func=lambda args: cmd_wrong(args, True))

    unwrong = subparsers.add_parser("unwrong", help="移出错题本；不传题目时默认使用当前题")
    unwrong.add_argument("problem", nargs="?", help="题号、slug 或标题")
    unwrong.add_argument("--note", help="覆盖备注")
    unwrong.set_defaults(func=lambda args: cmd_wrong(args, False))

    undo = subparsers.add_parser("undo", help="把某题恢复为 todo")
    undo.add_argument("problem", help="题号、slug 或标题")
    undo.add_argument("--note", help="覆盖备注")
    undo.set_defaults(func=lambda args: cmd_mark(args, "todo"))

    list_cmd = subparsers.add_parser("list", help="列出题目和状态")
    list_cmd.add_argument("--status", choices=("all",) + STATUSES, default="all", help="按状态过滤")
    list_cmd.add_argument("--wrong-only", action="store_true", help="只列出错题")
    list_cmd.add_argument("--show-topic", action="store_true", help="显示题型分组")
    list_cmd.add_argument("--with-url", action="store_true", help="同时打印链接")
    list_cmd.set_defaults(func=cmd_list)

    stats = subparsers.add_parser("stats", help="查看整体进度")
    stats.set_defaults(func=cmd_stats)

    reset = subparsers.add_parser("reset", help="清空本地进度和错题记录")
    reset.add_argument("--confirm", action="store_true", help="确认清空")
    reset.set_defaults(func=cmd_reset)

    export = subparsers.add_parser("export", help="导出 TSV，便于粘贴到表格")
    export.set_defaults(func=cmd_export)

    web = subparsers.add_parser("web", help="启动本地可视化界面")
    web.add_argument("--host", default="127.0.0.1", help="监听地址，默认 127.0.0.1")
    web.add_argument("--port", type=int, default=8765, help="监听端口，默认 8765；占用时自动顺延")
    web.add_argument("--web-root", type=Path, default=WEB_ROOT, help=f"前端目录，默认 {WEB_ROOT}")
    web.add_argument("--open", action="store_true", help="启动后自动打开浏览器")
    web.set_defaults(func=cmd_web)

    return parser


def main(argv: list[str] | None = None) -> int:
    parser = build_parser()
    args = parser.parse_args(argv)
    args.data = args.data.expanduser().resolve()
    args.state = args.state.expanduser().resolve()
    if hasattr(args, "web_root"):
        args.web_root = args.web_root.expanduser().resolve()
    try:
        return args.func(args)
    except Hot100Error as exc:
        print(f"错误：{exc}", file=sys.stderr)
        return 1


if __name__ == "__main__":
    raise SystemExit(main())