LWX/ai_teaching_system/scripts/check_analysis_consistency.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
校验 chemistry_question_bank 的解析字段一致性：
- 列 analysis
- JSON 字段 answer_schema.analysis

用法：
  python scripts/check_analysis_consistency.py
  python scripts/check_analysis_consistency.py --table chemistry_question_bank_test --limit 50
"""

from __future__ import annotations

import argparse
import json
import os
import re
from typing import Any, Dict, Optional

import pymysql


def _mysql_conn_from_uri() -> Dict[str, Any]:
    """从 MYSQL_URI 解析连接参数，解析失败时使用默认值。"""
    uri = os.getenv("MYSQL_URI", "mysql+pymysql://maple:123456@localhost:3306/ai_system")
    m = re.match(
        r"^mysql\+pymysql://(?P<user>[^:]+):(?P<pwd>[^@]*)@(?P<host>[^:]+):(?P<port>\d+)/(?P<db>[^?]+)",
        uri,
    )
    if not m:
        return {
            "host": "localhost",
            "port": 3306,
            "user": "maple",
            "password": "123456",
            "database": "ai_system",
        }
    return {
        "host": m.group("host"),
        "port": int(m.group("port")),
        "user": m.group("user"),
        "password": m.group("pwd"),
        "database": m.group("db"),
    }


def _normalize_text(v: Optional[str]) -> str:
    return (v or "").strip()


def _extract_json_analysis(raw: Any) -> str:
    if raw is None:
        return ""
    data: Optional[Dict[str, Any]] = None
    if isinstance(raw, dict):
        data = raw
    elif isinstance(raw, str):
        try:
            data = json.loads(raw)
        except json.JSONDecodeError:
            return ""
    else:
        return ""
    val = data.get("analysis") if isinstance(data, dict) else None
    return _normalize_text(str(val) if val is not None else "")


def main() -> None:
    parser = argparse.ArgumentParser(description="校验题库解析字段一致性")
    parser.add_argument(
        "--table",
        default="chemistry_question_bank",
        choices=["chemistry_question_bank", "chemistry_question_bank_test"],
        help="要校验的题库表",
    )
    parser.add_argument("--limit", type=int, default=20, help="展示不一致样本数量上限")
    args = parser.parse_args()

    conn_cfg = _mysql_conn_from_uri()
    conn = pymysql.connect(
        **conn_cfg,
        charset="utf8mb4",
        cursorclass=pymysql.cursors.DictCursor,
        autocommit=True,
    )
    try:
        with conn.cursor() as cursor:
            cursor.execute(
                f"""
                SELECT question_id, analysis, answer_schema
                FROM `{args.table}`
                """
            )
            rows = cursor.fetchall() or []
    finally:
        conn.close()

    total = len(rows)
    both_empty = 0
    equal_non_empty = 0
    mismatch = []
    only_col = 0
    only_json = 0

    for row in rows:
        col_analysis = _normalize_text(row.get("analysis"))
        json_analysis = _extract_json_analysis(row.get("answer_schema"))
        if not col_analysis and not json_analysis:
            both_empty += 1
            continue
        if col_analysis and json_analysis and col_analysis == json_analysis:
            equal_non_empty += 1
            continue
        if col_analysis and not json_analysis:
            only_col += 1
        elif (not col_analysis) and json_analysis:
            only_json += 1
        mismatch.append(
            {
                "question_id": row.get("question_id"),
                "analysis_col": col_analysis,
                "analysis_json": json_analysis,
            }
        )

    print(f"[INFO] 表：{args.table}")
    print(f"[INFO] 总题数：{total}")
    print(f"[INFO] 双空：{both_empty}")
    print(f"[INFO] 一致(非空)：{equal_non_empty}")
    print(f"[INFO] 仅列有值：{only_col}")
    print(f"[INFO] 仅JSON有值：{only_json}")
    print(f"[INFO] 不一致总数：{len(mismatch)}")

    if mismatch:
        print(f"[WARN] 以下展示前 {args.limit} 条不一致样本：")
        for item in mismatch[: max(args.limit, 0)]:
            qid = item["question_id"]
            col_preview = item["analysis_col"][:120]
            json_preview = item["analysis_json"][:120]
            print(
                f"  - question_id={qid}, "
                f"analysis_col='{col_preview}', "
                f"analysis_json='{json_preview}'"
            )


if __name__ == "__main__":
    main()