آموزش کامل Scrapy

python3 -m venv venv
source venv/bin/activate   # لینوکس/macOS
venv\Scripts\activate      # ویندوز

pip install scrapy

scrapy version

scrapy startproject myproject

myproject/
├── scrapy.cfg              # تنظیمات deploy
└── myproject/
    ├── __init__.py
    ├── items.py            # تعریف Item
    ├── middlewares.py      # Middleware سفارشی
    ├── pipelines.py        # Pipeline سفارشی
    ├── settings.py         # تنظیمات پروژه
    └── spiders/
        └── __init__.py

scrapy genspider example example.com

scrapy crawl example

scrapy crawl example -o output.json

import scrapy

class QuotesSpider(scrapy.Spider):
    name = "quotes"
    start_urls = [
        "https://quotes.toscrape.com/page/1/",
    ]

    def parse(self, response):
        for quote in response.css("div.quote"):
            yield {
                "text": quote.css("span.text::text").get(),
                "author": quote.css("small.author::text").get(),
                "tags": quote.css("div.tags a.tag::text").getall(),
            }

        next_page = response.css("li.next a::attr(href)").get()
        if next_page is not None:
            yield response.follow(next_page, callback=self.parse)

class MySpider(scrapy.Spider):
    name = "myspider"

    def __init__(self, category="", *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.start_urls = [f"https://example.com/{category}"]

# اجرا: scrapy crawl myspider -a category=books

# follow — دنبال کردن لینک
yield response.follow("/page/2/", callback=self.parse)

# follow_all — دنبال کردن همه لینک‌ها
yield from response.follow_all(links, callback=self.parse_detail)

# Request دستی
yield scrapy.Request(url, callback=self.parse_detail)

response.css("title::text").get()          # اولین نتیجه
response.css("title::text").getall()       # همه نتایج
response.css("img::attr(src)").get()        # مقدار ویژگی
response.css("div.quote").get()             # HTML کامل

response.xpath("//title/text()").get()
response.xpath('//span[@class="text"]/text()').getall()
response.xpath('//a/@href').get()

import scrapy

class ProductItem(scrapy.Item):
    name = scrapy.Field()
    price = scrapy.Field()
    url = scrapy.Field()
    description = scrapy.Field()

from myproject.items import ProductItem

def parse(self, response):
    item = ProductItem()
    item["name"] = response.css("h1::text").get()
    item["price"] = response.css(".price::text").get()
    item["url"] = response.url
    yield item

from scrapy.loader import ItemLoader
from myproject.items import ProductItem

loader = ItemLoader(item=ProductItem(), selector=response.css("div.product"))
loader.add_css("name", "h2::text")
loader.add_css("price", "span.price::text")
loader.add_value("url", response.url)
yield loader.load_item()

import json

class JsonWriterPipeline:
    def open_spider(self, spider):
        self.file = open("items.jsonl", "w", encoding="utf-8")

    def close_spider(self, spider):
        self.file.close()

    def process_item(self, item, spider):
        line = json.dumps(dict(item), ensure_ascii=False) + "\n"
        self.file.write(line)
        return item

ITEM_PIPELINES = {
    "myproject.pipelines.JsonWriterPipeline": 300,
}

class DuplicatesPipeline:
    def __init__(self):
        self.ids_seen = set()

    def process_item(self, item, spider):
        if item["id"] in self.ids_seen:
            raise DropItem(f"Duplicate item: {item['id']}")
        self.ids_seen.add(item["id"])
        return item

from scrapy.downloadermiddlewares.useragent import UserAgentMiddleware
import random

class RandomUserAgentMiddleware(UserAgentMiddleware):
    user_agent_list = [
        "Mozilla/5.0 (Windows NT 10.0; Win64; x64) ...",
        "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) ...",
    ]

    def process_request(self, request, spider):
        request.headers["User-Agent"] = random.choice(self.user_agent_list)

DOWNLOADER_MIDDLEWARES = {
    "myproject.middlewares.RandomUserAgentMiddleware": 400,
}

# رعایت robots.txt
ROBOTSTXT_OBEY = True

# تأخیر بین درخواست‌ها (ثانیه)
DOWNLOAD_DELAY = 2

# حداکثر درخواست همزمان به یک دامنه
CONCURRENT_REQUESTS_PER_DOMAIN = 2

# User-Agent پیش‌فرض
USER_AGENT = "mybot (+http://www.example.com)"

# فعال‌سازی AutoThrottle
AUTOTHROTTLE_ENABLED = True
AUTOTHROTTLE_START_DELAY = 1
AUTOTHROTTLE_MAX_DELAY = 10

# Cache
HTTPCACHE_ENABLED = True
HTTPCACHE_EXPIRATION_SECS = 86400

# Retry
RETRY_TIMES = 3
RETRY_HTTP_CODES = [500, 502, 503, 504, 408]

scrapy crawl myspider -s DOWNLOAD_DELAY=3 -s LOG_LEVEL=INFO

from scrapy.spiders import CrawlSpider, Rule
from scrapy.linkextractors import LinkExtractor

class MyCrawlSpider(CrawlSpider):
    name = "crawl_example"
    allowed_domains = ["example.com"]
    start_urls = ["https://example.com"]

    rules = (
        Rule(LinkExtractor(allow=r"/category/\d+"), callback="parse_category"),
        Rule(LinkExtractor(allow=r"/product/\d+"), callback="parse_product"),
    )

    def parse_product(self, response):
        yield {
            "title": response.css("h1::text").get(),
            "price": response.css(".price::text").get(),
        }

# Shell تعاملی برای تست
scrapy shell "https://example.com"

# در shell:
response.css("h1::text").get()
response.xpath("//title/text()").get()

# لیست Spiderها
scrapy list

# بررسی قراردادها
scrapy check

scrapy crawl myspider -o data.json       # JSON
scrapy crawl myspider -o data.csv        # CSV
scrapy crawl myspider -o data.xml        # XML
scrapy crawl myspider -o data.jsonl      # JSON Lines

yield scrapy.FormRequest.from_response(
    response,
    formdata={"username": "user", "password": "pass"},
    callback=self.after_login
)

yield scrapy.Request(
    url,
    meta={"proxy": "http://proxy.example.com:8080"},
    callback=self.parse
)