> ## Documentation Index
> Fetch the complete documentation index at: https://help.miyaip.com/llms.txt
> Use this file to discover all available pages before exploring further.

# Python 网页抓取代理配置：Requests 与 Scrapy

> 使用 Python Requests 和 Scrapy 配置代理、请求获准页面、解析字段并导出 JSON 或 CSV，处理重试、会话和抓取故障。

这份教程用一个获授权的公开页面完成代理配置、请求、解析和文件导出。先用 `example.com` 验证流程，再替换为获准的目标和运营联系地址，并遵守目标公布的请求限制。

## 选择代理与会话

| 任务 | 选择与配置 |
| - | - |
| 稳定且允许采集的公开端点 | 从数据中心代理或一个静态端点开始，便于监控和核算成本。 |
| 需要固定地区与连续身份的任务 | 从[静态住宅代理](/zh/proxy/static-residential/overview)提取连接信息；在同一任务中复用代理与会话。 |
| 相互独立、无状态的采集任务 | 参考[动态住宅代理](/zh/proxy/dynamic-residential/overview)配置地区和轮换；在任务之间轮换。 |
| 确实需要移动网络出口的地区测试 | 参考[移动代理](/zh/proxy/mobile/overview)选择对应运营商网络。 |

代理解决网络出口选择。你仍需控制请求速率、检查选择器并验证记录；HTTP 200 不能证明采集字段完整。

## 使用 Requests 完成采集与导出

<Steps>
  <Step title="创建 Python 环境">
    在新的工作目录运行以下命令。SOCKS 代理需要 `requests[socks]` 中的可选依赖。

    ```powershell theme={null}
    python -m venv .venv
    .\.venv\Scripts\Activate.ps1
    python -m pip install 'requests[socks]'
    ```

    macOS 或 Linux 的激活命令为 `. .venv/bin/activate`。下载 [web\_scraping.py](/examples/web_scraping.py) 和 [retry\_after.py](/examples/retry_after.py)，将两个文件放在同一目录。
  </Step>

  <Step title="设置目标和代理连接信息">
    以下都是占位值。将完整代理 URL 注入 `SCRAPING_PROXY_URL`，协议、主机、端口和认证信息必须与所提取的代理匹配。HTTP 代理使用 `http://`，SOCKS5 代理可使用 `socks5h://`。URL 中用户名或密码的特殊字符需要百分号编码。

    ```powershell theme={null}
    $env:SCRAPING_PROXY_URL = 'http://YOUR_USERNAME:YOUR_PASSWORD@proxy.example.com:1234'
    $env:TARGET_URL = 'https://example.com/'
    $env:ALLOWED_HOSTS = 'example.com'
    python -c "import os; assert os.environ.get('SCRAPING_PROXY_URL')"
    ```

    正式任务通过密钥管理工具注入真实凭据。不要把真实值写入共享脚本或打印到日志。修改目标时同步修改 `ALLOWED_HOSTS`；示例只允许白名单中的 HTTPS 主机和默认端口。将脚本中的 `contact@example.com` 替换为运营联系地址。
  </Step>

  <Step title="保存完整脚本">
    主程序复用一个 `requests.Session`，同时为 HTTP 与 HTTPS 请求设置代理，读取首个 `title` 和 `h1`，检查非空后导出 JSON 与 CSV。它关闭自动重定向，避免未经检查地切换目标。

    <CodeGroup>
      ```python web_scraping.py theme={null}
      """Collect an approved public HTML page through an explicitly configured proxy."""

      import csv
      import json
      import os
      from html.parser import HTMLParser
      from pathlib import Path
      from urllib.parse import urlsplit

      import requests

      from retry_after import (
          RetryAfterOutOfRange,
          RetryDeferred,
          retry_delay,
          wait_before_retry,
      )

      MAX_ATTEMPTS = 3
      MAX_WAIT_SECONDS = 60  # Waiting budget, not a cap on Retry-After.
      TIMEOUT = (10, 30)
      TRANSIENT_STATUSES = {429, 500, 502, 503, 504}


      class FetchFailed(RuntimeError):
          pass


      class PageParser(HTMLParser):
          def __init__(self):
              super().__init__()
              self.capture = None
              self.parts = []
              self.title = ""
              self.h1 = ""

          def handle_starttag(self, tag, attrs):
              if not self.capture and tag in ("title", "h1") and not getattr(self, tag):
                  self.capture = tag
                  self.parts = []

          def handle_data(self, data):
              if self.capture:
                  self.parts.append(data)

          def handle_endtag(self, tag):
              if tag == self.capture:
                  setattr(self, tag, " ".join("".join(self.parts).split()))
                  self.capture = None
                  self.parts = []


      def fetch(session, url, proxies, wait_budget=None):
          wait_budget = wait_budget if wait_budget is not None else {
              "remaining_seconds": MAX_WAIT_SECONDS
          }
          for attempt in range(1, MAX_ATTEMPTS + 1):
              try:
                  response = session.get(
                      url, proxies=proxies, timeout=TIMEOUT, allow_redirects=False
                  )
              except requests.exceptions.ProxyError:
                  raise FetchFailed("proxy_connection_or_authentication_failed") from None
              except requests.exceptions.Timeout:
                  if attempt == MAX_ATTEMPTS:
                      raise FetchFailed("timeout_attempts_exhausted") from None
                  wait_before_retry(2 ** (attempt - 1), wait_budget)
                  continue
              except requests.exceptions.RequestException:
                  raise FetchFailed("request_failed") from None

              status = response.status_code
              if status == 407:
                  response.close()
                  raise FetchFailed("proxy_authentication_failed")
              if status in TRANSIENT_STATUSES:
                  delay = retry_delay(response.headers.get("Retry-After"), 2 ** (attempt - 1))
                  response.close()
                  if attempt == MAX_ATTEMPTS:
                      raise RetryDeferred(delay, "maximum_attempts_reached")
                  wait_before_retry(delay, wait_budget)
                  continue
              if not 200 <= status < 300:
                  response.close()
                  raise FetchFailed("http_status_{}".format(status))
              return response
          raise FetchFailed("maximum_attempts_reached")


      def parse(response, url):
          parser = PageParser()
          parser.feed(response.text)
          if not parser.title or not parser.h1:
              raise FetchFailed("required_fields_missing")
          return {"url": url, "title": parser.title, "h1": parser.h1}


      def save(records, output_dir="output"):
          output = Path(output_dir)
          output.mkdir(parents=True, exist_ok=True)
          (output / "results.json").write_text(
              json.dumps(records, ensure_ascii=False, indent=2), encoding="utf-8"
          )
          with (output / "results.csv").open("w", newline="", encoding="utf-8-sig") as file:
              writer = csv.DictWriter(file, fieldnames=["url", "title", "h1"])
              writer.writeheader()
              writer.writerows(records)


      def main():
          proxy_url = os.environ.get("SCRAPING_PROXY_URL", "")
          proxy = urlsplit(proxy_url)
          if proxy.scheme not in {"http", "https", "socks5", "socks5h"} or not proxy.hostname:
              raise FetchFailed("invalid_or_missing_proxy_configuration")
          url = os.environ.get("TARGET_URL", "https://example.com/")
          target = urlsplit(url)
          allowed_hosts = set(os.environ.get("ALLOWED_HOSTS", "example.com").split(","))
          if target.scheme != "https" or target.hostname not in allowed_hosts:
              raise FetchFailed("target_not_allowlisted")
          if target.username or target.password or target.port not in {None, 443}:
              raise FetchFailed("invalid_target")

          budget = {"remaining_seconds": MAX_WAIT_SECONDS}
          proxies = {"http": proxy_url, "https": proxy_url}
          with requests.Session() as session:
              session.trust_env = False
              session.headers["User-Agent"] = "AuthorizedResearchBot/1.0 (+contact@example.com)"
              response = fetch(session, url, proxies, budget)
              try:
                  records = [parse(response, url)]
              finally:
                  response.close()
          save(records)
          print(json.dumps({"outcome": "success", "status": response.status_code,
                            "records": len(records), "output": "output/results.json,output/results.csv"}))


      if __name__ == "__main__":
          try:
              main()
          except RetryDeferred as error:
              print(json.dumps({"outcome": "deferred", "retry_at": error.retry_at,
                                "reason": error.reason}))
              raise SystemExit(2)
          except RetryAfterOutOfRange:
              print(json.dumps({"outcome": "failed", "reason": "retry_after_out_of_range",
                                "action": "manual_rescheduling_required"}))
              raise SystemExit(1)
          except Exception as error:
              reason = str(error) if isinstance(error, FetchFailed) else type(error).__name__
              print(json.dumps({"outcome": "failed", "reason": reason}))
              raise SystemExit(1)
      ```

      ```python retry_after.py theme={null}
      """Retry-After parsing shared by the Requests and Scrapy examples."""

      import math
      import time
      from datetime import datetime, timedelta, timezone
      from email.utils import parsedate_to_datetime


      class RetryAfterOutOfRange(RuntimeError):
          pass


      def utc_now():
          return datetime.now(timezone.utc)


      def retry_delay(value, fallback, now=None):
          """Return the full requested delay; never shorten a valid Retry-After."""
          now = now or utc_now()
          text = "" if value is None else str(value).strip()
          if text.isascii() and text.isdecimal():
              digits = text.lstrip("0") or "0"
              available = datetime.max.replace(tzinfo=timezone.utc) - now
              largest = str(available.days * 86400 + available.seconds)
              if len(digits) > len(largest) or (len(digits) == len(largest) and digits > largest):
                  raise RetryAfterOutOfRange("manual_rescheduling_required")
              return int(digits)
          if not text:
              return fallback
          try:
              retry_at = parsedate_to_datetime(text)
              if retry_at.tzinfo is None:
                  retry_at = retry_at.replace(tzinfo=timezone.utc)
              return max(math.ceil((retry_at - now).total_seconds()), 0)
          except (TypeError, ValueError, OverflowError):
              return fallback


      class RetryDeferred(RuntimeError):
          def __init__(self, delay, reason, now=None):
              try:
                  retry_at = (now or utc_now()) + timedelta(seconds=delay)
              except (OverflowError, ValueError):
                  raise RetryAfterOutOfRange("manual_rescheduling_required") from None
              self.retry_at = retry_at.isoformat().replace("+00:00", "Z")
              self.reason = reason
              super().__init__("retry_deferred")


      def wait_before_retry(delay, wait_budget, sleep=None):
          if delay > wait_budget["remaining_seconds"]:
              raise RetryDeferred(delay, "waiting_budget_exhausted")
          if delay > 0:
              (sleep or time.sleep)(delay)
          wait_budget["remaining_seconds"] -= delay
      ```
    </CodeGroup>
  </Step>

  <Step title="运行并检查结果">
    ```powershell theme={null}
    python web_scraping.py
    python --version
    python -m pip show requests
    ```

    成功运行后，打开 `output/results.json` 和 `output/results.csv`，核对 `url`、`title`、`h1` 的值、编码与记录数量。控制台的 `outcome=success` 来自本次运行，不是预置的验证结果。根据实际字段规则处理重复记录后，再将数据送到后续系统。
  </Step>
</Steps>

## 使用 Scrapy

这个单文件 Spider 通过真实请求的 `meta["proxy"]` 配置代理，并用下载器中间件覆盖机器人规则等附加请求。此路径使用 Scrapy 的 `HttpProxyMiddleware`，请提供 HTTP 或 HTTPS 代理 URL。将 [scrapy\_example.py](/examples/scrapy_example.py) 与 `retry_after.py` 放在同一目录，沿用上述环境变量。

```powershell theme={null}
python -m pip install scrapy
New-Item -ItemType Directory -Path output -Force
scrapy runspider scrapy_example.py -O output/scrapy-results.json
```

```python scrapy_example.py theme={null}
"""Run with: scrapy runspider scrapy_example.py -O output/scrapy-results.json"""

import json
import os
from urllib.parse import urlsplit

import scrapy
from scrapy.exceptions import CloseSpider, IgnoreRequest
from scrapy.utils.defer import deferred_from_coro

from retry_after import RetryAfterOutOfRange, RetryDeferred, retry_delay


class ProxyRouteMiddleware:
    def process_request(self, request, spider):
        if getattr(spider, "_stopped", False):
            raise IgnoreRequest("task_already_stopped")
        # Also proxy framework-generated requests, including robots.txt.
        request.meta["proxy"] = os.environ["SCRAPING_PROXY_URL"]

    def process_response(self, request, response, spider):
        # Apply server waits to auxiliary requests before their own callbacks run.
        if response.status in {407, 429, 500, 502, 503, 504}:
            reason = spider.stop_response(response)
            # Do not await closing here: this download is still active.
            deferred_from_coro(spider.crawler.engine.close_spider_async(reason=reason))
            raise IgnoreRequest(reason)
        return response


class ExampleSpider(scrapy.Spider):
    name = "authorized_example"
    handle_httpstatus_list = [407, 429, 500, 502, 503, 504]
    custom_settings = {
        "HTTPPROXY_ENABLED": True,
        "DOWNLOADER_MIDDLEWARES": {
            __name__ + ".ProxyRouteMiddleware": 740,
            "scrapy.downloadermiddlewares.httpproxy.HttpProxyMiddleware": 750,
        },
        "ROBOTSTXT_OBEY": True,
        "AUTOTHROTTLE_ENABLED": True,
        "AUTOTHROTTLE_START_DELAY": 2.0,
        "AUTOTHROTTLE_MAX_DELAY": 60.0,
        "AUTOTHROTTLE_TARGET_CONCURRENCY": 1.0,
        "CONCURRENT_REQUESTS_PER_DOMAIN": 2,
        "DOWNLOAD_DELAY": 1.0,
        "DOWNLOAD_TIMEOUT": 30,
        "RETRY_ENABLED": False,
        "REDIRECT_ENABLED": False,
        "LOG_ENABLED": False,
    }

    async def start(self):
        proxy_url = os.environ.get("SCRAPING_PROXY_URL", "")
        proxy = urlsplit(proxy_url)
        if proxy.scheme not in {"http", "https"} or not proxy.hostname:
            print(json.dumps({"outcome": "failed", "reason": "HTTP_proxy_required"}))
            raise CloseSpider("invalid_proxy_configuration")
        url = os.environ.get("TARGET_URL", "https://example.com/")
        target = urlsplit(url)
        allowed_hosts = set(os.environ.get("ALLOWED_HOSTS", "example.com").split(","))
        if target.scheme != "https" or target.hostname not in allowed_hosts:
            print(json.dumps({"outcome": "failed", "reason": "target_not_allowlisted"}))
            raise CloseSpider("target_not_allowlisted")
        if target.username or target.password or target.port not in {None, 443}:
            raise CloseSpider("invalid_target")
        yield scrapy.Request(
            url, meta={"proxy": proxy_url}, callback=self.parse, errback=self.failed
        )

    def failed(self, failure):
        if getattr(self, "_stopped", False):
            return
        print(json.dumps({"outcome": "failed", "reason": "request_failed"}))
        raise CloseSpider("request_failed")

    def stop_response(self, response):
        reason = "proxy_authentication_failed" if response.status == 407 else "reschedule_required"
        if getattr(self, "_stopped", False):
            return reason
        self._stopped = True
        if response.status == 407:
            print(json.dumps({"outcome": "failed", "reason": "proxy_authentication_failed"}))
        else:
            header = response.headers.get(b"Retry-After", b"").decode("ascii", "replace")
            try:
                deferred = RetryDeferred(retry_delay(header, 1), "reschedule_required")
                print(json.dumps({"outcome": "deferred", "retry_at": deferred.retry_at,
                                  "status": response.status, "reason": deferred.reason}))
            except RetryAfterOutOfRange:
                print(json.dumps({"outcome": "failed", "reason": "retry_after_out_of_range"}))
        return reason

    def parse(self, response):
        if getattr(self, "_stopped", False):
            return
        if response.status in {407, 429, 500, 502, 503, 504}:
            raise CloseSpider(self.stop_response(response))
        record = {"url": response.url,
                  "title": response.css("title::text").get(default="").strip(),
                  "h1": response.css("h1::text").get(default="").strip()}
        if not record["title"] or not record["h1"]:
            raise CloseSpider("required_fields_missing")
        print(json.dumps({"outcome": "success", "status": response.status, "records": 1}))
        yield record
```

Spider 使用异步 `start()`，启用机器人规则、AutoThrottle 和每个域名的并发限制。中间件对所有响应检查认证失败、429 和列出的暂时性服务错误，覆盖 `robots.txt` 等附加请求。遇到限流或暂时性错误时输出 UTC `retry_at`，设置停止标记并安排关闭爬虫；后续请求会被拒绝，留给外部调度器恢复。它关闭自动重试和重定向，避免其他组件提前重发请求。框架原始日志被关闭，控制台只输出预先定义的错误类别和实际导出结果。

## 控制重试与故障处理

Requests 示例每个 URL 最多尝试 3 次，连接和读取超时分别为 10 秒、30 秒，任务累计重试等待预算为 60 秒。预算限制本次进程愿意等待的时间，**不会把有效的 `Retry-After` 截短为 60 秒**。响应头可使用秒数或 HTTP 日期。超过剩余等待预算，或最后一次尝试仍遇到暂时性错误时，脚本输出 `outcome=deferred`、UTC `retry_at` 和原因，并以退出码 `2` 结束。只在响应头缺失或无效时使用有限指数退避。调度器必须在 `retry_at` 或之后恢复任务，示例本身不创建调度。

| 观察到的结果 | 处理方式 |
| - | - |
| 407 或 `requests.exceptions.ProxyError` | 停止；核对协议、主机、端口、用户名格式、百分号编码、账户状态与所选认证方式的 IP 白名单，不重复提交相同错误配置。 |
| 429、500、502、503、504 | 完整遵守 `Retry-After`；核对并发、下载间隔和每域名请求数，降低请求速率。超预算时延后任务，不立即更换 IP 重试。 |
| 请求超时 | 检查网关连通性、DNS 和路由，区分连接与读取耗时；在次数和等待预算内退避，次数耗尽后停止。 |
| `title` 或 `h1` 缺失 | 检查页面结构、内容类型、目标 URL、语言和选择器；隔离空记录，更新解析器后再接收数据。 |
| `retry_after_out_of_range` | 服务器要求的时间超出运行时 UTC 日期范围；停止并人工安排下次运行。 |

## 接入任务队列前验证

先运行一个获授权的请求，另外核对实际出口地区。检查两个输出文件的字段、编码和重复记录处理。用受控响应测试 407、429、HTTP 日期和超时路径，核对延后任务没有提前重试。

生产任务还应明确队列、目标规则、限速、会话所有权、记录校验与解析失败隔离。日志保留时间、状态类别和耗时，避免保存代理凭据。只在有效记录率、目标响应和成本都满足任务约束后提高并发。

代理配置参考 [Requests 官方文档](https://requests.readthedocs.io/en/latest/user/advanced/#proxies) 和 [Scrapy HttpProxyMiddleware](https://docs.scrapy.org/en/latest/topics/downloader-middleware.html#httpproxymiddleware)。重试时间格式见 [RFC 9110：Retry-After](https://www.rfc-editor.org/rfc/rfc9110.html#name-retry-after)。


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.