Agent/examples/mode_workflow/db.py

# -*- coding: utf-8 -*-
"""mode_workflow · MySQL 持久化(DB 为唯一事实源)
================================================================================
读 .env 的 MYSQL_* 连接 MySQL。四张表:
  search_process —— 每行一个 (query, 帖子):工序方向的搜索 + llm 评估结果
  search_tools   —— 同结构,工具方向的搜索结果(方向由表区分,不再用 mode_type 列)
  mode_process   —— 每行一个解构出的工序(steps 等嵌套结构存 JSON 列)
  mode_tools     —— 每行一个解构出的工具

与旧 fixed_query_eval/db.py 的关键差异:本系统 DB 是主存储,写入失败直接 raise,
不做"失败不阻断"。读侧保留防御(返回空/None)。

用法:
  python db.py init    # 建表(幂等)
  python db.py check   # 打印四表行数
  python db.py clear   # 清空四表数据(TRUNCATE)
"""
import json
import os
import sys
from datetime import datetime
from pathlib import Path

PROJECT_ROOT = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(PROJECT_ROOT))

from dotenv import load_dotenv
load_dotenv()

import pymysql
from pymysql.cursors import DictCursor
from dbutils.pooled_db import PooledDB

# ── 连接池 ──────────────────────────────────────────────────────────────────
# MySQL 是远程 RDS,每次 pymysql.connect() 的 TCP+鉴权握手 ~0.5s。旧实现每个
# 请求新建一条连接,一次"点开帖子"要 2~3 个请求 = 2~3 次握手 ≈ 1s。改用连接池
# 复用长连接后,握手只在池初始化时各发生一次,后续取连接近乎零开销。
# server.py 是 ThreadingHTTPServer(每请求一线程),PooledDB 线程安全,正好匹配。
# 注意:fetch_* 里的 conn.close() 在池连接上语义是"归还池中"而非真正断开。
_POOL = None


def _pool():
    global _POOL
    if _POOL is None:
        if not os.getenv("MYSQL_HOST"):
            raise RuntimeError("缺 MYSQL_HOST:检查 .env 的 MYSQL_* 配置")
        _POOL = PooledDB(
            creator=pymysql,
            mincached=2,          # 启动即预热 2 条,首点不再吃冷握手
            maxcached=5,          # 空闲保留上限
            maxconnections=20,    # 并发上限(ThreadingHTTPServer 线程数)
            blocking=True,        # 连接耗尽时等待而非报错
            ping=1,               # 取用前 ping,自动剔除被 RDS 掐断的死连接
            host=os.getenv("MYSQL_HOST"),
            port=int(os.getenv("MYSQL_PORT", 3306)),
            user=os.getenv("MYSQL_USER"),
            password=os.getenv("MYSQL_PASSWORD"),
            database=os.getenv("MYSQL_DATABASE"),
            charset="utf8mb4", cursorclass=DictCursor,
            autocommit=True, connect_timeout=10,
        )
    return _POOL


def _conn():
    """从池取一条连接;用法不变(with cursor / conn.close() 归还池)。"""
    return _pool().connection()


# ── DDL ──────────────────────────────────────────────────────────────────────

SEARCH_TABLES = {"process": "search_process", "tools": "search_tools"}
MODE_TABLES = {"process": "mode_process", "tools": "mode_tools"}


def _search_table(mode_or_table):
    """mode(process/tools)或表名 → 合法搜索表名(白名单,防 SQL 注入)。"""
    t = SEARCH_TABLES.get(mode_or_table, mode_or_table)
    if t not in SEARCH_TABLES.values():
        raise ValueError(f"未知搜索表/模式: {mode_or_table!r}")
    return t


def _mode_table(mode_or_table):
    """mode(process/tools)或表名 → 合法解构表名(白名单,防 SQL 注入)。"""
    t = MODE_TABLES.get(mode_or_table, mode_or_table)
    if t not in MODE_TABLES.values():
        raise ValueError(f"未知解构表/模式: {mode_or_table!r}")
    return t


def _ddl_search(table, direction):
    return f"""
CREATE TABLE IF NOT EXISTS {table} (
  id            BIGINT AUTO_INCREMENT PRIMARY KEY,
  query_id      VARCHAR(32)   NOT NULL COMMENT 'q0000',
  query_text    VARCHAR(512)  NULL,
  case_id       VARCHAR(128)  NOT NULL COMMENT 'platform_channelContentId',
  platform      VARCHAR(32)   NULL,
  channel_content_id VARCHAR(128) NULL,
  title         VARCHAR(512)  NULL,
  url           VARCHAR(1024) NULL,
  content_type  VARCHAR(32)   NULL,
  body          LONGTEXT      NULL,
  images        JSON          NULL,
  videos        JSON          NULL,
  like_count    INT           NULL,
  publish_time  VARCHAR(64)   NULL,
  quality_score FLOAT         NULL COMMENT 'post._quality_score',
  quality_grade VARCHAR(8)    NULL,
  found_by      JSON          NULL COMMENT '命中的措辞数组',
  knowledge_type JSON         NULL COMMENT '["能力","工序","工具"] 子集',
  overall_score FLOAT         NULL COMMENT '(相关均值+质量均值)/2',
  llm_evaluation JSON         NULL COMMENT '评估全量 blob',
  created_at    TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
  updated_at    TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  UNIQUE KEY uk_qid_case (query_id, case_id),
  KEY idx_platform (platform)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='搜索+评估结果({direction})';
"""

DDL_PROCESS = """
CREATE TABLE IF NOT EXISTS mode_process (
  id            BIGINT AUTO_INCREMENT PRIMARY KEY,
  query_id      VARCHAR(32)   NOT NULL,
  case_id       VARCHAR(128)  NOT NULL,
  platform      VARCHAR(32)   NULL,
  post_title    VARCHAR(512)  NULL,
  source        JSON          NULL COMMENT '解构返回的 source 块',
  procedure_id  VARCHAR(16)   NULL COMMENT 'p1,p2…',
  name          VARCHAR(255)  NULL,
  purpose       TEXT          NULL,
  category      VARCHAR(32)   NULL COMMENT '产物创造/资产建设/自动化/分析/学习',
  declarations  JSON          NULL,
  type_registry JSON          NULL,
  steps         JSON          NULL COMMENT '步骤数组全量',
  step_count    INT           NULL,
  tools_used    JSON          NULL COMMENT '从 steps[].via 去重提取',
  model         VARCHAR(64)   NULL,
  version       VARCHAR(32)   NULL COMMENT 'v_MMDDHHMM,保留历史;link_* 为跨 query 复制(cost=0)',
  cost_usd      DECIMAL(10,6) NULL COMMENT '本次解构调用成本(同版本各行相同,聚合需按 case+version 去重)',
  duration_s    FLOAT         NULL,
  seq           SMALLINT      NULL COMMENT '帖内序号(0-based);与 (query_id,case_id,version) 组唯一键防并发/重复写',
  created_at    TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
  UNIQUE KEY uk_q_case_ver_seq (query_id, case_id, version, seq),
  KEY idx_case_ver (case_id, version),
  KEY idx_qid (query_id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='工序解构结果(每行一个工序)';
"""

DDL_TOOLS = """
CREATE TABLE IF NOT EXISTS mode_tools (
  id            BIGINT AUTO_INCREMENT PRIMARY KEY,
  query_id      VARCHAR(32)   NOT NULL,
  case_id       VARCHAR(128)  NOT NULL,
  platform      VARCHAR(32)   NULL,
  post_title    VARCHAR(512)  NULL,
  source        JSON          NULL COMMENT '解构时帖子来源块(tool_extract._row_to_source 产出)',
  tool_name     VARCHAR(255)  NULL,
  substance_scope JSON        NULL COMMENT '实质作用域(数组)',
  form_scope    JSON          NULL COMMENT '形式作用域(数组或null)',
  creation_layer VARCHAR(32)  NULL COMMENT '制作层/创作层',
  source_link   VARCHAR(1024) NULL,
  input_desc    TEXT          NULL,
  output_desc   TEXT          NULL,
  usage_json    JSON          NULL,
  cases_json    JSON          NULL,
  defects_json  JSON          NULL,
  updated_time  VARCHAR(64)   NULL COMMENT '工具最新更新时间',
  model         VARCHAR(64)   NULL,
  version       VARCHAR(32)   NULL COMMENT 'v_MMDDHHMM;link_* 为跨 query 复制(cost=0)',
  cost_usd      DECIMAL(10,6) NULL COMMENT '同 mode_process,聚合按 case+version 去重',
  duration_s    FLOAT         NULL,
  seq           SMALLINT      NULL COMMENT '帖内序号(0-based);与 (query_id,case_id,version) 组唯一键防并发/重复写',
  created_at    TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
  UNIQUE KEY uk_q_case_ver_seq (query_id, case_id, version, seq),
  KEY idx_case_ver (case_id, version),
  KEY idx_qid (query_id),
  KEY idx_tool_name (tool_name)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='工具解构结果(每行一个工具)';
"""


# 工序知识「已导入知识库」台账:防重复上传(stages/import_process_knowledge.py 用)。
# 每条知识 = 某 case 的某个工序(proc_index 1-based)。记录导入时的 mode_process 版本:
# 版本变了(重解构)说明内容已变,应重导;版本不变即视为「已传过」,跳过。
# 选 DB 台账而非本地文件,是为了换机器/换链接后也不会重复写知识库。
# 注:工具知识用独立的 tools_ingest_log,不与本表混用(case_id 是帖子物理身份,
#     同帖可能既被工序解构又被工具解构,共表会在 (case_id, index) 上撞键)。
DDL_INGEST_LOG = """
CREATE TABLE IF NOT EXISTS knowledge_ingest_log (
  id            BIGINT AUTO_INCREMENT PRIMARY KEY,
  case_id       VARCHAR(128)  NOT NULL,
  proc_index    INT           NOT NULL COMMENT '工序序号(1-based),对齐导入脚本枚举',
  version       VARCHAR(32)   NULL COMMENT '导入时 mode_process 版本;变了应重导',
  knowledge_id  VARCHAR(128)  NULL COMMENT '接口返回的 knowledge_id',
  api_url       VARCHAR(255)  NULL,
  ingested_at   TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
  updated_at    TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  UNIQUE KEY uk_case_proc (case_id, proc_index)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='工序知识已导入台账(防重复上传)';
"""

# 工具知识「已导入知识库」台账:语义同 knowledge_ingest_log,但针对工具方向独立成表
# (stages/import_tools_knowledge.py 用)。每条知识 = 某 case 的某个工具(tool_index 1-based),
# 版本记录导入时的 mode_tools 版本;变了(重解构)应重导,不变即「已传过」跳过。
DDL_TOOLS_INGEST_LOG = """
CREATE TABLE IF NOT EXISTS tools_ingest_log (
  id            BIGINT AUTO_INCREMENT PRIMARY KEY,
  case_id       VARCHAR(128)  NOT NULL,
  tool_index    INT           NOT NULL COMMENT '工具序号(1-based),对齐导入脚本枚举',
  version       VARCHAR(32)   NULL COMMENT '导入时 mode_tools 版本;变了应重导',
  knowledge_id  VARCHAR(128)  NULL COMMENT '接口返回的 knowledge_id',
  api_url       VARCHAR(255)  NULL,
  ingested_at   TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
  updated_at    TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  UNIQUE KEY uk_case_tool (case_id, tool_index)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='工具知识已导入台账(防重复上传)';
"""

# 「解构知识(工具·仅实质/形式作用域 → what-制作还原)」已导入台账。
# 与 tools_ingest_log 分表:两者同源 mode_tools、同键 (case_id, tool_index),但上传到不同维度
# (本表 = what-制作还原,tools_ingest_log = how工具),共表会在 (case_id, tool_index) 撞键,
# 使一方上传后另一方被误判「已传过」而跳过。故独立成表(stages/import_destruction_knowledge.py 用)。
DDL_DESTRUCTION_INGEST_LOG = """
CREATE TABLE IF NOT EXISTS destruction_ingest_log (
  id            BIGINT AUTO_INCREMENT PRIMARY KEY,
  case_id       VARCHAR(128)  NOT NULL,
  tool_index    INT           NOT NULL COMMENT '工具序号(1-based),对齐导入脚本枚举',
  version       VARCHAR(32)   NULL COMMENT '导入时 mode_tools 版本;变了应重导',
  knowledge_id  VARCHAR(128)  NULL COMMENT '接口返回的 knowledge_id',
  api_url       VARCHAR(255)  NULL,
  ingested_at   TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
  updated_at    TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  UNIQUE KEY uk_case_tool (case_id, tool_index)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='解构知识(what-制作还原)已导入台账(防重复上传)';
"""


# step 维度归类结果独立表(3-I,见 docs/step_classification_3-I_设计.md):一行 = 某 case 某版本
# 某工序某 step 某维度某子项的命中分类。与 mode_process 解耦,归类重跑只覆盖本表、不重写 steps blob;
# 存 matched_path 全路径,使「某节点及其子树的所有帖」一句 SQL(LIKE 前缀)即可,且与显示同源。
DDL_STEP_CLASSIFICATION = """
CREATE TABLE IF NOT EXISTS step_classification (
  id            BIGINT AUTO_INCREMENT PRIMARY KEY,
  case_id       VARCHAR(128) NOT NULL,
  query_id      VARCHAR(32)  NULL COMMENT 'record 时的 post_id,便于溯源',
  version       VARCHAR(32)  NULL COMMENT '对齐 mode_process 最新真实版,重跑覆盖',
  procedure_id  VARCHAR(16)  NULL COMMENT 'p1,p2…',
  step_id       VARCHAR(16)  NULL COMMENT 's1,s2…',
  dimension     VARCHAR(8)   NOT NULL COMMENT '实质/形式/类型/作用/动作',
  sub_index     SMALLINT     NOT NULL COMMENT '原值「、」拆分后的子项下标',
  raw_term      VARCHAR(255) NULL COMMENT '原始词',
  matched_name  VARCHAR(255) NULL COMMENT '命中分类名(单一最优;无命中不入库)',
  matched_path  VARCHAR(512) NULL COMMENT '命中分类全路径,如 /表象/视觉/空间/空间环境',
  matched_id    INT          NULL COMMENT 'cat-api stable_id',
  score         FLOAT        NULL,
  match_run_at  TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  UNIQUE KEY uk_step_dim_sub (case_id, version, procedure_id, step_id, dimension, sub_index),
  KEY idx_case (case_id),
  KEY idx_dim_name (dimension, matched_name),
  KEY idx_dim_path (dimension, matched_path)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='step 维度归类结果(与 mode_process 解耦)';
"""


# 工序解构拆表(3-II,见 docs/拆表_3-II_设计.md):process_run → process_procedure → process_step 三层,
# 与旧 mode_process 并行。step_json 逐字保留原 step → ProcessPayload 还原零漂移;抽取列供 DB 层查询。
DDL_PROCESS_RUN = """
CREATE TABLE IF NOT EXISTS process_run (
  id          BIGINT AUTO_INCREMENT PRIMARY KEY,
  query_id    VARCHAR(32)   NOT NULL,
  case_id     VARCHAR(128)  NOT NULL,
  version     VARCHAR(32)   NOT NULL,
  is_latest   TINYINT       NOT NULL DEFAULT 0 COMMENT '该 case 最新真实版=1(link_ 恒 0)',
  platform    VARCHAR(32)   NULL,
  post_title  VARCHAR(512)  NULL,
  source      JSON          NULL,
  model       VARCHAR(64)   NULL,
  cost_usd    DECIMAL(10,6) NULL,
  duration_s  FLOAT         NULL,
  created_at  TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
  UNIQUE KEY uk_q_case_ver (query_id, case_id, version),
  KEY idx_case_latest (case_id, is_latest),
  KEY idx_qid (query_id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='工序解构运行级(一行=一帖一版本)';
"""

DDL_PROCESS_PROCEDURE = """
CREATE TABLE IF NOT EXISTS process_procedure (
  id            BIGINT AUTO_INCREMENT PRIMARY KEY,
  run_id        BIGINT        NOT NULL,
  case_id       VARCHAR(128)  NOT NULL,
  version       VARCHAR(32)   NOT NULL,
  procedure_id  VARCHAR(16)   NULL,
  seq           SMALLINT      NOT NULL,
  name          VARCHAR(255)  NULL,
  purpose       TEXT          NULL,
  category      VARCHAR(32)   NULL,
  declarations  JSON          NULL,
  type_registry JSON          NULL,
  tools_used    JSON          NULL,
  step_count    INT           NULL,
  UNIQUE KEY uk_run_seq (run_id, seq),
  KEY idx_run (run_id),
  KEY idx_case_ver (case_id, version)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='工序解构工序级(一行=一个 procedure)';
"""

DDL_PROCESS_STEP = """
CREATE TABLE IF NOT EXISTS process_step (
  id            BIGINT AUTO_INCREMENT PRIMARY KEY,
  procedure_pk  BIGINT        NOT NULL,
  case_id       VARCHAR(128)  NOT NULL,
  version       VARCHAR(32)   NOT NULL,
  step_id       VARCHAR(16)   NULL,
  seq           SMALLINT      NOT NULL,
  kind          VARCHAR(16)   NULL,
  grp           VARCHAR(16)   NULL COMMENT '原 step.group(group 为保留字)',
  via           VARCHAR(255)  NULL,
  effect        VARCHAR(255)  NULL,
  action        VARCHAR(255)  NULL,
  substance     VARCHAR(512)  NULL,
  form          VARCHAR(512)  NULL,
  step_json     JSON          NOT NULL COMMENT '原 step 逐字 JSON,payload 还原即用它',
  UNIQUE KEY uk_proc_seq (procedure_pk, seq),
  KEY idx_procedure (procedure_pk),
  KEY idx_case_ver (case_id, version),
  KEY idx_action (action), KEY idx_substance (substance), KEY idx_form (form)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='工序解构步骤级(一行=一个 step)';
"""


def _ensure_column(cur, table, column, column_ddl):
    """给已存在的表幂等补列:列已存在则跳过(MySQL ADD COLUMN 无 IF NOT EXISTS)。
    column_ddl 为 ADD COLUMN 后的完整定义,如 \"source JSON NULL ... AFTER post_title\"。"""
    cur.execute("""SELECT COUNT(*) AS n FROM information_schema.columns
                   WHERE table_schema=DATABASE() AND table_name=%s AND column_name=%s""",
                (table, column))
    if cur.fetchone()["n"] == 0:
        cur.execute(f"ALTER TABLE {table} ADD COLUMN {column_ddl}")


def _ensure_unique_index(cur, table, index_name, cols):
    """幂等加唯一索引:已存在则跳过(MySQL ADD INDEX 无 IF NOT EXISTS)。
    cols 为列表达式,如 "query_id, case_id, version, seq"。加之前需保证无冲突数据。"""
    cur.execute("""SELECT COUNT(*) AS n FROM information_schema.statistics
                   WHERE table_schema=DATABASE() AND table_name=%s AND index_name=%s""",
                (table, index_name))
    if cur.fetchone()["n"] == 0:
        cur.execute(f"ALTER TABLE {table} ADD UNIQUE KEY {index_name} ({cols})")


def init_tables():
    conn = _conn()
    try:
        with conn.cursor() as cur:
            cur.execute(_ddl_search("search_process", "工序方向"))
            cur.execute(_ddl_search("search_tools", "工具方向"))
            cur.execute(DDL_PROCESS)
            cur.execute(DDL_TOOLS)
            cur.execute(DDL_INGEST_LOG)
            cur.execute(DDL_TOOLS_INGEST_LOG)
            cur.execute(DDL_DESTRUCTION_INGEST_LOG)
            cur.execute(DDL_STEP_CLASSIFICATION)
            cur.execute(DDL_PROCESS_RUN)
            cur.execute(DDL_PROCESS_PROCEDURE)
            cur.execute(DDL_PROCESS_STEP)
            # 历史库迁移:version 由 VARCHAR(16) 放宽到 32,容纳 link_v_mopN_* 复制版本。
            # MODIFY 幂等(已是 32 则 MySQL 元数据无操作),建表后表必存在,可安全执行。
            for t in ("mode_process", "mode_tools"):
                cur.execute(f"ALTER TABLE {t} MODIFY COLUMN version VARCHAR(32) NULL")
            # 历史库迁移:给老 mode_tools 补 source 列(MySQL 的 ADD COLUMN 无 IF NOT EXISTS,
            # 故先查 information_schema 判存在,缺了才 ADD,幂等)。
            _ensure_column(cur, "mode_tools", "source",
                           "source JSON NULL COMMENT '解构时帖子来源块' AFTER post_title")
            # 历史库迁移:加 seq(帖内序号)+ (query_id,case_id,version,seq) 唯一键,防并发/重复
            # 写入产生重复行。顺序必须是 加列 → 回填 → 加唯一键。MySQL 5.7 无窗口函数,seq 在
            # 应用层按 (query_id,case_id,version) 内 id 升序回填(现有数据该粒度已无重复)。
            for t in ("mode_process", "mode_tools"):
                _ensure_column(cur, t, "seq",
                               "seq SMALLINT NULL COMMENT '帖内序号(0-based)' AFTER duration_s")
            for t in ("mode_process", "mode_tools"):
                cur.execute(f"""SELECT id, query_id, case_id, version FROM {t}
                                WHERE seq IS NULL ORDER BY query_id, case_id, version, id""")
                key, n, ups = None, 0, []
                for r in cur.fetchall():
                    k = (r["query_id"], r["case_id"], r["version"])
                    if k != key:
                        key, n = k, 0
                    ups.append((n, r["id"])); n += 1
                if ups:
                    cur.executemany(f"UPDATE {t} SET seq=%s WHERE id=%s", ups)
                    print(f"  ↳ {t}: 回填 seq {len(ups)} 行")
            for t in ("mode_process", "mode_tools"):
                _ensure_unique_index(cur, t, "uk_q_case_ver_seq",
                                     "query_id, case_id, version, seq")
        print("✅ 建表完成:search_process, search_tools, mode_process, mode_tools, "
              "knowledge_ingest_log, tools_ingest_log, step_classification")
    finally:
        conn.close()


def clear_tables():
    """清空四张表的数据(TRUNCATE,表结构保留)。"""
    conn = _conn()
    try:
        with conn.cursor() as cur:
            for t in ("search_process", "search_tools", "mode_process", "mode_tools",
                      "process_run", "process_procedure", "process_step", "step_classification"):
                cur.execute(f"TRUNCATE TABLE {t}")
                print(f"🧹 已清空 {t}")
    finally:
        conn.close()


# ── 工具函数 ──────────────────────────────────────────────────────────────────

def _loads(v, default=None):
    """pymysql 的 JSON 列可能返回字符串,统一解析。"""
    if v is None:
        return default
    if isinstance(v, (list, dict)):
        return v
    try:
        return json.loads(v)
    except Exception:
        return default


def _j(v):
    """写入 JSON 列:None 保持 NULL,其余 dumps。"""
    return None if v is None else json.dumps(v, ensure_ascii=False)


def _collect_scores(node):
    """递归收集嵌套评估里所有「得分」。LLM 直出的得分多为字符串("1"/"4"),
    个别为数字(如 时效性 10),统一按 float 解析;非数值(如 "N/A")跳过不计入。"""
    out = []
    if isinstance(node, dict):
        for k, v in node.items():
            if k == "得分":
                try:
                    out.append(float(v))
                except (TypeError, ValueError):
                    pass
            else:
                out.extend(_collect_scores(v))
    elif isinstance(node, list):
        for v in node:
            out.extend(_collect_scores(v))
    return out


def overall_score(e):
    """综合分 = (相关性各项均值 + 质量各项均值) / 可得部分数。算不出返回 None。"""
    parts = []
    for key in ("相关性", "质量"):
        scores = _collect_scores((e or {}).get(key))
        if scores:
            parts.append(sum(scores) / len(scores))
    return round(sum(parts) / len(parts), 2) if parts else None


def _recency_hard(date_str):
    """硬时效(同 mode_procedure/server.py:_recency_hard):半年内=3 / 两年内=2 / 更早=1。
    publish_time 头 10 字符按 YYYY-MM-DD 解析,失败返回 None(不参与判定)。"""
    try:
        d = datetime.strptime(str(date_str or "")[:10], "%Y-%m-%d")
    except (ValueError, TypeError):
        return None
    days = (datetime.now() - d).days
    if days <= 180:
        return 3
    if days <= 730:
        return 2
    return 1


def _fixed_dim_score(evaluation, name):
    """取 质量.固定维度.<name>.得分 标量,缺失/非数值返回 None(不参与判定)。"""
    v = (((evaluation or {}).get("质量") or {}).get("固定维度") or {}).get(name)
    if isinstance(v, dict):
        v = v.get("得分")
    try:
        return float(v) if v is not None else None
    except (TypeError, ValueError):
        return None


def _impl_score(evaluation):
    """取 质量.动态维度.工序.字段完整性.实现完整性.得分 标量,缺失/非数值返回 None。
    新版 prompt 把旧「可复现性」的硬封顶规则并入了「实现完整性」,故采纳门槛改读此处。"""
    v = ((((((evaluation or {}).get("质量") or {}).get("动态维度") or {})
           .get("工序") or {}).get("字段完整性") or {}).get("实现完整性"))
    if isinstance(v, dict):
        v = v.get("得分")
    try:
        return float(v) if v is not None else None
    except (TypeError, ValueError):
        return None


def _repro_score(evaluation):
    """采纳门槛用的「可复现/可实现」得分:优先旧版「可复现性」(固定维度),
    缺失则回退新版「实现完整性」(动态维度.工序)。这样新旧两套评估 blob 都能正确判定。"""
    v = _fixed_dim_score(evaluation, "可复现性")
    return v if v is not None else _impl_score(evaluation)


def is_adopted(overall, evaluation, publish_time):
    """采纳/命中判定,口径对齐 mode_procedure 的 decision=="report":
    制作相关性<4、可复现/实现完整性<4、发布超两年、综合分<6 —— 任一命中即不采纳;指标缺失不参与判定。
    (意图可控性暂只采分不设门槛,留待阈值标定后再开。)
    可复现/实现门槛兼容新旧 schema:旧版读「可复现性」,新版读「实现完整性」(见 _repro_score)。

    fail-closed:评估失败(_error)、blob 缺失/为空、或综合分算不出(None)→ 直接判不采纳。
    评不出的帖子不该混进命中集(此前 fail-open 会因各指标取不到值而误判采纳)。"""
    if not isinstance(evaluation, dict) or not evaluation or evaluation.get("_error"):
        return False
    if overall is None:
        return False
    rel = None
    v = ((evaluation or {}).get("相关性") or {}).get("和内容制作知识相关")
    if isinstance(v, dict):
        v = v.get("得分")
    try:
        rel = float(v) if v is not None else None
    except (TypeError, ValueError):
        rel = None
    if rel is not None and rel < 4:
        return False
    repro = _repro_score(evaluation)
    if repro is not None and repro < 4:
        return False
    rh = _recency_hard(publish_time)
    if rh is not None and rh < 2:
        return False
    if overall is not None and float(overall) < 6:
        return False
    return True


def is_adopted_rel(overall, rel, publish_time, repro=None):
    """is_adopted 的轻量版:相关性得分(rel)、可复现/实现门槛(repro)已由 SQL JSON_EXTRACT
    直接取出(repro 由 _REPRO_SQL 兼容新旧 schema 取值),无需传输/解析整块 llm_evaluation。
    判定口径与 is_adopted 完全一致(含 fail-closed:综合分算不出→不采纳;失败帖的 overall_score 列为 NULL)。"""
    if overall is None:
        return False
    try:
        rel = float(rel) if rel is not None else None
    except (TypeError, ValueError):
        rel = None
    if rel is not None and rel < 4:
        return False
    try:
        repro = float(repro) if repro is not None else None
    except (TypeError, ValueError):
        repro = None
    if repro is not None and repro < 4:
        return False
    rh = _recency_hard(publish_time)
    if rh is not None and rh < 2:
        return False
    if overall is not None and float(overall) < 6:
        return False
    return True


# ── search_process / search_tools ────────────────────────────────────────────

def upsert_search_posts(query_id, query_text, results, table="search_process"):
    """一组搜索结果写入指定搜索表(按 (query_id, case_id) upsert)。返回写入条数。
    table:search_process(工序方向) / search_tools(工具方向)。"""
    table = _search_table(table)
    if not results:
        return 0
    rows = []
    for r in results:
        post = r.get("post") or {}
        e = r.get("llm_evaluation") or {}
        rows.append((
            query_id, query_text, r.get("case_id"), r.get("platform"),
            r.get("channel_content_id"),
            (post.get("title") or post.get("desc") or "")[:500],
            r.get("source_url"), post.get("content_type"),
            post.get("body_text") or post.get("desc") or "",
            _j(post.get("images") or []), _j(post.get("videos") or []),
            post.get("like_count"),
            str(post.get("publish_time") or post.get("publish_timestamp") or "")[:64],
            post.get("_quality_score"), post.get("_quality_grade"),
            _j(r.get("found_by_queries") or []),
            _j(e.get("知识类型") or []),
            overall_score(e),
            _j(e),
        ))
    sql = f"""
    INSERT INTO {table}
      (query_id, query_text, case_id, platform, channel_content_id, title, url,
       content_type, body, images, videos, like_count, publish_time,
       quality_score, quality_grade, found_by, knowledge_type,
       overall_score, llm_evaluation)
    VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)
    ON DUPLICATE KEY UPDATE
      query_text=VALUES(query_text), platform=VALUES(platform),
      channel_content_id=VALUES(channel_content_id), title=VALUES(title), url=VALUES(url),
      content_type=VALUES(content_type), body=VALUES(body), images=VALUES(images),
      videos=VALUES(videos), like_count=VALUES(like_count), publish_time=VALUES(publish_time),
      quality_score=VALUES(quality_score), quality_grade=VALUES(quality_grade),
      found_by=VALUES(found_by), knowledge_type=VALUES(knowledge_type),
      overall_score=VALUES(overall_score), llm_evaluation=VALUES(llm_evaluation);
    """
    conn = _conn()
    try:
        with conn.cursor() as cur:
            cur.executemany(sql, rows)
        return len(rows)
    finally:
        conn.close()


# 占位帖 case_id:query 列表由 search_process 按 query_id 聚合得出(无独立 query 主表),
# 一个 query 要进列表必须至少有一行。为支持「只登记 query、不触发搜索」,给这类 query 写
# 一行哨兵帖,只承载 query_id+query_text。该哨兵行不属于任何真实帖子,故所有「帖子视图 /
# 统计」读取点都用 _REAL_POST 过滤掉它(fetch_queries 的 post_count、fetch_posts、
# fetch_all_posts、count_executed_queries、fetch_dashboard_rows)。真搜不会用到此 case_id。
PENDING_CASE_ID = "__pending__"
_REAL_POST = f"case_id <> '{PENDING_CASE_ID}'"


def add_pending_process_queries(texts):
    """把一批 query 词作为「占位 query」加入工序 query 列表(search_process),不触发搜索/解构。
    每条新增写一行哨兵帖(case_id=PENDING_CASE_ID,只填 query_id/query_text)。
    去重:① 文件内重复保序去重;② query_text 已存在于 search_process(含此前占位)则跳过。
    query_id 跨 process/tools 统一续号,避免与工具方向撞号。返回 (added, skipped)。"""
    seen, cleaned = set(), []
    for t in texts:
        t = (t or "").strip()
        if t and t not in seen:
            seen.add(t)
            cleaned.append(t)
    conn = _conn()
    try:
        with conn.cursor() as cur:
            cur.execute("SELECT DISTINCT query_text FROM search_process WHERE query_text IS NOT NULL")
            existing = {r["query_text"] for r in cur.fetchall()}
            cur.execute("SELECT query_id FROM search_process "
                        "UNION SELECT query_id FROM search_tools")
            nums = [int(r["query_id"][1:]) for r in cur.fetchall()
                    if r["query_id"] and r["query_id"].startswith("q") and r["query_id"][1:].isdigit()]
            nxt = (max(nums) + 1) if nums else 0
            rows = []
            for t in cleaned:
                if t in existing:
                    continue
                rows.append((f"q{nxt:04d}", t, PENDING_CASE_ID))
                nxt += 1
            if rows:
                cur.executemany(
                    "INSERT INTO search_process (query_id, query_text, case_id) "
                    "VALUES (%s,%s,%s)", rows)
        return len(rows), len(cleaned) - len(rows)
    finally:
        conn.close()


def fetch_queries(mode="process"):
    """某方向搜索表的 query 列表 + 帖子数 + 采纳/命中数 + 解构进度。"""
    table = _search_table(mode)
    conn = _conn()
    try:
        with conn.cursor() as cur:
            # post_count 只数真实帖,占位哨兵行不计(占位 query 显示为 0 帖);
            # GROUP BY 仍含占位 query_id,故无搜索的 query 也会出现在列表里。
            cur.execute(f"""SELECT query_id, MAX(query_text) AS query_text,
                                   COUNT(CASE WHEN {_REAL_POST} THEN 1 END) AS post_count
                            FROM {table} GROUP BY query_id ORDER BY query_id""")
            queries = cur.fetchall()
            # 采纳数:SQL 直取 rel/repro 标量算,**不拉整表 llm_evaluation**(旧版全表 blob,切 tab 巨慢)
            cur.execute(f"""SELECT query_id, overall_score, publish_time,
                                   {_REL_SQL} AS rel, {_REPRO_SQL} AS repro FROM {table}""")
            hits = {}
            for r in cur.fetchall():
                if is_adopted_rel(r["overall_score"], r["rel"], r["publish_time"], r["repro"]):
                    hits[r["query_id"]] = hits.get(r["query_id"], 0) + 1
            cur.execute("SELECT query_id, COUNT(DISTINCT case_id) AS n FROM process_run GROUP BY query_id")  # 3-II
            np = {r["query_id"]: r["n"] for r in cur.fetchall()}
            cur.execute("SELECT query_id, COUNT(DISTINCT case_id) AS n FROM mode_tools GROUP BY query_id")
            nt = {r["query_id"]: r["n"] for r in cur.fetchall()}
    finally:
        conn.close()
    for q in queries:
        q["hit_count"] = hits.get(q["query_id"], 0)
        q["process_done"] = np.get(q["query_id"], 0)
        q["tools_done"] = nt.get(q["query_id"], 0)
    return queries


def fetch_posts(query_id, mode="process"):
    """列表用:只取列表所需列 + SQL 直取 adopted 标量,**不拉 body/videos/llm_evaluation 大字段**
    (llm_evaluation ~1.5MB/帖,旧版 SELECT * 切 tab/选 query 要几十 MB 过远程 RDS,故慢)。
    正文/评分等详情按需走 fetch_post。带 adopted/has_process/has_tools;adopted 口径用
    is_adopted_rel(与 is_adopted 完全一致,rel/repro 由 _REL_SQL/_REPRO_SQL 直取标量)。"""
    table = _search_table(mode)
    conn = _conn()
    try:
        with conn.cursor() as cur:
            cur.execute(f"""SELECT id, query_id, query_text, case_id, platform, channel_content_id,
                                   title, url, content_type, images, like_count, publish_time,
                                   quality_score, quality_grade, found_by, knowledge_type, overall_score,
                                   {_REL_SQL} AS rel, {_REPRO_SQL} AS repro
                            FROM {table} WHERE query_id=%s AND {_REAL_POST}
                            ORDER BY overall_score DESC, id""", (query_id,))
            rows = cur.fetchall()
            cur.execute("SELECT DISTINCT case_id FROM process_run WHERE query_id=%s", (query_id,))  # 3-II
            hp = {r["case_id"] for r in cur.fetchall()}
            cur.execute("SELECT DISTINCT case_id FROM mode_tools WHERE query_id=%s", (query_id,))
            ht = {r["case_id"] for r in cur.fetchall()}
            # 已归类(工序):hp 中各 case 最新真实版的工序里有任一 step 含 substanceMatch(同归类回写口径)。
            # 3-II:改查三表——run 级聚合「该 run 是否有任一 step_json 含 substanceMatch」,逻辑同 _categorized_from_rows。
            hc = set()
            if hp:
                ph = ",".join(["%s"] * len(hp))
                cur.execute(f"""SELECT r.case_id, r.version, r.id,
                                       (LEFT(r.version,5)='link_') AS islink,
                                       EXISTS(SELECT 1 FROM process_step s
                                              JOIN process_procedure p ON s.procedure_pk=p.id
                                              WHERE p.run_id=r.id AND s.step_json LIKE %s) AS cat
                                FROM process_run r WHERE r.case_id IN ({ph})""",
                            ['%substanceMatch%'] + list(hp))
                hc = _categorized_from_rows(cur.fetchall())
    finally:
        conn.close()
    for r in rows:
        for col in ("images", "found_by", "knowledge_type"):
            r[col] = _loads(r[col])
        r["adopted"] = is_adopted_rel(r["overall_score"], r.pop("rel", None),
                                      r["publish_time"], r.pop("repro", None))
        r["has_process"] = r["case_id"] in hp
        r["has_tools"] = r["case_id"] in ht
        r["has_category"] = r["case_id"] in hc
    return rows


def fetch_post(query_id, case_id, table="search_process"):
    """指定搜索表的单帖完整行(给 pipeline 脚本重建 source 用)。无则 None。"""
    table = _search_table(table)
    conn = _conn()
    try:
        with conn.cursor() as cur:
            cur.execute(f"SELECT * FROM {table} WHERE query_id=%s AND case_id=%s",
                        (query_id, case_id))
            row = cur.fetchone()
    finally:
        conn.close()
    if not row:
        return None
    for col in ("images", "videos", "found_by", "knowledge_type", "llm_evaluation"):
        row[col] = _loads(row[col])
    return row


def fetch_all_posts(mode="process", *, query_ids=None, case_ids=None, adopted_only=False, distinct=False,
                    limit=None, offset=0):
    """某方向「全部帖子」:跨所有 query 的列表(瘦身列,口径同 fetch_posts,不拉
    body/videos/llm_evaluation 大字段)。fetch_posts 限定单 query,本函数默认取全表。
      - query_ids:选填 query_id 列表,传了就 WHERE query_id IN(...) 只取这些 query
        的帖子(SQL 层过滤,不拉全表);None=全部,[]=空结果。
      - adopted_only=True:只返回采纳帖(is_adopted_rel 口径,rel/repro 由
        _REL_SQL/_REPRO_SQL 直取标量算,不拉整表 blob)。
      - distinct=True:按 case_id 去重(同一帖被多个 query 搜到时,只保留
        overall_score 最高的一行——已按 score 降序,取首次出现即最高分)。
      - limit/offset:分页(limit=None 不分页)。
    返回 (total, rows):total 为过滤(+去重)后的总条数,rows 为本页切片。"""
    table = _search_table(mode)
    # 始终排除占位哨兵行(无搜索的 query 不在帖子视图里出现)
    where, params = f" WHERE {_REAL_POST}", []
    if query_ids is not None:
        if not query_ids:
            return 0, []   # 显式空列表:直接空结果,不必查库
        where += " AND query_id IN (" + ",".join(["%s"] * len(query_ids)) + ")"
        params = list(query_ids)
    if case_ids is not None:
        # case_ids:选填 case_id 列表(知识归类联动用——按「分类树节点的 knowledge_ids」
        # 取被归类进该节点的帖子,而非按 query 搜索结果)。None=不过滤,[]=空结果。
        if not case_ids:
            return 0, []
        where += " AND case_id IN (" + ",".join(["%s"] * len(case_ids)) + ")"
        params += list(case_ids)
    conn = _conn()
    try:
        with conn.cursor() as cur:
            cur.execute(f"""SELECT id, query_id, query_text, case_id, platform, channel_content_id,
                                   title, url, content_type, images, like_count, publish_time,
                                   quality_score, quality_grade, found_by, knowledge_type, overall_score,
                                   {_REL_SQL} AS rel, {_REPRO_SQL} AS repro
                            FROM {table}{where}
                            ORDER BY overall_score DESC, id""", params)
            rows = cur.fetchall()
            # has_process/has_tools 全局判定:跨 query 的「该帖是否已解构」,两张解构表各取一次
            cur.execute("SELECT DISTINCT case_id FROM process_run")  # 3-II
            hp = {r["case_id"] for r in cur.fetchall()}
            cur.execute("SELECT DISTINCT case_id FROM mode_tools")
            ht = {r["case_id"] for r in cur.fetchall()}
    finally:
        conn.close()
    out, seen = [], set()
    for r in rows:
        for col in ("images", "found_by", "knowledge_type"):
            r[col] = _loads(r[col])
        r["adopted"] = is_adopted_rel(r["overall_score"], r.pop("rel", None),
                                      r["publish_time"], r.pop("repro", None))
        if adopted_only and not r["adopted"]:
            continue
        if distinct:
            if r["case_id"] in seen:
                continue
            seen.add(r["case_id"])
        r["has_process"] = r["case_id"] in hp
        r["has_tools"] = r["case_id"] in ht
        out.append(r)
    total = len(out)
    if limit is not None:
        out = out[offset:offset + limit]
    elif offset:
        out = out[offset:]
    return total, out


def count_executed_queries(mode="process"):
    """该方向「已执行」的 query 数 = 搜索表里出现过的 distinct query_id 个数。
    注:一次搜索若 0 命中则不写任何行,故不计入(口径为「已产出结果的 query」)。"""
    table = _search_table(mode)
    conn = _conn()
    try:
        with conn.cursor() as cur:
            cur.execute(f"SELECT COUNT(DISTINCT query_id) AS n FROM {table} WHERE {_REAL_POST}")
            return cur.fetchone()["n"]
    finally:
        conn.close()


# ── mode_process ─────────────────────────────────────────────────────────────

def replace_process(query_id, case_id, platform, post_title, payload,
                    model, version, cost_usd, duration_s):
    """写入一帖某版本的工序解构结果(payload = {source, procedures})。
    删 (case_id, version) 旧行再插,同版本重跑幂等、跨版本保留历史。返回工序条数。"""
    source = payload.get("source")
    procedures = payload.get("procedures") or []
    conn = _conn()
    try:
        conn.begin()   # DELETE+INSERT 原子化:配合 uk_q_case_ver_seq,并发/重复写入不会留下重复行
        with conn.cursor() as cur:
            cur.execute("DELETE FROM mode_process WHERE case_id=%s AND version=%s",
                        (case_id, version))
            if procedures:
                rows = []
                for i, p in enumerate(procedures):
                    steps = p.get("steps") or []
                    vias = []
                    for s in steps:
                        v = s.get("via")
                        if v and v not in vias:
                            vias.append(v)
                    rows.append((
                        query_id, case_id, platform, (post_title or "")[:500],
                        _j(source), p.get("id"), (p.get("name") or "")[:250],
                        p.get("purpose"), p.get("category"),
                        _j(p.get("declarations")), _j(p.get("type_registry")),
                        _j(steps), len(steps), _j(vias),
                        model, version, cost_usd, duration_s, i,
                    ))
                cur.executemany("""
                INSERT INTO mode_process
                  (query_id, case_id, platform, post_title, source, procedure_id, name,
                   purpose, category, declarations, type_registry, steps, step_count,
                   tools_used, model, version, cost_usd, duration_s, seq)
                VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)
                """, rows)
            # 3-II 双写:同一事务内覆盖写 process_run/procedure/step(新表为主,mode_process 兜底)
            _split_write_run(cur, query_id, case_id, platform, post_title, source,
                             procedures, model, version, cost_usd, duration_s)
        conn.commit()
        return len(procedures)
    except Exception:
        conn.rollback()
        raise
    finally:
        conn.close()


def fetch_process_versions(case_id):
    # 3-II:版本列表读自三表(n=该版本工序数,口径同旧 mode_process 行数)
    conn = _conn()
    try:
        with conn.cursor() as cur:
            cur.execute("""SELECT r.version, COUNT(p.id) AS n, MAX(r.model) AS model
                           FROM process_run r
                           LEFT JOIN process_procedure p ON p.run_id = r.id
                           WHERE r.case_id=%s
                           GROUP BY r.version
                           ORDER BY (LEFT(r.version,5)='link_') ASC, MAX(r.id) DESC""", (case_id,))
            return cur.fetchall()
    finally:
        conn.close()


def fetch_process(case_id, version=None):
    """重建 {case_id, version, model, source, procedures:[...]}。version=None 取最新。
    3-II:读自 process_run/procedure/step(返回形状与旧 mode_process 口径完全一致)。"""
    return rebuild_process_from_split(case_id, version)


def fetch_all_process(case_ids=None, lite=False):
    """批量取多帖工序解构(每帖取最新真实版,link_ 排后),一次查询拍平。
      - case_ids:选填 case_id 列表;None=全表所有有解构的帖,[]=空(直接返回空)。
      - lite:True 走精简投影(丢大字段 + 截断 value),供工序库平铺表快速首屏。
    返回 {case_id: _proc_payload(...)};无解构记录的 case_id 不出现在结果里。"""
    if case_ids is not None and not case_ids:
        return {}
    # 3-II:读自三表,3 条批量 SQL(run → procedure → step),避免按 case 逐次往返(保首屏性能)。
    where, params = "", []
    if case_ids is not None:
        where = " WHERE case_id IN (" + ",".join(["%s"] * len(case_ids)) + ")"
        params = list(case_ids)
    conn = _conn()
    try:
        with conn.cursor() as cur:
            cur.execute(f"SELECT * FROM process_run{where} ORDER BY case_id, id", params)
            runs = cur.fetchall()
            best = {}  # case_id -> (key, run);口径同 fetch_process(is_latest→真实版→id 最大)
            for r in runs:
                c = r["case_id"]
                k = ((r.get("is_latest") or 0), not str(r["version"]).startswith("link_"), r["id"])
                if c not in best or k > best[c][0]:
                    best[c] = (k, r)
            run_ids = [run["id"] for _k, run in best.values()]
            procs, steps_by_proc = [], {}
            if run_ids:
                ph = ",".join(["%s"] * len(run_ids))
                cur.execute(f"SELECT * FROM process_procedure WHERE run_id IN ({ph}) "
                            "ORDER BY run_id, seq, id", run_ids)
                procs = cur.fetchall()
                pid = [p["id"] for p in procs]
                if pid:
                    ph2 = ",".join(["%s"] * len(pid))
                    cur.execute(f"SELECT procedure_pk, step_json FROM process_step "
                                f"WHERE procedure_pk IN ({ph2}) ORDER BY procedure_pk, seq, id", pid)
                    for s in cur.fetchall():
                        steps_by_proc.setdefault(s["procedure_pk"], []).append(
                            _loads(s["step_json"], {}))
    finally:
        conn.close()
    procs_by_run = {}
    for p in procs:
        procs_by_run.setdefault(p["run_id"], []).append(p)
    out = {}
    for cid, (_k, run) in best.items():
        rprocs = procs_by_run.get(run["id"], [])
        if lite:
            procedures = [{"id": p["procedure_id"], "name": p["name"],
                           "steps": _lite_steps(steps_by_proc.get(p["id"], []))} for p in rprocs]
            out[cid] = {"case_id": cid, "version": run["version"],
                        "title": run["post_title"], "procedures": procedures}
        else:
            procedures = [{
                "id": p["procedure_id"], "name": p["name"], "purpose": p["purpose"],
                "category": p["category"], "declarations": _loads(p["declarations"]),
                "type_registry": _loads(p["type_registry"]),
                "steps": steps_by_proc.get(p["id"], []),
                "tools_used": _loads(p["tools_used"], [])} for p in rprocs]
            out[cid] = {"case_id": cid, "version": run["version"], "platform": run["platform"],
                        "title": run["post_title"], "model": run["model"],
                        "cost_usd": float(run["cost_usd"]) if run["cost_usd"] is not None else None,
                        "duration_s": run["duration_s"], "source": _loads(run["source"]),
                        "procedures": procedures}
    return out


def fetch_all_process_pairs():
    """全量重跑归类用:每个有解构记录的 case 取其「最新真实版」(link_ 排后、id 最大)
    所在行的 (query_id, case_id)。选版口径与 fetch_process / fetch_all_process 完全一致,
    保证回写的版本即前端展示的版本。query_id 作 post_id 供 category-match record。
    返回 [(query_id, case_id)],按 case_id 排序、按 case 去重。"""
    conn = _conn()
    try:
        with conn.cursor() as cur:
            cur.execute("SELECT id, query_id, case_id, version FROM process_run ORDER BY case_id, id")
            rows = cur.fetchall()
    finally:
        conn.close()
    best = {}  # case_id -> ((is_real, id), query_id);取 max,与 fetch_all_process 同口径
    for r in rows:
        c = r["case_id"]
        key = (not str(r["version"]).startswith("link_"), r["id"])
        if c not in best or key > best[c][0]:
            best[c] = (key, r["query_id"])
    return [(qid, c) for c, (_k, qid) in sorted(best.items())]


# ── 拆表(3-II):mode_process → process_run/procedure/step ────────────────────────

def _jn(v):
    """JSON 列搬运规整:先 _loads 再 _j,无论 pymysql 返回 str 还是已解析对象都得到合法 JSON 文本/None。"""
    return _j(_loads(v))


def _t(s, n):
    """抽取列按字符数截断(只用于可查询列;完整值在 step_json,不丢)。None 原样。"""
    return s if (s is None or len(str(s)) <= n) else str(s)[:n]


def migrate_to_split_tables(truncate=True):
    """把 mode_process 灌入 process_run/procedure/step(加法,不动 mode_process)。
    幂等:默认先 TRUNCATE 三表再灌。step_json 逐字保留原 step。返回 {runs, procedures, steps}。"""
    conn = _conn()
    try:
        with conn.cursor() as cur:
            cur.execute("SELECT * FROM mode_process ORDER BY query_id, case_id, version, seq, id")
            rows = cur.fetchall()
    finally:
        conn.close()
    # 每 case 最新真实版(口径同 fetch_all_process),用于置 is_latest
    best = {}
    for r in rows:
        c = r["case_id"]; k = (not str(r["version"]).startswith("link_"), r["id"])
        if c not in best or k > best[c][0]:
            best[c] = (k, r["version"])
    latest_ver = {c: v for c, (_k, v) in best.items()}
    # 按 (query_id, case_id, version) 分组 = 一个 run(保序)
    runs = {}
    order = []
    for r in rows:
        key = (r["query_id"], r["case_id"], r["version"])
        if key not in runs:
            runs[key] = []; order.append(key)
        runs[key].append(r)

    nrun = nproc = nstep = 0
    conn = _conn()
    try:
        conn.begin()
        with conn.cursor() as cur:
            if truncate:
                for t in ("process_step", "process_procedure", "process_run"):
                    cur.execute(f"TRUNCATE TABLE {t}")
            for key in order:
                qid, cid, ver = key
                prows = sorted(runs[key],
                               key=lambda r: (r["seq"] if r["seq"] is not None else 0, r["id"]))
                first = prows[0]
                is_latest = 1 if (not str(ver).startswith("link_") and latest_ver.get(cid) == ver) else 0
                cur.execute("""INSERT INTO process_run
                    (query_id, case_id, version, is_latest, platform, post_title, source,
                     model, cost_usd, duration_s, created_at)
                    VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)""",
                    (qid, cid, ver, is_latest, first["platform"], first["post_title"],
                     _jn(first["source"]), first["model"], first["cost_usd"], first["duration_s"],
                     first["created_at"]))   # 保留原解构时间(Dashboard 成本趋势按此)
                run_id = cur.lastrowid; nrun += 1
                for pseq, pr in enumerate(prows):
                    steps = _loads(pr["steps"], [])
                    cur.execute("""INSERT INTO process_procedure
                        (run_id, case_id, version, procedure_id, seq, name, purpose, category,
                         declarations, type_registry, tools_used, step_count)
                        VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)""",
                        (run_id, cid, ver, pr["procedure_id"], pseq, pr["name"], pr["purpose"],
                         pr["category"], _jn(pr["declarations"]), _jn(pr["type_registry"]),
                         _jn(pr["tools_used"]), pr["step_count"]))
                    proc_pk = cur.lastrowid; nproc += 1
                    for sseq, st in enumerate(steps if isinstance(steps, list) else []):
                        if not isinstance(st, dict):
                            st = {"value": st}
                        cur.execute("""INSERT INTO process_step
                            (procedure_pk, case_id, version, step_id, seq, kind, grp, via,
                             effect, action, substance, form, step_json)
                            VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)""",
                            (proc_pk, cid, ver, _t(st.get("id"), 16), sseq,
                             _t(st.get("kind"), 16), _t(st.get("group"), 16), _t(st.get("via"), 255),
                             _t(st.get("effect"), 255), _t(st.get("action"), 255),
                             _t(st.get("substance"), 512), _t(st.get("form"), 512),
                             _j(st)))
                        nstep += 1
        conn.commit()
    except Exception:
        conn.rollback(); raise
    finally:
        conn.close()
    return {"runs": nrun, "procedures": nproc, "steps": nstep}


def _steps_by_proc(cur, run_id):
    """取某 run 下所有 step 的 step_json,按 procedure_pk 分组(组内按 seq 序)。"""
    cur.execute("""SELECT procedure_pk, step_json FROM process_step
                   WHERE procedure_pk IN (SELECT id FROM process_procedure WHERE run_id=%s)
                   ORDER BY procedure_pk, seq, id""", (run_id,))
    out = {}
    for s in cur.fetchall():
        out.setdefault(s["procedure_pk"], []).append(_loads(s["step_json"], {}))
    return out


def _split_payload(cur, run, lite=False):
    """run 行 → ProcessPayload(形状与 _proc_payload 完全一致)。需开放 cursor。无 run 返回 None。"""
    if not run:
        return None
    cur.execute("SELECT * FROM process_procedure WHERE run_id=%s ORDER BY seq, id", (run["id"],))
    procs = cur.fetchall()
    sbp = _steps_by_proc(cur, run["id"])
    case_id, version = run["case_id"], run["version"]
    if lite:
        procedures = [{
            "id": p["procedure_id"], "name": p["name"],
            "steps": _lite_steps(sbp.get(p["id"], [])),
        } for p in procs]
        return {"case_id": case_id, "version": version,
                "title": run["post_title"], "procedures": procedures}
    procedures = [{
        "id": p["procedure_id"], "name": p["name"], "purpose": p["purpose"],
        "category": p["category"], "declarations": _loads(p["declarations"]),
        "type_registry": _loads(p["type_registry"]), "steps": sbp.get(p["id"], []),
        "tools_used": _loads(p["tools_used"], []),
    } for p in procs]
    return {"case_id": case_id, "version": version, "platform": run["platform"],
            "title": run["post_title"], "model": run["model"],
            "cost_usd": float(run["cost_usd"]) if run["cost_usd"] is not None else None,
            "duration_s": run["duration_s"],
            "source": _loads(run["source"]), "procedures": procedures}


def _select_run(cur, case_id, version=None, query_id=None):
    """选 run:version 指定则精确取;否则取最新(is_latest 优先 → 非 link_ → id 最大,
    口径同旧 fetch_process)。query_id 给定则限定该 query。无则 None。"""
    where = "case_id=%s"; params = [case_id]
    if query_id is not None:
        where += " AND query_id=%s"; params.append(query_id)
    if version is not None:
        cur.execute(f"SELECT * FROM process_run WHERE {where} AND version=%s LIMIT 1",
                    params + [version])
    else:
        cur.execute(f"""SELECT * FROM process_run WHERE {where}
                        ORDER BY is_latest DESC, (LEFT(version,5)='link_') ASC, id DESC
                        LIMIT 1""", params)
    return cur.fetchone()


def rebuild_process_from_split(case_id, version=None):
    """从三表重建 ProcessPayload(迁移校验/读侧共用)。version=None 取最新。"""
    conn = _conn()
    try:
        with conn.cursor() as cur:
            return _split_payload(cur, _select_run(cur, case_id, version))
    finally:
        conn.close()


def _refresh_is_latest(cur, case_id):
    """重算某 case 各 run 的 is_latest:最新真实版(非 link_、id 最大)置 1,其余 0。
    口径同 fetch_process 选版。只有 link_ 时无 1(读侧 rebuild 仍能按排序兜底)。需在事务内调用。"""
    cur.execute("SELECT id, version FROM process_run WHERE case_id=%s", (case_id,))
    rows = cur.fetchall()
    if not rows:
        return
    cur.execute("UPDATE process_run SET is_latest=0 WHERE case_id=%s", (case_id,))
    best = max(rows, key=lambda r: (not str(r["version"]).startswith("link_"), r["id"]))
    if not str(best["version"]).startswith("link_"):
        cur.execute("UPDATE process_run SET is_latest=1 WHERE id=%s", (best["id"],))


def _split_write_run(cur, query_id, case_id, platform, post_title, source,
                     procedures, model, version, cost_usd, duration_s):
    """在三表覆盖写某 (query_id, case_id, version) 的 run/procedure/step(需在调用方事务内)。
    与 replace_process 写 mode_process 同口径(tools_used 从 steps[].via 去重)。维护 is_latest。"""
    cur.execute("SELECT id FROM process_run WHERE query_id=%s AND case_id=%s AND version=%s",
                (query_id, case_id, version))
    old = cur.fetchone()
    if old:
        cur.execute("""DELETE FROM process_step WHERE procedure_pk IN
                       (SELECT id FROM process_procedure WHERE run_id=%s)""", (old["id"],))
        cur.execute("DELETE FROM process_procedure WHERE run_id=%s", (old["id"],))
        cur.execute("DELETE FROM process_run WHERE id=%s", (old["id"],))
    cur.execute("""INSERT INTO process_run
        (query_id, case_id, version, is_latest, platform, post_title, source,
         model, cost_usd, duration_s)
        VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)""",
        (query_id, case_id, version, 0, platform, (post_title or "")[:500],
         _j(source), model, cost_usd, duration_s))
    run_id = cur.lastrowid
    for pseq, p in enumerate(procedures or []):
        steps = p.get("steps") or []
        vias = []
        for s in steps:
            v = s.get("via") if isinstance(s, dict) else None
            if v and v not in vias:
                vias.append(v)
        cur.execute("""INSERT INTO process_procedure
            (run_id, case_id, version, procedure_id, seq, name, purpose, category,
             declarations, type_registry, tools_used, step_count)
            VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)""",
            (run_id, case_id, version, p.get("id"), pseq, (p.get("name") or "")[:250],
             p.get("purpose"), p.get("category"), _j(p.get("declarations")),
             _j(p.get("type_registry")), _j(vias), len(steps)))
        proc_pk = cur.lastrowid
        for sseq, st in enumerate(steps):
            if not isinstance(st, dict):
                st = {"value": st}
            cur.execute("""INSERT INTO process_step
                (procedure_pk, case_id, version, step_id, seq, kind, grp, via,
                 effect, action, substance, form, step_json)
                VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)""",
                (proc_pk, case_id, version, _t(st.get("id"), 16), sseq,
                 _t(st.get("kind"), 16), _t(st.get("group"), 16), _t(st.get("via"), 255),
                 _t(st.get("effect"), 255), _t(st.get("action"), 255),
                 _t(st.get("substance"), 512), _t(st.get("form"), 512), _j(st)))
    _refresh_is_latest(cur, case_id)


def _split_update_steps(cur, case_id, version, steps_in_order):
    """归类回写:按工序顺序覆盖某 (case_id, version) 各 procedure 的 step_json(及抽取列)。
    steps_in_order 与 process_procedure(按 seq) 一一对应。需在事务内。返回更新的 step 行数。"""
    cur.execute("""SELECT id FROM process_procedure
                   WHERE case_id=%s AND version=%s ORDER BY seq, id""", (case_id, version))
    proc_ids = [r["id"] for r in cur.fetchall()]
    if len(proc_ids) != len(steps_in_order):
        raise ValueError(f"process_procedure 行数({len(proc_ids)})与工序数({len(steps_in_order)})不一致")
    n = 0
    for proc_pk, steps in zip(proc_ids, steps_in_order):
        cur.execute("DELETE FROM process_step WHERE procedure_pk=%s", (proc_pk,))
        for sseq, st in enumerate(steps or []):
            if not isinstance(st, dict):
                st = {"value": st}
            cur.execute("""INSERT INTO process_step
                (procedure_pk, case_id, version, step_id, seq, kind, grp, via,
                 effect, action, substance, form, step_json)
                VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)""",
                (proc_pk, case_id, version, _t(st.get("id"), 16), sseq,
                 _t(st.get("kind"), 16), _t(st.get("group"), 16), _t(st.get("via"), 255),
                 _t(st.get("effect"), 255), _t(st.get("action"), 255),
                 _t(st.get("substance"), 512), _t(st.get("form"), 512), _j(st)))
            n += 1
    return n


# ── step_classification(归类独立表,3-I)────────────────────────────────────────

def replace_step_classification(case_id, version, records):
    """覆盖写某 (case_id, version) 的 step 维度归类结果(DELETE+INSERT 原子,语义同 replace_process)。
    records:[{query_id, procedure_id, step_id, dimension, sub_index, raw_term,
              matched_name, matched_path, matched_id, score}](只含有命中的子项)。
    同版本重跑幂等、洗净旧归属。返回写入条数。"""
    conn = _conn()
    try:
        conn.begin()
        with conn.cursor() as cur:
            cur.execute("DELETE FROM step_classification WHERE case_id=%s AND version=%s",
                        (case_id, version))
            if records:
                rows = [(
                    case_id, r.get("query_id"), version,
                    r.get("procedure_id"), r.get("step_id"),
                    r.get("dimension"), r.get("sub_index"), r.get("raw_term"),
                    r.get("matched_name"), r.get("matched_path"),
                    r.get("matched_id"), r.get("score"),
                ) for r in records]
                cur.executemany("""
                INSERT INTO step_classification
                  (case_id, query_id, version, procedure_id, step_id, dimension,
                   sub_index, raw_term, matched_name, matched_path, matched_id, score)
                VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)
                """, rows)
        conn.commit()
        return len(records)
    except Exception:
        conn.rollback()
        raise
    finally:
        conn.close()


def fetch_classified_cases(dimension, path=None, name=None, subtree=False):
    """维度联动取帖:返回某维度下命中分类的 distinct case_id 集合。
      - path:给定分类全路径。**默认精确匹配本节点**(matched_path = path),与 cat-api 树徽标
        (节点 knowledge_count)同口径,保证取到的帖工序解构 substanceMatch 就是该节点名;
        subtree=True 时改取 **该节点及整棵子树**(matched_path = path 或 LIKE 'path/%')。
      - name:给定分类名精确匹配(matched_name = name)。
    path 与 name 至少给一个;都给则取并集。
    与 mode_process.steps[].substanceMatch 同源(同一次归类双写),不依赖 cat-api 树。"""
    conds, params = [], [dimension]
    if path:
        if subtree:
            conds.append("(matched_path = %s OR matched_path LIKE %s)")
            params += [path, path.rstrip("/") + "/%"]
        else:
            conds.append("matched_path = %s")
            params.append(path)
    if name:
        conds.append("matched_name = %s")
        params.append(name)
    if not conds:
        return set()
    where = "dimension=%s AND (" + " OR ".join(conds) + ")"
    conn = _conn()
    try:
        with conn.cursor() as cur:
            cur.execute(f"SELECT DISTINCT case_id FROM step_classification WHERE {where}", params)
            return {r["case_id"] for r in cur.fetchall()}
    finally:
        conn.close()


def fetch_process_by_query(query_id, case_id, version=None):
    """同 fetch_process,但用 (query_id, case_id) 精确定位某 query 下该帖的工序
    (category-match 用:post_id=query_id / knowledge_id=case_id)。
    version=None 取该 (query_id, case_id) 下最新真实版(link_ 排后)。无行返回 None。"""
    conn = _conn()
    try:
        with conn.cursor() as cur:
            return _split_payload(cur, _select_run(cur, case_id, version, query_id=query_id))
    finally:
        conn.close()


def update_process_steps_by_query(query_id, case_id, version, steps_in_order):
    """按工序顺序覆盖某 (query_id, case_id, version) 各行的 steps JSON 列。
    steps_in_order 必须与 fetch_process_by_query 返回的 procedures 同序(均按 seq, id 升序);
    按行 id 一一对应更新,稳健于 seq 不连续。行数与工序数不符则报错回滚。返回更新行数。"""
    conn = _conn()
    try:
        conn.begin()
        with conn.cursor() as cur:
            cur.execute("""SELECT id FROM mode_process
                           WHERE query_id=%s AND case_id=%s AND version=%s
                           ORDER BY seq, id""", (query_id, case_id, version))
            ids = [r["id"] for r in cur.fetchall()]
            if len(ids) != len(steps_in_order):
                raise ValueError(f"行数({len(ids)})与工序数({len(steps_in_order)})不一致")
            n = 0
            for row_id, steps in zip(ids, steps_in_order):
                cur.execute("UPDATE mode_process SET steps=%s WHERE id=%s", (_j(steps), row_id))
                n += cur.rowcount
            # 3-II 双写:同步更新 process_step.step_json(同事务)
            _split_update_steps(cur, case_id, version, steps_in_order)
        conn.commit()
        return n
    except Exception:
        conn.rollback()
        raise
    finally:
        conn.close()


def update_process_steps(case_id, version, steps_in_order):
    """按工序顺序覆盖某 (case_id, version) 各行的 steps JSON 列(不限 query_id)。
    与 fetch_process / fetch_extract 同口径(按 case 的某版本),保证归类回写的版本
    与前端 /api/extract 展示的版本一致(否则 link_ 复制帖会写错版本、前端看不到)。
    steps_in_order 须与 fetch_process(case_id, version).procedures 同序(按 id 升序)。
    行数与工序数不符则报错回滚。返回更新行数。"""
    conn = _conn()
    try:
        conn.begin()
        with conn.cursor() as cur:
            cur.execute("""SELECT id FROM mode_process WHERE case_id=%s AND version=%s
                           ORDER BY id""", (case_id, version))
            ids = [r["id"] for r in cur.fetchall()]
            if len(ids) != len(steps_in_order):
                raise ValueError(f"行数({len(ids)})与工序数({len(steps_in_order)})不一致")
            n = 0
            for row_id, steps in zip(ids, steps_in_order):
                cur.execute("UPDATE mode_process SET steps=%s WHERE id=%s", (_j(steps), row_id))
                n += cur.rowcount
            # 3-II 双写:同步更新 process_step.step_json(同事务)
            _split_update_steps(cur, case_id, version, steps_in_order)
        conn.commit()
        return n
    except Exception:
        conn.rollback()
        raise
    finally:
        conn.close()


def _categorized_from_rows(rows):
    """rows:[{case_id, version, id, islink(0/1), cat(0/1)}]。返回已归类 case 集合。
    口径:每 case 取最新真实版(真实版优先、id 最大),该版本**任一行** cat=1 即已归类。
    关键——不能只看「id 最大的那一行」:工序里可能有 steps 为空的 procedure(step_count=0),
    其行永远不含 substanceMatch,若恰好 id 最大会误判整条 case 未归类(见该函数修复缘由)。"""
    best, has = {}, {}
    for r in rows:
        c, v = r["case_id"], r["version"]
        sk = (1 if r["islink"] else 0, -r["id"])   # 真实版(islink=0)优先,其次 id 大;取 min
        if c not in best or sk < best[c][0]:
            best[c] = (sk, v)
        k = (c, v)
        has[k] = has.get(k, False) or bool(r["cat"])
    return {c for c, (sk, v) in best.items() if has.get((c, v))}


def fetch_categorized_cases(case_ids, mode="process"):
    """返回 case_ids 中「已归类」的子集:该 case 最新真实版(link_ 排后)的工序里有任一步骤
    含 substanceMatch(归类跑过的非空 step 一定带此 key)。与归类回写/前端展示同口径。
    供前端判断「是否已全部归类 → 提示重新归类」。仅工序方向有意义(mode_process)。"""
    if not case_ids:
        return set()
    ph = ",".join(["%s"] * len(case_ids))
    conn = _conn()
    try:
        with conn.cursor() as cur:
            if mode == "process":
                # 3-II:run 级聚合「该 run 是否有任一 step_json 含 substanceMatch」
                cur.execute(f"""SELECT r.case_id, r.version, r.id,
                                       (LEFT(r.version,5)='link_') AS islink,
                                       EXISTS(SELECT 1 FROM process_step s
                                              JOIN process_procedure p ON s.procedure_pk=p.id
                                              WHERE p.run_id=r.id AND s.step_json LIKE %s) AS cat
                                FROM process_run r WHERE r.case_id IN ({ph})""",
                            ['%substanceMatch%'] + list(case_ids))
            else:
                table = _mode_table(mode)
                cur.execute(f"""SELECT case_id, version, id,
                                       (LEFT(version,5)='link_') AS islink, (steps LIKE %s) AS cat
                                FROM {table} WHERE case_id IN ({ph})""",
                            ['%substanceMatch%'] + list(case_ids))
            rows = cur.fetchall()
    finally:
        conn.close()
    return _categorized_from_rows(rows)


_LITE_VALUE_LIMIT = 300  # lite 模式 输入/输出 value 截断字节上限


def _trunc(s, limit=_LITE_VALUE_LIMIT):
    """按 UTF-8 字节截断长文本(不切坏多字节字符),超限追加省略号。非字符串原样返回。"""
    if not isinstance(s, str):
        return s
    b = s.encode("utf-8")
    if len(b) <= limit:
        return s
    return b[:limit].decode("utf-8", "ignore") + "…"


def _lite_steps(steps):
    """lite 模式:仅截断每步 inputs/outputs 的 value(其余字段供平铺表/分组用,保留)。"""
    if not isinstance(steps, list):
        return steps
    for st in steps:
        if not isinstance(st, dict):
            continue
        for io_key in ("inputs", "outputs"):
            ios = st.get(io_key)
            if isinstance(ios, list):
                for io in ios:
                    if isinstance(io, dict) and "value" in io:
                        io["value"] = _trunc(io.get("value"))
    return steps


def _proc_payload(case_id, version, rows, lite=False):
    """mode_process 行集 → {case_id, version, …, procedures:[...]}。无行返回 None。
    lite=True:工序库平铺表只需 procedures[].{id,name,steps},丢弃大字段
    (source/declarations/type_registry/tools_used)并截断输入/输出 value;
    完整值由前端展开时按 case 调 /api/process 懒加载。"""
    if not rows:
        return None
    if lite:
        procedures = [{
            "id": r["procedure_id"], "name": r["name"],
            "steps": _lite_steps(_loads(r["steps"], [])),
        } for r in rows]
        return {"case_id": case_id, "version": version,
                "title": rows[0]["post_title"], "procedures": procedures}
    procedures = [{
        "id": r["procedure_id"], "name": r["name"], "purpose": r["purpose"],
        "category": r["category"], "declarations": _loads(r["declarations"]),
        "type_registry": _loads(r["type_registry"]), "steps": _loads(r["steps"], []),
        "tools_used": _loads(r["tools_used"], []),
    } for r in rows]
    return {"case_id": case_id, "version": version, "platform": rows[0]["platform"],
            "title": rows[0]["post_title"], "model": rows[0]["model"],
            "cost_usd": float(rows[0]["cost_usd"]) if rows[0]["cost_usd"] is not None else None,
            "duration_s": rows[0]["duration_s"],
            "source": _loads(rows[0]["source"]), "procedures": procedures}


# ── mode_tools ───────────────────────────────────────────────────────────────

def replace_tools(query_id, case_id, platform, post_title, tools,
                  model, version, cost_usd, duration_s, source=None):
    """写入一帖某版本的工具解构结果。语义同 replace_process。返回工具条数。
    source:帖子来源块(同 mode_process,每行重复存),供知识上传脚本重建 source 用。"""
    src = _j(source)
    conn = _conn()
    try:
        conn.begin()   # DELETE+INSERT 原子化:配合 uk_q_case_ver_seq,并发/重复写入不会留下重复行
        with conn.cursor() as cur:
            cur.execute("DELETE FROM mode_tools WHERE case_id=%s AND version=%s",
                        (case_id, version))
            if tools:
                rows = [(
                    query_id, case_id, platform, (post_title or "")[:500], src,
                    (t.get("工具名称") or "")[:250],
                    _j(t.get("实质作用域")), _j(t.get("形式作用域")),
                    t.get("创作层级"), t.get("来源链接"), t.get("输入"), t.get("输出"),
                    _j(t.get("用法")), _j(t.get("案例")), _j(t.get("缺点")),
                    t.get("最新更新时间"), model, version, cost_usd, duration_s, i,
                ) for i, t in enumerate(tools)]
                cur.executemany("""
                INSERT INTO mode_tools
                  (query_id, case_id, platform, post_title, source, tool_name, substance_scope,
                   form_scope, creation_layer, source_link, input_desc, output_desc,
                   usage_json, cases_json, defects_json, updated_time, model, version,
                   cost_usd, duration_s, seq)
                VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)
                """, rows)
        conn.commit()
        return len(tools)
    except Exception:
        conn.rollback()
        raise
    finally:
        conn.close()


def fetch_tools_versions(case_id):
    conn = _conn()
    try:
        with conn.cursor() as cur:
            cur.execute("""SELECT version, COUNT(*) AS n, MAX(model) AS model
                           FROM mode_tools WHERE case_id=%s
                           GROUP BY version
                           ORDER BY (LEFT(version,5)='link_') ASC, MAX(id) DESC""", (case_id,))
            return cur.fetchall()
    finally:
        conn.close()


def fetch_tools(case_id, version=None):
    """重建 {case_id, version, model, tool_count, tools:[...]}。version=None 取最新。"""
    conn = _conn()
    try:
        with conn.cursor() as cur:
            if version is None:
                cur.execute("""SELECT version FROM mode_tools WHERE case_id=%s
                               ORDER BY (LEFT(version,5)='link_') ASC, id DESC LIMIT 1""", (case_id,))
                row = cur.fetchone()
                if not row:
                    return None
                version = row["version"]
            cur.execute("""SELECT * FROM mode_tools WHERE case_id=%s AND version=%s
                           ORDER BY id""", (case_id, version))
            rows = cur.fetchall()
    finally:
        conn.close()
    return _tools_payload(case_id, version, rows)


def _tools_payload(case_id, version, rows):
    """mode_tools 行集 → {case_id, version, …, tools:[...]}。无行返回 None。"""
    if not rows:
        return None
    tools = [{
        "工具名称": r["tool_name"], "实质作用域": _loads(r["substance_scope"]),
        "形式作用域": _loads(r["form_scope"]), "创作层级": r["creation_layer"],
        "来源链接": r["source_link"], "输入": r["input_desc"], "输出": r["output_desc"],
        "用法": _loads(r["usage_json"]), "案例": _loads(r["cases_json"]),
        "缺点": _loads(r["defects_json"]), "最新更新时间": r["updated_time"],
    } for r in rows]
    return {"case_id": case_id, "version": version, "platform": rows[0]["platform"],
            "title": rows[0]["post_title"], "model": rows[0]["model"],
            "cost_usd": float(rows[0]["cost_usd"]) if rows[0]["cost_usd"] is not None else None,
            "duration_s": rows[0]["duration_s"],
            "source": _loads(rows[0].get("source")),
            "tool_count": len(tools), "tools": tools}


# ── 点击帖子合一查询(单连接,最少往返;远程 RDS 每次往返 ~80ms,故按次数优化)──

def fetch_extract(mode, case_id, version=None):
    """一次取版本列表 + 解构详情,复用同一条池连接、最少往返。
    返回 {versions, data, missing}。mode: process / tools。"""
    is_proc = mode != "tools"
    if is_proc:
        # 3-II:工序方向版本列表 + 详情都读自三表(形状不变)
        conn = _conn()
        try:
            with conn.cursor() as cur:
                cur.execute("""SELECT r.version, COUNT(p.id) AS n, MAX(r.model) AS model
                               FROM process_run r
                               LEFT JOIN process_procedure p ON p.run_id = r.id
                               WHERE r.case_id=%s GROUP BY r.version
                               ORDER BY (LEFT(r.version,5)='link_') ASC, MAX(r.id) DESC""", (case_id,))
                versions = cur.fetchall()
                target = version or (versions[0]["version"] if versions else None)
                payload = _split_payload(cur, _select_run(cur, case_id, target)) if target else None
        finally:
            conn.close()
        return {"versions": versions, "data": payload, "missing": payload is None}
    # 工具方向:仍读 mode_tools(本期不拆)
    mtable = _mode_table("tools")
    conn = _conn()
    try:
        with conn.cursor() as cur:
            cur.execute(f"""SELECT version, COUNT(*) AS n, MAX(model) AS model
                            FROM {mtable} WHERE case_id=%s
                            GROUP BY version
                            ORDER BY (LEFT(version,5)='link_') ASC, MAX(id) DESC""", (case_id,))
            versions = cur.fetchall()
            target = version or (versions[0]["version"] if versions else None)
            rows = []
            if target is not None:
                cur.execute(f"SELECT * FROM {mtable} WHERE case_id=%s AND version=%s ORDER BY id",
                            (case_id, target))
                rows = cur.fetchall()
    finally:
        conn.close()
    payload = _tools_payload(case_id, target, rows)
    return {"versions": versions, "data": payload, "missing": payload is None}


# ── 跨 query 去重 / link 复制(方案A:解构前先去重,避免重复花钱)──────────────
# case_id 是帖子物理身份(platform_channelContentId),与 query 无关。同一帖被多个
# query 搜到时只需真实解构一次;其余 query 用 link_* 复制行补齐关联(cost=0)。

def latest_real_version(case_id, mode="process"):
    """该 case 是否已有「真实」解构(任意 query;link_* 是复制品,不算源)。
    返回最新一行 {"version","query_id"} 或 None。给解构前去重判定用。"""
    table = "process_run" if mode == "process" else _mode_table(mode)  # 3-II:工序读三表的 run
    conn = _conn()
    try:
        with conn.cursor() as cur:
            cur.execute(f"""SELECT version, query_id FROM {table}
                            WHERE case_id=%s AND LEFT(version,5) <> 'link_'
                            ORDER BY id DESC LIMIT 1""", (case_id,))
            return cur.fetchone()
    finally:
        conn.close()


def link_process(query_id, case_id, mode="process"):
    """把 case 在别处最新「真实」版本的解构行复制到目标 query
    (version='link_'+源版本, cost_usd=0)。幂等(先删目标同版本)。
    返回复制行数;该 case 从未真实解构过则返回 0(无源可复制)。"""
    table = _mode_table(mode)
    conn = _conn()
    try:
        with conn.cursor() as cur:
            cur.execute(f"""SELECT version FROM {table}
                            WHERE case_id=%s AND LEFT(version,5) <> 'link_'
                            ORDER BY id DESC LIMIT 1""", (case_id,))
            r = cur.fetchone()
            if not r:
                return 0
            srcver = r["version"]
            newver = ("link_" + srcver)[:32]   # version 列 VARCHAR(32)
            # 复制除自增 id / 时间戳外的全部列,改写 query_id / version / cost。
            cur.execute(f"SHOW COLUMNS FROM {table}")
            cols = [c["Field"] for c in cur.fetchall()
                    if c["Field"] not in ("id", "created_at", "updated_at")]
            cur.execute(f"SELECT {','.join(cols)} FROM {table} WHERE case_id=%s AND version=%s",
                        (case_id, srcver))
            rows = cur.fetchall()
            cur.execute(f"DELETE FROM {table} WHERE query_id=%s AND case_id=%s AND version=%s",
                        (query_id, case_id, newver))
            for row in rows:
                row = dict(row)
                row["query_id"] = query_id
                row["version"] = newver
                row["cost_usd"] = 0
                cur.execute(
                    f"INSERT INTO {table} ({','.join(cols)}) VALUES ({','.join(['%s']*len(cols))})",
                    [row[k] for k in cols])
            # 3-II 双写(仅工序方向):把 link_ 复制版同步写进三表
            if mode == "process" and rows:
                srows = sorted(rows, key=lambda r: (r.get("seq") if r.get("seq") is not None else 0))
                f = srows[0]
                procedures = [{
                    "id": r.get("procedure_id"), "name": r.get("name"), "purpose": r.get("purpose"),
                    "category": r.get("category"), "declarations": _loads(r.get("declarations")),
                    "type_registry": _loads(r.get("type_registry")), "steps": _loads(r.get("steps"), []),
                } for r in srows]
                _split_write_run(cur, query_id, case_id, f.get("platform"), f.get("post_title"),
                                 _loads(f.get("source")), procedures, f.get("model"), newver,
                                 0, f.get("duration_s"))
            return len(rows)
    finally:
        conn.close()


# ── Dashboard 原始行(指标计算在 server.py)─────────────────────────────────────

# 采纳判定只需「和内容制作知识相关」的得分,用 SQL JSON_EXTRACT 直取这一个标量,
# 避免把整块 llm_evaluation(本库 ~1.5MB)拉到 Python 再解析。得分可能直接是数字,
# 也可能裹在 {"得分": x} 里,COALESCE 两条路径覆盖两种存法,口径同 is_adopted。
_REL_SQL = ("JSON_UNQUOTE(COALESCE("
            "JSON_EXTRACT(llm_evaluation,'$.\"相关性\".\"和内容制作知识相关\".\"得分\"'),"
            "JSON_EXTRACT(llm_evaluation,'$.\"相关性\".\"和内容制作知识相关\"')))")

# 可复现/实现门槛标量直取(口径同 is_adopted 的 _repro_score):兼容新旧 schema——
# 旧版「质量.固定维度.可复现性」,新版「质量.动态维度.工序.字段完整性.实现完整性」,COALESCE 依次回退。
_REPRO_SQL = ("JSON_UNQUOTE(COALESCE("
              "JSON_EXTRACT(llm_evaluation,'$.\"质量\".\"固定维度\".\"可复现性\".\"得分\"'),"
              "JSON_EXTRACT(llm_evaluation,'$.\"质量\".\"固定维度\".\"可复现性\"'),"
              "JSON_EXTRACT(llm_evaluation,'$.\"质量\".\"动态维度\".\"工序\".\"字段完整性\".\"实现完整性\".\"得分\"'),"
              "JSON_EXTRACT(llm_evaluation,'$.\"质量\".\"动态维度\".\"工序\".\"字段完整性\".\"实现完整性\"')))")


def fetch_adopted_process_cases(query_id=None):
    """返回「已采纳且有工序解构」的 case_id 列表(供知识上传脚本用)。

    采纳是帖子级属性(评估存在 search_process),工序解构存在 mode_process,故二者 JOIN:
    只取两边都有的 case,再用 is_adopted_rel(口径同 Dashboard)在 Python 侧过滤。
    relevance 得分由 _REL_SQL 直取标量,不传整块 llm_evaluation。
    query_id 给定时只看该搜索任务下的 case。返回去重、按 case_id 排序的列表。
    """
    sql = (f"SELECT DISTINCT s.case_id, s.overall_score, s.publish_time, "
           f"{_REL_SQL} AS rel, {_REPRO_SQL} AS repro "
           "FROM search_process s "
           "JOIN (SELECT DISTINCT case_id FROM process_run) m ON s.case_id = m.case_id")  # 3-II
    params = ()
    if query_id:
        sql += " WHERE s.query_id=%s"
        params = (query_id,)
    conn = _conn()
    try:
        with conn.cursor() as cur:
            cur.execute(sql, params)
            rows = cur.fetchall()
    finally:
        conn.close()
    cases = [r["case_id"] for r in rows
             if is_adopted_rel(r["overall_score"], r["rel"], r["publish_time"], r["repro"])]
    return sorted(set(cases))


def fetch_destructed_tools_cases(query_ids=None):
    """返回「有工具解构」的 case_id 列表(**不看采纳**),按 mode_tools.query_id 过滤。

    与 fetch_adopted_tools_cases 的区别:不 JOIN search_tools、不做 is_adopted_rel 过滤,
    直接取 mode_tools 里存在解构行的帖子。query_ids=None 取全部;给列表则 IN 过滤
    (可跨多个 query;同一 case 在多 query 下解构时按 case_id 去重)。去重、排序返回。
    供「所有已解构(含未采纳)」上传口径用。
    """
    sql = "SELECT DISTINCT case_id FROM mode_tools"
    params = ()
    if query_ids is not None:
        if not query_ids:
            return []   # 显式空列表:直接空结果
        sql += " WHERE query_id IN (" + ",".join(["%s"] * len(query_ids)) + ")"
        params = tuple(query_ids)
    conn = _conn()
    try:
        with conn.cursor() as cur:
            cur.execute(sql, params)
            return sorted({r["case_id"] for r in cur.fetchall()})
    finally:
        conn.close()


def fetch_adopted_tools_cases(query_id=None):
    """返回「已采纳且有工具解构」的 case_id 列表(供工具知识上传脚本用)。

    与 fetch_adopted_process_cases 完全同构,只把搜索/解构表换成工具方向:
    采纳是帖子级属性(评估存在 search_tools),工具解构存在 mode_tools,故二者 JOIN,
    只取两边都有的 case,再用 is_adopted_rel(口径同 Dashboard)在 Python 侧过滤。
    query_id 给定时只看该搜索任务下的 case。返回去重、按 case_id 排序的列表。
    """
    sql = (f"SELECT DISTINCT s.case_id, s.overall_score, s.publish_time, "
           f"{_REL_SQL} AS rel, {_REPRO_SQL} AS repro "
           "FROM search_tools s "
           "JOIN (SELECT DISTINCT case_id FROM mode_tools) m ON s.case_id = m.case_id")
    params = ()
    if query_id:
        sql += " WHERE s.query_id=%s"
        params = (query_id,)
    conn = _conn()
    try:
        with conn.cursor() as cur:
            cur.execute(sql, params)
            rows = cur.fetchall()
    finally:
        conn.close()
    cases = [r["case_id"] for r in rows
             if is_adopted_rel(r["overall_score"], r["rel"], r["publish_time"], r["repro"])]
    return sorted(set(cases))


def route_tables(knowledge_types):
    """知识类型标签 → 落表列表(有序去重)。
    工序/能力 → search_process;工具 → search_tools;两者都含写两表;空/None 兜底 search_process。
    评估是统一一套(同一 llm_evaluation blob)，故同帖落多表不重复打分,只是多写一行。"""
    kt = set(knowledge_types or [])
    tables = []
    if kt & {"工具"}:
        tables.append("search_tools")
    if (kt & {"工序", "能力"}) or not tables:   # 工序/能力,或没命中任何已知标签 → 兜底 process
        tables.insert(0, "search_process")
    return tables


# ── 评估去重:复用 query 无关分,只重算 query 相关分(search_eval.py 用)──────────

def fetch_existing_eval(case_id, table="search_process"):
    """返回该 case 在搜索表里最近一条「有效」评估 blob(任意 query)。
    评估去重用:同帖在别的相似 query 下评过时,复用其 query 无关分(质量/通用相关/时效),
    只重算「和 query 相关」。无有效评估(全是 _error 或没评过)返回 None。
    取最近若干条逐一挑出首个非 error、结构完整的 blob。"""
    table = _search_table(table)
    conn = _conn()
    try:
        with conn.cursor() as cur:
            cur.execute(f"""SELECT llm_evaluation FROM {table}
                            WHERE case_id=%s AND llm_evaluation IS NOT NULL
                            ORDER BY updated_at DESC, id DESC LIMIT 5""", (case_id,))
            rows = cur.fetchall()
    finally:
        conn.close()
    for r in rows:
        e = _loads(r["llm_evaluation"])
        if isinstance(e, dict) and not e.get("_error") and isinstance(e.get("相关性"), dict):
            return e
    return None


def fetch_existing_eval_any(case_id):
    """跨两张搜索表找该 case 最近一条有效评估 blob。
    评估与表无关(统一一套),任一表评过即可复用,避免同帖在两表各评一次。无则 None。"""
    for table in ("search_process", "search_tools"):
        e = fetch_existing_eval(case_id, table)
        if e:
            return e
    return None


def update_post_eval(query_id, case_id, evaluation, table="search_process"):
    """用新的评估 blob 覆盖某 (query, case) 行的 llm_evaluation,并同步重算派生列
    overall_score、knowledge_type(口径同 upsert_search_posts)。返回受影响行数。"""
    table = _search_table(table)
    overall = overall_score(evaluation)
    ktype = evaluation.get("知识类型") if isinstance(evaluation, dict) else None
    conn = _conn()
    try:
        with conn.cursor() as cur:
            n = cur.execute(
                f"UPDATE {table} SET llm_evaluation=%s, overall_score=%s, knowledge_type=%s "
                "WHERE query_id=%s AND case_id=%s",
                (_j(evaluation), overall, _j(ktype), query_id, case_id))
        return n
    finally:
        conn.close()


# ── 上传去重:知识库已导入台账(stages/import_process_knowledge.py 用)────────────────

def fetch_ingested_map(case_id):
    """返回 {proc_index: version} —— 该 case 各工序已导入知识库的版本。空表示没传过。"""
    conn = _conn()
    try:
        with conn.cursor() as cur:
            cur.execute("SELECT proc_index, version FROM knowledge_ingest_log WHERE case_id=%s",
                        (case_id,))
            return {r["proc_index"]: r["version"] for r in cur.fetchall()}
    finally:
        conn.close()


def mark_ingested(case_id, proc_index, version, knowledge_id=None, api_url=None):
    """记一条「已导入」台账(case_id+proc_index 唯一,重导同序号则更新版本/knowledge_id)。"""
    conn = _conn()
    try:
        with conn.cursor() as cur:
            cur.execute("""INSERT INTO knowledge_ingest_log
                             (case_id, proc_index, version, knowledge_id, api_url)
                           VALUES (%s,%s,%s,%s,%s)
                           ON DUPLICATE KEY UPDATE version=VALUES(version),
                             knowledge_id=VALUES(knowledge_id), api_url=VALUES(api_url)""",
                        (case_id, proc_index, version, knowledge_id, api_url))
    finally:
        conn.close()


def fetch_tools_ingested_map(case_id):
    """返回 {tool_index: version} —— 该 case 各工具已导入知识库的版本。空表示没传过。
    工具方向独立台账(tools_ingest_log),与工序的 knowledge_ingest_log 互不干扰。"""
    conn = _conn()
    try:
        with conn.cursor() as cur:
            cur.execute("SELECT tool_index, version FROM tools_ingest_log WHERE case_id=%s",
                        (case_id,))
            return {r["tool_index"]: r["version"] for r in cur.fetchall()}
    finally:
        conn.close()


def mark_tools_ingested(case_id, tool_index, version, knowledge_id=None, api_url=None):
    """记一条工具「已导入」台账(case_id+tool_index 唯一,重导同序号则更新版本/knowledge_id)。"""
    conn = _conn()
    try:
        with conn.cursor() as cur:
            cur.execute("""INSERT INTO tools_ingest_log
                             (case_id, tool_index, version, knowledge_id, api_url)
                           VALUES (%s,%s,%s,%s,%s)
                           ON DUPLICATE KEY UPDATE version=VALUES(version),
                             knowledge_id=VALUES(knowledge_id), api_url=VALUES(api_url)""",
                        (case_id, tool_index, version, knowledge_id, api_url))
    finally:
        conn.close()


def fetch_destruction_ingested_map(case_id):
    """返回 {tool_index: version} —— 该 case 各工具的解构知识(what-制作还原)已导入的版本。
    独立台账(destruction_ingest_log),与工具方向 tools_ingest_log 互不干扰(见 DDL 注释)。"""
    conn = _conn()
    try:
        with conn.cursor() as cur:
            cur.execute("SELECT tool_index, version FROM destruction_ingest_log WHERE case_id=%s",
                        (case_id,))
            return {r["tool_index"]: r["version"] for r in cur.fetchall()}
    finally:
        conn.close()


def mark_destruction_ingested(case_id, tool_index, version, knowledge_id=None, api_url=None):
    """记一条解构知识「已导入」台账(case_id+tool_index 唯一,重导同序号则更新版本/knowledge_id)。"""
    conn = _conn()
    try:
        with conn.cursor() as cur:
            cur.execute("""INSERT INTO destruction_ingest_log
                             (case_id, tool_index, version, knowledge_id, api_url)
                           VALUES (%s,%s,%s,%s,%s)
                           ON DUPLICATE KEY UPDATE version=VALUES(version),
                             knowledge_id=VALUES(knowledge_id), api_url=VALUES(api_url)""",
                        (case_id, tool_index, version, knowledge_id, api_url))
    finally:
        conn.close()


def fetch_dashboard_rows():
    """拉 Dashboard 计算所需的轻量行。数据量级:百~千行,Python 聚合足够。
    优化:① 不传 llm_evaluation 整块,SQL 只取采纳判定要的相关性得分;
    ② steps 只取每个 case 的最新版本(覆盖度只看最新版),历史/link_ 版本不传 steps。"""
    conn = _conn()
    try:
        with conn.cursor() as cur:
            # 进度分母走「采纳」口径;mode 标方向(工序帖来自 search_process)。
            cols = (f"query_id, case_id, platform, overall_score, publish_time, "
                    f"{_REL_SQL} AS rel, {_REPRO_SQL} AS repro")
            cur.execute(f"SELECT {cols} FROM search_process WHERE {_REAL_POST}")
            posts = cur.fetchall()
            for p in posts:
                p["mode"] = "process"
            cur.execute(f"SELECT {cols} FROM search_tools")
            st = cur.fetchall()
            for p in st:
                p["mode"] = "tools"
            posts += st
            # 3-II:成本/耗时按 run(=case+version)取自 process_run;steps 仅最新版(is_latest)需要,
            # 其余 run steps=[] 省传输。返回形状与旧 mode_process 段一致(id/case_id/version/cost/dur/created_at/steps)。
            cur.execute("""SELECT id, case_id, version, cost_usd, duration_s, created_at, is_latest
                           FROM process_run ORDER BY id""")
            run_rows = cur.fetchall()
            latest_ids = [r["id"] for r in run_rows if r["is_latest"]]
            steps_by_run = {}
            if latest_ids:
                ph = ",".join(["%s"] * len(latest_ids))
                cur.execute(f"""SELECT p.run_id, s.step_json FROM process_step s
                                JOIN process_procedure p ON s.procedure_pk = p.id
                                WHERE p.run_id IN ({ph})
                                ORDER BY p.run_id, p.seq, s.seq, s.id""", latest_ids)
                for row in cur.fetchall():
                    steps_by_run.setdefault(row["run_id"], []).append(_loads(row["step_json"], {}))
            procs = [{"id": r["id"], "case_id": r["case_id"], "version": r["version"],
                      "cost_usd": r["cost_usd"], "duration_s": r["duration_s"],
                      "created_at": r["created_at"],
                      "steps": steps_by_run.get(r["id"], [])} for r in run_rows]
            cur.execute("""SELECT id, case_id, version, tool_name, substance_scope,
                                  form_scope, cost_usd, duration_s, created_at
                           FROM mode_tools""")
            tools = cur.fetchall()
    finally:
        conn.close()
    for p in posts:
        # 采纳判定:口径同帖子列表(is_adopted),作为「需解构」分母依据
        p["adopted"] = is_adopted_rel(p["overall_score"], p["rel"], p["publish_time"], p["repro"])
    for r in procs:
        r["steps"] = _loads(r["steps"], [])
        r["cost_usd"] = float(r["cost_usd"]) if r["cost_usd"] is not None else None
        r["created_at"] = str(r["created_at"]) if r["created_at"] else None
    for r in tools:
        r["substance_scope"] = _loads(r["substance_scope"], [])
        r["form_scope"] = _loads(r["form_scope"], [])
        r["cost_usd"] = float(r["cost_usd"]) if r["cost_usd"] is not None else None
        r["created_at"] = str(r["created_at"]) if r["created_at"] else None
    return posts, procs, tools


def check():
    conn = _conn()
    try:
        with conn.cursor() as cur:
            for t in ("search_process", "search_tools", "mode_process", "mode_tools",
                      "process_run", "process_procedure", "process_step", "step_classification"):
                cur.execute(f"SELECT COUNT(*) AS n FROM {t}")
                print(f"{t}: {cur.fetchone()['n']} 行")
    finally:
        conn.close()


if __name__ == "__main__":
    cmd = sys.argv[1] if len(sys.argv) > 1 else ""
    if cmd == "init":
        init_tables()
    elif cmd == "check":
        check()
    elif cmd == "clear":
        clear_tables()
    else:
        print("用法:\n  python db.py init    # 建表\n  python db.py check   # 四表行数\n  python db.py clear   # 清空四表数据")