---
phase: 01-poc
plan: 03
type: execute
wave: 3
depends_on:
  - 01-01
  - 01-02
files_modified:
  - govcrawler/fetcher/__init__.py
  - govcrawler/fetcher/browser.py
  - govcrawler/fetcher/throttle.py
  - govcrawler/storage/__init__.py
  - govcrawler/storage/paths.py
  - govcrawler/storage/files.py
  - govcrawler/storage/attachments.py
  - govcrawler/storage/repo.py
  - govcrawler/pipeline.py
  - govcrawler/cli.py
  - govcrawler/__main__.py
  - govcrawler/sites/gdqy.py
  - tests/test_paths.py
  - tests/test_throttle.py
  - tests/test_filename_parse.py
  - tests/test_pipeline_mocked.py
  - tests/test_gdqy_smoke.py
autonomous: true
requirements:
  - FETCH-02
  - STORE-01
  - STORE-02
  - STORE-03
  - STORE-04
  - STORE-05
tags:
  - fetcher
  - patchright
  - storage
  - cli
  - end-to-end

must_haves:
  truths:
    - "开发者执行 `uv run python -m govcrawler fetch gdqy szfwj post_2136593` 能从 gdqy.gov.cn 端到端抓到一篇文章：自动过 ctct 盾、解析、落盘、入库"
    - "执行成功后 data/govcrawler/raw_html/gdqy/szfwj/2026/04/post_2136593.html 存在且非空"
    - "执行成功后 data/govcrawler/articles_text/gdqy/szfwj/2026/04/post_2136593.txt 存在且包含 '无人驾驶航空器'"
    - "执行成功后 PG articles 表有一行，url_hash=sha256(normalize_url(target))，content_simhash 非空非 '0'*16，status='ready'，raw_html_path/text_path 指向上面两个文件（POSIX 形式）"
    - "执行成功后 PG crawl_logs 表有一行 success=true, strategy='playwright', http_status=200（或 0 表示 response 对象缺失但拿到 HTML）"
    - "当传入一篇带 PDF 附件的 gdqy 文章 URL，PDF 被流式下载到 attachments/ 目录且 attachments 表有一行，file_hash 是文件 SHA256"
    - "同一 URL 再抓一次触发 url_hash UNIQUE 冲突，CLI 打印 '[skip] duplicate url_hash' 并以 exit 0 结束（不重复抓）"
    - "节流：连续两次对同一 host 的 fetch 间隔 >= 5 秒（throttle 单测证明）"
    - "当 detail_parser.content_text < 50 字时，status 写为 'failed' 且 crawl_logs.error_msg 记录原因"
  artifacts:
    - path: "govcrawler/fetcher/browser.py"
      provides: "fetch_html(url) -> FetchResult via patchright（domcontentloaded + selector wait + page.content()）"
      contains: "async_playwright"
    - path: "govcrawler/fetcher/throttle.py"
      provides: "HostThrottle — 同 host 间隔 >= 5s + 20% jitter"
      contains: "class HostThrottle"
    - path: "govcrawler/storage/paths.py"
      provides: "build_relpath + to_os_path — POSIX 稳定路径"
      contains: "PurePosixPath"
    - path: "govcrawler/storage/files.py"
      provides: "write_raw_html / write_article_text — 原子写"
      contains: "def write_raw_html"
    - path: "govcrawler/storage/attachments.py"
      provides: "download_attachment — httpx 流式、Content-Disposition 解析、SHA256"
      contains: "httpx.stream"
    - path: "govcrawler/storage/repo.py"
      provides: "insert_article / insert_attachments / insert_crawl_log / get_article_by_url_hash"
      contains: "def insert_article"
    - path: "govcrawler/pipeline.py"
      provides: "fetch_and_store(site_id, column_id, article_key, target_url) — 端到端编排"
      contains: "def fetch_and_store"
    - path: "govcrawler/cli.py"
      provides: "argparse: fetch <site> <column> <article_key> [--url <url>]"
      contains: "def main"
    - path: "govcrawler/__main__.py"
      provides: "python -m govcrawler 入口"
      contains: "from govcrawler.cli import main"
  key_links:
    - from: "govcrawler/cli.py"
      to: "govcrawler/pipeline.py"
      via: "cli.main() -> fetch_and_store(site, column, article_key, target_url)"
      pattern: "fetch_and_store"
    - from: "govcrawler/pipeline.py"
      to: "govcrawler/fetcher/browser.py"
      via: "asyncio.run(fetch_html(url))"
      pattern: "fetch_html|asyncio\\.run"
    - from: "govcrawler/pipeline.py"
      to: "govcrawler/parser/detail_parser.py"
      via: "parse_detail(html, base_url, DETAIL_SELECTORS)"
      pattern: "parse_detail"
    - from: "govcrawler/pipeline.py"
      to: "govcrawler/storage/files.py"
      via: "write_raw_html + write_article_text"
      pattern: "write_raw_html|write_article_text"
    - from: "govcrawler/pipeline.py"
      to: "govcrawler/storage/attachments.py"
      via: "for each attachment_url: download_attachment(...)"
      pattern: "download_attachment"
    - from: "govcrawler/pipeline.py"
      to: "govcrawler/storage/repo.py"
      via: "insert_article + insert_attachments + insert_crawl_log (单事务)"
      pattern: "insert_article|insert_attachments|insert_crawl_log"
    - from: "govcrawler/pipeline.py"
      to: "govcrawler/utils/url_norm.py"
      via: "url_hash(target_url) 作 articles.url_hash"
      pattern: "url_hash"
    - from: "govcrawler/pipeline.py"
      to: "govcrawler/utils/simhash_utils.py"
      via: "content_simhash(content_text) 写 articles.content_simhash"
      pattern: "content_simhash"
---

<objective>
把 Plan 01 的 DB schema + Plan 02 的 parser/utils 粘合成一条端到端流水线：CLI → Fetcher (patchright) → Parser → Storage (files + PG) → CrawlLog。支持对 gdqy.gov.cn 目标文章 `post_2136593` 的一次成功抓取。

Purpose: 这是 Phase 1 的"交付时刻"——覆盖剩余 6 个 REQ-ID（FETCH-02, STORE-01..05），并通过 ROADMAP Phase 1 的 5 条成功标准。**本 plan 不引入新的解析/去重算法**（都在 Plan 02），只做 orchestration。

Output: 可运行的 CLI + 端到端流水线 + 真实网络 smoke test（可选跳过）。
</objective>

<execution_context>
@$HOME/.claude/get-shit-done/workflows/execute-plan.md
@$HOME/.claude/get-shit-done/templates/summary.md
</execution_context>

<context>
@.planning/PROJECT.md
@.planning/ROADMAP.md
@.planning/STATE.md
@.planning/phases/01-poc/01-CONTEXT.md
@.planning/phases/01-poc/01-RESEARCH.md

<!-- 引用 Plan 01/02 的关键契约供执行者直接 wire，不用翻源码 -->

<interfaces>
<!-- From Plan 01 (01-01-PLAN.md interfaces section): -->

```python
# govcrawler/settings.py (from Plan 01)
from govcrawler.settings import get_settings
s = get_settings()
s.db_url           # postgresql+psycopg://...
s.data_dir         # "./data/govcrawler"
s.user_agent       # "GovCrawlerBot/1.0 (contact: xxx@example.com)"

# govcrawler/db.py (from Plan 01)
from govcrawler.db import get_sessionmaker
Session = get_sessionmaker()

# govcrawler/models.py (from Plan 01) — SQLAlchemy 2.0
from govcrawler.models import Article, Attachment, CrawlLog, Site, Column_
# Article fields (relevant to this plan):
#   site_id, column_id, category, url, url_hash, content_simhash,
#   title, publish_time (DateTime|None), source, content_text,
#   raw_html_path, text_path, has_attachment, status, fetch_strategy,
#   fetched_at (server_default=now()), exported_to_rag_at
# Attachment fields:
#   article_id (FK), file_name, file_ext, size_bytes, file_path, file_hash, downloaded_at
# CrawlLog fields:
#   site_id, column_id, article_url, strategy, http_status,
#   duration_ms, success, error_msg, occurred_at
```

<!-- From Plan 02 (01-02-PLAN.md interfaces section): -->

```python
from govcrawler.utils.url_norm import normalize_url, url_hash, sha256_bytes
from govcrawler.utils.simhash_utils import content_simhash
from govcrawler.parser.list_parser import parse_list, ListItem
from govcrawler.parser.detail_parser import parse_detail, DetailFields
from govcrawler.parser.cleaner import html_to_text
from govcrawler.sites.gdqy import (
    SITE_ID, SITE_NAME, SITE_BASE_URL,
    COLUMN_ID, COLUMN_NAME, COLUMN_CATEGORY, COLUMN_LIST_URL,
    TARGET_ARTICLE_URL, TARGET_ARTICLE_KEY,
    LIST_SELECTORS, DETAIL_SELECTORS,
)
```
</interfaces>
</context>

<tasks>

<task type="auto">
  <name>Task 1: Fetcher (patchright) + Throttle + Storage paths/files/attachments</name>
  <files>
    govcrawler/fetcher/__init__.py,
    govcrawler/fetcher/browser.py,
    govcrawler/fetcher/throttle.py,
    govcrawler/storage/__init__.py,
    govcrawler/storage/paths.py,
    govcrawler/storage/files.py,
    govcrawler/storage/attachments.py,
    tests/test_paths.py,
    tests/test_throttle.py,
    tests/test_filename_parse.py
  </files>
  <read_first>
    - /Users/alphaPlanet/vibecoding/website/.planning/phases/01-poc/01-RESEARCH.md  (§Code Examples · patchright Fetcher / §挑战页检测 / §附件流式下载 / §存储路径构建 / §Common Pitfalls #2 networkidle / #4 Windows 路径 / #5 Content-Disposition)
    - /Users/alphaPlanet/vibecoding/website/.planning/phases/01-poc/01-CONTEXT.md  (§Fetcher LOCKED / §Storage 布局 LOCKED)
    - /Users/alphaPlanet/vibecoding/website/.planning/phases/01-poc/01-01-PLAN.md  (interfaces: Settings / get_sessionmaker)
    - /Users/alphaPlanet/vibecoding/website/.planning/phases/01-poc/01-02-PLAN.md  (不直接消费，但确认 utils.url_norm 已提供 sha256_bytes)
  </read_first>
  <action>
  本任务交付三大模块：patchright Fetcher、Host Throttle、Storage 三件套（paths/files/attachments）。不碰 DB、不碰 CLI。

  **1. `govcrawler/fetcher/__init__.py`**：空。

  **2. `govcrawler/fetcher/browser.py`** — 基于 RESEARCH §Code Examples · patchright Fetcher 启动，改成**同步包装**（内部 asyncio.run），外部 Pipeline 直接同步调：
  ```python
  from __future__ import annotations
  import asyncio
  import time
  from dataclasses import dataclass
  from govcrawler.settings import get_settings

  CHALLENGE_MARKERS = ("ctct-slider-canvas", "请稍候")

  @dataclass
  class FetchResult:
      url: str
      final_url: str
      status: int
      html: str
      fetched_at: float
      duration_ms: int
      is_challenge: bool
      error: str | None = None

  def is_challenge_page(status: int, html: str) -> bool:
      if status == 412:
          return True
      return any(m in (html or "") for m in CHALLENGE_MARKERS)

  async def _fetch_async(url: str, *, user_agent: str, timeout_ms: int = 30000) -> FetchResult:
      from patchright.async_api import async_playwright
      t0 = time.time()
      try:
          async with async_playwright() as p:
              browser = await p.chromium.launch(
                  headless=True,
                  args=["--disable-blink-features=AutomationControlled"],
              )
              try:
                  context = await browser.new_context(
                      user_agent=user_agent,
                      locale="zh-CN",
                      timezone_id="Asia/Shanghai",
                      viewport={"width": 1440, "height": 900},
                  )
                  page = await context.new_page()
                  try:
                      # RESEARCH §Common Pitfalls #2 + CONTEXT §Fetcher LOCKED (2026-04-22 revision):
                      # 两步 fallback — domcontentloaded 主路径 + networkidle 回退 + 最终 page.content() 兜底。
                      DETAIL_SELECTOR = "div.article-content, div.content, div.TRS_Editor, body"

                      # Step 1: domcontentloaded 主路径
                      response = await page.goto(url, wait_until="domcontentloaded", timeout=timeout_ms)

                      # Step 2: wait_for_selector 5s
                      selector_ok = False
                      try:
                          await page.wait_for_selector(DETAIL_SELECTOR, timeout=5000)
                          selector_ok = True
                      except Exception:
                          selector_ok = False

                      # Step 3: 若 selector 超时，回退到 networkidle (45s) 再次尝试
                      if not selector_ok:
                          try:
                              response = await page.goto(url, wait_until="networkidle", timeout=45000)
                              try:
                                  await page.wait_for_selector(DETAIL_SELECTOR, timeout=5000)
                                  selector_ok = True
                              except Exception:
                                  selector_ok = False
                          except Exception:
                              # networkidle 本身超时也不抛——继续走 Step 4 读当前 DOM
                              selector_ok = False

                      # Step 4: 无论 selector 是否命中，读当前 page.content() 交给 parser+GNE 兜底
                      html = await page.content()
                      final = page.url
                      status = response.status if response else 0
                      challenge = is_challenge_page(status, html)
                      return FetchResult(
                          url=url, final_url=final, status=status, html=html,
                          fetched_at=time.time(), duration_ms=int((time.time() - t0) * 1000),
                          is_challenge=challenge,
                      )
                  finally:
                      await context.close()
              finally:
                  await browser.close()
      except Exception as e:
          return FetchResult(
              url=url, final_url=url, status=0, html="",
              fetched_at=time.time(), duration_ms=int((time.time() - t0) * 1000),
              is_challenge=False, error=f"{type(e).__name__}: {e}",
          )

  def fetch_html(url: str, *, timeout_ms: int = 30000) -> FetchResult:
      """Sync wrapper. Pipeline 层同步调用即可。"""
      settings = get_settings()
      return asyncio.run(_fetch_async(url, user_agent=settings.user_agent, timeout_ms=timeout_ms))
  ```

  **3. `govcrawler/fetcher/throttle.py`** — 按 CONTEXT §LOCKED "请求间隔 ≥ 5s + 20% 抖动"：
  ```python
  from __future__ import annotations
  import random
  import time
  from urllib.parse import urlparse

  DEFAULT_INTERVAL_S = 5.0
  JITTER_PCT = 0.20   # ±20%

  class HostThrottle:
      """Per-host minimum interval gate. In-process only (Phase 2 will use Valkey).

      Usage:
          t = HostThrottle(interval_s=5.0)
          t.wait(url)   # blocks until safe
      """
      def __init__(self, interval_s: float = DEFAULT_INTERVAL_S, jitter_pct: float = JITTER_PCT):
          self.interval_s = interval_s
          self.jitter_pct = jitter_pct
          self._last_by_host: dict[str, float] = {}

      def _host(self, url: str) -> str:
          return urlparse(url).netloc.lower()

      def wait(self, url: str, *, sleep=time.sleep, now=time.monotonic) -> float:
          """Block until interval elapsed. Returns seconds slept (0 if no wait)."""
          host = self._host(url)
          now_t = now()
          last = self._last_by_host.get(host)
          sleep_s = 0.0
          if last is not None:
              jitter = 1.0 + random.uniform(-self.jitter_pct, self.jitter_pct)
              due = last + self.interval_s * jitter
              sleep_s = max(0.0, due - now_t)
              if sleep_s > 0:
                  sleep(sleep_s)
          self._last_by_host[host] = now()
          return sleep_s
  ```

  **4. `govcrawler/storage/__init__.py`**：空。

  **5. `govcrawler/storage/paths.py`** — 照搬 RESEARCH §Code Examples · 存储路径构建，加 kind 白名单：
  ```python
  from __future__ import annotations
  from datetime import datetime
  from pathlib import Path, PurePosixPath

  KINDS = frozenset({"raw_html", "articles_text", "attachments"})

  def build_reldir(site: str, column: str, when: datetime, kind: str) -> PurePosixPath:
      """Return POSIX-style relative directory (not including filename).
      Example: raw_html/gdqy/szfwj/2026/04
      """
      if kind not in KINDS:
          raise ValueError(f"unknown kind: {kind!r}; expected one of {sorted(KINDS)}")
      return PurePosixPath(kind, site, column, f"{when:%Y}", f"{when:%m}")

  def to_os_path(root: Path | str, rel: PurePosixPath) -> Path:
      """Convert (data_dir, posix-relpath) -> OS-native absolute Path."""
      return Path(root).joinpath(*rel.parts)
  ```

  **6. `govcrawler/storage/files.py`** — 原子写 raw_html / article_text。**DB 里存 POSIX 相对路径**（RESEARCH §Common Pitfalls #4）：
  ```python
  from __future__ import annotations
  import os
  import tempfile
  from datetime import datetime
  from pathlib import Path, PurePosixPath

  from govcrawler.settings import get_settings
  from govcrawler.storage.paths import build_reldir, to_os_path

  def _atomic_write_bytes(path: Path, data: bytes) -> None:
      path.parent.mkdir(parents=True, exist_ok=True)
      fd, tmp = tempfile.mkstemp(prefix=".tmp_", dir=str(path.parent))
      try:
          with os.fdopen(fd, "wb") as f:
              f.write(data)
          os.replace(tmp, path)
      except Exception:
          try:
              os.unlink(tmp)
          except FileNotFoundError:
              pass
          raise

  def write_raw_html(site: str, column: str, when: datetime, article_key: str, html: str) -> PurePosixPath:
      """Write raw html; return POSIX relative path (stored in DB).
      Absolute path on disk = data_dir / reldir / filename."""
      reldir = build_reldir(site, column, when, "raw_html")
      relpath = reldir / f"{article_key}.html"
      abs_path = to_os_path(get_settings().data_dir, relpath)
      _atomic_write_bytes(abs_path, html.encode("utf-8"))
      return relpath

  def write_article_text(site: str, column: str, when: datetime, article_key: str, text: str) -> PurePosixPath:
      reldir = build_reldir(site, column, when, "articles_text")
      relpath = reldir / f"{article_key}.txt"
      abs_path = to_os_path(get_settings().data_dir, relpath)
      _atomic_write_bytes(abs_path, text.encode("utf-8"))
      return relpath
  ```

  **7. `govcrawler/storage/attachments.py`** — httpx 流式下载 + Content-Disposition 解析 + SHA256：
  ```python
  from __future__ import annotations
  import hashlib
  import os
  import re
  import tempfile
  from dataclasses import dataclass
  from datetime import datetime
  from email.message import Message
  from pathlib import Path, PurePosixPath
  from urllib.parse import unquote

  import httpx
  from slugify import slugify

  from govcrawler.settings import get_settings
  from govcrawler.storage.paths import build_reldir, to_os_path
  from govcrawler.utils.url_norm import normalize_url

  MAX_ATTACHMENT_BYTES = 200 * 1024 * 1024   # 200 MB, RESEARCH §Security Domain
  CHUNK = 64 * 1024
  SAFE_EXT = {"pdf", "doc", "docx", "xls", "xlsx", "zip", "rar", "txt"}

  @dataclass
  class DownloadedAttachment:
      file_name: str         # cleaned final filename (without dir)
      file_ext: str
      size_bytes: int
      file_hash: str         # sha256 hex of content
      file_path: PurePosixPath  # POSIX relpath stored in DB

  def parse_disposition_filename(header: str | None) -> str | None:
      if not header:
          return None
      m = Message()
      m["content-disposition"] = header
      name = m.get_param("filename*") or m.get_param("filename")
      if not name:
          return None
      if isinstance(name, tuple):
          # RFC 5987: ("utf-8", "", "%E5%85%AC%E5%91%8A.pdf")
          _, _, raw = name
          try:
              return unquote(raw)
          except Exception:
              return raw
      return name

  def safe_filename(name: str, max_len: int = 180) -> str:
      cleaned = slugify(name, allow_unicode=True, separator="_",
                        regex_pattern=r'[\\/:*?"<>|\x00-\x1f]')
      if len(cleaned) > max_len:
          cleaned = cleaned[:max_len]
      return cleaned or "attachment"

  def _ext_of(name: str) -> str:
      _, dot, ext = name.rpartition(".")
      return ext.lower() if dot else ""

  def download_attachment(
      url: str, *, site: str, column: str, when: datetime, article_key: str,
  ) -> DownloadedAttachment:
      """Stream-download; enforce size cap + path-traversal guard (SRE §V5)."""
      norm = normalize_url(url)   # raises on file:// / javascript:
      reldir = build_reldir(site, column, when, "attachments")
      abs_dir = to_os_path(get_settings().data_dir, reldir)
      abs_dir.mkdir(parents=True, exist_ok=True)

      hasher = hashlib.sha256()
      size = 0
      headers = {"User-Agent": get_settings().user_agent}
      # tempfile in the target dir so os.replace is atomic
      fd, tmp = tempfile.mkstemp(prefix=".dl_", dir=str(abs_dir))
      raw_name_fallback = norm.rsplit("/", 1)[-1].split("?", 1)[0] or "attachment.bin"
      raw_name: str = raw_name_fallback
      try:
          with httpx.stream("GET", norm, headers=headers, follow_redirects=True, timeout=60) as r:
              r.raise_for_status()
              disp_name = parse_disposition_filename(r.headers.get("content-disposition"))
              raw_name = disp_name or raw_name_fallback
              with os.fdopen(fd, "wb") as f:
                  for chunk in r.iter_bytes(chunk_size=CHUNK):
                      if not chunk:
                          continue
                      size += len(chunk)
                      if size > MAX_ATTACHMENT_BYTES:
                          raise ValueError(f"attachment exceeds {MAX_ATTACHMENT_BYTES} bytes: {norm}")
                      hasher.update(chunk)
                      f.write(chunk)
      except Exception:
          try:
              os.unlink(tmp)
          except FileNotFoundError:
              pass
          raise

      ext = _ext_of(raw_name)
      final_name = f"{article_key}_{safe_filename(raw_name)}"
      final_abs = abs_dir / final_name
      # Path-traversal guard
      try:
          final_abs.resolve().relative_to(abs_dir.resolve())
      except Exception:
          os.unlink(tmp)
          raise ValueError(f"path traversal blocked for filename={raw_name!r}")
      os.replace(tmp, final_abs)

      return DownloadedAttachment(
          file_name=final_name,
          file_ext=ext,
          size_bytes=size,
          file_hash=hasher.hexdigest(),
          file_path=reldir / final_name,
      )
  ```

  **8. `tests/test_paths.py`** — 跨平台稳定性：
  ```python
  from datetime import datetime
  from pathlib import PurePosixPath
  from govcrawler.storage.paths import build_reldir, to_os_path

  def test_build_reldir_posix():
      d = datetime(2026, 4, 10, 12, 0, 0)
      r = build_reldir("gdqy", "szfwj", d, "raw_html")
      assert isinstance(r, PurePosixPath)
      assert str(r) == "raw_html/gdqy/szfwj/2026/04"

  def test_build_reldir_invalid_kind():
      import pytest
      with pytest.raises(ValueError):
          build_reldir("gdqy", "szfwj", datetime(2026,1,1), "nonsense")

  def test_to_os_path_joins():
      d = datetime(2026, 4, 10)
      r = build_reldir("gdqy", "szfwj", d, "articles_text")
      p = to_os_path("/tmp/data", r)
      # on both Windows and posix, last parts are consistent
      assert p.parts[-5:] == ("articles_text", "gdqy", "szfwj", "2026", "04")
  ```

  **9. `tests/test_throttle.py`** — 用注入 sleep/now 单测，不真正 sleep 5s：
  ```python
  from govcrawler.fetcher.throttle import HostThrottle

  class FakeClock:
      def __init__(self): self.t = 1000.0; self.slept = []
      def now(self): return self.t
      def sleep(self, s):
          self.slept.append(s)
          self.t += s

  def test_first_call_no_wait():
      c = FakeClock()
      t = HostThrottle(interval_s=5.0, jitter_pct=0.0)
      slept = t.wait("https://a.com/x", sleep=c.sleep, now=c.now)
      assert slept == 0.0

  def test_second_call_same_host_waits_at_least_interval():
      c = FakeClock()
      t = HostThrottle(interval_s=5.0, jitter_pct=0.0)
      t.wait("https://a.com/x", sleep=c.sleep, now=c.now)
      # advance 1s
      c.t += 1.0
      t.wait("https://a.com/y", sleep=c.sleep, now=c.now)
      # second call should have slept ~4s (5 - 1)
      assert sum(c.slept) >= 4.0

  def test_different_host_no_wait():
      c = FakeClock()
      t = HostThrottle(interval_s=5.0, jitter_pct=0.0)
      t.wait("https://a.com/x", sleep=c.sleep, now=c.now)
      slept = t.wait("https://b.com/x", sleep=c.sleep, now=c.now)
      assert slept == 0.0
  ```

  **10. `tests/test_filename_parse.py`**：
  ```python
  from govcrawler.storage.attachments import parse_disposition_filename, safe_filename

  def test_plain_filename():
      assert parse_disposition_filename('attachment; filename="notice.pdf"') == "notice.pdf"

  def test_rfc5987_utf8():
      h = "attachment; filename*=UTF-8''%E5%85%AC%E5%91%8A.pdf"
      assert parse_disposition_filename(h) == "公告.pdf"

  def test_missing_header_returns_none():
      assert parse_disposition_filename(None) is None
      assert parse_disposition_filename("") is None

  def test_safe_filename_strips_illegal():
      assert "/" not in safe_filename("a/b.pdf")
      assert "\\" not in safe_filename("a\\b.pdf")

  def test_safe_filename_keeps_chinese():
      out = safe_filename("公告全文.pdf")
      assert "公告" in out
  ```

  约束：
  - Fetcher 同步包装：用 `asyncio.run` 最简单可靠；本阶段单次抓一篇不存在事件循环嵌套问题。
  - Fetcher 两步 fallback（CONTEXT §Fetcher LOCKED + RESEARCH §Common Pitfalls #2 已明确）：`domcontentloaded`(30s) → `wait_for_selector`(5s) → `networkidle`(45s) 二次 `goto` → 再 `wait_for_selector`(5s) → 最终 `page.content()` 兜底。任何一步的异常都不应向上抛，最差情况返回"selector 未命中但拿到 DOM"的 FetchResult，交给 parser + GNE 兜底处理。
  - Throttle 的 sleep/now 注入点让单测秒级跑完，不真正等 5s。
  - `path-traversal guard` 在 `download_attachment` 里显式调 `resolve().relative_to(abs_dir.resolve())`，防御 Content-Disposition 注入（RESEARCH §Security Domain V5）。
  - `MAX_ATTACHMENT_BYTES = 200MB` 防 DoS（RESEARCH §Security Domain）。
  - 所有落盘路径返回 `PurePosixPath`（RESEARCH §Common Pitfalls #4）。
  </action>
  <verify>
    <automated>python -c "import ast; [ast.parse(open(f).read()) for f in ['govcrawler/fetcher/browser.py','govcrawler/fetcher/throttle.py','govcrawler/storage/paths.py','govcrawler/storage/files.py','govcrawler/storage/attachments.py','tests/test_paths.py','tests/test_throttle.py','tests/test_filename_parse.py']]" && uv run pytest tests/test_paths.py tests/test_throttle.py tests/test_filename_parse.py -x -q</automated>
  </verify>
  <acceptance_criteria>
    - `uv run pytest tests/test_paths.py tests/test_throttle.py tests/test_filename_parse.py -x -q` 全绿（3+3+5=11 case）
    - `from govcrawler.fetcher.browser import fetch_html, FetchResult, is_challenge_page` 成功 import
    - `from govcrawler.storage.attachments import download_attachment, DownloadedAttachment, parse_disposition_filename` 成功 import
    - `parse_disposition_filename("attachment; filename*=UTF-8''%E5%85%AC%E5%91%8A.pdf")` 返回 "公告.pdf"
    - `HostThrottle` 对同 host 第二次调用会 sleep >= 4s（interval=5 jitter=0，1s 后再调）
    - `build_reldir("gdqy","szfwj",datetime(2026,4,10),"raw_html")` 返回 `PurePosixPath("raw_html/gdqy/szfwj/2026/04")`
    - `build_reldir(...,"nonsense")` 抛 ValueError
    - `download_attachment` 源码中包含 `resolve().relative_to` 防穿越 + `MAX_ATTACHMENT_BYTES` 判断
    - `govcrawler/fetcher/browser.py` 源码实现完整两步 fallback：grep 该文件必须同时出现 `wait_until="domcontentloaded"`、`wait_until="networkidle"`、两处 `wait_for_selector`、且最终调用 `page.content()` 兜底（即使 selector 两次都超时也不抛 exception）
  </acceptance_criteria>
  <done>Fetcher + Throttle + Storage 三件套就绪；11 个 pytest case 全绿；Task 2 可把它们与 Parser/DB wire 起来。</done>
</task>

<task type="auto">
  <name>Task 2: DB Repo + Pipeline + CLI + Smoke test</name>
  <files>
    govcrawler/storage/repo.py,
    govcrawler/pipeline.py,
    govcrawler/cli.py,
    govcrawler/__main__.py,
    govcrawler/sites/gdqy.py,
    tests/test_pipeline_mocked.py,
    tests/test_gdqy_smoke.py
  </files>
  <read_first>
    - /Users/alphaPlanet/vibecoding/website/.planning/phases/01-poc/01-RESEARCH.md  (§System Architecture Diagram / §Decision points 1 和 2 / §Pattern 1-2)
    - /Users/alphaPlanet/vibecoding/website/.planning/phases/01-poc/01-CONTEXT.md  (§CLI 入口 LOCKED / §成功标准映射)
    - /Users/alphaPlanet/vibecoding/website/.planning/phases/01-poc/01-01-PLAN.md  (interfaces: Article/Attachment/CrawlLog 字段)
    - /Users/alphaPlanet/vibecoding/website/.planning/phases/01-poc/01-02-PLAN.md  (interfaces: parse_detail / gdqy selectors)
    - /Users/alphaPlanet/vibecoding/website/.planning/ROADMAP.md  (Phase 1 success criteria)
  </read_first>
  <action>
  本任务把所有模块粘起来 + 交付 CLI + 真实 smoke test。

  **1. `govcrawler/storage/repo.py`** — 薄层封装 insert + lookup：
  ```python
  from __future__ import annotations
  from datetime import datetime
  from typing import Sequence
  from sqlalchemy import select
  from sqlalchemy.orm import Session

  from govcrawler.models import Article, Attachment, CrawlLog

  def get_article_by_url_hash(session: Session, url_hash: str) -> Article | None:
      stmt = select(Article).where(Article.url_hash == url_hash)
      return session.execute(stmt).scalar_one_or_none()

  def insert_article(session: Session, **kwargs) -> Article:
      """Relevant kwargs: site_id, column_id, category, url, url_hash,
      content_simhash, title, publish_time, source, content_text,
      raw_html_path (str), text_path (str), has_attachment, status,
      fetch_strategy. fetched_at is server_default."""
      a = Article(**kwargs)
      session.add(a)
      session.flush()   # so a.id populated
      return a

  def insert_attachments(session: Session, article_id: int, records: Sequence[dict]) -> list[Attachment]:
      """records keys: file_name, file_ext, size_bytes, file_path (str POSIX), file_hash."""
      out: list[Attachment] = []
      for r in records:
          att = Attachment(article_id=article_id, **r)
          session.add(att)
          out.append(att)
      session.flush()
      return out

  def insert_crawl_log(session: Session, *, site_id: str, column_id: str, article_url: str,
                       strategy: str, http_status: int | None, duration_ms: int,
                       success: bool, error_msg: str | None) -> CrawlLog:
      c = CrawlLog(
          site_id=site_id, column_id=column_id, article_url=article_url,
          strategy=strategy, http_status=http_status, duration_ms=duration_ms,
          success=success, error_msg=error_msg,
      )
      session.add(c)
      session.flush()
      return c
  ```

  **2. `govcrawler/sites/gdqy.py`** — **APPEND ONLY**（Plan 02 已创建）：加一个 secondary test URL 字段供附件 smoke test。Executor 按真实 gdqy 列表页挑一个带 PDF 附件的 URL（若找不到合适的，留 None + 在 SUMMARY 里说明，附件路径仅通过 `tests/test_pipeline_mocked.py` 的 mocked httpx.stream 验证）。
  ```python
  # Append to existing govcrawler/sites/gdqy.py:

  # Optional: 带附件的文章 URL (for STORE-03/05 secondary smoke test).
  # If None, attachment path is covered only by tests/test_pipeline_mocked.py (httpx mocked).
  # Executor may set this to a real URL found at COLUMN_LIST_URL on execution day.
  ATTACHMENT_TEST_ARTICLE_URL: str | None = None
  ATTACHMENT_TEST_ARTICLE_KEY: str | None = None
  ```

  **3. `govcrawler/pipeline.py`** — 端到端编排，单事务：
  ```python
  from __future__ import annotations
  import logging
  from dataclasses import asdict
  from datetime import datetime
  from pathlib import PurePosixPath
  import re

  from govcrawler.db import get_sessionmaker
  from govcrawler.fetcher.browser import fetch_html, FetchResult
  from govcrawler.fetcher.throttle import HostThrottle
  from govcrawler.parser.detail_parser import parse_detail
  from govcrawler.parser.cleaner import html_to_text
  from govcrawler.sites.gdqy import DETAIL_SELECTORS
  from govcrawler.storage.files import write_raw_html, write_article_text
  from govcrawler.storage.attachments import download_attachment
  from govcrawler.storage.repo import (
      get_article_by_url_hash, insert_article, insert_attachments, insert_crawl_log,
  )
  from govcrawler.utils.simhash_utils import content_simhash
  from govcrawler.utils.url_norm import url_hash as compute_url_hash

  log = logging.getLogger(__name__)

  MIN_CONTENT_TEXT_CHARS = 50   # Decision point 2: 正文文本 < 50 字 → status='failed'

  def _parse_publish_time(raw: str) -> datetime | None:
      """Try common gdqy formats: '2026-04-10 16:34:22', '2026-04-10', '2026/04/10'.
      Return None if no format matches (字符串仍保留在 content 里不会丢)."""
      if not raw:
          return None
      # Normalize separators
      s = raw.strip().replace("/", "-").replace("年", "-").replace("月", "-").replace("日", " ")
      s = re.sub(r"\s+", " ", s).strip()
      for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%d %H:%M", "%Y-%m-%d"):
          try:
              return datetime.strptime(s[:len(fmt) + 4], fmt)
          except Exception:
              continue
      return None

  def fetch_and_store(site_id: str, column_id: str, article_key: str, target_url: str,
                      *, throttle: HostThrottle | None = None) -> dict:
      """End-to-end: fetch → parse → write files → download attachments → insert rows.
      Returns a result dict suitable for CLI printing.

      Duplicate behavior: if url_hash already exists, log 'skip' and return without re-fetching.
      """
      throttle = throttle or HostThrottle()
      url_h = compute_url_hash(target_url)
      Session = get_sessionmaker()

      # Pre-check duplicate (fast path — avoids spinning up Chromium)
      with Session() as s:
          existing = get_article_by_url_hash(s, url_h)
          if existing:
              return {"status": "skipped", "reason": "duplicate_url_hash", "url_hash": url_h,
                      "article_id": existing.id}

      # Throttle before network
      throttle.wait(target_url)

      fr: FetchResult = fetch_html(target_url)
      strategy = "playwright"
      occurred_at = datetime.utcnow()

      # Challenge or hard error → log + exit
      if fr.error or fr.is_challenge or not fr.html:
          with Session() as s:
              insert_crawl_log(
                  s, site_id=site_id, column_id=column_id, article_url=target_url,
                  strategy=strategy, http_status=fr.status, duration_ms=fr.duration_ms,
                  success=False,
                  error_msg=(fr.error or ("ctct_challenge_unresolved" if fr.is_challenge else "empty_html")),
              )
              s.commit()
          return {"status": "failed",
                  "reason": fr.error or ("challenge" if fr.is_challenge else "empty_html"),
                  "http_status": fr.status}

      # Parse
      fields = parse_detail(fr.html, target_url, DETAIL_SELECTORS)

      # Decide status before writing DB (we still write files for forensics even if failed)
      pub_dt = _parse_publish_time(fields.publish_time_raw) or datetime.utcnow()
      status = "ready" if len(fields.content_text) >= MIN_CONTENT_TEXT_CHARS else "failed"

      # File writes — use pub_dt for path partition
      raw_rel = write_raw_html(site_id, column_id, pub_dt, article_key, fr.html)
      text_rel = write_article_text(site_id, column_id, pub_dt, article_key, fields.content_text)

      # Attachments — download BEFORE DB so file_hash available
      downloaded = []
      for au in fields.attachment_urls:
          try:
              d = download_attachment(au, site=site_id, column=column_id,
                                      when=pub_dt, article_key=article_key)
              downloaded.append(d)
          except Exception as e:
              log.warning("attachment download failed url=%s err=%s", au, e)

      # DB writes (single txn)
      with Session() as s:
          article = insert_article(
              s,
              site_id=site_id, column_id=column_id, category=None,
              url=target_url, url_hash=url_h,
              content_simhash=content_simhash(fields.content_text),
              title=fields.title or None,
              publish_time=_parse_publish_time(fields.publish_time_raw),
              source=fields.source or None,
              content_text=fields.content_text or None,
              raw_html_path=str(raw_rel),
              text_path=str(text_rel),
              has_attachment=bool(downloaded),
              status=status,
              fetch_strategy=strategy,
          )
          att_records = [{
              "file_name": d.file_name, "file_ext": d.file_ext,
              "size_bytes": d.size_bytes, "file_path": str(d.file_path),
              "file_hash": d.file_hash,
          } for d in downloaded]
          if att_records:
              insert_attachments(s, article.id, att_records)
          error_msg = None if status == "ready" else f"content_text_too_short:{len(fields.content_text)}"
          insert_crawl_log(
              s, site_id=site_id, column_id=column_id, article_url=target_url,
              strategy=strategy, http_status=fr.status, duration_ms=fr.duration_ms,
              success=(status == "ready"), error_msg=error_msg,
          )
          s.commit()
          article_id = article.id

      return {
          "status": status, "article_id": article_id, "url_hash": url_h,
          "title": fields.title, "used_fallback": fields.used_fallback,
          "fallback_engine": fields.fallback_engine,
          "raw_html_path": str(raw_rel), "text_path": str(text_rel),
          "attachments_downloaded": len(downloaded),
          "http_status": fr.status, "duration_ms": fr.duration_ms,
      }
  ```

  **4. `govcrawler/cli.py`** — argparse：
  ```python
  from __future__ import annotations
  import argparse
  import json
  import logging
  import sys

  from govcrawler.pipeline import fetch_and_store
  from govcrawler.sites.gdqy import (
      SITE_ID as GDQY_SITE_ID, COLUMN_ID as GDQY_COLUMN_ID,
      TARGET_ARTICLE_URL as GDQY_TARGET_URL, TARGET_ARTICLE_KEY as GDQY_TARGET_KEY,
  )

  def _build_parser() -> argparse.ArgumentParser:
      p = argparse.ArgumentParser(prog="govcrawler", description="政务网站采集系统 (PoC)")
      sub = p.add_subparsers(dest="cmd", required=True)

      fetch = sub.add_parser("fetch", help="抓取一篇文章")
      fetch.add_argument("site", help="site_id, e.g. gdqy")
      fetch.add_argument("column", help="column_id, e.g. szfwj")
      fetch.add_argument("article_key", help="article key, e.g. post_2136593")
      fetch.add_argument("--url", default=None,
                        help="目标 URL；未给时使用该 site+column 硬编码的 TARGET_ARTICLE_URL")
      return p

  def _resolve_url(site: str, column: str, article_key: str, explicit: str | None) -> str:
      if explicit:
          return explicit
      if site == GDQY_SITE_ID and column == GDQY_COLUMN_ID and article_key == GDQY_TARGET_KEY:
          return GDQY_TARGET_URL
      raise SystemExit(f"no hardcoded URL for {site}/{column}/{article_key}; pass --url explicitly")

  def main(argv: list[str] | None = None) -> int:
      logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s")
      args = _build_parser().parse_args(argv)
      if args.cmd == "fetch":
          url = _resolve_url(args.site, args.column, args.article_key, args.url)
          result = fetch_and_store(args.site, args.column, args.article_key, url)
          if result.get("status") == "skipped":
              print(f"[skip] duplicate url_hash article_id={result['article_id']}")
              return 0
          print(json.dumps(result, ensure_ascii=False, indent=2))
          return 0 if result.get("status") == "ready" else 2
      return 1

  if __name__ == "__main__":
      sys.exit(main())
  ```

  **5. `govcrawler/__main__.py`**：
  ```python
  from govcrawler.cli import main
  import sys
  if __name__ == "__main__":
      sys.exit(main())
  ```

  **6. `tests/test_pipeline_mocked.py`** — 不真跑 Chromium，全面 mock 验证 orchestration 正确性（**可在无 DB、无网络的 CI 下跑——用 monkeypatch 把 Session/fetch_html/download_attachment 都替换掉**）：
  ```python
  import io
  from datetime import datetime
  from unittest.mock import MagicMock
  import pytest

  from govcrawler import pipeline
  from govcrawler.fetcher.browser import FetchResult
  from govcrawler.parser.detail_parser import DetailFields
  from govcrawler.storage.attachments import DownloadedAttachment
  from pathlib import PurePosixPath

  @pytest.fixture
  def fake_session_factory(monkeypatch):
      """Replace get_sessionmaker with a MagicMock session recorder."""
      articles = []
      attachments = []
      crawl_logs = []

      class FakeSession:
          def __init__(self):
              self._dup = None
              self.committed = False
          def __enter__(self): return self
          def __exit__(self, *a): pass
          def execute(self, stmt):
              # Return empty => no duplicate
              res = MagicMock(); res.scalar_one_or_none.return_value = self._dup
              return res
          def add(self, obj):
              cls_name = type(obj).__name__
              if cls_name == "Article":
                  obj.id = len(articles) + 1
                  articles.append(obj)
              elif cls_name == "Attachment":
                  obj.id = len(attachments) + 1
                  attachments.append(obj)
              elif cls_name == "CrawlLog":
                  obj.id = len(crawl_logs) + 1
                  crawl_logs.append(obj)
          def flush(self): pass
          def commit(self): self.committed = True

      def factory():
          return lambda: FakeSession()   # Session() callable

      monkeypatch.setattr(pipeline, "get_sessionmaker", factory)
      return {"articles": articles, "attachments": attachments, "crawl_logs": crawl_logs}

  def test_happy_path_writes_article_and_log(monkeypatch, tmp_path, fake_session_factory):
      # mock fetch_html
      html = '<html><body><h1 class="article-title">测试标题</h1>' \
             '<span class="time">2026-04-10 16:34:22</span>' \
             '<div class="article-content"><p>正文内容，足够长，足够长，足够长，足够长，足够长，足够长。</p></div>' \
             '</body></html>'
      monkeypatch.setattr(pipeline, "fetch_html",
                          lambda url: FetchResult(url=url, final_url=url, status=200, html=html,
                                                  fetched_at=0.0, duration_ms=100, is_challenge=False))
      # redirect writes to tmp_path via settings
      monkeypatch.setenv("DB_URL", "postgresql+psycopg://x/x")   # get_settings() needs it
      monkeypatch.setenv("DATA_DIR", str(tmp_path))
      monkeypatch.setenv("USER_AGENT", "TestBot/1.0")

      # no attachments to download
      r = pipeline.fetch_and_store("gdqy", "szfwj", "post_test",
                                   "https://www.gdqy.gov.cn/x/post_test.html")
      assert r["status"] == "ready"
      assert r["article_id"] == 1
      assert len(fake_session_factory["articles"]) == 1
      assert len(fake_session_factory["crawl_logs"]) == 1
      assert fake_session_factory["crawl_logs"][0].success is True

      # files written
      raw = tmp_path / r["raw_html_path"]
      txt = tmp_path / r["text_path"]
      assert raw.exists() and raw.read_text("utf-8").startswith("<html")
      assert txt.exists() and "正文内容" in txt.read_text("utf-8")

  def test_challenge_page_logged_as_failure(monkeypatch, tmp_path, fake_session_factory):
      monkeypatch.setenv("DB_URL", "postgresql+psycopg://x/x")
      monkeypatch.setenv("DATA_DIR", str(tmp_path))
      monkeypatch.setenv("USER_AGENT", "TestBot/1.0")
      monkeypatch.setattr(pipeline, "fetch_html",
                          lambda url: FetchResult(url=url, final_url=url, status=412,
                                                  html="<html><title>请稍候</title></html>",
                                                  fetched_at=0.0, duration_ms=50, is_challenge=True))
      r = pipeline.fetch_and_store("gdqy", "szfwj", "post_x",
                                   "https://www.gdqy.gov.cn/x/post_x.html")
      assert r["status"] == "failed"
      assert len(fake_session_factory["articles"]) == 0
      assert len(fake_session_factory["crawl_logs"]) == 1
      assert fake_session_factory["crawl_logs"][0].success is False
      assert "challenge" in (fake_session_factory["crawl_logs"][0].error_msg or "").lower()

  def test_attachment_downloaded_and_recorded(monkeypatch, tmp_path, fake_session_factory):
      """Path B coverage (per 01-03-PLAN §附件链路验证).

      CRITICAL: only mock httpx.stream — DO NOT mock download_attachment itself.
      The real streaming/hashing/write code must run, so tmp_path/attachments/... 
      actually contains the file and file_hash is the sha256 of the real bytes.
      """
      import hashlib
      import os
      from contextlib import contextmanager
      from govcrawler.storage import attachments as att_mod

      monkeypatch.setenv("DB_URL", "postgresql+psycopg://x/x")
      monkeypatch.setenv("DATA_DIR", str(tmp_path))
      monkeypatch.setenv("USER_AGENT", "TestBot/1.0")
      html = '<html><body><h1 class="article-title">带附件</h1>' \
             '<span class="time">2026-04-10 10:00:00</span>' \
             '<div class="article-content"><p>正文正文正文正文正文正文正文正文正文正文。</p>' \
             '<a href="https://www.gdqy.gov.cn/attach/a.pdf">公告全文.pdf</a></div></body></html>'
      monkeypatch.setattr(pipeline, "fetch_html",
                          lambda url: FetchResult(url=url, final_url=url, status=200, html=html,
                                                  fetched_at=0.0, duration_ms=100, is_challenge=False))

      # Mock ONLY httpx.stream — the real download_attachment logic (slugify, path-traversal
      # guard, sha256, atomic rename, PG record) still runs against tmp_path.
      FAKE_BYTES = b"%PDF-1.4\n<fake pdf bytes for Path B test>\n" * 32
      EXPECTED_SHA = hashlib.sha256(FAKE_BYTES).hexdigest()

      class _FakeResp:
          headers = {"content-disposition": 'attachment; filename="公告全文.pdf"'}
          def raise_for_status(self): pass
          def iter_bytes(self, chunk_size=65536):
              # yield in two chunks to exercise loop
              mid = len(FAKE_BYTES) // 2
              yield FAKE_BYTES[:mid]
              yield FAKE_BYTES[mid:]

      @contextmanager
      def fake_stream(method, url, **kw):
          yield _FakeResp()

      monkeypatch.setattr(att_mod.httpx, "stream", fake_stream)

      r = pipeline.fetch_and_store("gdqy", "szfwj", "post_test",
                                   "https://www.gdqy.gov.cn/x/post_test.html")
      assert r["status"] == "ready"
      assert r["attachments_downloaded"] == 1
      assert len(fake_session_factory["attachments"]) == 1

      att = fake_session_factory["attachments"][0]
      # PG record assertions
      assert att.file_ext == "pdf"
      assert att.file_hash == EXPECTED_SHA, \
          f"file_hash must be sha256 of real bytes, got {att.file_hash}"
      assert att.size_bytes == len(FAKE_BYTES)
      # file_path stored as POSIX
      assert "/" in att.file_path and "\\" not in att.file_path

      # Real file written under tmp_path/attachments/gdqy/szfwj/2026/04/
      written = tmp_path / att.file_path
      assert written.exists(), f"expected real file at {written}"
      assert written.stat().st_size == len(FAKE_BYTES)
      # sha256 of on-disk bytes matches recorded file_hash
      with open(written, "rb") as f:
          assert hashlib.sha256(f.read()).hexdigest() == EXPECTED_SHA

  def test_duplicate_url_hash_skips(monkeypatch, tmp_path, fake_session_factory):
      monkeypatch.setenv("DB_URL", "postgresql+psycopg://x/x")
      monkeypatch.setenv("DATA_DIR", str(tmp_path))
      monkeypatch.setenv("USER_AGENT", "TestBot/1.0")
      # Make the fake session return an existing article
      existing = MagicMock(); existing.id = 42
      import govcrawler.storage.repo as repo_mod
      monkeypatch.setattr(repo_mod, "get_article_by_url_hash", lambda s, h: existing)
      # fetch_html should NOT be called; make it blow up if called
      def _boom(url): raise AssertionError("fetch_html should not be called on duplicate")
      monkeypatch.setattr(pipeline, "fetch_html", _boom)
      r = pipeline.fetch_and_store("gdqy", "szfwj", "post_dup",
                                   "https://www.gdqy.gov.cn/x/post_dup.html")
      assert r["status"] == "skipped"
      assert r["article_id"] == 42
  ```

  **7. `tests/test_gdqy_smoke.py`** — 真实网络 smoke，**默认跳过**（通过 env var 开启）：
  ```python
  import os
  import pytest
  from sqlalchemy import select

  from govcrawler.pipeline import fetch_and_store
  from govcrawler.db import get_sessionmaker
  from govcrawler.models import Article, CrawlLog
  from govcrawler.sites.gdqy import (
      SITE_ID, COLUMN_ID, TARGET_ARTICLE_KEY, TARGET_ARTICLE_URL,
  )

  RUN_NETWORK = os.environ.get("GOVCRAWLER_RUN_SMOKE") == "1"

  pytestmark = pytest.mark.skipif(
      not RUN_NETWORK,
      reason="real-network smoke; set GOVCRAWLER_RUN_SMOKE=1 to run",
  )

  def test_smoke_gdqy_post_2136593():
      """End-to-end: requires docker-compose db up + alembic upgrade head + patchright chromium."""
      r = fetch_and_store(SITE_ID, COLUMN_ID, TARGET_ARTICLE_KEY, TARGET_ARTICLE_URL)
      assert r["status"] == "ready", f"unexpected: {r}"
      assert r["article_id"]

      # Verify DB row
      Session = get_sessionmaker()
      with Session() as s:
          a = s.execute(select(Article).where(Article.id == r["article_id"])).scalar_one()
          assert a.status == "ready"
          assert a.url_hash and len(a.url_hash) == 64
          assert a.content_simhash and a.content_simhash != "0" * 16
          assert a.raw_html_path and a.text_path
          assert "无人驾驶航空器" in (a.content_text or "") or "无人机" in (a.content_text or "")
          cl = s.execute(select(CrawlLog).where(CrawlLog.article_url == TARGET_ARTICLE_URL)
                         .order_by(CrawlLog.id.desc())).scalars().first()
          assert cl is not None
          assert cl.success is True
          assert cl.strategy == "playwright"
  ```

  约束/注意：
  - `fetch_and_store` 的流程顺序：**文件先落、附件再下、最后一次性写 DB（单事务）**——符合 RESEARCH §System Architecture Diagram。
  - 重复 URL 的快速路径在 `get_article_by_url_hash` 命中后**不启动 Chromium**，节省资源（覆盖 INC-01 的精神，但 INC-01 正式属于 Phase 2；这里只是合理默认）。
  - 挑战页检测路径会写 crawl_logs.success=false + error_msg='ctct_challenge_unresolved'，不写 articles（覆盖 Decision Point 1）。
  - 正文过短（<50 字）时 status='failed' 但仍写 articles，方便人工排查（覆盖 Decision Point 2）。
  - Smoke test 默认跳过：本地开发者执行 `GOVCRAWLER_RUN_SMOKE=1 uv run pytest tests/test_gdqy_smoke.py -s` 才真跑。
  - CLI 对 `duplicate url_hash` 打印 `[skip] duplicate url_hash` 并以 exit 0 结束（`must_haves.truths` 第 7 条）。
  </action>
  <verify>
    <automated>python -c "import ast; [ast.parse(open(f).read()) for f in ['govcrawler/storage/repo.py','govcrawler/pipeline.py','govcrawler/cli.py','govcrawler/__main__.py','tests/test_pipeline_mocked.py','tests/test_gdqy_smoke.py']]" && uv run pytest tests/test_pipeline_mocked.py -x -q && uv run python -m govcrawler fetch --help</automated>
  </verify>
  <acceptance_criteria>
    - `uv run pytest tests/test_pipeline_mocked.py -x -q` 4 个 case 全绿
    - `uv run python -m govcrawler fetch --help` 输出 argparse 帮助（不报 ImportError）
    - `uv run python -m govcrawler fetch` 不带参数退出非 0（argparse required）
    - smoke test 默认被 skip：`uv run pytest tests/test_gdqy_smoke.py -v` 输出 "SKIPPED"
    - **端到端 smoke（手工执行者验证，记录到 SUMMARY）**：
      1. `docker compose up -d db && uv run alembic upgrade head`
      2. `uv run patchright install chromium`（如未跑过）
      3. `GOVCRAWLER_RUN_SMOKE=1 uv run pytest tests/test_gdqy_smoke.py -v` 通过
      4. 或直接 `uv run python -m govcrawler fetch gdqy szfwj post_2136593` 退出 0 且 stdout JSON 含 `"status": "ready"`
      5. `ls data/govcrawler/raw_html/gdqy/szfwj/2026/*/post_2136593.html` 存在且 >10KB
      6. `ls data/govcrawler/articles_text/gdqy/szfwj/2026/*/post_2136593.txt` 存在且含 "无人驾驶航空器"
      7. `psql ...` 执行 `SELECT status, url_hash, content_simhash, raw_html_path, text_path FROM articles WHERE url LIKE '%post_2136593%'` 返回 1 行，status='ready'，url_hash 64 字符，content_simhash 16 字符且非全零，路径为 POSIX 形式（含 /，不含 \）
      8. `SELECT success, strategy, http_status FROM crawl_logs ORDER BY id DESC LIMIT 1` 返回 (true, 'playwright', 200 或 0)
      9. 再跑一次同样命令，CLI 输出 `[skip] duplicate url_hash` 且不启动 Chromium
    - **附件链路验证（二选一，SUMMARY 必须记录选哪条 + 证据日志）**：
      - (A) **真实 gdqy 文章**：执行者从 `https://www.gdqy.gov.cn/gdqy/newxxgk/fgwj/szfwj/` 列表页找到一篇带 PDF 附件的文章，设置 `ATTACHMENT_TEST_ARTICLE_URL/KEY` 后跑 `uv run python -m govcrawler fetch gdqy szfwj <key> --url <url>`。Assert：
        - `attachments_downloaded >= 1`
        - `SELECT COUNT(*) FROM attachments WHERE article_id=<id>` ≥ 1
        - `file_hash` 为 64 字符 hex
        - `data/govcrawler/attachments/gdqy/szfwj/YYYY/MM/<article_key>_<safe_name>.pdf` 实际存在且 size > 0
        - SUMMARY 贴 stdout 的 attachment JSON + `sha256sum` 命令输出作为证据
      - (B) **mocked 测试（仅当 Path A 无合适候选时）**：`tests/test_pipeline_mocked.py::test_attachment_downloaded_and_recorded` 必须**真实写入** `tmp_path/attachments/` 下的文件并断言：
        - 只 mock `httpx.stream`（返回固定 bytes，如 `b"%PDF-1.4\n<fake pdf bytes>\n"`），**不 mock** `download_attachment` 本身——让真实的流式写入逻辑跑完
        - 断言 `os.path.exists(tmp_path / "attachments/gdqy/szfwj/2026/04/<article_key>_<safe_name>.pdf")` 为 True
        - 断言该文件 `size_bytes > 0` 且与 `DownloadedAttachment.size_bytes` 一致
        - 断言 `hashlib.sha256(open(file,"rb").read()).hexdigest() == attachment.file_hash`（证明 `file_hash` 是真实字节的哈希，不是随便填的占位）
        - 断言 `attachment.file_path` 在 PG attachments row 里是 POSIX 形式（含 `/` 不含 `\`）
        - SUMMARY 必须写明 "Path B chosen, evidence: <test log snippet 显示 tmp_path/attachments/... 真实文件路径 + size + sha256>" 并说明为何 Path A 不可用
  </acceptance_criteria>
  <done>
    Phase 1 全部 10 个 REQ-ID 达成：
    - FETCH-02 → fetcher/browser.py + smoke 实测过 ctct 盾
    - PARSE-01..04 → 复用 Plan 02
    - STORE-01 → write_raw_html + 实文件
    - STORE-02 → write_article_text + 实文件
    - STORE-03 → download_attachment + 实文件（或 mocked 覆盖）
    - STORE-04 → articles 表写入 + url_hash + content_simhash
    - STORE-05 → attachments 表写入 + file_hash
    ROADMAP Phase 1 成功标准 1/2/3/5 全部由 smoke test 证明；成功标准 4（选择器失效 GNE 兜底）由 Plan 02 的 tests/test_gne_fallback.py 证明。
  </done>
</task>

</tasks>

<threat_model>
## Trust Boundaries

| Boundary | Description |
|----------|-------------|
| gdqy.gov.cn → Chromium | 对方 HTML/JS 视为不可信，Chromium sandbox 作为第一道防线 |
| Chromium → pipeline.html | 字符串，交给 parsel/GNE/trafilatura |
| Attachment URL → httpx stream → local file | URL 可能含恶意 filename，需 normalize + slugify + path-traversal 校验 |
| pipeline → PostgreSQL | 全走 SQLAlchemy ORM 参数化；无 raw SQL 拼字符串 |

## STRIDE Threat Register

| Threat ID | Category | Component | Disposition | Mitigation Plan |
|-----------|----------|-----------|-------------|-----------------|
| T-01-03-01 | Tampering | 附件文件名含 `../` 或绝对路径（Content-Disposition 注入） | mitigate | `download_attachment` 先 `safe_filename(slugify)` → 再 `final_abs.resolve().relative_to(abs_dir.resolve())` 显式校验，失败 raise |
| T-01-03-02 | Tampering | 附件 URL 为 `javascript:` 或 `file://` | mitigate | `download_attachment` 入口 `normalize_url(url)`，scheme 白名单触发 ValueError（Plan 02 Task 1 已实现白名单，此处复用） |
| T-01-03-03 | DoS | 超大附件把磁盘打爆 | mitigate | `MAX_ATTACHMENT_BYTES = 200MB`，流式边收边累计，超阈值 raise |
| T-01-03-04 | DoS | 对 gdqy.gov.cn 过于频繁请求触发 WAF 封禁 | mitigate | `HostThrottle.wait(url)` 每次请求前调用，默认 5s + 20% jitter，实测本 plan 只抓 1-2 篇 |
| T-01-03-05 | Information Disclosure | DB 密码/UA 中的联系邮箱被日志 leaked | accept | 本阶段日志只打 URL/status/duration；Settings 的 db_url 不会被 print。DB 密码只在 .env（已 .gitignore） |
| T-01-03-06 | Tampering | SQL Injection via title / url / error_msg | mitigate | 全走 SQLAlchemy ORM（`Article(title=...)`），参数化绑定，禁止 raw SQL |
| T-01-03-07 | Spoofing | 挑战页被误判为正文（status 200 软挑战） | mitigate | `is_challenge_page(status, html)` 同时检查 412 与 HTML markers `{"ctct-slider-canvas","请稍候"}`；Phase 1 若仍被绕过，crawl_logs 的 error_msg 会捕获到（`content_text_too_short`） |
| T-01-03-08 | Repudiation | 抓取动作无审计 | mitigate | 每次 fetch 无论成功失败都写 `crawl_logs` 一行 |
</threat_model>

<verification>
- Task 1: fetcher/throttle/storage 模块 11 个 pytest 全绿
- Task 2: pipeline mocked 4 个 pytest 全绿 + CLI help 可用
- **端到端（执行者 MUST 手工跑一次并记录到 SUMMARY）**：
  1. `docker compose up -d db && uv run alembic upgrade head`
  2. `uv run patchright install chromium`
  3. `uv run python -m govcrawler fetch gdqy szfwj post_2136593`
  4. 文件三件套 + PG articles/crawl_logs 三查询全部满足 acceptance_criteria
  5. 重跑 → `[skip] duplicate url_hash`
- Phase 1 全量 pytest：`uv run pytest tests/ -x -q`（排除 smoke）应报告 44 个 PASSED（Plan 02 的 29 + 本 plan 的 15）
</verification>

<success_criteria>
- [ ] `uv run pytest tests/ -x -q --deselect tests/test_gdqy_smoke.py` 全绿
- [ ] `uv run python -m govcrawler fetch gdqy szfwj post_2136593` 端到端跑通，stdout JSON `status=ready`
- [ ] data/govcrawler/raw_html / articles_text / attachments 三目录出现对应文件
- [ ] PG articles 表有 1 行，字段齐全，raw_html_path/text_path 是 POSIX 形式
- [ ] PG crawl_logs 表有 1 行，success=true, strategy='playwright'
- [ ] 重复 URL 触发 skip 分支，不启动 Chromium
- [ ] 附件链路在 mocked 或 real 二选一下被验证
- [ ] **REQ 覆盖自检**：
  - FETCH-02 → fetcher/browser.py + test_gdqy_smoke.py（真网络验证）
  - STORE-01 → write_raw_html + smoke 手工 ls
  - STORE-02 → write_article_text + smoke 手工 ls
  - STORE-03 → download_attachment + test_pipeline_mocked.py attachment case
  - STORE-04 → insert_article + smoke 手工 SELECT
  - STORE-05 → insert_attachments + test_pipeline_mocked.py attachment case（file_hash 字段）
</success_criteria>

<output>
完成后创建 `.planning/phases/01-poc/01-03-SUMMARY.md`，记录：
- 端到端 smoke 的实际输出（stdout JSON + SQL SELECT 的行）
- 附件链路用的是 Path A 还是 Path B（必填）；若 A 则记录 `ATTACHMENT_TEST_ARTICLE_URL` + stdout JSON + `sha256sum` 输出；若 B 则贴 `pytest -v` 对 `test_attachment_downloaded_and_recorded` 的输出（包含 tmp_path 下真实文件路径、size、sha256 三项证据）并解释为何 Path A 不可用
- patchright 实测是否自动过 ctct 盾（是 → Phase 1 完美；否 → 记录挑战页 HTML 节选供 Phase 2 优化）
- 任何对 gdqy selectors 的实时调整（Plan 02 sites/gdqy.py 是否需要补丁）
- 跑完后 `uv run pytest tests/ -x -q` 的 PASSED 数量
- Phase 1 ROADMAP 成功标准 1-5 的逐条确认
</output>
