Scrapy代理接入与完整爬虫

export PROXY_USER="替换为控制台代理用户名"
export PROXY_PASS="替换为控制台代理密码"
# 隧道代理固定网关：proxy.123proxy.cn:36923

$env:PROXY_USER="替换为控制台代理用户名"
$env:PROXY_PASS="替换为控制台代理密码"
# 隧道代理固定网关：proxy.123proxy.cn:36923

# requirements:
#   python -m pip install scrapy
# run:
#   scrapy runspider books_spider.py
import os
from urllib.parse import quote

import scrapy
from scrapy.crawler import CrawlerProcess

PROXY_HOST = "proxy.123proxy.cn"
PROXY_PORT = 36923
PROXY_USER = os.environ["PROXY_USER"]
PROXY_PASS = os.environ["PROXY_PASS"]

proxy_url = "http://{}:{}@{}:{}".format(
    quote(PROXY_USER, safe=""),
    quote(PROXY_PASS, safe=""),
    PROXY_HOST,
    PROXY_PORT,
)


class BooksSpider(scrapy.Spider):
    name = "books_with_123proxy"
    allowed_domains = ["books.toscrape.com"]
    start_urls = [
        "https://books.toscrape.com/catalogue/page-1.html"
    ]
    custom_settings = {
        "CONCURRENT_REQUESTS": 8,
        "CONCURRENT_REQUESTS_PER_DOMAIN": 4,
        "DOWNLOAD_DELAY": 0.25,
        "DOWNLOAD_TIMEOUT": 30,
        "RETRY_TIMES": 3,
        "RETRY_HTTP_CODES": [429, 500, 502, 503, 504],
        "ROBOTSTXT_OBEY": True,
        "USER_AGENT": "123Proxy-Developer-Example/1.0",
        "FEEDS": {
            "books.jsonl": {
                "format": "jsonlines",
                "encoding": "utf8",
                "overwrite": True,
            }
        },
        "LOG_LEVEL": "INFO",
    }

    def start_requests(self):
        for url in self.start_urls:
            yield scrapy.Request(
                url,
                callback=self.parse,
                meta={"proxy": proxy_url},
            )

    def parse(self, response):
        for book in response.css("article.product_pod"):
            detail_url = response.urljoin(
                book.css("h3 a::attr(href)").get()
            )
            yield scrapy.Request(
                detail_url,
                callback=self.parse_book,
                meta={
                    "proxy": proxy_url,
                    "list_price": book.css(
                        "p.price_color::text"
                    ).get(),
                },
            )

        next_href = response.css("li.next a::attr(href)").get()
        if next_href:
            yield scrapy.Request(
                response.urljoin(next_href),
                callback=self.parse,
                meta={"proxy": proxy_url},
            )

    def parse_book(self, response):
        table = {
            row.css("th::text").get(): row.css("td::text").get()
            for row in response.css("table.table tr")
        }
        yield {
            "title": response.css("div.product_main h1::text").get(),
            "price": response.meta["list_price"],
            "availability": response.css(
                "div.product_main p.availability::text"
            ).getall()[-1].strip(),
            "upc": table.get("UPC"),
            "product_type": table.get("Product Type"),
            "url": response.url,
        }


if __name__ == "__main__":
    process = CrawlerProcess()
    process.crawl(BooksSpider)
    process.start()