Agent/examples/process_pipeline/script/search_eval/fixed_query_eval/db.py

# -*- coding: utf-8 -*-
"""fixed_query_eval · MySQL 持久化（双写：本地 json + 数据库）
================================================================================

读 .env 的 MYSQL_* 连接 MySQL（仓库 MySQL 约定是 pymysql）。两张表：
  fqe_posts —— 每行一个 (query, 帖子)：搜索 + llm 评分结果
  fqe_tools —— 每行一个解构出的工具：工具解构结果

设计原则：
- **失败不阻断**：所有写入用 try/except 包，DB 挂了不影响本地 json 写入（文件是主存储）。
- **幂等**：posts 用 (q, case_id) 唯一键 upsert；tools 先按 (q, case_id) 删旧再插新（重新解构会覆盖）。
- 建表：init_tables() 用 CREATE TABLE IF NOT EXISTS，跑 `python db.py init` 即建表。
"""
import os
import json
import sys
from pathlib import Path

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

from dotenv import load_dotenv
load_dotenv()

try:
    import pymysql
    from pymysql.cursors import DictCursor
except ImportError:
    pymysql = None


def _enabled() -> bool:
    return pymysql is not None and bool(os.getenv("MYSQL_HOST"))


def _conn():
    return pymysql.connect(
        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,
    )


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

DDL_POSTS = """
CREATE TABLE IF NOT EXISTS fqe_posts (
  id            BIGINT AUTO_INCREMENT PRIMARY KEY,
  q             VARCHAR(16)   NOT NULL COMMENT 'query 目录名 q0000',
  query_text    VARCHAR(255)  NULL     COMMENT '基准 query（如 GPT image2 评测）',
  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,
  body          MEDIUMTEXT    NULL,
  images        JSON          NULL     COMMENT '图片 URL 数组',
  like_count    INT           NULL,
  publish_time  VARCHAR(64)   NULL,
  found_by      JSON          NULL     COMMENT '命中的措辞数组',
  knowledge_type JSON         NULL     COMMENT 'llm 判定的知识类型',
  overall_score FLOAT         NULL     COMMENT '综合分（相关性两子项均值，便于排序）',
  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_q_case (q, case_id),
  KEY idx_platform (platform),
  KEY idx_query_text (query_text)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='固定query搜索+评分结果';
"""

DDL_TOOLS = """
CREATE TABLE IF NOT EXISTS fqe_tools (
  id            BIGINT AUTO_INCREMENT PRIMARY KEY,
  q             VARCHAR(16)   NOT NULL,
  case_id       VARCHAR(128)  NOT NULL,
  platform      VARCHAR(32)   NULL,
  post_title    VARCHAR(512)  NULL,
  tool_name     VARCHAR(255)  NULL     COMMENT '工具名称',
  substance_scope VARCHAR(255) NULL    COMMENT '实质作用域',
  form_scope    VARCHAR(255)  NULL     COMMENT '形式作用域',
  creation_layer VARCHAR(32)  NULL     COMMENT '创作层级：制作层/创作层',
  source_link   VARCHAR(1024) NULL     COMMENT '来源链接',
  input_desc    TEXT          NULL     COMMENT '输入',
  output_desc   TEXT          NULL     COMMENT '输出',
  usage_json    JSON          NULL     COMMENT '用法数组',
  cases_json    JSON          NULL     COMMENT '案例数组',
  defects_json  JSON          NULL     COMMENT '缺点数组',
  updated_time  VARCHAR(64)   NULL     COMMENT '工具最新更新时间',
  model         VARCHAR(64)   NULL     COMMENT '解构模型',
  version       VARCHAR(16)   NULL     COMMENT '解构版本号 v_MMDDHHMM（每次解构生成；保留历史，多版本共存）',
  created_at    TIMESTAMP     DEFAULT CURRENT_TIMESTAMP,
  KEY idx_q_case (q, case_id),
  KEY idx_q_case_ver (q, case_id, version),
  KEY idx_tool_name (tool_name)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='工具解构结果（每行一个工具）';
"""


def init_tables():
    """建表（幂等）。"""
    if not _enabled():
        print("⚠️ MySQL 未启用（缺 pymysql 或 MYSQL_HOST），跳过建表")
        return False
    conn = _conn()
    try:
        with conn.cursor() as cur:
            cur.execute(DDL_POSTS)
            cur.execute(DDL_TOOLS)
        print("✅ 建表完成：fqe_posts, fqe_tools")
        return True
    finally:
        conn.close()


# ── 写入 ─────────────────────────────────────────────────────────────────────

def _overall_from_eval(e):
    """从 mod schema 评分粗算综合分（相关性两子项均值）。算不出返回 None。"""
    try:
        rel = (e or {}).get("相关性") or {}
        vals = []
        for k in ("和内容制作知识相关", "和 query 相关"):
            v = (rel.get(k) or {}).get("得分")
            if v is not None:
                vals.append(float(v))
        return round(sum(vals) / len(vals), 2) if vals else None
    except Exception:
        return None


def upsert_posts(q, query_text, results):
    """把一组搜索结果写入 fqe_posts（按 (q, case_id) upsert）。返回写入条数；失败返回 0。"""
    if not _enabled() or not results:
        return 0
    rows = []
    for r in results:
        post = r.get("post") or {}
        e = r.get("llm_evaluation") or {}
        rows.append((
            q, 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("body_text") or post.get("desc") or "",
            json.dumps(post.get("images") or [], ensure_ascii=False),
            post.get("like_count"),
            str(post.get("publish_time") or post.get("publish_timestamp") or "")[:64],
            json.dumps(r.get("found_by_queries") or [], ensure_ascii=False),
            json.dumps(e.get("知识类型") or [], ensure_ascii=False),
            _overall_from_eval(e),
            json.dumps(e, ensure_ascii=False),
        ))
    sql = """
    INSERT INTO fqe_posts
      (q, query_text, case_id, platform, channel_content_id, title, url, body,
       images, like_count, publish_time, found_by, knowledge_type, overall_score, llm_evaluation)
    VALUES (%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),
      body=VALUES(body), images=VALUES(images), like_count=VALUES(like_count),
      publish_time=VALUES(publish_time), found_by=VALUES(found_by),
      knowledge_type=VALUES(knowledge_type), overall_score=VALUES(overall_score),
      llm_evaluation=VALUES(llm_evaluation);
    """
    try:
        conn = _conn()
        try:
            with conn.cursor() as cur:
                cur.executemany(sql, rows)
            return len(rows)
        finally:
            conn.close()
    except Exception as ex:
        print(f"⚠️ fqe_posts 写库失败（不影响本地 json）：{ex}")
        return 0


def upsert_tools(q, case_id, model, tools, version, platform=None, post_title=None):
    """写入一帖某个版本的工具解构结果。**保留历史**：不删其它版本，只删本 (q,case_id,version)
    的旧行再插（保证同一版本重跑幂等）。失败返回 0。"""
    if not _enabled():
        return 0
    try:
        conn = _conn()
        try:
            with conn.cursor() as cur:
                cur.execute("DELETE FROM fqe_tools WHERE q=%s AND case_id=%s AND version=%s",
                            (q, case_id, version))
                if tools:
                    rows = [(
                        q, case_id, platform, (post_title or "")[:500],
                        t.get("工具名称"), t.get("实质作用域"), t.get("形式作用域"),
                        t.get("创作层级"), t.get("来源链接"), t.get("输入"), t.get("输出"),
                        json.dumps(t.get("用法"), ensure_ascii=False) if t.get("用法") is not None else None,
                        json.dumps(t.get("案例"), ensure_ascii=False) if t.get("案例") is not None else None,
                        json.dumps(t.get("缺点"), ensure_ascii=False) if t.get("缺点") is not None else None,
                        t.get("最新更新时间"), model, version,
                    ) for t in tools]
                    cur.executemany("""
                    INSERT INTO fqe_tools
                      (q, case_id, platform, post_title, tool_name, substance_scope, form_scope,
                       creation_layer, source_link, input_desc, output_desc,
                       usage_json, cases_json, defects_json, updated_time, model, version)
                    VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)
                    """, rows)
            return len(tools)
        finally:
            conn.close()
    except Exception as ex:
        print(f"⚠️ fqe_tools 写库失败（不影响本地 json）：{ex}")
        return 0


# ── 读取（本地 runs_full 被清空时，server 回退读库重建视图）────────────────────────

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 fetch_posts_grouped():
    """从 fqe_posts 重建 {q: {query_text, results:[r...]}}（r 的结构对齐 form_A.json 的 result，
    便于喂给 server 的 adapt()）。库不可用/空/异常返回 {}。"""
    if not _enabled():
        return {}
    try:
        conn = _conn()
        try:
            with conn.cursor() as cur:
                cur.execute("SELECT * FROM fqe_posts ORDER BY q, overall_score DESC")
                rows = cur.fetchall()
        finally:
            conn.close()
    except Exception as ex:
        print(f"⚠️ 读 fqe_posts 失败：{ex}")
        return {}
    out = {}
    for row in rows:
        q = row["q"]
        r = {
            "case_id": row["case_id"], "platform": row["platform"],
            "channel_content_id": row["channel_content_id"], "source_url": row["url"],
            "found_by_queries": _loads(row["found_by"], []),
            "llm_evaluation": _loads(row["llm_evaluation"], {}),
            "post": {
                "title": row["title"], "body_text": row["body"],
                "images": _loads(row["images"], []), "like_count": row["like_count"],
                "publish_timestamp": row["publish_time"],
            },
        }
        out.setdefault(q, {"query_text": row["query_text"], "results": []})["results"].append(r)
    return out


def fetch_tool_versions(q, case_id):
    """列出某帖的全部解构版本（降序，最新在前）。无则返回 []。"""
    if not _enabled():
        return []
    try:
        conn = _conn()
        try:
            with conn.cursor() as cur:
                cur.execute("SELECT DISTINCT version FROM fqe_tools WHERE q=%s AND case_id=%s "
                            "ORDER BY version DESC", (q, case_id))
                return [r["version"] for r in cur.fetchall() if r["version"]]
        finally:
            conn.close()
    except Exception as ex:
        print(f"⚠️ 读 fqe_tools 版本失败：{ex}")
        return []


def fetch_tools(q, case_id, version=None):
    """从 fqe_tools 重建 {case_id, version, model, tool_count, tools:[...]}。
    version=None 取最新版本（v_MMDDHHMM 字符串降序）；指定则取该版本。无记录返回 None。"""
    if not _enabled():
        return None
    try:
        conn = _conn()
        try:
            with conn.cursor() as cur:
                if version is None:
                    cur.execute("SELECT version FROM fqe_tools WHERE q=%s AND case_id=%s "
                                "ORDER BY version DESC, id DESC LIMIT 1", (q, case_id))
                    row = cur.fetchone()
                    if not row:
                        return None
                    version = row["version"]
                cur.execute("SELECT * FROM fqe_tools WHERE q=%s AND case_id=%s AND version=%s "
                            "ORDER BY id", (q, case_id, version))
                rows = cur.fetchall()
        finally:
            conn.close()
    except Exception as ex:
        print(f"⚠️ 读 fqe_tools 失败：{ex}")
        return None
    if not rows:
        return None
    tools = [{
        "工具名称": r["tool_name"], "实质作用域": r["substance_scope"],
        "形式作用域": 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"],
            "tool_count": len(tools), "tools": tools}


def has_tools(q, case_id):
    """库里是否有该帖的工具解构记录。"""
    if not _enabled():
        return False
    try:
        conn = _conn()
        try:
            with conn.cursor() as cur:
                cur.execute("SELECT 1 FROM fqe_tools WHERE q=%s AND case_id=%s LIMIT 1", (q, case_id))
                return cur.fetchone() is not None
        finally:
            conn.close()
    except Exception:
        return False


def rebuild_local(runs_dir=None):
    """从数据库重建本地 runs_full/{q}/form_A.json 和 tools/{case_id}.json。
    用于本地文件丢失后恢复（界面回退只让界面有数据；提取脚本仍读本地文件，故提供本命令）。"""
    if not _enabled():
        print("⚠️ MySQL 未启用，无法重建"); return
    runs_dir = Path(runs_dir) if runs_dir else (Path(__file__).resolve().parent / "runs_full")
    grouped = fetch_posts_grouped()
    np = nt = 0
    for q, g in grouped.items():
        d = {"form": "A", "query": g["query_text"], "original_q": g["query_text"] or "",
             "platforms": [], "total": len(g["results"]), "failed": 0, "results": g["results"]}
        p = runs_dir / q / "form_A.json"
        p.parent.mkdir(parents=True, exist_ok=True)
        p.write_text(json.dumps(d, ensure_ascii=False, indent=2), encoding="utf-8")
        np += 1
    # tools
    try:
        conn = _conn()
        try:
            with conn.cursor() as cur:
                cur.execute("SELECT DISTINCT q, case_id FROM fqe_tools")
                pairs = cur.fetchall()
        finally:
            conn.close()
        for row in pairs:
            data = fetch_tools(row["q"], row["case_id"])
            if data:
                tp = runs_dir / row["q"] / "tools" / f"{row['case_id']}.json"
                tp.parent.mkdir(parents=True, exist_ok=True)
                tp.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")
                nt += 1
    except Exception as ex:
        print(f"⚠️ 重建 tools 失败：{ex}")
    print(f"✅ 从库重建本地：{np} 个 query 的 form_A.json，{nt} 帖工具结果 → {runs_dir}")


if __name__ == "__main__":
    cmd = sys.argv[1] if len(sys.argv) > 1 else ""
    if cmd == "init":
        init_tables()
    elif cmd == "rebuild":
        rebuild_local()
    else:
        print("用法：\n  python db.py init      # 建表\n  python db.py rebuild   # 从库重建本地 runs_full（本地文件丢失后恢复）")