| 1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447 |
- # -*- coding: utf-8 -*-
- """搜索评估案例查看 server。
- 沿用 图文排版搜索评估.html 的版式(卡片 + dialog 详情 + rubric 评分条),
- 数据实时扫描 runs_full/*/form_*.json —— runs_full 下每新增一个 q 文件夹,刷新即出现。
- 分页:query → 三种形式(A/B/C) → 三个渠道 三行从上到下。
- 用法:python server.py [port] 默认 8770,浏览器开 http://0.0.0.0:8770
- """
- import json, re, glob, sys, pathlib, subprocess, threading
- from datetime import datetime
- from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
- from urllib.parse import urlparse, parse_qs
- try: # Windows 控制台默认 cp1252,中文 print 会崩,统一切 utf-8
- sys.stdout.reconfigure(encoding="utf-8")
- except Exception:
- pass
- HERE = pathlib.Path(__file__).parent
- SE = HERE.parent # search_eval:引擎脚本/procedure-dsl/evaluation 都在父目录
- sys.path.insert(0, str(HERE))
- sys.path.insert(0, str(SE)) # 复用兄弟模块 batch_extract_procedures / build_workflows
- import argparse
- _parser = argparse.ArgumentParser(description="工序解构 server (mode_procedure:写死 new 模式 + procedure 直写版)")
- _parser.add_argument("port", type=int, nargs="?", default=8771, help="运行端口(默认 8771)")
- _parser.add_argument("--mode", choices=["legacy", "new"], default="new", help="(mode_procedure 固定 new)")
- _args, _ = _parser.parse_known_args()
- PORT = _args.port
- MODE = "new" # mode_procedure 写死 new 模式
- RUNS_DIR_NAME = "runs_new"
- RUNS_DIR = HERE / RUNS_DIR_NAME # 产出目录在本文件夹 mode_procedure/runs_new
- PLAT = {"xhs": "小红书", "gzh": "公众号", "zhihu": "知乎", "x": "X", "bili": "B站", "douyin": "抖音",
- "sph": "视频号", "youtube": "YouTube", "github": "GitHub", "toutiao": "头条", "weibo": "微博"}
- KT = {"procedure": "工序", "step": "步骤", "tool": "工具"}
- # 从 taxonomy 取动作叶子/类型名,用于把 original_q 解析回原始维度(动作×类型 正交)
- # 路径优先级:search_eval/evaluation/(主源,IDE 编辑那份就是 runtime 实际读的)
- # → test_script/evaluation/(历史副本兜底)→ script/evaluation/(更老兜底)
- # 谁也找不到时整目录扫空,server 仍能起。
- EVALDIR = HERE / "evaluation"
- if not EVALDIR.exists():
- EVALDIR = HERE.parent.parent / "test_script" / "evaluation"
- if not EVALDIR.exists():
- EVALDIR = HERE.parent / "evaluation"
- try:
- _jm = json.load(open(EVALDIR / "judged_matrix.json", encoding="utf-8"))
- ACT_L1 = {a["name"]: a["l1"] for a in _jm["actions"]}
- ACTION_SET = set(ACT_L1)
- TYPE_SET = {t["name"] for t in _jm["types"]}
- ACTIONS_TAX = [{"name": a["name"], "l1": a["l1"], "l2": a.get("l2", "")} for a in _jm["actions"]]
- TYPES_TAX = [{"name": t["name"], "l1": t["l1"]} for t in _jm["types"]]
- # taxonomy 顺序沿用 judged_matrix(严格版);矩阵分值改用 type_action_scores(宽松版) —
- # 两份是同一组 27×50 cell 的独立 gemini judging,前者只 53 格到 tier3,后者 156 格到 score3
- _tas = json.load(open(EVALDIR / "type_action_scores.json", encoding="utf-8"))["scores"]
- _MATRIX = []
- for a in _jm["actions"]:
- row = []
- for t in _jm["types"]:
- rec = _tas.get(t["name"], {}).get(a["name"])
- row.append({"tier": rec["score"], "r": rec.get("reason", "")} if rec else {})
- _MATRIX.append(row)
- except Exception:
- ACT_L1, ACTION_SET, TYPE_SET, ACTIONS_TAX, TYPES_TAX, _MATRIX = {}, set(), set(), [], [], []
- ACTIVE_TASKS = {}
- ACTIVE_REEVALS = {}
- ACTIVE_SEARCH_EVALS = {}
- ACTIVE_BATCH_TASKS = {}
- def _gen_version() -> str:
- """生成版本号 v_月日时分。"""
- return "v_" + datetime.now().strftime("%m%d%H%M")
- def _read_meta_version(out_dir: pathlib.Path):
- """读 out_dir/_meta.json 里的版本号(用于把旧产物归档到 history/{旧版本}/)。"""
- try:
- meta = json.loads((out_dir / "_meta.json").read_text(encoding="utf-8"))
- return meta.get("version")
- except Exception:
- return None
- def _stamp_version(out_dir: pathlib.Path, version: str):
- """把版本号写进 out_dir/_meta.json(_write_meta 之后调用)。"""
- meta_path = out_dir / "_meta.json"
- try:
- meta = json.loads(meta_path.read_text(encoding="utf-8")) if meta_path.is_file() else {}
- except Exception:
- meta = {}
- meta["version"] = version
- try:
- meta_path.write_text(json.dumps(meta, ensure_ascii=False, indent=2), encoding="utf-8")
- except Exception:
- pass
- def _store_dsl(table: str, q: str, case_id: str, version: str, model: str, out_dir: pathlib.Path):
- """把刚生成的 workflow.json 写入数据库(agent_dsl / mode_dsl)。失败不阻断。"""
- try:
- import db
- wf = json.loads((out_dir / "workflow.json").read_text(encoding="utf-8"))
- src = None
- sp = out_dir / "_source.json"
- if sp.is_file():
- try:
- src = json.loads(sp.read_text(encoding="utf-8"))
- except Exception:
- src = None
- platform = case_id.split("_")[0] if "_" in case_id else None
- db.upsert_dsl(table, q, case_id, version, model, wf, source=src,
- platform=platform,
- post_title=(src or {}).get("title"),
- source_link=(src or {}).get("link"))
- except Exception as ex:
- print(f"[db] _store_dsl({table}) 失败(不阻断):{ex}")
- def backup_procedure_history(out_dir: pathlib.Path):
- if not out_dir.is_dir():
- return
- files_to_backup = [f for f in out_dir.iterdir() if f.is_file()]
- if not files_to_backup:
- return
- history_dir = out_dir / "history"
- # 按旧产物的版本号归档(v_MMDDHHMM);无版本号(历史遗留)回退时间戳
- label = _read_meta_version(out_dir) or datetime.now().strftime("%Y%m%d_%H%M%S")
- version_dir = history_dir / label
- version_dir.mkdir(parents=True, exist_ok=True)
- import shutil
- for f in files_to_backup:
- try:
- shutil.move(str(f), str(version_dir / f.name))
- except Exception as e:
- print(f"[backup] failed to move procedure file {f.name}: {e}")
- def backup_reeval_history(q_dir: pathlib.Path):
- if not q_dir.is_dir():
- return
- files_to_backup = list(q_dir.glob("form_*.json"))
- if not files_to_backup:
- return
- history_dir = q_dir / "history"
- timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
- version_dir = history_dir / timestamp
- version_dir.mkdir(parents=True, exist_ok=True)
- import shutil
- for f in files_to_backup:
- try:
- shutil.copy(str(f), str(version_dir / f.name))
- except Exception as e:
- print(f"[backup] failed to copy reeval file {f.name}: {e}")
- from batch_extract_procedures import _short_case, _source_to_dsl_input, _write_meta, _composite_score
- def _backend_note(model, engine="cyber_runner"):
- """根据引擎/模型判断计费后端,明确扣 OpenRouter 余额还是 Claude 订阅。"""
- if engine and engine != "cyber_runner":
- return "Claude SDK / OAuth(扣 Claude 订阅额度)"
- if "qwen" in (model or "").lower():
- return "Qwen 原生(阿里云,不走 OpenRouter)"
- return "OpenRouter(扣 OpenRouter 余额,非 Claude 订阅)"
- def _log_header(log_fh, title, model, engine="cyber_runner"):
- """在子进程日志开头写一行:本次用什么模型、走什么后端。"""
- line = f"[server] {title} · 模型 {model} · 后端 {_backend_note(model, engine)}\n"
- log_fh.write(line); log_fh.write("-" * 60 + "\n"); log_fh.flush()
- def run_model_extraction_task(q, folder_name, case_id, src_path, out_dir, model, version):
- """模型提取:跑 procedure_model_extract.py(单次大模型直出 workflow.json + 渲染)。
- 产物写 procedures/{folder}/_model/ 子目录,task_key 带 /_model 后缀,与 Agent 工序分开。"""
- task_key = f"{q}/{folder_name}/_model"
- log_path = out_dir / "_extract.log"
- try:
- out_dir.mkdir(parents=True, exist_ok=True)
- cmd = [sys.executable, "-u", str(HERE / "mode-dsl" / "procedure_model_extract.py"),
- str(src_path), "--out-dir", str(out_dir), "--model", model, "--case-id", folder_name]
- flags = subprocess.CREATE_NEW_PROCESS_GROUP if sys.platform == "win32" else 0
- with open(log_path, "w", encoding="utf-8", buffering=1) as log_fh:
- _log_header(log_fh, "🤖 工序提取(模型直出,无 agent/validate 循环)", model, "cyber_runner")
- proc = subprocess.Popen(cmd, stdout=log_fh, stderr=subprocess.STDOUT,
- cwd=str(HERE), creationflags=flags)
- ACTIVE_TASKS[task_key]["pid"] = proc.pid
- proc.wait()
- if proc.returncode == 0:
- # 模型工序独立展示在「大模型工序」子 tab(按 _model/*.html 检测),不进 gallery 合并
- _stamp_version(out_dir, version)
- _store_dsl("mode_dsl", q, case_id, version, model, out_dir) # 写库
- ACTIVE_TASKS[task_key]["status"] = "success"
- else:
- ACTIVE_TASKS[task_key]["status"] = "failed"
- ACTIVE_TASKS[task_key]["error"] = f"model extract exit code {proc.returncode}"
- except Exception as e:
- ACTIVE_TASKS[task_key]["status"] = "failed"
- ACTIVE_TASKS[task_key]["error"] = str(e)
- try:
- with open(log_path, "a", encoding="utf-8") as f_err:
- f_err.write(f"\n[server error] model extract failed: {e}\n")
- except Exception:
- pass
- def run_extraction_task(q, folder_name, case_id, src_path, out_dir, engine, model, version):
- task_key = f"{q}/{folder_name}"
- log_path = out_dir / "_extract.log"
- try:
- out_dir.mkdir(parents=True, exist_ok=True)
- if engine == "cyber_runner":
- script_path = HERE / "procedure-dsl" / "run_cyber.py"
- else:
- script_path = HERE / "procedure-dsl" / "run_procedure_dsl.py"
-
- cmd = [
- sys.executable, "-u", str(script_path),
- str(src_path),
- "--out-dir", str(out_dir),
- "--model", model,
- "--max-turns", "300"
- ]
- if engine != "cyber_runner":
- cmd.extend(["--max-retries", "3"])
- else:
- cmd.extend(["--skill", "procedure"]) # mode_procedure:run_cyber 用直写版 skill
- flags = subprocess.CREATE_NEW_PROCESS_GROUP if sys.platform == "win32" else 0
- with open(log_path, "w", encoding="utf-8", buffering=1) as log_fh:
- _log_header(log_fh, "🔧 工序提取(单帖)", model, engine)
- proc = subprocess.Popen(cmd, stdout=log_fh, stderr=subprocess.STDOUT,
- cwd=str(HERE / "procedure-dsl"), creationflags=flags) # cwd=本地 procedure-dsl(run_cyber 假设的工作目录,agent 相对路径才对)
- ACTIVE_TASKS[task_key]["pid"] = proc.pid
- proc.wait()
-
- if proc.returncode == 0:
- try:
- import build_workflows
- build_workflows.write_one(q, folder_name, runs_dir=RUNS_DIR)
- _stamp_version(out_dir, version)
- _store_dsl("agent_dsl", q, case_id, version, model, out_dir) # 写库
- ACTIVE_TASKS[task_key]["status"] = "success"
- except Exception as ex:
- ACTIVE_TASKS[task_key]["status"] = "failed"
- ACTIVE_TASKS[task_key]["error"] = f"Workflow compilation failed: {ex}"
- with open(log_path, "a", encoding="utf-8") as f_err:
- f_err.write(f"\n[server error] Workflow compilation failed: {ex}\n")
- else:
- ACTIVE_TASKS[task_key]["status"] = "failed"
- ACTIVE_TASKS[task_key]["error"] = f"Runner failed with exit code {proc.returncode}"
- except Exception as e:
- ACTIVE_TASKS[task_key]["status"] = "failed"
- ACTIVE_TASKS[task_key]["error"] = str(e)
- try:
- with open(log_path, "a", encoding="utf-8") as f_err:
- f_err.write(f"\n[server error] Extraction failed: {e}\n")
- except Exception:
- pass
- MODSET = {"文", "图", "视频", "音频"}
- TOOLQUAL = {"AI": "AI 模型", "软件": "桌面 APP", "电脑端": "桌面 APP", "在线": "云端 Web",
- "网页版": "云端 Web", "代码": "API·CLI", "命令行": "API·CLI", "插件": "插件扩展"}
- def parse_dims(oq):
- """把组合 query(如 '文 元素生成 提示词 教程')解析回 {动作, 类型, 动作L1, 约束}。"""
- toks = (oq or "").split()
- action = next((t for t in toks if t in ACTION_SET), None)
- type_ = next((t for t in toks if t in TYPE_SET), None)
- cons = None
- if toks:
- t0 = toks[0]
- if t0 in MODSET:
- cons = {"kind": "模态", "value": t0}
- elif t0 in TOOLQUAL:
- cons = {"kind": "工具类型", "value": TOOLQUAL[t0]}
- return {"action": action, "type": type_, "action_l1": ACT_L1.get(action, ""), "constraint": cons}
- def flat_scores(sc):
- f = {}
- for k, v in (sc or {}).items():
- if isinstance(v, dict):
- for kk, vv in v.items():
- try: f[kk] = int(vv)
- except Exception: pass
- else:
- try: f[k] = int(v)
- except Exception: pass
- return f
- def _recency_hard(date_str):
- """按 publish_timestamp 头 10 字符(YYYY-MM-DD)算硬时效:半年内=3 / 两年内=2 / 更早=1。
- 取代原 LLM 评的 recency 维度——脚本算更稳,发布时间在帖子抓取时就有,无需 LLM token。
- """
- try:
- d = datetime.strptime((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 adapt(r, run, form_name=None):
- p = r.get("post", {}); e = r.get("llm_evaluation", {})
-
- # 1. 解析 知识类型 (knowledge_type)
- kt = []
- kt_raw = e.get("知识类型") or e.get("knowledge_type") or []
- for k in kt_raw:
- if k in ("工序", "procedure"): kt.append("procedure")
- elif k in ("能力", "步骤", "step"): kt.append("step")
- elif k in ("工具", "tool"): kt.append("tool")
- fs = {}
- score_reasons = {}
- # 检测是否为 eval_prompt_sample-mod 里的新版 0-10 分数 schema
- is_mod_schema = "相关性" in e and isinstance(e["相关性"], dict) and ("和内容制作知识相关" in e["相关性"] or "和 query 相关" in e["相关性"])
- if is_mod_schema:
- # 新版 0-10 分数格式解析
- # 1. 相关性
- rel = e.get("相关性") or {}
- for subkey, item in rel.items():
- if isinstance(item, dict):
- score_val = item.get("得分")
- reason_val = item.get("理由")
- code_key = None
- if "内容制作" in subkey or "知识" in subkey:
- code_key = "relevance_production"
- elif "query" in subkey or "检索" in subkey:
- code_key = "relevance_query"
- if code_key and score_val is not None:
- try:
- fs[code_key] = float(score_val)
- if reason_val:
- score_reasons[code_key] = reason_val
- except Exception:
- pass
- # 2. 质量
- q_block = e.get("质量") or {}
- fixed = q_block.get("固定维度") or {}
-
- # 固定维度
- fixed_keys = {
- "时效性": "recency",
- "热度性": "popularity",
- "评论反馈": "feedback"
- }
- for cn, code in fixed_keys.items():
- item = fixed.get(cn)
- if isinstance(item, dict):
- score_val = item.get("得分")
- reason_val = item.get("理由")
- if score_val is not None:
- try:
- fs[code] = float(score_val)
- if reason_val:
- score_reasons[code] = reason_val
- except Exception:
- pass
-
- # 用例 (真实感, 表现力)
- usecase = fixed.get("用例") or {}
- usecase_keys = {
- "真实感": "realism",
- "表现力": "expressiveness"
- }
- for cn, code in usecase_keys.items():
- item = usecase.get(cn)
- if isinstance(item, dict):
- score_val = item.get("得分")
- reason_val = item.get("理由")
- if score_val is not None:
- try:
- fs[code] = float(score_val)
- if reason_val:
- score_reasons[code] = reason_val
- except Exception:
- pass
- # 动态维度
- dynamic = q_block.get("动态维度") or {}
-
- # 工序
- proc = dynamic.get("工序") or {}
- if proc:
- item = proc.get("流程完整性")
- if isinstance(item, dict):
- score_val = item.get("得分")
- reason_val = item.get("理由")
- if score_val is not None:
- try:
- fs["procedure_completeness"] = float(score_val)
- if reason_val:
- score_reasons["procedure_completeness"] = reason_val
- except Exception:
- pass
- field = proc.get("字段完整性") or {}
- field_keys = {
- "输入完整性": "procedure_input",
- "实现完整性": "procedure_implementation",
- "输出完整性": "procedure_output"
- }
- for cn, code in field_keys.items():
- item = field.get(cn)
- if isinstance(item, dict):
- score_val = item.get("得分")
- reason_val = item.get("理由")
- if score_val is not None:
- try:
- fs[code] = float(score_val)
- if reason_val:
- score_reasons[code] = reason_val
- except Exception:
- pass
- item = proc.get("泛化性")
- if isinstance(item, dict):
- score_val = item.get("得分")
- reason_val = item.get("理由")
- if score_val is not None:
- try:
- fs["procedure_generality"] = float(score_val)
- if reason_val:
- score_reasons["procedure_generality"] = reason_val
- except Exception:
- pass
- # 能力
- cap = dynamic.get("能力") or dynamic.get("步骤") or {}
- if cap:
- field = cap.get("字段完整性") or {}
- field_keys = {
- "输入完整性": "step_input",
- "实现完整性": "step_implementation",
- "输出完整性": "step_output"
- }
- for cn, code in field_keys.items():
- item = field.get(cn)
- if isinstance(item, dict):
- score_val = item.get("得分")
- reason_val = item.get("理由")
- if score_val is not None:
- try:
- fs[code] = float(score_val)
- if reason_val:
- score_reasons[code] = reason_val
- except Exception:
- pass
- item = cap.get("泛化性")
- if isinstance(item, dict):
- score_val = item.get("得分")
- reason_val = item.get("理由")
- if score_val is not None:
- try:
- fs["step_generality"] = float(score_val)
- if reason_val:
- score_reasons["step_generality"] = reason_val
- except Exception:
- pass
- # 工具
- tool = dynamic.get("工具") or {}
- if tool:
- tool_keys = {
- "能力边界覆盖": "tool_boundary",
- "有效比较": "tool_comparison",
- "参数/接口具体性": "tool_specificity",
- "实操示例": "tool_example",
- "版本&限制": "tool_limits"
- }
- for cn, code in tool_keys.items():
- item = tool.get(cn)
- if isinstance(item, dict):
- score_val = item.get("得分")
- reason_val = item.get("理由")
- if score_val is not None:
- try:
- fs[code] = float(score_val)
- if reason_val:
- score_reasons[code] = reason_val
- except Exception:
- pass
- else:
- # 兼容老版 1-5 分数 schema (带 "评分" 或 old-style flatness)
- is_new_schema = "评分" in e or "知识类型" in e or "制作相关性" in e
- CN_TO_EN = {
- "相关性": "relevance",
- "成品质量": "result_quality",
- "可信度": "credibility",
- "具体用例": "concrete_use_case",
- "完整性": "completeness",
- "步骤结构": "step_structure",
- "步骤可复现": "step_reproducibility",
- "步骤可复现性": "step_reproducibility",
- "能力定义": "capability_definition",
- "实现深度": "implementation_depth",
- "边界失败": "boundary_failure_eval",
- "通用性": "generality",
- "能力覆盖": "capability_coverage",
- "有效对比": "effective_comparison",
- "参数具体": "param_specificity",
- "实操示例": "worked_example",
- "实操用例": "worked_example",
- "示例完整": "worked_example",
- "版本限制": "version_limits",
- "版本说明": "version_limits",
- "限制说明": "version_limits",
- }
-
- if is_new_schema:
- pf = e.get("评分") or {}
- for cat, metrics in pf.items():
- if isinstance(metrics, dict):
- for metric, val in metrics.items():
- en_key = CN_TO_EN.get(metric, metric)
- if isinstance(val, dict) and "得分" in val:
- try: fs[en_key] = int(val["得分"])
- except Exception: pass
- elif isinstance(val, (int, float)):
- fs[en_key] = int(val)
-
- if isinstance(val, dict) and "理由" in val:
- score_reasons[en_key] = val["理由"]
- else:
- fs = flat_scores(e.get("scores", {}))
-
- # 计算均分 (overall)
- if is_mod_schema:
- rel_keys = {"relevance_production", "relevance_query"}
- rel_vals = [v for k, v in fs.items() if k in rel_keys]
- qual_vals = [v for k, v in fs.items() if k not in rel_keys]
-
- rel_avg = sum(rel_vals) / len(rel_vals) if rel_vals else None
- qual_avg = sum(qual_vals) / len(qual_vals) if qual_vals else None
-
- if rel_avg is not None and qual_avg is not None:
- overall = round((rel_avg + qual_avg) / 2, 1)
- elif rel_avg is not None:
- overall = round(rel_avg, 1)
- elif qual_avg is not None:
- overall = round(qual_avg, 1)
- else:
- overall = 0.0
- else:
- overall = round(sum(fs.values()) / len(fs), 1) if fs else 0
- anomaly = bool(e.get("error")) or not fs
- grade = p.get("_quality_grade", "")
- fb = r.get("found_by_queries", [])
-
- # 4. 解析 制作相关性 (production_relevance)
- if is_mod_schema:
- # 新版使用 "相关性" 中的 "和内容制作知识相关" 代表制作相关性
- production_relevance = fs.get("relevance_production")
- else:
- if is_new_schema:
- pr_block = e.get("制作相关性") or {}
- pr_raw = pr_block.get("得分") if isinstance(pr_block, dict) else pr_block
- if isinstance(pr_block, dict) and "理由" in pr_block:
- score_reasons["production_relevance"] = pr_block["理由"]
- else:
- pr_raw = e.get("production_relevance")
-
- try: production_relevance = int(float(pr_raw)) if pr_raw is not None else None
- except (TypeError, ValueError): production_relevance = None
-
- recency_hard = _recency_hard(p.get("publish_timestamp", ""))
-
- # 5. 解析 判定决策 (decision) 和 理由 (reason)
- reason = e.get("判定理由") or e.get("reason") or ""
-
- # 根据过滤指标决定是否保留 (过滤指标判定逻辑优先,不依赖文字匹配)
- is_discard = False
-
- # 制作相关性低于阈值则丢弃 (新版 0-10 满分,因此低于 4 丢弃;老版低于 2 丢弃)
- if production_relevance is not None:
- threshold = 4 if is_mod_schema else 2
- if production_relevance < threshold:
- is_discard = True
-
- # 时效性低于 2 被丢弃(发布时间超两年的老帖)
- if recency_hard is not None and recency_hard < 2:
- is_discard = True
-
- # 综合均分低于阈值被丢弃 (新版低于 6 丢弃;老版低于 3 丢弃)
- if overall is not None:
- threshold_ov = 6 if is_mod_schema else 3
- if overall < threshold_ov:
- is_discard = True
-
- decision = "discard" if is_discard else "report"
- # Find matching procedure html
- procedure_html = None
- case_id = r.get("case_id", "")
- title = p.get("title", "")
- run_dir = RUNS_DIR / run
- if run_dir.is_dir():
- # 1. 优先扫描该帖子对应的文件夹下的任何 HTML 文件 (不限名称)
- # 文件夹名格式: {form}_{platform}_{channel_content_id[:8]}
- content_id = r.get("channel_content_id") or ""
- if not content_id and case_id and "_" in case_id:
- content_id = case_id.split("_", 1)[1]
- plat_key = r.get("platform") or ""
-
- if form_name and plat_key and content_id:
- folder_name = f"{form_name}_{plat_key}_{content_id[:8]}"
- case_dir = run_dir / "procedures" / folder_name
- if case_dir.is_dir():
- html_files = list(case_dir.glob("*.html"))
- if html_files:
- procedure_html = f"runs_full/{run}/procedures/{folder_name}/{html_files[0].name}"
- # 1b. 兼容 fallback: 批量提取模式下直接以 case_id 命名的文件夹
- if not procedure_html and case_id:
- case_dir = run_dir / "procedures" / case_id
- if case_dir.is_dir():
- html_files = list(case_dir.glob("*.html"))
- if html_files:
- procedure_html = f"runs_full/{run}/procedures/{case_id}/{html_files[0].name}"
- # 2. 其次匹配标准文件名: case-{case_id}.html 或 {case_id}.html
- candidate_dirs = [run_dir, run_dir / "procedures"]
- if not procedure_html and case_id:
- named_files = [f"case-{case_id}.html", f"{case_id}.html"]
- for d_dir in candidate_dirs:
- if d_dir.is_dir():
- for name in named_files:
- if (d_dir / name).is_file():
- procedure_html = f"runs_full/{run}/procedures/{name}" if d_dir.name == "procedures" else f"runs_full/{run}/{name}"
- break
- if procedure_html:
- break
- # 3. 再次匹配 HTML 内部的标准声明 (meta 标签或 HTML 注释)
- if not procedure_html and case_id:
- for d_dir in candidate_dirs:
- if d_dir.is_dir():
- for html_path in d_dir.glob("*.html"):
- try:
- content = html_path.read_text(encoding="utf-8")
- if f'name="case-id" content="{case_id}"' in content or \
- f'name="case_id" content="{case_id}"' in content or \
- f'<!-- case_id: {case_id} -->' in content or \
- f'<!-- case-id: {case_id} -->' in content:
- procedure_html = f"runs_full/{run}/procedures/{html_path.name}" if d_dir.name == "procedures" else f"runs_full/{run}/{html_path.name}"
- break
- except Exception:
- continue
- if procedure_html:
- break
- # 4. 最后使用标题作为兜底模糊匹配
- if not procedure_html and title:
- for d_dir in candidate_dirs:
- if d_dir.is_dir():
- for html_path in d_dir.glob("*.html"):
- try:
- content = html_path.read_text(encoding="utf-8")
- if title in content:
- procedure_html = f"runs_full/{run}/procedures/{html_path.name}" if d_dir.name == "procedures" else f"runs_full/{run}/{html_path.name}"
- break
- except Exception:
- continue
- if procedure_html:
- break
- if procedure_html:
- procedure_html = procedure_html.replace("runs_full/", f"{RUNS_DIR_NAME}/")
- return {
- "case_id": r.get("case_id", ""),
- "platform": PLAT.get(r.get("platform"), r.get("platform")), "platformKey": r.get("platform"),
- "title": p.get("title", "") or "(无标题)", "date": (p.get("publish_timestamp", "") or "")[:10],
- "url": r.get("source_url", ""), "engagement": f'{p.get("like_count", 0)} 赞',
- "knowledge_type": kt, "decision": decision,
- "tools": [KT.get(k, k) for k in kt] + ([f"质量 {grade}"] if grade else []), "found_by": fb,
- "images": (p.get("images") or [])[:6], "text": p.get("body_text", "") or "",
- "scores": fs, "overall": overall, "reason": reason, "score_reasons": score_reasons,
- "grade": grade, "qscore": p.get("_quality_score", 0), "anomaly": anomaly,
- "production_relevance": production_relevance, "recency_hard": recency_hard,
- "run": run, "procedure_html": procedure_html,
- }
- def scan_runs():
- runs = {}
- for f in sorted(glob.glob(str(RUNS_DIR / "*" / "form_*.json"))):
- try:
- d = json.load(open(f, encoding="utf-8"))
- except Exception:
- continue
- run = pathlib.Path(f).parent.name
- form_name = d.get("form") or ""
- results = [adapt(r, run, form_name) for r in d.get("results", [])]
- report_val = sum(1 for r in results if r.get("decision") == "report" and not r.get("anomaly"))
- discard_val = sum(1 for r in results if r.get("decision") == "discard" and not r.get("anomaly"))
-
- runs.setdefault(run, []).append({
- "form": d.get("form"), "query": d.get("query"), "original_q": d.get("original_q", ""),
- "requirement": d.get("requirement", ""),
- "platforms": d.get("platforms", []), "total": d.get("total"),
- "report": report_val, "discard": discard_val,
- "results": results,
- })
- for v in runs.values():
- v.sort(key=lambda x: x.get("form") or "")
- if MODE == "new":
- def _mtime(name):
- try:
- return (RUNS_DIR / name).stat().st_mtime
- except Exception:
- return 0
- sorted_keys = sorted(runs.keys(), key=_mtime, reverse=True)
- else:
- def _qnum(name):
- m = re.search(r"\d+", name)
- return (int(m.group()) if m else 0, name)
- sorted_keys = [kv[0] for kv in sorted(runs.items(), key=lambda kv: _qnum(kv[0]))]
- out = []
- for k in sorted_keys:
- v = runs[k]
- oq = v[0].get("original_q") or v[0].get("query") or ""
- seen, hits = set(), 0 # 知识命中数 = 各形式采纳(report)且非异常、按 url 去重后的帖子数
- for f in v:
- for r in f.get("results", []):
- if r.get("decision") == "report" and not r.get("anomaly") and r.get("url") not in seen:
- seen.add(r.get("url")); hits += 1
- out.append({"key": k, "forms": v, "dims": parse_dims(oq), "original_q": oq,
- "hits": hits, "tot": sum((f.get("total") or 0) for f in v)})
- active_reevals = {k: v["status"] for k, v in ACTIVE_REEVALS.items()}
- active_batch_tasks = {k: v["status"] for k, v in ACTIVE_BATCH_TASKS.items()}
- return {"queries": out, "actions": ACTIONS_TAX, "types": TYPES_TAX, "matrix": _MATRIX, "active_reevals": active_reevals, "active_batch_tasks": active_batch_tasks}
- class H(BaseHTTPRequestHandler):
- def _send(self, code, body, ctype):
- b = body.encode("utf-8") if isinstance(body, str) else body
- self.send_response(code)
- if ctype.startswith("text/") or ctype == "application/json" or ctype == "application/javascript":
- self.send_header("Content-Type", ctype + "; charset=utf-8")
- else:
- self.send_header("Content-Type", ctype)
- self.send_header("Content-Length", str(len(b))); self.end_headers(); self.wfile.write(b)
- def do_GET(self):
- parsed = urlparse(self.path)
- path = parsed.path
- params = parse_qs(parsed.query)
- if path in ("/", "/index.html"):
- try:
- html_file = "new_query.html" if MODE == "new" else "index.html"
- page = (HERE / html_file).read_text(encoding="utf-8")
- self._send(200, page, "text/html")
- except Exception as e:
- self._send(500, f"Error reading page: {e}", "text/plain")
- elif path == "/api/data":
- self._send(200, json.dumps(scan_runs(), ensure_ascii=False), "application/json")
- elif path == "/api/procedure_status":
- q = (params.get("q") or [""])[0].strip()
- form = (params.get("form") or [""])[0].strip()
- case_id = (params.get("case_id") or [""])[0].strip()
-
- if not q or not form or not case_id:
- self._send(400, "missing q, form, or case_id", "text/plain")
- return
-
- variant = (params.get("variant") or ["agent"])[0].strip()
- req_version = (params.get("version") or [""])[0].strip()
- folder_name = f"{form}_{_short_case(case_id)}"
- out_dir = RUNS_DIR / q / "procedures" / folder_name
- rel_base = f"{RUNS_DIR_NAME}/{q}/procedures/{folder_name}"
- task_key = f"{q}/{folder_name}"
- if variant == "model": # 大模型工序在 _model/ 子目录
- out_dir = out_dir / "_model"
- rel_base = rel_base + "/_model"
- task_key = task_key + "/_model"
- # 指定了非当前版本 → 直接从 history/{version}/ 取该版本的 html
- if req_version and req_version != _read_meta_version(out_dir):
- hist = out_dir / "history" / req_version
- hf = list(hist.glob("*.html")) if hist.is_dir() else []
- if hf:
- self._send(200, json.dumps({"status": "success",
- "procedure_html": f"{rel_base}/history/{req_version}/{hf[0].name}"},
- ensure_ascii=False), "application/json")
- else:
- self._send(200, json.dumps({"status": "not_started",
- "error": f"版本 {req_version} 的页面文件已不存在"}, ensure_ascii=False), "application/json")
- return
- if task_key in ACTIVE_TASKS:
- task = ACTIVE_TASKS[task_key]
- res = {"status": task["status"], "error": task["error"]}
- if task["status"] == "success":
- html_files = list(out_dir.glob("*.html")) if out_dir.is_dir() else []
- if html_files:
- res["procedure_html"] = f"{rel_base}/{html_files[0].name}"
- self._send(200, json.dumps(res, ensure_ascii=False), "application/json")
- return
- html_files = list(out_dir.glob("*.html")) if out_dir.is_dir() else []
- if html_files:
- self._send(200, json.dumps({
- "status": "success",
- "procedure_html": f"{rel_base}/{html_files[0].name}"
- }, ensure_ascii=False), "application/json")
- return
- # 仅 agent 变体兼容旧的 case_id 命名目录
- if variant != "model" and case_id:
- fallback_dir = RUNS_DIR / q / "procedures" / case_id
- fb_html = list(fallback_dir.glob("*.html")) if fallback_dir.is_dir() else []
- if fb_html:
- self._send(200, json.dumps({
- "status": "success",
- "procedure_html": f"{RUNS_DIR_NAME}/{q}/procedures/{case_id}/{fb_html[0].name}"
- }, ensure_ascii=False), "application/json")
- return
- log_path = out_dir / "_extract.log"
- if not log_path.is_file() and variant != "model" and case_id:
- log_path = RUNS_DIR / q / "procedures" / case_id / "_extract.log"
- if log_path.is_file():
- self._send(200, json.dumps({"status": "failed", "error": "Not running, but no HTML output found (possibly crashed)."}, ensure_ascii=False), "application/json")
- return
- self._send(200, json.dumps({"status": "not_started"}, ensure_ascii=False), "application/json")
- elif path == "/api/procedure_versions":
- # 列某帖某变体的所有版本(DB 为主,文件系统兜底)。latest = 当前 _meta 版本。
- q = (params.get("q") or [""])[0].strip()
- form = (params.get("form") or [""])[0].strip()
- case_id = (params.get("case_id") or [""])[0].strip()
- variant = (params.get("variant") or ["agent"])[0].strip()
- folder_name = f"{form}_{_short_case(case_id)}"
- out_dir = RUNS_DIR / q / "procedures" / folder_name
- if variant == "model":
- out_dir = out_dir / "_model"
- latest = _read_meta_version(out_dir)
- by_ver = {}
- # DB(带 model / 工序数 / 时间)
- try:
- import db
- for r in db.fetch_versions("mode_dsl" if variant == "model" else "agent_dsl", q, case_id):
- if r.get("version"):
- by_ver[r["version"]] = {"version": r["version"], "model": r.get("model"),
- "procedure_count": r.get("procedure_count"),
- "created_time": str(r.get("created_time") or "")}
- except Exception:
- pass
- # 文件系统兜底(当前 + history/v_*)
- for v in ([latest] if latest else []) + (
- [d.name for d in (out_dir / "history").iterdir()
- if d.is_dir() and d.name.startswith("v_")] if (out_dir / "history").is_dir() else []):
- by_ver.setdefault(v, {"version": v, "model": None, "procedure_count": None, "created_time": ""})
- versions = sorted(by_ver.values(), key=lambda x: x["version"], reverse=True)
- for v in versions:
- v["is_latest"] = (v["version"] == latest)
- self._send(200, json.dumps({"versions": versions, "latest": latest}, ensure_ascii=False), "application/json")
- elif path == "/api/procedure_log":
- q = (params.get("q") or [""])[0].strip()
- form = (params.get("form") or [""])[0].strip()
- case_id = (params.get("case_id") or [""])[0].strip()
-
- if not q or not form or not case_id:
- self._send(400, "missing q, form, or case_id", "text/plain")
- return
-
- variant = (params.get("variant") or ["agent"])[0].strip()
- folder_name = f"{form}_{_short_case(case_id)}"
- if variant == "model":
- log_path = RUNS_DIR / q / "procedures" / folder_name / "_model" / "_extract.log"
- else:
- log_path = RUNS_DIR / q / "procedures" / folder_name / "_extract.log"
- if not log_path.is_file() and case_id:
- log_path = RUNS_DIR / q / "procedures" / case_id / "_extract.log"
-
- if not log_path.is_file():
- self._send(200, json.dumps({"log": ""}, ensure_ascii=False), "application/json")
- return
-
- try:
- content = log_path.read_text(encoding="utf-8", errors="replace")
- self._send(200, json.dumps({"log": content}, ensure_ascii=False), "application/json")
- except Exception as e:
- self._send(500, json.dumps({"error": str(e)}, ensure_ascii=False), "application/json")
- elif path == "/api/spec_content":
- file_name = (params.get("file") or [""])[0].strip()
- allowed = [
- "README.md",
- "tools.md",
- "extraction/phase1-skeleton.md",
- "extraction/phase2-normalize.md",
- "extraction/phase3-finalize.md",
- "taxonomy/type_suggestions.md"
- ]
- if file_name not in allowed:
- self._send(400, "invalid file parameter", "text/plain")
- return
- target_path = HERE / "procedure-dsl" / "spec" / file_name
- if not target_path.is_file():
- self._send(404, "spec file not found", "text/plain")
- return
- try:
- content = target_path.read_text(encoding="utf-8", errors="replace")
- self._send(200, json.dumps({"content": content}, ensure_ascii=False), "application/json")
- except Exception as e:
- self._send(500, json.dumps({"error": str(e)}, ensure_ascii=False), "application/json")
- elif path == "/api/reeval_status":
- q = (params.get("q") or [""])[0].strip()
- if not q:
- self._send(400, "missing q", "text/plain")
- return
- if q in ACTIVE_REEVALS:
- self._send(200, json.dumps({
- "status": ACTIVE_REEVALS[q]["status"],
- "error": ACTIVE_REEVALS[q].get("error")
- }, ensure_ascii=False), "application/json")
- else:
- self._send(200, json.dumps({"status": "not_started"}, ensure_ascii=False), "application/json")
- elif path == "/api/search_eval_status":
- q = (params.get("q") or [""])[0].strip()
- if not q:
- self._send(400, "missing q", "text/plain")
- return
- if q in ACTIVE_SEARCH_EVALS:
- self._send(200, json.dumps({
- "status": ACTIVE_SEARCH_EVALS[q]["status"],
- "error": ACTIVE_SEARCH_EVALS[q].get("error")
- }, ensure_ascii=False), "application/json")
- else:
- self._send(200, json.dumps({"status": "not_started"}, ensure_ascii=False), "application/json")
- elif path == "/api/batch_generate_status":
- q = (params.get("q") or [""])[0].strip()
- if not q:
- self._send(400, "missing q", "text/plain")
- return
- if q in ACTIVE_BATCH_TASKS:
- self._send(200, json.dumps({
- "status": ACTIVE_BATCH_TASKS[q]["status"],
- "error": ACTIVE_BATCH_TASKS[q].get("error")
- }, ensure_ascii=False), "application/json")
- else:
- self._send(200, json.dumps({"status": "not_started"}, ensure_ascii=False), "application/json")
- elif path == "/api/batch_generate_log":
- q = (params.get("q") or [""])[0].strip()
- if not q:
- self._send(400, "missing q", "text/plain")
- return
- safe_q = re.sub(r'[\x00-\x1f\\/*?:"<>|]', '', q).strip()
- log_path = RUNS_DIR / safe_q / "procedures" / "_batch_extract.log"
- if not log_path.is_file():
- self._send(200, json.dumps({"log": ""}, ensure_ascii=False), "application/json")
- return
- try:
- content = log_path.read_text(encoding="utf-8", errors="replace")
- self._send(200, json.dumps({"log": content}, ensure_ascii=False), "application/json")
- except Exception as e:
- self._send(500, json.dumps({"error": str(e)}, ensure_ascii=False), "application/json")
- elif path == "/api/search_eval_log":
- q = (params.get("q") or [""])[0].strip()
- if not q:
- self._send(400, "missing q", "text/plain")
- return
- safe_q = re.sub(r'[\x00-\x1f\\/*?:"<>|]', '', q).strip()
- log_path = RUNS_DIR / safe_q / "_search_eval.log"
- if not log_path.is_file():
- self._send(200, json.dumps({"log": ""}, ensure_ascii=False), "application/json")
- return
- try:
- content = log_path.read_text(encoding="utf-8", errors="replace")
- self._send(200, json.dumps({"log": content}, ensure_ascii=False), "application/json")
- except Exception as e:
- self._send(500, json.dumps({"error": str(e)}, ensure_ascii=False), "application/json")
- elif self.path.startswith("/runs_full/") or self.path.startswith("/runs_new/"):
- try:
- import urllib.parse
- clean_path = urllib.parse.unquote(self.path.split("?")[0])
- parts = clean_path.strip("/").split("/")
- target_file = HERE
- for part in parts:
- target_file = target_file / part
- runs_dir = HERE / parts[0]
- if runs_dir.resolve() in target_file.resolve().parents and target_file.is_file():
- content = target_file.read_bytes()
- ext = target_file.suffix.lower()
- ctype = "text/html"
- if ext in (".png", ".webp"):
- ctype = f"image/{ext[1:]}"
- elif ext in (".jpg", ".jpeg"):
- ctype = "image/jpeg"
- elif ext == ".json":
- ctype = "application/json"
- elif ext == ".js":
- ctype = "application/javascript"
- elif ext == ".css":
- ctype = "text/css"
- self._send(200, content, ctype)
- else:
- self._send(404, "not found", "text/plain")
- except Exception as e:
- self._send(500, f"Error: {e}", "text/plain")
- else:
- self._send(404, "not found", "text/plain")
- def do_POST(self):
- if self.path == "/api/run_search_eval":
- length = int(self.headers.get("Content-Length") or 0)
- raw = self.rfile.read(length).decode("utf-8") if length > 0 else "{}"
- try:
- payload = json.loads(raw)
- except Exception as e:
- self._send(400, json.dumps({"error": f"bad json: {e}"}), "application/json"); return
-
- q = (payload.get("query") or "").strip()
- platforms = (payload.get("platforms") or "xhs,zhihu").strip()
-
- if not q:
- self._send(400, json.dumps({"error": "missing query"}, ensure_ascii=False), "application/json"); return
-
- safe_q = re.sub(r'[\x00-\x1f\\/*?:"<>|]', '', q).strip()
- if not safe_q:
- self._send(400, json.dumps({"error": "invalid query name"}, ensure_ascii=False), "application/json"); return
-
- q_dir = RUNS_DIR / safe_q
- q_dir.mkdir(parents=True, exist_ok=True)
-
- temp_queries_file = q_dir / "temp_queries.json"
- try:
- with open(temp_queries_file, "w", encoding="utf-8") as f:
- json.dump([q], f, ensure_ascii=False)
- except Exception as e:
- self._send(500, json.dumps({"error": f"failed to write temp query: {e}"}), "application/json"); return
-
- log_path = q_dir / "_search_eval.log"
-
- ACTIVE_SEARCH_EVALS[q] = {
- "status": "running",
- "pid": None,
- "error": None
- }
-
- def run_search_eval_task(query_text, safe_query_name, queries_file, out_dir, target_platforms, log_file):
- try:
- cmd = [
- sys.executable, "-u", str(SE / "search_and_evaluate.py"),
- "--queries", str(queries_file),
- "--platforms", target_platforms,
- "--output-dir", str(out_dir),
- "--max-count", "20"
- ]
- import os
- env = os.environ.copy()
- env["PYTHONIOENCODING"] = "utf-8"
- env["PYTHONUTF8"] = "1"
- # 搜索评估未传 --eval-model → 用 search_and_evaluate 默认 gemini-flash-lite(OpenRouter)
- _se_model = cmd[cmd.index("--eval-model") + 1] if "--eval-model" in cmd else "google/gemini-3.1-flash-lite (默认)"
- flags = subprocess.CREATE_NEW_PROCESS_GROUP if sys.platform == "win32" else 0
- with open(log_file, "w", encoding="utf-8", buffering=1) as log_fh:
- _log_header(log_fh, "🔎 搜索评估", _se_model)
- proc = subprocess.Popen(cmd, stdout=log_fh, stderr=subprocess.STDOUT,
- cwd=str(SE), env=env, creationflags=flags)
- ACTIVE_SEARCH_EVALS[query_text]["pid"] = proc.pid
- proc.wait()
-
- if proc.returncode == 0:
- evaluated_file = out_dir / "evaluated.json"
- if evaluated_file.is_file():
- with open(evaluated_file, "r", encoding="utf-8") as rf:
- eval_data = json.load(rf)
- eval_data["form"] = "A"
- eval_data["query"] = query_text
- eval_data["original_q"] = query_text
-
- with open(out_dir / "form_A.json", "w", encoding="utf-8") as wf:
- json.dump(eval_data, wf, ensure_ascii=False, indent=2)
- ACTIVE_SEARCH_EVALS[query_text]["status"] = "success"
- else:
- ACTIVE_SEARCH_EVALS[query_text]["status"] = "failed"
- ACTIVE_SEARCH_EVALS[query_text]["error"] = "evaluated.json not found after execution"
- with open(log_file, "a", encoding="utf-8") as f_err:
- f_err.write("\n[server error] evaluated.json not found after execution\n")
- else:
- ACTIVE_SEARCH_EVALS[query_text]["status"] = "failed"
- ACTIVE_SEARCH_EVALS[query_text]["error"] = f"search_and_evaluate.py exited with code {proc.returncode}"
- except Exception as ex:
- ACTIVE_SEARCH_EVALS[query_text]["status"] = "failed"
- ACTIVE_SEARCH_EVALS[query_text]["error"] = str(ex)
- try:
- with open(log_file, "a", encoding="utf-8") as f_err:
- f_err.write(f"\n[server error] Exception: {ex}\n")
- except Exception:
- pass
- finally:
- try:
- if queries_file.is_file():
- queries_file.unlink()
- except Exception:
- pass
-
- t = threading.Thread(target=run_search_eval_task, args=(q, safe_q, temp_queries_file, q_dir, platforms, log_path))
- t.daemon = True
- t.start()
-
- self._send(200, json.dumps({
- "status": "started",
- "query": q,
- "log": f"{RUNS_DIR_NAME}/{safe_q}/_search_eval.log"
- }, ensure_ascii=False), "application/json")
- elif self.path == "/api/generate_procedure":
- length = int(self.headers.get("Content-Length") or 0)
- raw = self.rfile.read(length).decode("utf-8") if length > 0 else "{}"
- try:
- payload = json.loads(raw)
- except Exception as e:
- self._send(400, json.dumps({"error": f"bad json: {e}"}), "application/json"); return
-
- q = (payload.get("q") or "").strip()
- form = (payload.get("form") or "").strip()
- case_id = (payload.get("case_id") or "").strip()
- engine = (payload.get("engine") or "cyber_runner").strip()
- model = (payload.get("model") or "google/gemini-3.1-flash-lite").strip()
-
- if MODE != "new" and not re.match(r"^q\d+$", q):
- self._send(400, json.dumps({"error": f"bad q (expect 'qNN'): {q!r}"}, ensure_ascii=False), "application/json"); return
- if form not in ("A", "B", "C"):
- self._send(400, json.dumps({"error": f"bad form: {form!r}"}, ensure_ascii=False), "application/json"); return
- if not case_id:
- self._send(400, json.dumps({"error": "missing case_id"}, ensure_ascii=False), "application/json"); return
-
- q_dir = RUNS_DIR / q
- form_file = q_dir / f"form_{form}.json"
- if not form_file.is_file():
- self._send(404, json.dumps({"error": f"form file not found: {form_file.name}"}, ensure_ascii=False), "application/json"); return
-
- try:
- with open(form_file, encoding="utf-8") as f:
- form_data = json.load(f)
- except Exception as e:
- self._send(500, json.dumps({"error": f"failed to read form: {e}"}, ensure_ascii=False), "application/json"); return
-
- matching_result = None
- for r in form_data.get("results", []):
- if r.get("case_id") == case_id:
- matching_result = r
- break
-
- if not matching_result:
- self._send(404, json.dumps({"error": f"case_id {case_id} not found in form {form}"}, ensure_ascii=False), "application/json"); return
-
- folder_name = f"{form}_{_short_case(case_id)}"
- out_dir = q_dir / "procedures" / folder_name
- backup_procedure_history(out_dir)
- out_dir.mkdir(parents=True, exist_ok=True)
-
- src_path = out_dir / "_source.json"
- try:
- with open(src_path, "w", encoding="utf-8") as f:
- json.dump(_source_to_dsl_input(matching_result), f, ensure_ascii=False, indent=2)
-
- score = _composite_score(matching_result.get("llm_evaluation") or {})
- _write_meta(out_dir, case_id=case_id, from_q=q, form=form, score=score)
- version = _gen_version()
- _stamp_version(out_dir, version)
- except Exception as e:
- self._send(500, json.dumps({"error": f"failed to write inputs: {e}"}, ensure_ascii=False), "application/json"); return
- task_key = f"{q}/{folder_name}"
- ACTIVE_TASKS[task_key] = {
- "status": "running",
- "start_time": datetime.now().isoformat(),
- "pid": None,
- "error": None
- }
- t = threading.Thread(target=run_extraction_task, args=(q, folder_name, case_id, src_path, out_dir, engine, model, version))
- t.daemon = True
- t.start()
-
- self._send(200, json.dumps({
- "status": "started",
- "task_key": task_key,
- "log": f"{RUNS_DIR_NAME}/{q}/procedures/{folder_name}/_extract.log"
- }, ensure_ascii=False), "application/json")
- elif self.path == "/api/extract_procedure_model":
- # 模型提取:单次大模型直出 workflow.json(无 agent / 无 validate 循环)。
- # 写同一个 procedures/{form}_{short_case}/,复用 procedure_status/log/展示。
- length = int(self.headers.get("Content-Length") or 0)
- raw = self.rfile.read(length).decode("utf-8") if length > 0 else "{}"
- try:
- payload = json.loads(raw)
- except Exception as e:
- self._send(400, json.dumps({"error": f"bad json: {e}"}), "application/json"); return
- q = (payload.get("q") or "").strip()
- form = (payload.get("form") or "").strip()
- case_id = (payload.get("case_id") or "").strip()
- model = (payload.get("model") or "anthropic/claude-sonnet-4-6").strip() # 大模型工序默认 claude
- if form not in ("A", "B", "C"):
- self._send(400, json.dumps({"error": f"bad form: {form!r}"}, ensure_ascii=False), "application/json"); return
- if not case_id:
- self._send(400, json.dumps({"error": "missing case_id"}, ensure_ascii=False), "application/json"); return
- q_dir = RUNS_DIR / q
- form_file = q_dir / f"form_{form}.json"
- if not form_file.is_file():
- self._send(404, json.dumps({"error": f"form file not found: {form_file.name}"}, ensure_ascii=False), "application/json"); return
- try:
- with open(form_file, encoding="utf-8") as f:
- form_data = json.load(f)
- except Exception as e:
- self._send(500, json.dumps({"error": f"failed to read form: {e}"}, ensure_ascii=False), "application/json"); return
- matching_result = next((r for r in form_data.get("results", []) if r.get("case_id") == case_id), None)
- if not matching_result:
- self._send(404, json.dumps({"error": f"case_id {case_id} not found in form {form}"}, ensure_ascii=False), "application/json"); return
- folder_name = f"{form}_{_short_case(case_id)}"
- out_dir = q_dir / "procedures" / folder_name / "_model" # 大模型工序独立子目录,不覆盖 Agent 工序
- backup_procedure_history(out_dir)
- out_dir.mkdir(parents=True, exist_ok=True)
- src_path = out_dir / "_source.json"
- try:
- with open(src_path, "w", encoding="utf-8") as f:
- json.dump(_source_to_dsl_input(matching_result), f, ensure_ascii=False, indent=2)
- _write_meta(out_dir, case_id=case_id, from_q=q, form=form,
- score=_composite_score(matching_result.get("llm_evaluation") or {}))
- version = _gen_version()
- _stamp_version(out_dir, version)
- except Exception as e:
- self._send(500, json.dumps({"error": f"failed to write inputs: {e}"}, ensure_ascii=False), "application/json"); return
- task_key = f"{q}/{folder_name}/_model"
- ACTIVE_TASKS[task_key] = {"status": "running", "start_time": datetime.now().isoformat(), "pid": None, "error": None}
- t = threading.Thread(target=run_model_extraction_task, args=(q, folder_name, case_id, src_path, out_dir, model, version))
- t.daemon = True
- t.start()
- self._send(200, json.dumps({
- "status": "started", "task_key": task_key,
- "log": f"{RUNS_DIR_NAME}/{q}/procedures/{folder_name}/_model/_extract.log"
- }, ensure_ascii=False), "application/json")
- elif self.path == "/api/reeval":
- length = int(self.headers.get("Content-Length") or 0)
- raw = self.rfile.read(length).decode("utf-8") if length > 0 else "{}"
- try:
- payload = json.loads(raw)
- except Exception as e:
- self._send(400, json.dumps({"error": f"bad json: {e}"}), "application/json"); return
- q = (payload.get("q") or "").strip()
- if MODE != "new" and not re.match(r"^q\d+$", q):
- self._send(400, json.dumps({"error": f"bad q (expect 'qNN'): {q!r}"},
- ensure_ascii=False), "application/json"); return
- q_dir = RUNS_DIR / q
- if not q_dir.is_dir():
- self._send(404, json.dumps({"error": f"{RUNS_DIR_NAME}/{q} not found"}, ensure_ascii=False),
- "application/json"); return
- backup_reeval_history(q_dir)
- log_path = q_dir / "_reeval.log"
- try:
- log_fh = open(log_path, "w", encoding="utf-8", buffering=1)
- cmd = [sys.executable, "-u", str(SE / "batch_3forms.py"),
- "--reeval", "--reeval-q", q, "--output-dir", str(RUNS_DIR)]
- flags = subprocess.CREATE_NEW_PROCESS_GROUP if sys.platform == "win32" else 0
- proc = subprocess.Popen(cmd, stdout=log_fh, stderr=subprocess.STDOUT,
- cwd=str(SE), creationflags=flags)
-
- ACTIVE_REEVALS[q] = {
- "status": "running",
- "pid": proc.pid,
- "error": None
- }
-
- def wait_reeval(q_key, p_obj, fh):
- try:
- p_obj.wait()
- if p_obj.returncode == 0:
- ACTIVE_REEVALS[q_key]["status"] = "success"
- else:
- ACTIVE_REEVALS[q_key]["status"] = "failed"
- ACTIVE_REEVALS[q_key]["error"] = f"Subprocess exited with code {p_obj.returncode}"
- except Exception as ex:
- ACTIVE_REEVALS[q_key]["status"] = "failed"
- ACTIVE_REEVALS[q_key]["error"] = str(ex)
- finally:
- try:
- fh.close()
- except Exception:
- pass
-
- t = threading.Thread(target=wait_reeval, args=(q, proc, log_fh))
- t.daemon = True
- t.start()
-
- self._send(200, json.dumps(
- {"status": "started", "pid": proc.pid, "q": q,
- "log": str(log_path.relative_to(HERE))},
- ensure_ascii=False), "application/json")
- except Exception as e:
- self._send(500, json.dumps({"error": f"failed to start: {e}"},
- ensure_ascii=False), "application/json")
- elif self.path == "/api/batch_generate_procedure":
- length = int(self.headers.get("Content-Length") or 0)
- raw = self.rfile.read(length).decode("utf-8") if length > 0 else "{}"
- try:
- payload = json.loads(raw)
- except Exception as e:
- self._send(400, json.dumps({"error": f"bad json: {e}"}), "application/json"); return
-
- q = (payload.get("q") or "").strip()
- form = (payload.get("form") or "A").strip()
- engine = (payload.get("engine") or "cyber_runner").strip()
- model = (payload.get("model") or "google/gemini-3.1-flash-lite").strip()
- try:
- concurrency = int(payload.get("concurrency", 4))
- except Exception:
- concurrency = 4
- concurrency = max(1, min(16, concurrency))
-
- if not q:
- self._send(400, json.dumps({"error": "missing q"}, ensure_ascii=False), "application/json"); return
-
- q_dir = RUNS_DIR / q
- form_file = q_dir / f"form_{form}.json"
- if not form_file.is_file():
- self._send(404, json.dumps({"error": f"form file not found: {form_file.name}"}, ensure_ascii=False), "application/json"); return
-
- try:
- with open(form_file, encoding="utf-8") as f:
- form_data = json.load(f)
- except Exception as e:
- self._send(500, json.dumps({"error": f"failed to read form: {e}"}, ensure_ascii=False), "application/json"); return
-
- valid_results = []
- import copy
- for r in form_data.get("results", []):
- adapted = adapt(r, q, form)
- if adapted.get("decision") == "report" and not adapted.get("anomaly"):
- r_copy = copy.deepcopy(r)
- original_case_id = r.get("case_id", "")
- r_copy["case_id"] = f"{form}_{_short_case(original_case_id)}"
- valid_results.append(r_copy)
-
- if not valid_results:
- self._send(400, json.dumps({"error": "没有可提取工序的帖子(所有帖子都已被过滤,或都存在解析异常)"}, ensure_ascii=False), "application/json"); return
-
- procedures_dir = q_dir / "procedures"
- procedures_dir.mkdir(parents=True, exist_ok=True)
- batch_posts_path = procedures_dir / "temp_batch.json"
- try:
- with open(batch_posts_path, "w", encoding="utf-8") as f:
- json.dump({"results": valid_results}, f, ensure_ascii=False, indent=2)
- except Exception as e:
- self._send(500, json.dumps({"error": f"failed to write temp_batch.json: {e}"}, ensure_ascii=False), "application/json"); return
-
- ACTIVE_BATCH_TASKS[q] = {
- "status": "running",
- "start_time": datetime.now().isoformat(),
- "pid": None,
- "error": None
- }
-
- def run_batch_extraction():
- try:
- log_path = procedures_dir / "_batch_extract.log"
- cmd = [
- sys.executable, "-u", str(HERE / "procedure-dsl" / "run_cyber.py"),
- str(batch_posts_path),
- "--out-dir", str(procedures_dir),
- "--batch",
- "--batch-workers", str(concurrency),
- "--model", model,
- "--skill", "procedure", # mode_procedure:用直写版 skill
- "--max-turns", "300"
- ]
-
- flags = subprocess.CREATE_NEW_PROCESS_GROUP if sys.platform == "win32" else 0
- with open(log_path, "w", encoding="utf-8", buffering=1) as log_fh:
- _log_header(log_fh, "🔧 工序提取(批量 --skill procedure)", model, engine)
- proc = subprocess.Popen(cmd, stdout=log_fh, stderr=subprocess.STDOUT,
- cwd=str(HERE / "procedure-dsl"), creationflags=flags) # cwd=本地 procedure-dsl(run_cyber 假设的工作目录,agent 相对路径才对)
- ACTIVE_BATCH_TASKS[q]["pid"] = proc.pid
- proc.wait()
-
- if proc.returncode == 0:
- try:
- import build_workflows
- build_workflows.write_run(q, runs_dir=RUNS_DIR)
- ACTIVE_BATCH_TASKS[q]["status"] = "success"
- except Exception as ex:
- ACTIVE_BATCH_TASKS[q]["status"] = "failed"
- ACTIVE_BATCH_TASKS[q]["error"] = f"Workflows compilation failed: {ex}"
- else:
- ACTIVE_BATCH_TASKS[q]["status"] = "failed"
- ACTIVE_BATCH_TASKS[q]["error"] = f"Batch runner exited with code {proc.returncode}"
- except Exception as ex:
- ACTIVE_BATCH_TASKS[q]["status"] = "failed"
- ACTIVE_BATCH_TASKS[q]["error"] = str(ex)
- finally:
- try:
- if batch_posts_path.is_file():
- batch_posts_path.unlink()
- except Exception:
- pass
-
- t = threading.Thread(target=run_batch_extraction)
- t.daemon = True
- t.start()
-
- self._send(200, json.dumps({
- "status": "started",
- "q": q,
- "log": f"{RUNS_DIR_NAME}/{q}/procedures/_batch_extract.log"
- }, ensure_ascii=False), "application/json")
- elif self.path == "/api/save_spec":
- length = int(self.headers.get("Content-Length") or 0)
- raw = self.rfile.read(length).decode("utf-8") if length > 0 else "{}"
- try:
- payload = json.loads(raw)
- except Exception as e:
- self._send(400, json.dumps({"error": f"bad json: {e}"}), "application/json"); return
- file_name = (payload.get("file") or "").strip()
- content = payload.get("content") or ""
- allowed = [
- "README.md",
- "tools.md",
- "extraction/phase1-skeleton.md",
- "extraction/phase2-normalize.md",
- "extraction/phase3-finalize.md",
- "taxonomy/type_suggestions.md"
- ]
- if file_name not in allowed:
- self._send(400, json.dumps({"error": "invalid file parameter"}), "application/json"); return
- target_path = HERE / "procedure-dsl" / "spec" / file_name
- try:
- target_path.parent.mkdir(parents=True, exist_ok=True)
- target_path.write_text(content, encoding="utf-8")
- self._send(200, json.dumps({"status": "ok"}, ensure_ascii=False), "application/json")
- except Exception as e:
- self._send(500, json.dumps({"error": str(e)}, ensure_ascii=False), "application/json")
- else:
- self._send(404, json.dumps({"error": "not found"}), "application/json")
- def log_message(self, *a): pass
- if __name__ == "__main__":
- n = len(scan_runs()["queries"])
- print(f"搜索评估查看 server ({MODE} 模式):http://0.0.0.0:{PORT} ({RUNS_DIR_NAME}/ 下 {n} 个 query,实时扫描)")
- ThreadingHTTPServer(("0.0.0.0", PORT), H).serve_forever()
|