如何使用 Python 和 Scrapeless 抓取多个网页

import csv
import hashlib
import json
import os
from datetime import datetime, timezone
from pathlib import Path
from urllib.parse import urlsplit, urlunsplit
import requests
from bs4 import BeautifulSoup

INPUT = [
    "https://quotes.toscrape.com/page/1/",
    "https://quotes.toscrape.com/page/2/",
    "https://quotes.toscrape.com/page/2/#top",
    "https://quotes.toscrape.com/page/99/",
    "https://quotes.toscrape.com/missing/"
]
MODE = os.environ.get("FETCH_MODE", "direct")
if MODE not in {"direct", "unlocker"}:
    raise ValueError("Unsupported acquisition mode")

def normalize(url):
    parts = urlsplit(url)
    if parts.scheme != "https" or parts.hostname != "quotes.toscrape.com":
        raise ValueError("URL is outside the permitted demo scope")
    return urlunsplit((parts.scheme, parts.netloc.lower(),
                      parts.path or "/", parts.query, ""))

queue = list(dict.fromkeys(normalize(url) for url in INPUT))
pages = []
records = {}
duplicate_records = 0
with requests.Session() as client:
    for url in queue:
        page = {"requested_url": url}
        try:
            if MODE == "unlocker":
                response = client.post(
                    "https://api.scrapeless.com/api/v2/unlocker/request",
                    headers={"x-api-token": os.environ["SCRAPELESS_API_KEY"]},
                    json={"actor": "unlocker.webunlocker",
                          "proxy": {"country": "ANY"},
                          "input": {"url": url, "jsRender": {
                              "enabled": True, "response": {"type": "html"}}}},
                    timeout=60
                )
                response.raise_for_status()
                envelope = response.json()
                html = envelope.get("data")
                if envelope.get("code") != 200 or not isinstance(html, str):
                    raise ValueError("Expected a successful HTML envelope")
                page["final_url"] = None
            else:
                response = client.get(url, timeout=30)
                response.raise_for_status()
                html = response.text
                page["final_url"] = response.url
            soup = BeautifulSoup(html, "html.parser")
            heading = soup.select_one("h1 a")
            grids = soup.select("div.container > div.row > div.col-md-8")
            if not heading or heading.get_text(strip=True) != "Quotes to Scrape" or not grids:
                raise ValueError("Source identity does not match the demo contract")
            grid = grids[-1]
            cards = grid.select("div.quote")
            if not cards and (grid.select_one("ul.pager") is None or
                              "No quotes found!" not in grid.get_text(" ", strip=True)):
                raise ValueError("Empty state is not confirmed")
            extracted = []
            for card in cards:
                text_node = card.select_one("span.text")
                author_node = card.select_one("small.author")
                if not text_node or not author_node:
                    raise ValueError("Required source field missing")
                text = text_node.get_text(strip=True)
                author = author_node.get_text(strip=True)
                if not text or not author:
                    raise ValueError("Required source field empty")
                digest = hashlib.sha256((text + "\n" + author).encode()).hexdigest()
                extracted.append({"record_id": digest, "author": author,
                                  "tags": [x.get_text(strip=True) for x in card.select("a.tag")],
                                  "source_urls": [url]})
            for record in extracted:
                digest = record["record_id"]
                if digest in records:
                    duplicate_records += 1
                    records[digest]["source_urls"].append(url)
                else:
                    records[digest] = record
            page.update(status="completed" if cards else "empty", records=len(cards),
                        captured_at=datetime.now(timezone.utc).isoformat())
        except (requests.RequestException, ValueError, KeyError) as error:
            page.update(status="failed", reason=type(error).__name__)
        pages.append(page)

rows = list(records.values())
with Path("records.jsonl").open("w", encoding="utf-8") as output:
    for row in rows:
        output.write(json.dumps(row, ensure_ascii=False) + "\n")
with Path("records.csv").open("w", encoding="utf-8", newline="") as output:
    writer = csv.DictWriter(output, fieldnames=["record_id", "author", "tags", "source_urls"])
    writer.writeheader()
    for row in rows:
        writer.writerow({**row, "tags": json.dumps(row["tags"], ensure_ascii=False),
                         "source_urls": json.dumps(row["source_urls"])})
with Path("records.csv").open(encoding="utf-8", newline="") as source:
    csv_rows = list(csv.DictReader(source))
jsonl_rows = [json.loads(line) for line in Path("records.jsonl").read_text().splitlines()]
assert [r["record_id"] for r in csv_rows] == [r["record_id"] for r in jsonl_rows]
summary = {"mode": MODE, "input_urls": len(INPUT), "duplicate_inputs": len(INPUT)-len(queue),
           "queued_pages": len(queue), "completed": sum(p["status"]=="completed" for p in pages),
           "empty": sum(p["status"]=="empty" for p in pages),
           "failed": sum(p["status"]=="failed" for p in pages),
           "unique_records": len(rows), "duplicate_records": duplicate_records,
           "csv_rows": len(csv_rows), "jsonl_rows": len(jsonl_rows)}
assert summary["completed"] + summary["empty"] + summary["failed"] == len(queue)
Path("pages.json").write_text(json.dumps(pages, indent=2), encoding="utf-8")
Path("summary.json").write_text(json.dumps(summary, indent=2), encoding="utf-8")
print(json.dumps(summary))