diff --git a/app/question_gen/run_store.py b/app/question_gen/run_store.py new file mode 100644 index 0000000..b1a07ba --- /dev/null +++ b/app/question_gen/run_store.py @@ -0,0 +1,380 @@ +"""SQLite 出题管线日志记录器 — 记录批次运行与逐题门判定。 + +写入频率低（每批次 ~240 题），使用同步 sqlite3 即可。 +schema 定义参照 research-wiki/schemas/question-gen-runs.md 和 question-gen-items.md。 +""" + +from __future__ import annotations + +import sqlite3 +from dataclasses import dataclass +from datetime import UTC, datetime +from typing import TYPE_CHECKING, Protocol, runtime_checkable + +from loguru import logger + +if TYPE_CHECKING: + from pathlib import Path + + +# --------------------------------------------------------------------------- +# GateReport Protocol（Task 6 尚未实现，此处声明 duck-type 接口） +# --------------------------------------------------------------------------- + + +@runtime_checkable +class _GateResultLike(Protocol): + """单门判定结果的最小接口。""" + + @property + def verdict(self) -> object: + """PASS / FAIL / SKIP 枚举值，.value 为小写字符串。""" + ... + + @property + def reason(self) -> str: + """判定理由。""" + ... + + +@runtime_checkable +class GateReportLike(Protocol): + """GateReport 的最小 duck-type 接口。""" + + @property + def key_verify(self) -> _GateResultLike: ... + + @property + def blind_answer(self) -> _GateResultLike: ... + + @property + def multi_true(self) -> _GateResultLike: ... + + @property + def leak_test(self) -> _GateResultLike: ... + + @property + def passed(self) -> bool: ... + + @property + def reject_reason(self) -> str | None: ... + + +# --------------------------------------------------------------------------- +# 数据类 +# --------------------------------------------------------------------------- + + +@dataclass(frozen=True) +class RunStats: + """批次运行的统计摘要。 + + Attributes + ---------- + total_slots : int + 目标题数（slot 总数）。 + accepted : int + 最终通过门判定的题数。 + rejected : int + 最终被拒绝的题数。 + heavy_sampled : int + 进行重量抽检的题数。 + """ + + total_slots: int + accepted: int + rejected: int + heavy_sampled: int + + +# --------------------------------------------------------------------------- +# DDL +# --------------------------------------------------------------------------- + +_DDL_RUNS = """ +CREATE TABLE IF NOT EXISTS question_gen_runs ( + run_id TEXT PRIMARY KEY, + git_sha TEXT NOT NULL, + config_snapshot TEXT, + started_at TEXT NOT NULL, + finished_at TEXT, + status TEXT NOT NULL DEFAULT 'running', + total_slots INTEGER, + accepted INTEGER, + rejected INTEGER, + heavy_sampled INTEGER +); +""" + +_DDL_ITEMS = """ +CREATE TABLE IF NOT EXISTS question_gen_items ( + item_id TEXT PRIMARY KEY, + run_id TEXT NOT NULL REFERENCES question_gen_runs(run_id), + slot_id TEXT NOT NULL, + video_id TEXT NOT NULL, + family TEXT NOT NULL, + task_type TEXT NOT NULL, + skill_target TEXT NOT NULL, + attempt INTEGER NOT NULL, + question_text TEXT NOT NULL, + gate_key_verify TEXT, + gate_blind_answer TEXT, + gate_multi_true TEXT, + gate_leak_test TEXT, + gate_reject_reason TEXT, + final_status TEXT NOT NULL DEFAULT 'pending', + difficulty_steps INTEGER, + created_at TEXT NOT NULL DEFAULT (datetime('now')) +); +""" + +_DDL_INDEXES = [ + "CREATE INDEX IF NOT EXISTS idx_qgr_status ON question_gen_runs(status);", + "CREATE INDEX IF NOT EXISTS idx_qgi_run ON question_gen_items(run_id);", + "CREATE INDEX IF NOT EXISTS idx_qgi_slot ON question_gen_items(slot_id);", + "CREATE INDEX IF NOT EXISTS idx_qgi_status ON question_gen_items(final_status);", + "CREATE INDEX IF NOT EXISTS idx_qgi_family ON question_gen_items(family);", + "CREATE INDEX IF NOT EXISTS idx_qgi_task_type ON question_gen_items(task_type);", +] + + +# --------------------------------------------------------------------------- +# Store 实现 +# --------------------------------------------------------------------------- + + +class QuestionGenStore: + """SQLite 出题管线日志记录器。 + + 记录每次出题批次的元数据（run）和每题的门判定结果（item）。 + 使用同步 sqlite3，写入频率低无需异步。 + + Parameters + ---------- + db_path : Path + SQLite 数据库文件路径。父目录必须存在。 + """ + + def __init__(self, db_path: Path) -> None: + self._db_path = db_path + self._conn = sqlite3.connect( + str(db_path), + check_same_thread=False, + timeout=10.0, + ) + self._conn.execute("PRAGMA journal_mode=WAL") + self._conn.execute("PRAGMA busy_timeout=5000") + self._conn.execute("PRAGMA foreign_keys=ON") + self._init_schema() + logger.debug("QuestionGenStore 已初始化: {}", db_path) + + def _init_schema(self) -> None: + """幂等创建表和索引。多次调用安全。""" + self._conn.execute(_DDL_RUNS) + self._conn.execute(_DDL_ITEMS) + for idx_sql in _DDL_INDEXES: + self._conn.execute(idx_sql) + self._conn.commit() + + def record_run_start(self, run_id: str, git_sha: str, config_snapshot: str) -> None: + """记录批次开始。 + + Parameters + ---------- + run_id : str + 批次唯一标识（UUID）。 + git_sha : str + 当前代码版本 HEAD commit。 + config_snapshot : str + 科研配置快照（JSON 序列化字符串）。 + """ + now = datetime.now(tz=UTC).isoformat(timespec="seconds") + self._conn.execute( + """ + INSERT INTO question_gen_runs (run_id, git_sha, config_snapshot, started_at, status) + VALUES (?, ?, ?, ?, 'running') + """, + (run_id, git_sha, config_snapshot, now), + ) + self._conn.commit() + logger.info("出题批次已开始: run_id={}", run_id) + + def record_run_end(self, run_id: str, status: str, stats: RunStats) -> None: + """记录批次结束，更新状态与统计。 + + Parameters + ---------- + run_id : str + 批次唯一标识。 + status : str + 最终状态（completed / failed）。 + stats : RunStats + 批次统计摘要。 + """ + now = datetime.now(tz=UTC).isoformat(timespec="seconds") + self._conn.execute( + """ + UPDATE question_gen_runs + SET finished_at=?, status=?, total_slots=?, accepted=?, rejected=?, heavy_sampled=? + WHERE run_id=? + """, + ( + now, + status, + stats.total_slots, + stats.accepted, + stats.rejected, + stats.heavy_sampled, + run_id, + ), + ) + self._conn.commit() + logger.info( + "出题批次已结束: run_id={}, status={}, accepted={}/{}", + run_id, + status, + stats.accepted, + stats.total_slots, + ) + + def record_item( + self, + item_id: str, + run_id: str, + slot_id: str, + video_id: str, + family: str, + task_type: str, + skill_target: str, + attempt: int, + question_text: str, + ) -> None: + """记录一道新生成的题目（初始状态 pending）。 + + Parameters + ---------- + item_id : str + 题目唯一 ID（每轮独立）。 + run_id : str + 关联的批次 ID。 + slot_id : str + 逻辑 slot 标识（同 slot 多次重出共享）。 + video_id : str + 视频 ID。 + family : str + 题族（retrieval/reasoning/enumeration/visual/spatial）。 + task_type : str + Video-MME 12 类主标签。 + skill_target : str + M1-M5 + 题族子标签。 + attempt : int + 当前重出轮次（1-based）。 + question_text : str + 题目文本。 + """ + now = datetime.now(tz=UTC).isoformat(timespec="seconds") + self._conn.execute( + """ + INSERT INTO question_gen_items + (item_id, run_id, slot_id, video_id, family, task_type, + skill_target, attempt, question_text, created_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + item_id, + run_id, + slot_id, + video_id, + family, + task_type, + skill_target, + attempt, + question_text, + now, + ), + ) + self._conn.commit() + + def update_gates(self, item_id: str, report: GateReportLike) -> None: + """更新门判定结果及 final_status。 + + Parameters + ---------- + item_id : str + 题目唯一 ID。 + report : GateReportLike + 门判定报告（duck-type，需有 key_verify/blind_answer/multi_true/leak_test + 属性，每个属性具有 .verdict.value 和 .reason；以及 passed/reject_reason 属性）。 + """ + final_status = "accepted" if report.passed else "rejected" + self._conn.execute( + """ + UPDATE question_gen_items + SET gate_key_verify=?, gate_blind_answer=?, gate_multi_true=?, + gate_leak_test=?, gate_reject_reason=?, final_status=? + WHERE item_id=? + """, + ( + report.key_verify.verdict.value, + report.blind_answer.verdict.value, + report.multi_true.verdict.value, + report.leak_test.verdict.value, + report.reject_reason, + final_status, + item_id, + ), + ) + self._conn.commit() + + def update_difficulty(self, item_id: str, difficulty_steps: int) -> None: + """更新重量抽检产出的 Agent 步数。 + + Parameters + ---------- + item_id : str + 题目唯一 ID。 + difficulty_steps : int + Agent 完成该题所需步数。 + """ + self._conn.execute( + "UPDATE question_gen_items SET difficulty_steps=? WHERE item_id=?", + (difficulty_steps, item_id), + ) + self._conn.commit() + + def get_run_stats(self, run_id: str) -> RunStats: + """查询批次统计摘要。 + + Parameters + ---------- + run_id : str + 批次唯一标识。 + + Returns + ------- + RunStats + 该批次的统计数据。 + + Raises + ------ + ValueError + run_id 不存在时抛出。 + """ + row = self._conn.execute( + "SELECT total_slots, accepted, rejected, heavy_sampled " + "FROM question_gen_runs WHERE run_id=?", + (run_id,), + ).fetchone() + if row is None: + raise ValueError(f"run_id 不存在: {run_id}") + return RunStats( + total_slots=row[0] or 0, + accepted=row[1] or 0, + rejected=row[2] or 0, + heavy_sampled=row[3] or 0, + ) + + def close(self) -> None: + """关闭数据库连接。""" + self._conn.close() + logger.debug("QuestionGenStore 已关闭") diff --git a/tests/unit/test_run_store.py b/tests/unit/test_run_store.py new file mode 100644 index 0000000..902a0d7 --- /dev/null +++ b/tests/unit/test_run_store.py @@ -0,0 +1,244 @@ +"""QuestionGenStore 单元测试 — SQLite 出题管线日志记录器。 + +覆盖 schema 幂等性、run 生命周期、item 写入与门判定更新、难度更新。 +""" + +from __future__ import annotations + +from dataclasses import dataclass +from enum import Enum +from typing import TYPE_CHECKING + +import pytest + +if TYPE_CHECKING: + from pathlib import Path + +from app.question_gen.run_store import QuestionGenStore, RunStats + +# --------------------------------------------------------------------------- +# GateReport 测试替身（Task 6 尚未实现，此处构造兼容 duck-type） +# --------------------------------------------------------------------------- + + +class _Verdict(Enum): + """门判定枚举替身。""" + + PASS = "pass" + FAIL = "fail" + SKIP = "skip" + + +@dataclass(frozen=True) +class _GateResult: + """单门判定结果替身。""" + + verdict: _Verdict + reason: str + + +@dataclass +class _MockGateReport: + """GateReport 测试替身，duck-type 兼容。""" + + key_verify: _GateResult + blind_answer: _GateResult + multi_true: _GateResult + leak_test: _GateResult + + @property + def passed(self) -> bool: + """所有门均 PASS 或 SKIP 时视为通过。""" + results = [self.key_verify, self.blind_answer, self.multi_true, self.leak_test] + return all(r.verdict != _Verdict.FAIL for r in results) + + @property + def reject_reason(self) -> str | None: + """返回首个 FAIL 门的 reason，若全部通过则 None。""" + for r in [self.key_verify, self.blind_answer, self.multi_true, self.leak_test]: + if r.verdict == _Verdict.FAIL: + return r.reason + return None + + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + + +@pytest.fixture() +def store(tmp_path: Path) -> QuestionGenStore: + """创建临时 SQLite 上的 QuestionGenStore 实例。""" + db_path = tmp_path / "question_gen.db" + s = QuestionGenStore(db_path=db_path) + yield s + s.close() + + +# --------------------------------------------------------------------------- +# TestQuestionGenStore +# --------------------------------------------------------------------------- + + +class TestQuestionGenStore: + """QuestionGenStore 全功能验证。""" + + def test_schema_idempotent(self, tmp_path: Path) -> None: + """连续两次初始化同一 db 不报错，表结构不变。""" + db_path = tmp_path / "idem.db" + s1 = QuestionGenStore(db_path=db_path) + s2 = QuestionGenStore(db_path=db_path) + # 第二次初始化不应抛异常 + s2.record_run_start(run_id="r-idem", git_sha="abc123", config_snapshot="{}") + s1.close() + s2.close() + + def test_record_run_lifecycle(self, store: QuestionGenStore) -> None: + """start -> record items -> end -> get_stats 完整生命周期。""" + run_id = "run-001" + store.record_run_start( + run_id=run_id, + git_sha="deadbeef", + config_snapshot='{"n_slots": 240}', + ) + + # 写入几个 item 来模拟统计 + for i in range(5): + store.record_item( + item_id=f"item-{i}", + run_id=run_id, + slot_id=f"slot-{i}", + video_id="v001", + family="retrieval", + task_type="Object Recognition", + skill_target="M1", + attempt=1, + question_text=f"问题{i}", + ) + + # 结束 run + stats = RunStats(total_slots=10, accepted=5, rejected=3, heavy_sampled=2) + store.record_run_end(run_id=run_id, status="completed", stats=stats) + + # 查询 stats + retrieved = store.get_run_stats(run_id) + assert retrieved.total_slots == 10 + assert retrieved.accepted == 5 + assert retrieved.rejected == 3 + assert retrieved.heavy_sampled == 2 + + def test_record_item_and_gates(self, store: QuestionGenStore) -> None: + """record_item 后 update_gates 正确更新门判定列与 final_status。""" + run_id = "run-gates" + store.record_run_start(run_id=run_id, git_sha="aaa111", config_snapshot="{}") + item_id = "item-gate-1" + store.record_item( + item_id=item_id, + run_id=run_id, + slot_id="slot-0", + video_id="v002", + family="reasoning", + task_type="Action Sequence", + skill_target="M2", + attempt=1, + question_text="为什么这样做？", + ) + + # 所有门通过 + report_pass = _MockGateReport( + key_verify=_GateResult(_Verdict.PASS, "ok"), + blind_answer=_GateResult(_Verdict.PASS, "ok"), + multi_true=_GateResult(_Verdict.SKIP, "n/a"), + leak_test=_GateResult(_Verdict.PASS, "ok"), + ) + store.update_gates(item_id=item_id, report=report_pass) + + # 验证数据库中的值 + import sqlite3 + + conn = sqlite3.connect(str(store._db_path)) + conn.row_factory = sqlite3.Row + row = conn.execute( + "SELECT * FROM question_gen_items WHERE item_id=?", (item_id,) + ).fetchone() + conn.close() + + assert row["gate_key_verify"] == "pass" + assert row["gate_blind_answer"] == "pass" + assert row["gate_multi_true"] == "skip" + assert row["gate_leak_test"] == "pass" + assert row["gate_reject_reason"] is None + assert row["final_status"] == "accepted" + + def test_update_gates_rejected(self, store: QuestionGenStore) -> None: + """门判定失败时 final_status 为 rejected 且记录 reject_reason。""" + run_id = "run-rej" + store.record_run_start(run_id=run_id, git_sha="bbb222", config_snapshot="{}") + item_id = "item-rej-1" + store.record_item( + item_id=item_id, + run_id=run_id, + slot_id="slot-1", + video_id="v003", + family="visual", + task_type="Scene Transition", + skill_target="M3", + attempt=2, + question_text="画面中有什么？", + ) + + report_fail = _MockGateReport( + key_verify=_GateResult(_Verdict.PASS, "ok"), + blind_answer=_GateResult(_Verdict.FAIL, "答案可由常识推断"), + multi_true=_GateResult(_Verdict.SKIP, "n/a"), + leak_test=_GateResult(_Verdict.SKIP, "n/a"), + ) + store.update_gates(item_id=item_id, report=report_fail) + + import sqlite3 + + conn = sqlite3.connect(str(store._db_path)) + conn.row_factory = sqlite3.Row + row = conn.execute( + "SELECT * FROM question_gen_items WHERE item_id=?", (item_id,) + ).fetchone() + conn.close() + + assert row["final_status"] == "rejected" + assert row["gate_reject_reason"] == "答案可由常识推断" + + def test_update_difficulty(self, store: QuestionGenStore) -> None: + """update_difficulty 正确写入 difficulty_steps。""" + run_id = "run-diff" + store.record_run_start(run_id=run_id, git_sha="ccc333", config_snapshot="{}") + item_id = "item-diff-1" + store.record_item( + item_id=item_id, + run_id=run_id, + slot_id="slot-2", + video_id="v004", + family="enumeration", + task_type="Counting", + skill_target="M4", + attempt=1, + question_text="有多少个？", + ) + + store.update_difficulty(item_id=item_id, difficulty_steps=7) + + import sqlite3 + + conn = sqlite3.connect(str(store._db_path)) + conn.row_factory = sqlite3.Row + row = conn.execute( + "SELECT difficulty_steps FROM question_gen_items WHERE item_id=?", + (item_id,), + ).fetchone() + conn.close() + + assert row["difficulty_steps"] == 7 + + def test_get_run_stats_not_found(self, store: QuestionGenStore) -> None: + """查询不存在的 run_id 应报错。""" + with pytest.raises(ValueError, match="run_id"): + store.get_run_stats("nonexistent-run")