123 lines
5.0 KiB
Python
123 lines
5.0 KiB
Python
|
#!/usr/bin/env python3
|
|||
|
|
"""scripts/fetch_format_rules.py —— 格式规则自动获取（第一批，人工核验用 CLI）。
|
|||
|
|
|
|||
|
|
用法（在 review_service/ 下）：
|
|||
|
|
/home/xzq/5090/teamWorkspace/zh/plantai-rag-new-venv/bin/python \
|
|||
|
|
scripts/fetch_format_rules.py <journal_id> <guidelines_url>
|
|||
|
|
|
|||
|
|
流程：
|
|||
|
|
1. fetch_submission_guidelines(url) # robots.txt + 同域 ≥2s 限速 + 正文转 Markdown
|
|||
|
|
2. extract_format_rules(markdown, journal_name) # DeepSeek 抽取（llm.py）
|
|||
|
|
3. 存 data/format_cache/{journal_id}.json（失败也存，便于批量对账）
|
|||
|
|
4. 打印完整结果 JSON 供人工核验
|
|||
|
|
|
|||
|
|
凭证：结果文件顶层 source_url（步骤 1 的实际抓取地址）+ source_quotes（LLM
|
|||
|
|
逐字引证，按字段分组）；非 null 字段缺引证 → 抽取判失败（ok=false）。
|
|||
|
|
|
|||
|
|
约束（第一批）：只跑指定 3 本；不触 API / 前端 / journals.json（format_rules 只读
|
|||
|
|
查名，不写回）。journal_name 优先取 data/journals/journals.json 的 name（只读）。
|
|||
|
|
"""
|
|||
|
|
from __future__ import annotations
|
|||
|
|
|
|||
|
|
import asyncio
|
|||
|
|
import json
|
|||
|
|
import os
|
|||
|
|
import re
|
|||
|
|
import sys
|
|||
|
|
from datetime import datetime, timezone
|
|||
|
|
from pathlib import Path
|
|||
|
|
|
|||
|
|
_ROOT = Path(__file__).resolve().parents[1] # review_service/
|
|||
|
|
sys.path.insert(0, str(_ROOT))
|
|||
|
|
|
|||
|
|
_JOURNALS_JSON = _ROOT / "data" / "journals" / "journals.json"
|
|||
|
|
_CACHE_DIR = _ROOT / "data" / "format_cache"
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _load_env() -> None:
|
|||
|
|
"""镜像 restart.sh 的 `set -a; source .env`：REVIEW_LLM_*/DOCGEN_LLM_* 裸键
|
|||
|
|
由 services/llm.py 在每次调用时读 os.environ（不缓存），故调用前灌入即可。"""
|
|||
|
|
env_path = _ROOT / ".env"
|
|||
|
|
if not env_path.is_file():
|
|||
|
|
return
|
|||
|
|
for raw in env_path.read_text("utf-8").splitlines():
|
|||
|
|
line = raw.strip()
|
|||
|
|
if not line or line.startswith("#") or "=" not in line:
|
|||
|
|
continue
|
|||
|
|
if line.startswith("export "):
|
|||
|
|
line = line[7:].strip()
|
|||
|
|
key, _, val = line.partition("=")
|
|||
|
|
os.environ.setdefault(key.strip(), val.strip().strip("\"'"))
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _journal_name(journal_id: str) -> str:
|
|||
|
|
"""journals.json 只读查名；查不到 → id 连字符转空格并首字母大写（LLM 上下文可读）。"""
|
|||
|
|
try:
|
|||
|
|
data = json.loads(_JOURNALS_JSON.read_text("utf-8"))
|
|||
|
|
for j in data.get("journals") or []:
|
|||
|
|
if j.get("id") == journal_id:
|
|||
|
|
return str(j.get("name") or journal_id)
|
|||
|
|
except Exception: # noqa: BLE001
|
|||
|
|
pass
|
|||
|
|
return journal_id.replace("-", " ").title()
|
|||
|
|
|
|||
|
|
|
|||
|
|
def main() -> int:
|
|||
|
|
if len(sys.argv) != 3:
|
|||
|
|
print("用法: fetch_format_rules.py <journal_id> <guidelines_url>", file=sys.stderr)
|
|||
|
|
return 2
|
|||
|
|
journal_id = re.sub(r"[^\w.-]", "_", sys.argv[1].strip())
|
|||
|
|
url = sys.argv[2].strip()
|
|||
|
|
_load_env()
|
|||
|
|
|
|||
|
|
from app.services.format_extractor import extract_format_rules
|
|||
|
|
from app.services.format_fetcher import fetch_submission_guidelines
|
|||
|
|
|
|||
|
|
async def run() -> tuple[dict, dict]:
|
|||
|
|
fetched = await fetch_submission_guidelines(url)
|
|||
|
|
print(f"[1/2] 抓取 {url} → ok={fetched['ok']}"
|
|||
|
|
+ (f"，markdown={len(fetched['markdown'])} 字符"
|
|||
|
|
if fetched["ok"] else f"，error={fetched['error']}"))
|
|||
|
|
if not fetched["ok"]:
|
|||
|
|
return fetched, {"ok": False, "error": f"抓取失败：{fetched['error']}",
|
|||
|
|
"rules": None, "source_quotes": None,
|
|||
|
|
"confidence": None, "notes": None, "raw": None}
|
|||
|
|
extracted = await extract_format_rules(fetched["markdown"], _journal_name(journal_id))
|
|||
|
|
if extracted["ok"]:
|
|||
|
|
filled = sum(1 for v in extracted["rules"].values() if v is not None)
|
|||
|
|
print(f"[2/2] LLM 抽取 → ok=True，8 字段中 {filled} 个非 null")
|
|||
|
|
else:
|
|||
|
|
print(f"[2/2] LLM 抽取 → ok=False，error={extracted['error']}")
|
|||
|
|
return fetched, extracted
|
|||
|
|
|
|||
|
|
fetched, extracted = asyncio.run(run())
|
|||
|
|
|
|||
|
|
result = {
|
|||
|
|
"journal_id": journal_id,
|
|||
|
|
"journal_name": _journal_name(journal_id),
|
|||
|
|
"source_url": url,
|
|||
|
|
"fetched_at": fetched.get("fetched_at"),
|
|||
|
|
"extracted_at": datetime.now(timezone.utc).isoformat(timespec="seconds"),
|
|||
|
|
"markdown_chars": len(fetched.get("markdown") or ""),
|
|||
|
|
"ok": bool(extracted.get("ok")),
|
|||
|
|
"error": extracted.get("error"),
|
|||
|
|
"format_rules": ({**extracted["rules"], "rules_source": "fetched"}
|
|||
|
|
if extracted.get("ok") else None),
|
|||
|
|
"source_quotes": extracted.get("source_quotes"),
|
|||
|
|
"confidence": extracted.get("confidence"),
|
|||
|
|
"notes": extracted.get("notes"),
|
|||
|
|
"raw_llm": extracted.get("raw"),
|
|||
|
|
}
|
|||
|
|
_CACHE_DIR.mkdir(parents=True, exist_ok=True)
|
|||
|
|
out_path = _CACHE_DIR / f"{journal_id}.json"
|
|||
|
|
out_path.write_text(json.dumps(result, ensure_ascii=False, indent=2) + "\n", "utf-8")
|
|||
|
|
|
|||
|
|
print(f"[save] {out_path.relative_to(_ROOT)}")
|
|||
|
|
print("===== 完整结果（人工核验）=====")
|
|||
|
|
print(json.dumps(result, ensure_ascii=False, indent=2))
|
|||
|
|
return 0 if result["ok"] else 1
|
|||
|
|
|
|||
|
|
|
|||
|
|
if __name__ == "__main__":
|
|||
|
|
sys.exit(main())
|