vent_agent_2.0/db/main_fan_store.py

"""主通风机文档识别缓存和确定性研判原件，存入本应用数据库。"""
import json

from db import db_conn

CACHE_VERSION = "main-fan-facts-v4"


def save_document(sha256, facts, method):
    key = f"{CACHE_VERSION}:{sha256}"
    content = json.dumps(facts, ensure_ascii=False, allow_nan=False)
    with db_conn.transaction():
        db_conn.execute("DELETE FROM main_fan_documents WHERE id = ?", (key,))
        db_conn.execute("INSERT INTO main_fan_documents (id, sha256, facts_json, method, created_at) VALUES (?, ?, ?, ?, ?)",
                        (key, sha256, content, method, db_conn.fmt_dt()))


def document(sha256, *, allow_manual=True):
    row = db_conn.query_one("SELECT * FROM main_fan_documents WHERE id = ?", (f"{CACHE_VERSION}:{sha256}",))
    if row and row["method"] == "manual_transcription" and not allow_manual:
        return None
    return json.loads(row["facts_json"]) if row else None


def save_analysis(user_name, session_id, result):
    value = dict(result)
    value["analysis_id"] = db_conn.new_id()
    db_conn.execute("INSERT INTO main_fan_analyses (id, user_name, session_id, result_json, created_at) VALUES (?, ?, ?, ?, ?)",
                    (value["analysis_id"], user_name, session_id,
                     json.dumps(value, ensure_ascii=False, allow_nan=False), db_conn.fmt_dt()))
    return value


def canonical_result(value, user_name, session_id):
    """正式筛查只接受本次用户/会话实际运行工具生成的原件。"""
    if not value.get("analysis_id"):
        if value.get("status") == "completed":
            raise ValueError("主通风机已完成结果缺少工具生成的 analysis_id")
        return value
    row = db_conn.query_one("SELECT * FROM main_fan_analyses WHERE id = ? AND user_name = ? AND session_id = ?",
                            (value["analysis_id"], user_name, session_id))
    if not row:
        raise ValueError("主通风机研判原件不存在或不属于当前会话")
    # 用数据库原件替换模型转述，禁止模型改写数值/状态后对外推送。
    return json.loads(row["result_json"])


def create_workflow(user_name, session_id, context, stages):
    identifier, now = db_conn.new_id(), db_conn.fmt_dt()
    db_conn.execute("INSERT INTO main_fan_workflows (id,user_name,session_id,context_json,stages_json,status,created_at,updated_at) VALUES (?,?,?,?,?,?,?,?)",
        (identifier,user_name,session_id,json.dumps(context,ensure_ascii=False,allow_nan=False),
         json.dumps(stages,ensure_ascii=False,allow_nan=False),'running',now,now))
    return identifier


def update_workflow(identifier, *, context=None, stages=None, result=None, status="running"):
    values = {"status":status,"updated_at":db_conn.fmt_dt()}
    for name, value in (("context_json",context),("stages_json",stages),("result_json",result)):
        if value is not None:
            values[name]=json.dumps(value,ensure_ascii=False,allow_nan=False)
    sql="UPDATE main_fan_workflows SET "+", ".join(key+" = ?" for key in values)+" WHERE id = ?"
    db_conn.execute(sql,(*values.values(),identifier))


def workflow(identifier, user_name, session_id):
    row=db_conn.query_one("SELECT * FROM main_fan_workflows WHERE id = ? AND user_name = ? AND session_id = ?",
                         (identifier,user_name,session_id))
    if not row:
        raise ValueError("研判流程不存在或不属于当前会话")
    return {**row,"context":json.loads(row["context_json"]),"stages":json.loads(row["stages_json"]),
            "result":json.loads(row["result_json"]) if row["result_json"] else None}