pipeline-bidding/scripts/import_init_bidding.py

#!/usr/bin/env python3
"""投标产线 init 数据导入（码表 + 产线描述），幂等可重复执行。

独立于宿主 scripts/import_init.py（其 INIT_MODULES 漏了本模块），
由模块自己维护，部署时在宿主应用根目录执行：
  cd <APP_ROOT> && py3/bin/python pkgs/pipeline-bidding/scripts/import_init_bidding.py
"""
import os
import subprocess
import sys

SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
ROOT_DIR = os.path.abspath(os.path.join(SCRIPT_DIR, "..", "..", ".."))
sys.path.insert(0, ROOT_DIR)

from appPublic.jsonConfig import getConfig
from appPublic.aes import aes_decode_b64
import pymysql


def pymysql_connect(kw, pwd):
    return pymysql.connect(host=str(kw.host), port=int(kw.port),
                           user=str(kw.user), password=pwd,
                           database=str(kw.db), charset='utf8mb4')


def mysql_exec(kw, pwd, sql):
    r = subprocess.run(
        ["mysql", "-h", str(kw.host), "-P", str(kw.port),
         "-u", str(kw.user), "-p%s" % pwd, str(kw.db), "-e", sql],
        capture_output=True, text=True)
    if r.returncode != 0:
        raise RuntimeError(r.stderr[:300])
    return r.stdout.strip()


def main():
    cfg = getConfig(ROOT_DIR, {"workdir": ROOT_DIR})
    kw = cfg.databases["pipeline"].kwargs
    pwd = aes_decode_b64(cfg.password_key, kw.password)

    # ── 1. bid_qc_type：QC 审核对象码表（幂等：按组合键）──
    qc_items = [
        ("scoring_items", "评分项+得分规则"),
        ("qualifications", "所需资质"),
        ("doc_requirements", "投标文件要求"),
        ("chapter_outline", "章节骨架"),
        ("cost_benefit", "成本收益分析"),
        ("tech_items", "技术评估项（纯技术方案流程，对标技术需求书）"),
    ]
    mysql_exec(kw, pwd,
               "INSERT IGNORE INTO appcodes (id, name, hierarchy_flg) "
               "VALUES ('bid_qc_type', 'QC审核对象', '0')")
    for k, v in qc_items:
        mysql_exec(kw, pwd,
                   "INSERT IGNORE INTO appcodes_kv (id, parentid, k, v) "
                   "VALUES ('bid_qc_type_%s', 'bid_qc_type', '%s', '%s')" % (k, k, v))
    print("bid_qc_type: %d items ensured" % len(qc_items))

    # ── 2. bid_tender_source：旧值 mail(邮件采集) 已废弃 → 替换为 tender_doc ──
    mysql_exec(kw, pwd,
               "DELETE FROM appcodes_kv WHERE id='bid_tender_source_mail'")
    mysql_exec(kw, pwd,
               "INSERT IGNORE INTO appcodes_kv (id, parentid, k, v) "
               "VALUES ('bid_tender_source_tender_doc', 'bid_tender_source', "
               "'tender_doc', '招标文件解析')")
    print("bid_tender_source: mail removed, tender_doc ensured")

    # ── 3. 已废弃的招标审批状态码表清理（前三步归商机产线）──
    mysql_exec(kw, pwd, "DELETE FROM appcodes_kv WHERE parentid='bid_tender_status'")
    mysql_exec(kw, pwd, "DELETE FROM appcodes WHERE id='bid_tender_status'")
    print("bid_tender_status: deprecated codes cleaned")

    # ── 4. 产线描述更新 ──
    desc = ("上传招标文件→解析（评分项/得分规则/资质/投标文件要求/章节骨架）"
            "→QC契合度审核→资质准备→分章节编写→章节评审→合成标书→整书评分→交付。"
            "招标信息的采集/摘要/审批归商机产线。")
    mysql_exec(kw, pwd,
               "UPDATE pipelines SET description='%s' WHERE id='bidding_general'" % desc)
    print("pipelines.bidding_general: description updated")

    # ── 5. bid_analysis_history：QC 不通过清空前的产出快照表（2026-09-03，保留失败轮次输出）──
    mysql_exec(kw, pwd, """
        CREATE TABLE IF NOT EXISTS bid_analysis_history (
          id VARCHAR(32) NOT NULL,
          project_id VARCHAR(32) NOT NULL,
          qc_type VARCHAR(32) NOT NULL,
          round INT NOT NULL DEFAULT 0,
          task_id VARCHAR(32) NOT NULL DEFAULT '',
          fit_score DECIMAL(5,2) NOT NULL DEFAULT 0,
          improvement TEXT,
          payload LONGTEXT,
          created_at DATETIME NOT NULL,
          PRIMARY KEY (id),
          KEY idx_bah_project (project_id, qc_type, round)
        ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4""")
    print("bid_analysis_history: ensured")

    # ── 6. 存量任务状态回填：approved 但实际触发过重做的分析任务 → qc_rejected（幂等）──
    # 规则：同维度内其后还有任务 ⇒ 该轮失败触发过重做 ⇒ qc_rejected；
    # 末位任务看该维度最新一轮 QC：passed=0 ⇒ qc_rejected。只改 approved/completed。
    import json as _json
    conn = pymysql_connect(kw, pwd)
    cur = conn.cursor()
    cur.execute("SELECT DISTINCT tenant_id FROM pipeline_tasks WHERE role='agent.tender_analyst'")
    pids = [r[0] for r in cur.fetchall()]
    DIM_QC = {'scoring': ('scoring_items',), 'quals': ('qualifications',),
              'reqs_outline': ('doc_requirements', 'chapter_outline'),
              'cost_benefit': ('cost_benefit',)}
    n_fix = 0
    for pid in pids:
        cur.execute("SELECT id, state, params, created_at FROM pipeline_tasks "
                    "WHERE tenant_id=%s AND role='agent.tender_analyst' ORDER BY created_at", (pid,))
        tasks = cur.fetchall()
        groups = {}
        for tid, st, params, cat in tasks:
            try:
                dim = (_json.loads(params if isinstance(params, str) else (params or '{}')) or {}).get('analysis_dim') or ''
            except Exception:
                dim = ''
            if dim:
                groups.setdefault(dim, []).append((tid, st, cat))
        for dim, ts in groups.items():
            for i, (tid, st, cat) in enumerate(ts):
                new_state = None
                if i < len(ts) - 1 and st in ('approved', 'completed'):
                    new_state = 'qc_rejected'
                elif i == len(ts) - 1 and st in ('approved', 'completed'):
                    qcs = DIM_QC.get(dim, ())
                    if qcs:
                        fmt = ','.join(['%s'] * len(qcs))
                        cur.execute("SELECT passed FROM bid_qc_reviews WHERE project_id=%s "
                                    "AND qc_type IN (" + fmt + ") ORDER BY round DESC LIMIT 1",
                                    (pid,) + qcs)
                        rr = cur.fetchone()
                        if rr and str(rr[0]) != '1':
                            new_state = 'qc_rejected'
                if new_state:
                    cur.execute("UPDATE pipeline_tasks SET state=%s WHERE id=%s", (new_state, tid))
                    n_fix += 1
    conn.commit()
    conn.close()
    print("backfill qc_rejected: %d tasks" % n_fix)


if __name__ == "__main__":
    main()