333 lines
11 KiB
Python
333 lines
11 KiB
Python
#!/usr/bin/env python3
|
||
"""双语术语表事实核查脚本。
|
||
|
||
输入：
|
||
- <project>/phase4/glossary.json（来自 translate.py 累积的初版术语表）
|
||
- 可选：--extra terms.txt（每行一个英文术语，补充进来一起核查）
|
||
|
||
流程（每个术语独立可并行）：
|
||
1. 用 SearchClient（Exa > Tavily）搜一次（query = "<term> <domain hint>"）
|
||
2. 把 top 3-5 snippet 喂给 Haiku，让模型返回 {zh, en_full, confidence, issue}
|
||
3. 合并回 glossary，字段扩展：
|
||
{
|
||
"Mabwell": {
|
||
"zh": "迈威生物",
|
||
"en_full": "Mabwell (Shanghai) Bioscience Co., Ltd.",
|
||
"confidence": "high",
|
||
"issue": "...",
|
||
"verified_at": "2026-04-22",
|
||
"sources": ["https://mabwell.com/", ...]
|
||
}
|
||
}
|
||
|
||
用法：
|
||
uv run python scripts/build_glossary.py <project_slug>
|
||
uv run python scripts/build_glossary.py <project_slug> --workers 4
|
||
uv run python scripts/build_glossary.py <project_slug> --only "Mabwell,Maywavee"
|
||
uv run python scripts/build_glossary.py <project_slug> --force
|
||
|
||
断点续传：已核查过的条目（有 verified_at 字段）默认跳过；--force 全部重跑。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import concurrent.futures
|
||
import datetime as dt
|
||
import json
|
||
import sys
|
||
import time
|
||
from pathlib import Path
|
||
|
||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||
|
||
from scripts.lib.search_client import SearchClient, SearchError
|
||
from scripts.lib.zenmux_client import ZenMuxClient, ZenMuxError, load_secrets
|
||
|
||
DEFAULT_MODEL = "anthropic/claude-haiku-4.5"
|
||
PROMPT_FILE = Path(__file__).parent / "prompts" / "glossary_system.txt"
|
||
|
||
|
||
def resolve_project(arg: str) -> Path:
|
||
p = Path(arg)
|
||
if p.is_dir():
|
||
return p
|
||
cand = Path.cwd() / "projects" / arg
|
||
if cand.is_dir():
|
||
return cand
|
||
raise SystemExit(f"project not found: {arg}")
|
||
|
||
|
||
def load_glossary(path: Path) -> dict:
|
||
if not path.exists():
|
||
return {}
|
||
return json.loads(path.read_text(encoding="utf-8"))
|
||
|
||
|
||
def save_glossary(path: Path, glossary: dict) -> None:
|
||
path.parent.mkdir(parents=True, exist_ok=True)
|
||
path.write_text(
|
||
json.dumps(glossary, ensure_ascii=False, indent=2, sort_keys=True) + "\n",
|
||
encoding="utf-8",
|
||
)
|
||
|
||
|
||
def migrate_legacy_entry(value) -> dict:
|
||
"""旧版 glossary 里 value 是字符串；升级为对象格式。"""
|
||
if isinstance(value, str):
|
||
return {"zh": value}
|
||
if isinstance(value, dict):
|
||
return value
|
||
return {"zh": str(value)}
|
||
|
||
|
||
def build_query(term: str, domain: str) -> str:
|
||
if domain:
|
||
return f"{term} {domain} 中文名"
|
||
return term
|
||
|
||
|
||
def parse_json_line(text: str) -> dict:
|
||
"""模型返回的单行 JSON。容忍前后额外字符。"""
|
||
s = text.strip()
|
||
l = s.find("{")
|
||
r = s.rfind("}")
|
||
if l == -1 or r == -1:
|
||
raise ValueError(f"no JSON object: {text[:200]}")
|
||
obj = json.loads(s[l : r + 1])
|
||
if not isinstance(obj, dict):
|
||
raise ValueError("top-level JSON not object")
|
||
return obj
|
||
|
||
|
||
def build_user_prompt(term: str, domain: str, current_zh: str, hits: list) -> str:
|
||
hits_text = "\n\n".join(
|
||
f"[{i+1}] {h.title}\n URL: {h.url}\n {h.snippet[:500]}"
|
||
for i, h in enumerate(hits[:5])
|
||
)
|
||
if not hits_text:
|
||
hits_text = "(无搜索结果)"
|
||
return (
|
||
f"term: {term}\n"
|
||
f"domain: {domain or '(未指定)'}\n"
|
||
f"current_zh: {current_zh or '(空)'}\n\n"
|
||
f"search_hits:\n{hits_text}\n"
|
||
)
|
||
|
||
|
||
def verify_term(
|
||
term: str,
|
||
current_zh: str,
|
||
domain: str,
|
||
search_client: SearchClient,
|
||
llm_client: ZenMuxClient,
|
||
*,
|
||
model: str,
|
||
system_prompt: str,
|
||
) -> dict:
|
||
try:
|
||
hits = search_client.search(
|
||
build_query(term, domain),
|
||
num_results=4,
|
||
)
|
||
except SearchError as e:
|
||
return {
|
||
"zh": current_zh or "",
|
||
"en_full": term,
|
||
"confidence": "low",
|
||
"issue": f"搜索失败：{e}",
|
||
"sources": [],
|
||
}
|
||
|
||
user = build_user_prompt(term, domain, current_zh, hits)
|
||
try:
|
||
raw = llm_client.chat_complete(
|
||
model=model,
|
||
system=system_prompt,
|
||
user=user,
|
||
temperature=0.1,
|
||
max_tokens=800,
|
||
tag=f"glossary:{term[:30]}",
|
||
)
|
||
obj = parse_json_line(raw)
|
||
except (ZenMuxError, ValueError) as e:
|
||
return {
|
||
"zh": current_zh or "",
|
||
"en_full": term,
|
||
"confidence": "low",
|
||
"issue": f"LLM 核查失败：{e}",
|
||
"sources": [h.url for h in hits[:3]],
|
||
}
|
||
|
||
# 规范化输出
|
||
return {
|
||
"zh": str(obj.get("zh", current_zh or "")),
|
||
"en_full": str(obj.get("en_full", term)),
|
||
"confidence": str(obj.get("confidence", "low")),
|
||
"issue": str(obj.get("issue", "")),
|
||
"sources": [h.url for h in hits[:3]],
|
||
}
|
||
|
||
|
||
def main() -> int:
|
||
parser = argparse.ArgumentParser(description="双语术语表事实核查（Haiku + Exa）")
|
||
parser.add_argument("project", help="项目 slug 或完整路径")
|
||
parser.add_argument(
|
||
"--input",
|
||
default="phase4/glossary.json",
|
||
help="初版术语表路径（相对项目根）",
|
||
)
|
||
parser.add_argument(
|
||
"--output",
|
||
default="phase4/glossary.json",
|
||
help="输出路径（默认原地覆盖）",
|
||
)
|
||
parser.add_argument("--model", default=DEFAULT_MODEL, help="LLM 模型")
|
||
parser.add_argument(
|
||
"--workers", type=int, default=4, help="并发度（默认 4；网络不稳时建议降到 3）"
|
||
)
|
||
parser.add_argument(
|
||
"--force", action="store_true",
|
||
help="忽略已核查状态，全部重跑",
|
||
)
|
||
parser.add_argument(
|
||
"--only",
|
||
default=None,
|
||
help="只核查指定术语（逗号分隔，大小写敏感）",
|
||
)
|
||
parser.add_argument(
|
||
"--extra",
|
||
default=None,
|
||
help="每行一个英文术语的文本文件，补充进术语表一起核查",
|
||
)
|
||
parser.add_argument(
|
||
"--domain",
|
||
default=None,
|
||
help="术语领域提示（默认自动读 manifest.topic）",
|
||
)
|
||
args = parser.parse_args()
|
||
|
||
load_secrets()
|
||
project_root = resolve_project(args.project)
|
||
input_path = project_root / args.input
|
||
output_path = project_root / args.output
|
||
|
||
manifest_path = project_root / "manifest.json"
|
||
domain = args.domain
|
||
if not domain and manifest_path.exists():
|
||
m = json.loads(manifest_path.read_text(encoding="utf-8"))
|
||
domain = m.get("topic") or m.get("report_title") or ""
|
||
|
||
glossary = load_glossary(input_path)
|
||
# migration
|
||
for k, v in list(glossary.items()):
|
||
glossary[k] = migrate_legacy_entry(v)
|
||
|
||
if args.extra:
|
||
extra_path = Path(args.extra)
|
||
if not extra_path.exists():
|
||
raise SystemExit(f"--extra 文件不存在：{extra_path}")
|
||
for line in extra_path.read_text(encoding="utf-8").splitlines():
|
||
term = line.strip()
|
||
if term and term not in glossary:
|
||
glossary[term] = {"zh": ""}
|
||
|
||
only_terms: set[str] | None = None
|
||
if args.only:
|
||
only_terms = {t.strip() for t in args.only.split(",") if t.strip()}
|
||
|
||
system_prompt = PROMPT_FILE.read_text(encoding="utf-8")
|
||
|
||
# 筛选需要核查的
|
||
todo: list[str] = []
|
||
for term, entry in glossary.items():
|
||
if only_terms is not None and term not in only_terms:
|
||
continue
|
||
if not args.force and entry.get("verified_at"):
|
||
continue
|
||
todo.append(term)
|
||
|
||
print(f"Project: {project_root.name}")
|
||
print(f"Domain hint: {domain or '(none)'}")
|
||
print(f"Glossary size: {len(glossary)} | to verify: {len(todo)} | workers: {args.workers}")
|
||
print(f"Model: {args.model}")
|
||
if not todo:
|
||
print(" 没有需要核查的条目（使用 --force 强制重跑）")
|
||
save_glossary(output_path, glossary)
|
||
return 0
|
||
|
||
today = dt.date.today().isoformat()
|
||
logs_dir = project_root / "phase4" / "logs"
|
||
log_file = logs_dir / "glossary.jsonl"
|
||
|
||
start = time.time()
|
||
done_count = 0
|
||
failed: list[str] = []
|
||
|
||
# 并发执行：SearchClient/ZenMuxClient 都是 thread-safe（httpx.Client 支持）
|
||
with SearchClient() as search_client, ZenMuxClient(log_file=log_file) as llm_client:
|
||
|
||
def worker(term: str) -> tuple[str, dict]:
|
||
current_zh = glossary.get(term, {}).get("zh", "")
|
||
result = verify_term(
|
||
term,
|
||
current_zh,
|
||
domain or "",
|
||
search_client,
|
||
llm_client,
|
||
model=args.model,
|
||
system_prompt=system_prompt,
|
||
)
|
||
return term, result
|
||
|
||
with concurrent.futures.ThreadPoolExecutor(max_workers=args.workers) as pool:
|
||
futures = {pool.submit(worker, term): term for term in todo}
|
||
for fut in concurrent.futures.as_completed(futures):
|
||
term = futures[fut]
|
||
try:
|
||
_term, result = fut.result()
|
||
except Exception as e:
|
||
print(f" [FAIL] {term}: {e}")
|
||
failed.append(term)
|
||
continue
|
||
# 合并到 glossary
|
||
old = glossary.get(term, {})
|
||
old.update(result)
|
||
old["verified_at"] = today
|
||
glossary[term] = old
|
||
done_count += 1
|
||
conf = result.get("confidence", "?")
|
||
issue = result.get("issue", "")
|
||
zh = result.get("zh") or "(保留英文)"
|
||
marker = {"high": "✓", "medium": "~", "low": "?"}.get(conf, " ")
|
||
issue_str = f" ⚠ {issue[:80]}" if issue else ""
|
||
print(f" [{marker}] {term:<35} → {zh}{issue_str}")
|
||
# 阶段性存盘，避免中途挂掉丢数据
|
||
if done_count % 10 == 0:
|
||
save_glossary(output_path, glossary)
|
||
|
||
save_glossary(output_path, glossary)
|
||
|
||
elapsed = time.time() - start
|
||
print(f"\n完成：{done_count}/{len(todo)}（失败 {len(failed)}，耗时 {elapsed:.1f}s）")
|
||
print(f"术语表：{output_path.relative_to(project_root)}")
|
||
print(llm_client.usage.summary())
|
||
|
||
# 高警示项汇总
|
||
issues = [
|
||
(k, v) for k, v in glossary.items()
|
||
if v.get("issue") and v.get("confidence") != "high"
|
||
]
|
||
if issues:
|
||
print(f"\n⚠ 低置信度或带问题的条目（{len(issues)} 条）：")
|
||
for k, v in issues[:15]:
|
||
print(f" - {k:<30} [{v.get('confidence','?')}] {v.get('issue','')[:100]}")
|
||
if len(issues) > 15:
|
||
print(f" … 还有 {len(issues) - 15} 条，见 {output_path.name}")
|
||
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|