استخراج داده از وبسایتهای مدرن SPA (Next.js & React) بدون نیاز به مرورگر

import json
import re
import asyncio
from typing import Dict, Any, List, Optional
import httpx

class NextJsDirectExtractor:
    NEXT_DATA_REGEX = re.compile(r'<script id="__NEXT_DATA__" type="application/json"[^>]*>(.*?)</script>', re.DOTALL)
    RSC_FLIGHT_REGEX = re.compile(r'self\.__next_f\.push\(\[\d+,\s*"(.*?)"\]\)', re.DOTALL)

    def __init__(self, base_url: str):
        self.base_url = base_url.rstrip('/')
        self.build_id: Optional[str] = None
        self.client = httpx.AsyncClient(http2=True, timeout=15.0, headers={
            'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36'
        })

    async def refresh_build_id(self, sample_path: str = '/') -> Optional[str]:
        resp = await self.client.get(f'{self.base_url}{sample_path}')
        resp.raise_for_status()
        match = self.NEXT_DATA_REGEX.search(resp.text)
        if match:
            payload = json.loads(match.group(1))
            self.build_id = payload.get('buildId')
            return self.build_id
        return None

    async def fetch_page_props(self, route_path: str, query_params: Optional[Dict[str, str]] = None) -> Dict[str, Any]:
        clean_route = route_path.strip('/') or 'index'
        if not self.build_id:
            await self.refresh_build_id()

        if self.build_id:
            data_url = f'{self.base_url}/_next/data/{self.build_id}/{clean_route}.json'
            resp = await self.client.get(data_url, params=query_params)
            if resp.status_code == 404:
                # Target deployed a new build; refresh buildId once and retry
                await self.refresh_build_id()
                data_url = f'{self.base_url}/_next/data/{self.build_id}/{clean_route}.json'
                resp = await self.client.get(data_url, params=query_params)
            if resp.status_code == 200:
                return resp.json().get('pageProps', {})

        # Fallback to HTML extraction
        full_url = f'{self.base_url}/{route_path.lstrip("/")}'
        resp = await self.client.get(full_url, params=query_params)
        resp.raise_for_status()
        match = self.NEXT_DATA_REGEX.search(resp.text)
        if match:
            return json.loads(match.group(1)).get('props', {}).get('pageProps', {})
        raise ValueError('No __NEXT_DATA__ found; target may use App Router RSC.')

    async def fetch_rsc_flight_stream(self, route_path: str) -> List[str]:
        """Fetches raw React Server Components payload using the RSC header."""
        full_url = f'{self.base_url}/{route_path.lstrip("/")}'
        resp = await self.client.get(full_url, headers={'RSC': '1'})
        if resp.status_code == 200 and 'text/x-component' in resp.headers.get('content-type', ''):
            return resp.text.splitlines()
        return [m.encode('utf-8').decode('unicode_escape') for m in self.RSC_FLIGHT_REGEX.findall(resp.text)]

    async def close(self):
        await self.client.aclose()