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

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

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