feat(review): journal format-rule fetch scripts (offline CLI)

- 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)

.gitignore

@@ -53,3 +53,4 @@ review_service/data/reformat_jobs/
 !review_service/data/
 review_service/tools/tectonic
 review_service/data/tectonic-cache/
+review_service/data/format_cache/

review_service/app/services/format_extractor.py

@@ -0,0 +1,232 @@
+"""services/format_extractor.py —— 投稿指南 → 格式规则 JSON（LLM 抽取，第一批）。
+
+- 调 services/llm.py 的 complete_json（DeepSeek json_object + 修复重试，零新依赖）
+- Prompt = 「格式规则抽取器」模板：目标字段与 journals.json format_rules_default
+  同构（max_pages / min_font_pt / min_margin_mm / min_line_spacing / min_figures /
+  max_fonts / refs_style / anonymous），强制每字段带 source_quotes + confidence
+- 诚实原则：正文没写 → null，绝不编造（与 format_check 诚实原则一致）
+- 解析/校验失败 → {"ok": False, "error": ...}
+
+返回：
+  成功 {"ok": True, "rules": {...}, "source_quotes": {...},
+        "confidence": {...}, "notes": str, "raw": LLM原始JSON, "error": None}
+  失败 {"ok": False, "error": "...", 其余键为 None}
+source_url 由调用方（scripts/fetch_format_rules.py）在落盘时补入结果文件。
+"""
+from __future__ import annotations
+
+import logging
+import re
+from typing import Any
+
+from .llm import LLMError, complete_json
+
+logger = logging.getLogger(__name__)
+
+# 「格式规则抽取器」模板（system prompt）
+_SYSTEM = """你是「格式规则抽取器」：从期刊官方投稿指南的 Markdown 正文中抽取格式规则，供投稿排版检查引擎使用。
+
+# 目标字段（rules 对象，严格用以下字段名）
+- max_pages：正文页数上限（整数；原文无明确上限 → null）
+- min_font_pt：最小字号（数字，单位 pt；原文若用 cm/mm/inch 给出则换算为 pt）
+- min_margin_mm：最小页边距（数字，单位 mm；原文若用 inch 给出则 ×25.4 换算）
+- min_line_spacing：最小行距倍数（数字；如 "double-spaced" → 2.0，"1.5 spacing" → 1.5）
+- min_figures：研究论文最少图数（整数；无明确要求 → null）
+- max_fonts：全文最多字体种类（整数；无明确要求 → null）
+- refs_style：参考文献样式，枚举 apa | gb7714 | author_year | numbered | vancouver | mla | chicago | other（无法判断 → null）
+- anonymous：是否要求匿名/双盲评审（true / false / null）
+
+# 凭证（强制）
+- 每个非 null 字段必须给 source_quotes[字段名]：从正文【逐字】摘录的 1-3 句原文，
+  不改写、不翻译、不丢掉含关键数字的句子。
+- source_quotes["_general"]：整体最有凭证价值的 1-3 条原文摘录。
+- confidence：{"overall": 0-1, "per_field": {"字段名": 0-1}}。
+  直接原文命中 ≥0.9；单位换算 0.6-0.9；靠上下文推断 <0.6。
+
+# 诚实原则
+- 正文没写到的字段一律 null，绝不用常识或训练记忆补值。
+- 单位换算允许做，但该字段 confidence 不得超过 0.9。
+- 换算过程、拿不准的点、页面疑似不完整，写进 notes。
+
+# 输出
+只输出一个 JSON 对象，不要代码块、不要解释文字：
+{"rules": {"max_pages": null, "min_font_pt": null, "min_margin_mm": null,
+  "min_line_spacing": null, "min_figures": null, "max_fonts": null,
+  "refs_style": null, "anonymous": null},
+ "source_quotes": {"字段名": ["原文1", "..."], "_general": ["原文1", "..."]},
+ "confidence": {"overall": 0.0, "per_field": {"字段名": 0.0}},
+ "notes": "一句话说明抽取依据与不确定点"}"""
+
+_USER_TMPL = """期刊：{journal_name}
+
+=== 官方投稿指南正文（Markdown，已截断） ===
+{markdown}"""
+
+_RULE_INT_KEYS = ("max_pages", "min_figures", "max_fonts")
+_REFS_ALLOWED = {"apa", "gb7714", "author_year", "numbered", "vancouver",
+                 "mla", "chicago", "other"}
+_REFS_ALIAS = {
+    "apa-7": "apa", "apa7": "apa", "apa 7th": "apa", "apa_7th": "apa",
+    "harvard": "author_year", "author-year": "author_year", "authoryear": "author_year",
+    "gb/t 7714": "gb7714", "gb/t-7714": "gb7714", "gb7714-2015": "gb7714",
+    "gb/t7714": "gb7714", "chicago author-date": "chicago",
+}
+
+
+# 字符串值带显式单位时的目标单位换算表（LLM 若回传原文单位的兜底；无单位=数值即目标单位）
+_UNIT_TO_MM = {"mm": 1.0, "cm": 10.0, "in": 25.4, "inch": 25.4, "inches": 25.4, "pt": 25.4 / 72}
+_UNIT_TO_PT = {"pt": 1.0, "mm": 72 / 25.4, "cm": 72 / 2.54, "in": 72.0, "inch": 72.0, "inches": 72.0}
+
+
+def _to_int(v: Any) -> int | None:
+    if isinstance(v, bool) or v is None:
+        return None
+    if isinstance(v, int):
+        return v
+    if isinstance(v, float):
+        return int(round(v))
+    if isinstance(v, str):
+        m = re.search(r"-?\d+", v)
+        if m:
+            return int(m.group(0))
+    return None
+
+
+def _to_float(v: Any, units: dict[str, float] | None = None) -> float | None:
+    if isinstance(v, bool) or v is None:
+        return None
+    if isinstance(v, (int, float)):
+        return float(v)
+    if isinstance(v, str):
+        m = re.search(r"(-?\d+(?:\.\d+)?)\s*([a-zA-Z]+)?", v)
+        if not m:
+            return None
+        x = float(m.group(1))
+        unit = (m.group(2) or "").lower()
+        if units and unit:
+            factor = units.get(unit)
+            if factor is None:
+                return None  # 未知单位不猜
+            x *= factor
+        return x
+    return None
+
+
+def _to_bool(v: Any) -> bool | None:
+    if isinstance(v, bool):
+        return v
+    if isinstance(v, (int, float)) and not isinstance(v, bool):
+        if v in (0, 1):
+            return bool(v)
+    if isinstance(v, str):
+        low = v.strip().lower()
+        if low in ("true", "yes", "required"):
+            return True
+        if low in ("false", "no", "not required"):
+            return False
+    return None
+
+
+def _norm_refs(v: Any) -> str | None:
+    if not isinstance(v, str) or not v.strip():
+        return None
+    low = v.strip().lower()
+    low = _REFS_ALIAS.get(low, low)
+    return low if low in _REFS_ALLOWED else "other"
+
+
+def _norm_conf(v: Any) -> float | None:
+    if isinstance(v, bool) or not isinstance(v, (int, float)):
+        return None
+    return max(0.0, min(1.0, float(v)))
+
+
+def _normalize(raw: dict[str, Any]) -> dict[str, Any]:
+    """白名单归一：类型错的值置 null（宁缺勿错，raw 全量保留在结果里）。"""
+    src = raw.get("rules")
+    src = src if isinstance(src, dict) else {}
+    rules: dict[str, Any] = {}
+    for k in _RULE_INT_KEYS:
+        rules[k] = _to_int(src.get(k))
+    rules["min_font_pt"] = _to_float(src.get("min_font_pt"), _UNIT_TO_PT)
+    rules["min_margin_mm"] = _to_float(src.get("min_margin_mm"), _UNIT_TO_MM)
+    rules["min_line_spacing"] = _to_float(src.get("min_line_spacing"))
+    rules["refs_style"] = _norm_refs(src.get("refs_style"))
+    rules["anonymous"] = _to_bool(src.get("anonymous"))
+
+    sq_src = raw.get("source_quotes")
+    sq_src = sq_src if isinstance(sq_src, dict) else {}
+    quotes: dict[str, list[str]] = {}
+    for key, val in sq_src.items():
+        if key != "_general" and key not in rules:
+            continue
+        items = val if isinstance(val, list) else [val]
+        cleaned = [str(x).strip() for x in items if isinstance(x, (str, int, float))
+                   and str(x).strip()]
+        if cleaned:
+            quotes[key] = cleaned
+
+    conf_src = raw.get("confidence")
+    conf_src = conf_src if isinstance(conf_src, dict) else {}
+    pf_src = conf_src.get("per_field")
+    pf_src = pf_src if isinstance(pf_src, dict) else {}
+    per_field: dict[str, float] = {}
+    for k, v in pf_src.items():
+        c = _norm_conf(v)
+        if c is not None and (k in rules or k == "_general"):
+            per_field[k] = c
+    overall = _norm_conf(conf_src.get("overall"))
+    if overall is None and per_field:
+        overall = sum(per_field.values()) / len(per_field)
+    confidence = {"overall": overall, "per_field": per_field}
+
+    notes = raw.get("notes")
+    return {
+        "rules": rules,
+        "source_quotes": quotes,
+        "confidence": confidence,
+        "notes": str(notes) if notes else "",
+    }
+
+
+def _validation_error(norm: dict[str, Any]) -> str | None:
+    """强制凭证：任一非 null 字段缺 source_quotes → 拒收（ok=False）。"""
+    missing = [k for k, v in norm["rules"].items()
+               if v is not None and k not in norm["source_quotes"]]
+    if missing:
+        return f"LLM 输出缺少非 null 字段的 source_quotes：{', '.join(missing)}"
+    if not norm["source_quotes"].get("_general"):
+        return "LLM 输出缺少 source_quotes._general 整体引证"
+    if any(v is not None for v in norm["rules"].values()):
+        return None
+    return "LLM 未抽出任何非 null 字段（rules 全空）"
+
+
+async def extract_format_rules(markdown: str, journal_name: str) -> dict:
+    """markdown（≤30000 字符）+ 期刊名 → 抽取结果（schema 见模块 docstring）。"""
+    if not markdown or not markdown.strip():
+        return {"ok": False, "error": "markdown 为空", "rules": None,
+                "source_quotes": None, "confidence": None, "notes": None, "raw": None}
+    user = _USER_TMPL.format(journal_name=journal_name, markdown=markdown)
+    try:
+        raw = await complete_json(_SYSTEM, user, temperature=0.1, max_tokens=8000)
+    except LLMError as exc:
+        return {"ok": False, "error": f"LLM 调用失败：{exc}", "rules": None,
+                "source_quotes": None, "confidence": None, "notes": None, "raw": None}
+    if not isinstance(raw, dict):
+        return {"ok": False, "error": f"LLM 输出非 JSON 对象：{type(raw).__name__}",
+                "rules": None, "source_quotes": None, "confidence": None,
+                "notes": None, "raw": None}
+
+    norm = _normalize(raw)
+    err = _validation_error(norm)
+    if err:
+        logger.warning("[extract] %s 校验失败：%s", journal_name, err)
+        return {"ok": False, "error": err, "rules": None,
+                "source_quotes": norm["source_quotes"], "confidence": norm["confidence"],
+                "notes": norm["notes"], "raw": raw}
+    filled = sum(1 for v in norm["rules"].values() if v is not None)
+    logger.info("[extract] %s → 8 字段中 %d 个非 null", journal_name, filled)
+    return {"ok": True, "error": None, "rules": norm["rules"],
+            "source_quotes": norm["source_quotes"],
+            "confidence": norm["confidence"], "notes": norm["notes"], "raw": raw}

review_service/app/services/format_fetcher.py

@@ -0,0 +1,356 @@
+"""services/format_fetcher.py —— 投稿指南页面抓取（格式规则自动获取·第一批）。
+
+约束（用户拍板 2026-10-10）：
+- 零新依赖：httpx + beautifulsoup4（venv 已装 4.15.0，用户明确指定使用；
+  lxml 6.1.2 在 requirements.lock 内，作 bs4 解析器）
+- 同域限速 ≥ 2s（robots.txt 与正文请求共用计时器）
+- 抓正文前先 GET /robots.txt，命中 Disallow → ok=False 跳过
+- UA: "PlantAI-RAG-Bot/1.0 (+contact: admin@example.com)"
+- 只提取正文（去导航/页脚/脚本），截断至 30000 字符
+
+返回固定四键：{"ok": bool, "markdown": str, "error": str | None, "fetched_at": str}
+"""
+from __future__ import annotations
+
+import asyncio
+import logging
+import re
+import time
+from datetime import datetime, timezone
+from urllib.parse import urlparse
+
+import httpx
+from bs4 import BeautifulSoup, NavigableString
+
+logger = logging.getLogger(__name__)
+
+USER_AGENT = "PlantAI-RAG-Bot/1.0 (+contact: admin@example.com)"
+TIMEOUT = 30.0
+MAX_CHARS = 30_000
+MIN_INTERVAL_PER_HOST = 2.0  # 同域请求间隔 ≥ 2s（robots 与正文共用）
+_MIN_TEXT_CHARS = 200        # 正文短于此 → 疑似 JS 渲染/反爬页
+_HEADERS = {
+    "User-Agent": USER_AGENT,
+    "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
+    "Accept-Language": "en-US,en;q=0.9",
+}
+
+# 同域限速：host → 上次请求完成时刻（monotonic）
+_last_hit: dict[str, float] = {}
+_throttle_lock = asyncio.Lock()
+
+
+def _now_iso() -> str:
+    return datetime.now(timezone.utc).isoformat(timespec="seconds")
+
+
+def _fail(error: str) -> dict:
+    return {"ok": False, "markdown": "", "error": error, "fetched_at": _now_iso()}
+
+
+async def _throttle(host: str) -> None:
+    """同域串行 + ≥2s 间隔（全局锁：本批抓取量小，简单可靠优先）。"""
+    async with _throttle_lock:
+        now = time.monotonic()
+        wait = _last_hit.get(host, now) + MIN_INTERVAL_PER_HOST - now
+        if wait > 0:
+            await asyncio.sleep(wait)
+        _last_hit[host] = time.monotonic()
+
+
+# ---------------------------------------------------------------- robots.txt
+
+def _parse_robots(text: str) -> list[tuple[list[str], list[tuple[str, str]]]]:
+    """→ [(agents, [(kind, pattern)...]), ...]（连续 UA 行同组；规则行后出现 UA=新组）。"""
+    groups: list[tuple[list[str], list[tuple[str, str]]]] = []
+    agents: list[str] = []
+    rules: list[tuple[str, str]] = []
+    for raw in text.splitlines():
+        line = raw.split("#", 1)[0].strip()
+        if not line or ":" not in line:
+            continue
+        key, _, value = line.partition(":")
+        key, value = key.strip().lower(), value.strip()
+        if key == "user-agent":
+            if agents and rules:  # 上一组已出规则 → 当前 UA 行开新组
+                groups.append((agents, rules))
+                agents, rules = [], []
+            agents.append(value)
+        elif key in ("allow", "disallow"):
+            if not agents:  # 无前置 UA 的规则按 * 组处理
+                agents = ["*"]
+            rules.append((key, value))
+    if agents:
+        groups.append((agents, rules))
+    return groups
+
+
+def _robots_path_allowed(groups, path: str) -> tuple[bool, str]:
+    """最长匹配决胜（同长 Allow 优先；`$` 结尾=精确匹配；空 Disallow 跳过）。"""
+
+    def matched_by(pat: str) -> bool:
+        if pat.endswith("$"):
+            return path == pat[:-1]
+        return path.startswith(pat)
+
+    token = "plantai-rag-bot"
+    matched = [rules for agents, rules in groups
+               if any(token in a.lower() for a in agents)]
+    if not matched:
+        matched = [rules for agents, rules in groups
+                   if any(a.lower() == "*" for a in agents)]
+    if not matched:
+        return True, "robots.txt 无匹配组（本 bot 不受限）"
+
+    best_len, best_allow, best_pat = -1, False, ""
+    for rules in matched:
+        for kind, pat in rules:
+            if not pat:  # "Disallow:" 空值 = 放行一切（该组无需处理）
+                continue
+            if not matched_by(pat):
+                continue
+            length = len(pat)
+            allow = kind == "allow"
+            if length > best_len or (length == best_len and allow and not best_allow):
+                best_len, best_allow, best_pat = length, allow, pat
+    if best_len < 0:
+        return True, "robots.txt 允许（无路径匹配）"
+    if best_allow:
+        return True, f"robots.txt Allow 命中（{best_pat}）"
+    return False, f"robots.txt Disallow 命中（{best_pat}）"
+
+
+async def _robots_allows(client: httpx.AsyncClient, url: str) -> tuple[bool, str]:
+    p = urlparse(url)
+    robots_url = f"{p.scheme}://{p.netloc}/robots.txt"
+    await _throttle(p.netloc.lower())
+    try:
+        resp = await client.get(robots_url, headers=_HEADERS)
+    except httpx.HTTPError as exc:
+        return True, f"robots.txt 获取失败（{type(exc).__name__}），按允许处理"
+    if resp.status_code != 200:
+        return True, f"robots.txt HTTP {resp.status_code}，按允许处理"
+    groups = _parse_robots(resp.text)
+    return _robots_path_allowed(groups, p.path or "/")
+
+
+# ---------------------------------------------------------------- 正文提取
+
+_STRIP_TAGS = ("script", "style", "noscript", "iframe", "form", "nav",
+               "footer", "aside", "button", "select", "textarea", "input", "svg")
+_STRIP_ROLES = ("navigation", "contentinfo", "complementary", "search")
+# 正文容器候选（按优先级；首个文本 ≥500 字符者胜出）
+_CONTENT_SELECTORS = (
+    "main", "article", '[role="main"]', "#content", "#main", "#main-content",
+    ".article-body", "#article-body", ".article__body", ".hub-article-body",
+    ".page-content", ".main-content", ".content-area",
+)
+_CHALLENGE_MARKERS = ("just a moment", "attention required", "captcha",
+                      "access denied", "ddos protection", "verify you are human")
+
+
+def _decode_html(resp: httpx.Response) -> str:
+    """header charset → <meta charset> → utf-8（errors=replace），防 OUP/Wiley 无 header 乱码。"""
+    enc = None
+    m = re.search(r"charset=([\w.-]+)", resp.headers.get("content-type", ""), re.I)
+    if m:
+        enc = m.group(1)
+    if not enc:
+        m = re.search(rb"""charset\s*=\s*["']?([\w.-]+)""", resp.content[:4096], re.I)
+        if m:
+            enc = m.group(1).decode("ascii", "ignore")
+    try:
+        return resp.content.decode(enc or "utf-8", errors="replace")
+    except LookupError:
+        return resp.content.decode("utf-8", errors="replace")
+
+
+def _squeeze(s: str) -> str:
+    """行内空白压缩 + 去空行（保留单换行）。"""
+    lines = [re.sub(r"[ \t\r\f\v]+", " ", line).strip() for line in s.split("\n")]
+    return "\n".join(line for line in lines if line)
+
+
+def _inline(el) -> str:
+    """行内渲染：a→md链接 / strong→** / em→* / code→` / br→换行；跳过列表与图片。"""
+    parts: list[str] = []
+    for child in el.children:
+        name = getattr(child, "name", None)
+        if name is None:  # NavigableString
+            parts.append(str(child))
+            continue
+        if name in ("ul", "ol", "img", "picture", "figure"):
+            continue
+        if name == "a":
+            href = (child.get("href") or "").strip()
+            txt = _inline(child).strip()
+            if not txt:
+                continue
+            parts.append(f"[{txt}]({href})" if href.startswith(("http://", "https://")) else txt)
+        elif name in ("strong", "b"):
+            parts.append(f"**{_inline(child)}**")
+        elif name in ("em", "i"):
+            parts.append(f"*{_inline(child)}*")
+        elif name == "code":
+            parts.append(f"`{_inline(child).strip()}`")
+        elif name == "br":
+            parts.append("\n")
+        else:
+            parts.append(_inline(child))
+    return "".join(parts)
+
+
+def _list_md(el, ordered: bool) -> str:
+    lines: list[str] = []
+    idx = 1
+    for li in el.find_all("li", recursive=False):
+        # li 文本不含嵌套列表（嵌套列表随后缩进输出）
+        buf = []
+        for child in li.children:
+            if getattr(child, "name", None) in ("ul", "ol"):
+                continue
+            buf.append(_inline(child) if getattr(child, "name", None) else str(child))
+        txt = " ".join("".join(buf).split())
+        if txt:
+            lines.append(f"{idx}. {txt}" if ordered else f"- {txt}")
+            idx += 1
+        for nested in li.find_all(("ul", "ol"), recursive=False):
+            sub = _list_md(nested, nested.name == "ol")
+            if sub:
+                lines.extend("  " + line for line in sub.split("\n"))
+    return "\n".join(lines)
+
+
+def _table_md(table) -> str | None:
+    rows: list[list[str]] = []
+    for tr in table.find_all("tr"):
+        cells = [" ".join(_squeeze(_inline(c)).split())
+                 for c in tr.find_all(("td", "th"))]
+        if cells:
+            rows.append(cells)
+    if not rows:
+        return None
+    out = ["| " + " | ".join(rows[0]) + " |"]
+    if table.find("th"):  # 首行含表头 → 补分隔行（合法 GFM 表）
+        out.append("| " + " | ".join(["---"] * len(rows[0])) + " |")
+    out += ["| " + " | ".join(r) + " |" for r in rows[1:]]
+    return "\n".join(out)
+
+
+def _blocks(el, out: list[str]) -> None:
+    for child in el.children:
+        name = getattr(child, "name", None)
+        if name is None:  # 裸文本
+            txt = _squeeze(str(child))
+            if txt:
+                out.append(txt)
+            continue
+        if name in _STRIP_TAGS:
+            continue
+        if name in ("h1", "h2", "h3", "h4", "h5", "h6"):
+            txt = " ".join(_squeeze(_inline(child)).split())
+            if txt:
+                out.append("#" * int(name[1]) + " " + txt)
+        elif name == "p":
+            txt = _squeeze(_inline(child))
+            if txt:
+                out.append(txt)
+        elif name in ("ul", "ol"):
+            md = _list_md(child, name == "ol")
+            if md:
+                out.append(md)
+        elif name == "table":
+            md = _table_md(child)
+            if md:
+                out.append(md)
+        elif name == "blockquote":
+            txt = _squeeze(_inline(child))
+            if txt:
+                out.append("\n".join("> " + line for line in txt.split("\n")))
+        elif name == "pre":
+            code = child.get_text().strip("\n")
+            if code.strip():
+                out.append("```\n" + code + "\n```")
+        elif name == "hr":
+            out.append("---")
+        else:  # div/section/article/main/figure/dl/details/… 容器递归
+            _blocks(child, out)
+
+
+def _html_to_markdown(html: str) -> tuple[str, str | None]:
+    """→ (markdown, error)。选正文容器 → 去壳 → DOM 递归转 Markdown。"""
+    soup = BeautifulSoup(html, "lxml")  # lxml 在 lock 内；缺时 bs4 自动回退 html.parser
+
+    title = (soup.title.get_text() if soup.title else "") or ""
+    for tag in soup.find_all(_STRIP_TAGS):
+        tag.decompose()
+    for tag in soup.find_all(attrs={"role": _STRIP_ROLES}):
+        tag.decompose()
+
+    container = None
+    fallback_hits = []
+    for sel in _CONTENT_SELECTORS:
+        for node in soup.select(sel):
+            text_len = len(node.get_text(" ", strip=True))
+            fallback_hits.append((text_len, node))
+            if text_len >= 500:
+                container = node
+                break
+        if container is not None:
+            break
+    if container is None:
+        if fallback_hits:
+            container = max(fallback_hits, key=lambda x: x[0])[1]
+        else:
+            container = soup.body or soup
+
+    blocks: list[str] = []
+    _blocks(container, blocks)
+    md = "\n\n".join(blocks)
+
+    probe = (title + " " + md[:500]).lower()
+    marker = next((m for m in _CHALLENGE_MARKERS if m in probe), None)
+    if marker:
+        return "", f"疑似反爬/挑战页（命中 {marker!r}，title={title[:80]!r}）"
+    if len(md) < _MIN_TEXT_CHARS:
+        return "", f"正文提取过短（{len(md)} < {_MIN_TEXT_CHARS} 字符），疑似 JS 渲染或空壳页"
+    if len(md) > MAX_CHARS:
+        md = md[:MAX_CHARS - 1] + "…"
+    return md, None
+
+
+# ---------------------------------------------------------------- 入口
+
+async def fetch_submission_guidelines(url: str) -> dict:
+    """抓投稿指南页面 → 正文 Markdown。
+
+    流程：robots.txt 检查（≥2s 限速）→ 禁止即跳过 → GET 正文（≥2s 限速）
+    → 状态/类型校验 → 正文提取 → 截断 30000 字符。
+    """
+    parsed = urlparse(url)
+    if parsed.scheme not in ("http", "https") or not parsed.netloc:
+        return _fail(f"非法 URL：{url!r}")
+    host = parsed.netloc.lower()
+
+    async with httpx.AsyncClient(follow_redirects=True, timeout=TIMEOUT) as client:
+        allowed, robots_reason = await _robots_allows(client, url)
+        if not allowed:
+            logger.info("[fetch] robots 禁止，跳过 %s（%s）", url, robots_reason)
+            return _fail(robots_reason)
+
+        await _throttle(host)
+        try:
+            resp = await client.get(url, headers=_HEADERS)
+        except httpx.HTTPError as exc:
+            return _fail(f"请求失败（{type(exc).__name__}）：{exc}")
+        if resp.status_code != 200:
+            return _fail(f"HTTP {resp.status_code}")
+        ctype = resp.headers.get("content-type", "")
+        if "html" not in ctype and "xhtml" not in ctype:
+            return _fail(f"非 HTML 响应：{ctype or '无 Content-Type'}")
+
+        markdown, err = _html_to_markdown(_decode_html(resp))
+        if err:
+            return _fail(f"{err}（robots: {robots_reason}）")
+        logger.info("[fetch] %s → %d 字符（robots: %s）", url, len(markdown), robots_reason)
+        return {"ok": True, "markdown": markdown, "error": None, "fetched_at": _now_iso()}

review_service/scripts/README.md

@@ -0,0 +1,17 @@
+# scripts/ —— 离线工具（不被 app/ 服务代码引用，独立 CLI）
+
+## fetch_format_rules.py —— 投稿指南自动抓取（探索性，第一批）
+
+- 依赖：venv 已装的 httpx + beautifulsoup4 4.15.0（用户拍板 2026-10-10）；
+  解析器 lxml 6.1.2 在 `tooling/requirements.lock` 内。
+- 用法与流程见 `fetch_format_rules.py` 模块 docstring（CLI 一次一本期刊）。
+- 产物：`data/format_cache/{journal_id}.json`（运行产物，已 gitignore，不入库）。
+
+### 现状（2026-10-10）
+
+OUP / Wiley / Cell 官网对本运行环境返回 403，自动抓取路线失败；
+Genome Biology / BMC Plant Biology / Frontiers / Rice 可抓但 LLM 抽取结果
+仍需人工对照官网。当前 `journals.json` 中 5 本精选期刊的 `format_rules`
+为**人工核验官网原文后写入**（`rules_source=official` + `source_url` +
+`last_verified`），**不是**本脚本产物。本工具保留作为后续换网络环境 /
+换可达站点后的重跑手段。

review_service/scripts/fetch_format_rules.py

@@ -0,0 +1,122 @@
+#!/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())