""" Moteur Scrapy - spider, middlewares, pipelines, runner asyncio-compatible. L'astuce clé : Scrapy tourne sur Twisted, mais on l'intègre dans FastAPI (asyncio) via scrapy.utils.reactor.install_reactor('twisted.internet.asyncioreactor') appelé AVANT tout import de Scrapy. Le runner est lancé dans un thread dédié pour ne pas bloquer la boucle asyncio principale. """ from __future__ import annotations import asyncio import logging import time from concurrent.futures import ThreadPoolExecutor from typing import Any, Optional from urllib.parse import urlparse # Doit être le tout premier import Scrapy pour choisir l'asyncio reactor import scrapy.utils.reactor # noqa: F401 (side-effect import order matters) from scrapy import Spider, signals from scrapy.crawler import CrawlerRunner from scrapy.http import Response from scrapy.item import Field, Item from scrapy.utils.project import get_project_settings from twisted.internet import defer from models import settings logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # Scrapy Items # --------------------------------------------------------------------------- class PageItem(Item): """Item Scrapy transportant toutes les données d'une page scrappée.""" url = Field() final_url = Field() status_code = Field() html = Field() download_time = Field() error = Field() # --------------------------------------------------------------------------- # Scrapy Settings factory # --------------------------------------------------------------------------- def make_scrapy_settings(request_timeout: int, verify_ssl: bool) -> dict[str, Any]: """Construit les settings Scrapy depuis notre config Pydantic.""" return { # Identification "BOT_NAME": "sota-scraper", "USER_AGENT": settings.user_agent, # Concurrence et politesse "CONCURRENT_REQUESTS": settings.scrapy_concurrent_requests, "CONCURRENT_REQUESTS_PER_DOMAIN": settings.scrapy_concurrent_per_domain, "DOWNLOAD_DELAY": settings.scrapy_download_delay, "RANDOMIZE_DOWNLOAD_DELAY": True, # ±50% du DOWNLOAD_DELAY # Timeouts "DOWNLOAD_TIMEOUT": request_timeout, # Redirections "REDIRECT_ENABLED": settings.follow_redirects, "REDIRECT_MAX_TIMES": 10, # SSL "VERIFY_SSL": verify_ssl, # Retry middleware intégré "RETRY_ENABLED": True, "RETRY_TIMES": settings.max_retries, "RETRY_HTTP_CODES": [429, 500, 502, 503, 504, 522, 524, 408], "RETRY_BACKOFF_BASE": settings.retry_backoff, # Compression automatique "COMPRESSION_ENABLED": True, # Désactiver robots.txt (scraper général) "ROBOTSTXT_OBEY": False, # Cookies désactivés par défaut (moins de fingerprinting) "COOKIES_ENABLED": False, # Logging minimal (structlog gère le reste) "LOG_ENABLED": False, # Telnet console désactivée "TELNETCONSOLE_ENABLED": False, # Reactor asyncio "TWISTED_REACTOR": "twisted.internet.asyncioreactor.AsyncioSelectorReactor", # Fingerprinter v2.7 — supprime le DeprecationWarning de la valeur par défaut '2.6' "REQUEST_FINGERPRINTER_IMPLEMENTATION": "2.7", # Middlewares downloader (ordre : 100 = premier) "DOWNLOADER_MIDDLEWARES": { "scrapy.downloadermiddlewares.useragent.UserAgentMiddleware": None, "scrapy.downloadermiddlewares.retry.RetryMiddleware": 550, "scrapy.downloadermiddlewares.redirect.RedirectMiddleware": 600, "scrapy.downloadermiddlewares.httpcompression.HttpCompressionMiddleware": 810, }, # Pas de pipeline ici — on récupère l'item via signal dans le runner "ITEM_PIPELINES": {}, } # --------------------------------------------------------------------------- # Spider générique # --------------------------------------------------------------------------- class SinglePageSpider(Spider): """ Spider minimaliste : une URL → un PageItem. Utilisé par ScrapyRunner pour chaque requête individuelle. On lui passe les headers custom via `custom_headers`. """ name = "single_page" # custom_settings est surchargé à l'instanciation via kwargs def __init__( self, url: str, custom_headers: Optional[dict[str, str]] = None, *args: Any, **kwargs: Any, ): super().__init__(*args, **kwargs) self.start_urls = [url] self.custom_headers = custom_headers or {} self._download_start: float = 0.0 def start_requests(self): headers = { "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8", "Accept-Language": "en-US,en;q=0.9,fr;q=0.8", "Accept-Encoding": "gzip, deflate, br", "DNT": "1", "Connection": "keep-alive", "Upgrade-Insecure-Requests": "1", **self.custom_headers, } self._download_start = time.perf_counter() yield scrapy.Request( url=self.start_urls[0], headers=headers, callback=self.parse, errback=self.errback, dont_filter=True, ) def parse(self, response: Response): download_time = time.perf_counter() - self._download_start yield PageItem( url=self.start_urls[0], final_url=str(response.url), status_code=response.status, html=response.text, download_time=round(download_time, 4), error=None, ) def errback(self, failure): download_time = time.perf_counter() - self._download_start logger.warning("scrapy_error url=%s err=%s", self.start_urls[0], repr(failure)) yield PageItem( url=self.start_urls[0], final_url=self.start_urls[0], status_code=None, html="", download_time=round(download_time, 4), error=str(failure.value), ) # --------------------------------------------------------------------------- # Runner asyncio-compatible # --------------------------------------------------------------------------- class ScrapyRunner: """ Exécute un crawl Scrapy et retourne le premier PageItem via un Future asyncio. Architecture : - Scrapy / Twisted tourne dans un ThreadPoolExecutor dédié (1 thread). - La boucle asyncio principale attend via asyncio.Future. - Les résultats sont transmis via call_soon_threadsafe pour rester thread-safe. """ def __init__(self) -> None: self._executor = ThreadPoolExecutor(max_workers=1, thread_name_prefix="scrapy") self._loop: Optional[asyncio.AbstractEventLoop] = None def _get_loop(self) -> asyncio.AbstractEventLoop: if self._loop is None: self._loop = asyncio.get_event_loop() return self._loop async def fetch( self, url: str, *, timeout: int = 30, verify_ssl: bool = True, custom_headers: Optional[dict[str, str]] = None, ) -> PageItem: """ Lance un crawl Scrapy pour `url` et retourne le PageItem résultant. Lève une exception si le crawl échoue complètement. """ loop = asyncio.get_running_loop() future: asyncio.Future[PageItem] = loop.create_future() def _run_in_thread() -> None: """Exécution bloquante dans le thread Scrapy/Twisted.""" try: from twisted.internet import reactor # type: ignore scrapy_cfg = get_project_settings() scrapy_cfg.update(make_scrapy_settings(timeout, verify_ssl)) runner = CrawlerRunner(scrapy_cfg) collected: list[PageItem] = [] crawler = runner.create_crawler(SinglePageSpider) def _on_item(item: PageItem, response: Any, spider: Any) -> None: collected.append(item) def _on_finished(_: Any) -> None: result = collected[0] if collected else PageItem( url=url, final_url=url, status_code=None, html="", download_time=0.0, error="Aucun item collecté par Scrapy", ) loop.call_soon_threadsafe( future.set_result, result # type: ignore[arg-type] ) def _on_error(failure: Any) -> None: exc = failure.value if hasattr(failure, "value") else Exception(str(failure)) loop.call_soon_threadsafe(future.set_exception, exc) crawler.signals.connect(_on_item, signal=signals.item_scraped) d: defer.Deferred = runner.crawl( crawler, url=url, custom_headers=custom_headers or {}, ) d.addCallback(_on_finished) d.addErrback(_on_error) # Démarrer le reactor si pas déjà démarré if not reactor.running: # type: ignore[attr-defined] reactor.run(installSignalHandlers=False) # type: ignore[attr-defined] except Exception as exc: loop.call_soon_threadsafe(future.set_exception, exc) await loop.run_in_executor(self._executor, _run_in_thread) return await future async def shutdown(self) -> None: self._executor.shutdown(wait=False) # Singleton partagé par l'application scrapy_runner = ScrapyRunner()