如何为 AI 智能体设计网页抓取的访问层与提取层

from dataclasses import dataclass
from datetime import date
from hashlib import sha256
from html.parser import HTMLParser
import json
import time


class PipelineError(Exception):
    def __init__(self, stage, reason, retry_after=""):
        self.stage, self.reason = stage, reason
        self.retry_after = retry_after
        super().__init__(f"{stage}:{reason}")


@dataclass(frozen=True)
class Reply:
    status: int
    body: str = ""
    content_type: str = "text/html"
    challenge: bool = False
    ready: bool = True
    retry_after: str = ""


def access(fetch, wait=time.sleep):
    # fetch is a read-only adapter; all values below are application policy.
    for attempt in range(2):
        try:
            reply = fetch()
        except TimeoutError:
            if attempt == 0:
                wait(0.5)
                continue
            raise PipelineError("access", "timeout_exhausted")
        if reply.status == 429:
            # Pass Retry-After to a shared scheduler; do not retry here.
            raise PipelineError("access", "defer_rate_limit", reply.retry_after)
        if reply.status in (401, 403):
            raise PipelineError("access", "authorization_review")
        if reply.challenge:
            raise PipelineError("access", "challenge_review")
        if reply.status == 503 and reply.retry_after:
            raise PipelineError("access", "defer_service", reply.retry_after)
        if reply.status in (502, 503, 504) and attempt == 0:
            wait(0.5)
            continue
        if reply.status != 200:
            raise PipelineError("access", "http_status")
        if reply.content_type.split(";")[0].strip().lower() != "text/html":
            raise PipelineError("access", "representation_mismatch")
        if not reply.ready or not reply.body.strip():
            raise PipelineError("access", "incomplete_snapshot")
        return reply.body
    raise PipelineError("access", "attempts_exhausted")


class BulletinParser(HTMLParser):
    # This small parser supports only the documented fixture markup.
    def __init__(self):
        super().__init__(convert_charrefs=True)
        self.rows, self.current, self.in_title = [], None, False

    def handle_starttag(self, tag, attrs):
        attrs = dict(attrs)
        if tag == "article":
            if self.current is not None:
                raise PipelineError("extraction", "nested_record")
            self.current = {"id": attrs.get("data-id", ""),
                            "published": attrs.get("data-published", ""),
                            "title_parts": [], "title_count": 0}
        elif tag == "h2" and self.current is not None:
            self.current["title_count"] += 1
            self.in_title = True

    def handle_data(self, data):
        if self.current is not None and self.in_title:
            self.current["title_parts"].append(data)

    def handle_endtag(self, tag):
        if tag == "h2":
            self.in_title = False
        if tag == "article" and self.current is not None:
            self.rows.append(self.current)
            self.current, self.in_title = None, False


def extract(html):
    parser = BulletinParser()
    parser.feed(html)
    parser.close()
    if parser.current is not None or not parser.rows:
        raise PipelineError("extraction", "record_structure")
    records, seen = [], set()
    for row in parser.rows:
        title = " ".join("".join(row["title_parts"]).split())
        if not row["id"].strip() or not title or row["title_count"] != 1:
            raise PipelineError("extraction", "required_field")
        try:
            published = date.fromisoformat(row["published"]).isoformat()
        except ValueError:
            raise PipelineError("extraction", "invalid_date")
        if row["id"] in seen:
            raise PipelineError("extraction", "duplicate_id")
        seen.add(row["id"])
        records.append({"id": row["id"], "title": title,
                        "published": published})
    return records


def run(fetch, archive, wait=time.sleep):
    html = access(fetch, wait)
    digest = sha256(html.encode("utf-8")).hexdigest()
    archive[digest] = html  # In-memory evidence retained even if parsing fails.
    records = extract(html)
    return {"schema_version": "bulletins.v1", "source_id": "fixture:bulletins",
            "snapshot_sha256": digest,
            "records": records}


if __name__ == "__main__":
    html = ('<article data-id="notice-1" data-published="2026-09-10">'
            '<h2>Maintenance window announced</h2></article>')
    replies = iter([Reply(503), Reply(200, html)])
    archive = {}
    output = run(lambda: next(replies), archive, wait=lambda seconds: None)
    print(json.dumps(output, indent=2))

{
  "schema_version": "bulletins.v1",
  "source_id": "fixture:bulletins",
  "snapshot_sha256": "6ed8df5a98ee53e2889feb5ef7ed4dd8d549dba82882580418d4ca9656b7d46b",
  "records": [
    {
      "id": "notice-1",
      "title": "Maintenance window announced",
      "published": "2026-09-10"
    }
  ]
}