""" 把数据库里「已采纳」的工序解构(mode_process)批量导入到知识导入接口。 与 Downloads/import/how_process_knowledge/main.py 的区别只在数据来源: 参考实现 —— 扫 data/*/_model/workflow.json,读本地 JSON 文件。 本脚本 —— 从 MySQL 取数:db.fetch_adopted_process_cases() 拿采纳 case_id, 再 db.fetch_process(case_id) 重建 {source, procedures}。 两边的数据结构一致(source 块 + procedures[].steps[]),故 payload 组装逻辑完全复用。 字段映射(同参考实现): source.id ← case_id(DB 主键,无需从 JSON 兜底目录名) source.source_type ← 固定 "post" source.title/author ← source.title / source.author 每个 procedure → 一条知识: title ← procedure.name;为空回退 "来源标题 — 工序N" content ← 整个 procedure 对象的 JSON 串 dim_attributes ← 固定 ["how工序"] dim_creations ← 固定 ["制作"] scopes ← 各步骤 substance/form 去重 custom_ext ← 各步骤 effect→作用 action→动作 via→工具 去重 采纳口径:db.is_adopted_rel(相关性<4 / 发布超两年 / 综合分<6 任一命中即不采纳)。 用法: python import_process_knowledge.py # 真实导入(采纳工序全量) python import_process_knowledge.py --dry-run # 只组装+打印,不调接口 python import_process_knowledge.py --dry-run --verbose # 打印完整 payload JSON python import_process_knowledge.py --query-id q0001 # 只传某搜索任务下的采纳 case python import_process_knowledge.py --limit 5 # 只处理前 5 个 case(调试) python import_process_knowledge.py --api-url http://... # 指定后端地址 python import_process_knowledge.py --delay 200 # 每次调用间隔 200ms """ import argparse import json import logging import sys import time import requests import db # ── 配置(对齐参考实现)──────────────────────────────────────────────────────── DEFAULT_API_URL = "http://47.236.83.130:8001" INGEST_ENDPOINT = "/api/v1/knowledge/ingest" DIM_ATTRIBUTES = ["how工序"] DIM_CREATIONS = ["制作"] EXT_KEY_EFFECT = "作用" EXT_KEY_ACTION = "动作" EXT_KEY_TOOL = "工具" # ── 日志 ────────────────────────────────────────────────────────────────────── logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s", datefmt="%H:%M:%S", ) logger = logging.getLogger(__name__) # ── 数据来源:从 DB 取采纳工序(替代参考实现的本地文件扫描)────────────────────── def iter_cases_from_db(query_id=None, limit=None): """产出 (case_id, source_data, procedures, version)。 先取采纳 case_id 列表,再逐个 fetch_process 重建解构详情(取最新版本)。 fetch_process 返回 None(无解构行)或 procedures 为空的 case 自动跳过。 version 用于上传去重:同 case 同工序、版本未变即视为已传过,跳过。 """ case_ids = db.fetch_adopted_process_cases(query_id) if limit: case_ids = case_ids[:limit] for case_id in case_ids: payload = db.fetch_process(case_id) # 最新版本 if not payload: continue yield (case_id, (payload.get("source") or {}), (payload.get("procedures") or []), payload.get("version")) # ── 作用域提取(原样复用参考实现)────────────────────────────────────────────── def _split_values(raw): """按顿号分割,括号内的顿号不作为分隔符,结果去重保序。 "高保真线框图、UI设计稿" → ["高保真线框图", "UI设计稿"] "修改后的照片(发型、服装)、二次元服装" → ["修改后的照片(发型、服装)", "二次元服装"] """ parts, current, depth = [], [], 0 for ch in raw: if ch in ("(", "（"): depth += 1 current.append(ch) elif ch in (")", "）"): depth -= 1 current.append(ch) elif ch == "、" and depth == 0: part = "".join(current).strip() if part: parts.append(part) current = [] else: current.append(ch) part = "".join(current).strip() if part: parts.append(part) seen, result = set(), [] for p in parts: if p not in seen: seen.add(p) result.append(p) return result def build_scopes(procedure): """从所有步骤收集 substance / form,各自去重后返回 scope 列表。""" seen_sub, seen_form, scopes = set(), set(), [] for step in (procedure.get("steps") or []): for sub in _split_values((step.get("substance") or "").strip()): if sub not in seen_sub: scopes.append({"scope_type": "substance", "value": sub}) seen_sub.add(sub) for form in _split_values((step.get("form") or "").strip()): if form not in seen_form: scopes.append({"scope_type": "form", "value": form}) seen_form.add(form) return scopes def build_custom_ext(procedure): """从所有步骤提取 effect/action/via,值按顿号分割,同 key 相同值去重。""" ext = [] seen = {} def add(key, raw): if not raw: return for value in _split_values(raw): seen.setdefault(key, set()) if value not in seen[key]: seen[key].add(value) ext.append({"key": key, "type": "str", "value": value}) for step in (procedure.get("steps") or []): add(EXT_KEY_EFFECT, (step.get("effect") or "").strip()) add(EXT_KEY_ACTION, (step.get("action") or "").strip()) via = (step.get("via") or "").strip() if via and via != "human": add(EXT_KEY_TOOL, via) return ext # ── 单条 payload 组装(复用参考实现;source_id 直接用 case_id)───────────────────── def build_payload(source_id, source_data, procedure, proc_index): source_title = (source_data.get("title") or "").strip() source_author = (source_data.get("author") or "").strip() or None proc_name = (procedure.get("name") or "").strip() if proc_name: knowledge_title = proc_name elif source_title: knowledge_title = f"{source_title} — 工序{proc_index}" else: knowledge_title = f"工序{proc_index}" content = json.dumps(procedure, ensure_ascii=False) source_metadata = { "platform": source_data.get("platform") or "", "date": source_data.get("date") or "", "url": source_data.get("url") or None, "excerpt": (source_data.get("excerpt") or "")[:500], "procedure_id": procedure.get("id") or "", "procedure_name": proc_name, } payload = { "source": { "id": source_id, "source_type": "post", "title": source_title or None, "author": source_author, "source_metadata": source_metadata, }, "title": knowledge_title[:512], "content": content, "dim_attributes": DIM_ATTRIBUTES, "dim_creations": DIM_CREATIONS, } scopes = build_scopes(procedure) if scopes: payload["scopes"] = scopes custom_ext = build_custom_ext(procedure) if custom_ext: payload["custom_ext"] = custom_ext return payload # ── 单条写入(原样复用参考实现)──────────────────────────────────────────────── def ingest_one(api_url, payload, dry_run): """调用导入接口写入一条知识,返回 (success, info_message, knowledge_id)。""" if dry_run: return True, "(dry-run, skipped)", None url = api_url.rstrip("/") + INGEST_ENDPOINT try: resp = requests.post(url, json=payload, timeout=30) if resp.status_code == 201: kid = resp.json().get("knowledge_id", "?") return True, f"knowledge_id={kid}", kid try: detail = resp.json().get("detail", resp.text[:300]) except Exception: detail = resp.text[:300] return False, f"HTTP {resp.status_code}: {detail}", None except requests.Timeout: return False, "超时(30s)", None except requests.RequestException as exc: return False, str(exc), None # ── 主循环 ──────────────────────────────────────────────────────────────────── def run(api_url, dry_run, verbose, delay_ms, query_id, limit, force): cases = list(iter_cases_from_db(query_id, limit)) if not cases: scope = f"(query_id={query_id})" if query_id else "" logger.error("DB 中未发现任何「已采纳且有工序解构」的 case %s", scope) sys.exit(1) mode_tag = " [DRY-RUN]" if dry_run else "" force_tag = " [FORCE]" if force else "" logger.info("发现 %d 个采纳 case。目标接口:%s%s%s", len(cases), api_url, mode_tag, force_tag) ok_count = fail_count = skip_count = dup_count = 0 for case_id, source_data, procedures, version in cases: logger.info("── %s", case_id) if not procedures: logger.warning(" 无 procedures,跳过") skip_count += 1 continue # 去重台账:该 case 各工序已导入的版本。force 时清空(强制重导)。 ingested = {} if force else db.fetch_ingested_map(case_id) logger.info(" source_id=%-45s procedures=%d version=%s", case_id, len(procedures), version) for idx, procedure in enumerate(procedures, 1): # 已导入且版本未变 → 跳过,杜绝重复上传(版本变了说明重解构过,应重导)。 if not force and ingested.get(idx) == version: logger.info(" ♻️ [%d/%d] 已导入(版本 %s),跳过", idx, len(procedures), version) dup_count += 1 continue payload = build_payload(case_id, source_data, procedure, idx) title = payload["title"] n_scopes = len(payload.get("scopes", [])) n_ext = len(payload.get("custom_ext", [])) n_steps = len(procedure.get("steps") or []) if dry_run and verbose: print(f"\n{'=' * 60}") print(f"[{case_id}] 工序 {idx}/{len(procedures)}") print(json.dumps(payload, ensure_ascii=False, indent=2)) ok, msg, kid = ingest_one(api_url, payload, dry_run) status_icon = "✓" if ok else "✗" level = logging.INFO if ok else logging.WARNING logger.log( level, " %s [%d/%d] title=%r steps=%d scopes=%d ext=%d %s", status_icon, idx, len(procedures), title[:40], n_steps, n_scopes, n_ext, msg, ) if ok: ok_count += 1 # 仅真实导入成功才记台账(dry-run 不写,免污染去重状态) if not dry_run: try: db.mark_ingested(case_id, idx, version, kid, api_url) except Exception as exc: logger.warning(" ⚠️ 台账写入失败(不影响本次导入):%s", exc) else: fail_count += 1 if delay_ms > 0 and not dry_run: time.sleep(delay_ms / 1000) logger.info( "完成。成功=%d 失败=%d 无工序跳过=%d 已传过跳过=%d 合计导入=%d", ok_count, fail_count, skip_count, dup_count, ok_count, ) if fail_count: sys.exit(1) # ── CLI ─────────────────────────────────────────────────────────────────────── def main(): parser = argparse.ArgumentParser( description="把 DB 中已采纳的工序解构(mode_process)批量导入知识接口", formatter_class=argparse.RawDescriptionHelpFormatter, ) parser.add_argument("--api-url", default=DEFAULT_API_URL, metavar="URL", help=f"后端 API 根地址(默认:{DEFAULT_API_URL})") parser.add_argument("--dry-run", action="store_true", help="仅从 DB 取数并组装 payload,不实际调用接口") parser.add_argument("--verbose", "-v", action="store_true", help="dry-run 时打印完整 payload JSON") parser.add_argument("--delay", type=int, default=100, metavar="MS", help="两次 API 调用之间的间隔毫秒数(默认:100)") parser.add_argument("--query-id", default=None, metavar="QID", help="只导入该搜索任务(query_id)下的采纳 case") parser.add_argument("--limit", type=int, default=None, metavar="N", help="只处理前 N 个 case(调试用)") parser.add_argument("--force", action="store_true", help="忽略去重台账,强制重导(换 prompt/模型、需覆盖时用)") args = parser.parse_args() run( api_url=args.api_url, dry_run=args.dry_run, verbose=args.verbose, delay_ms=args.delay, query_id=args.query_id, limit=args.limit, force=args.force, ) if __name__ == "__main__": main()