potaser
55410b663a
- fetch_format_rules.py CLI: robots.txt-aware fetcher (httpx + bs4, >=2s per-host throttle) + DeepSeek quote-backed rule extractor - Status: OUP/Wiley/Cell return 403 from this environment; the five featured journals' format_rules in journals.json were manually verified against official sites instead. Tooling kept for future re-runs; data/format_cache/ is runtime output and gitignored - Service code never imports this toolchain (standalone CLI only)
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())
|