""" 0pt Scraper Engine v2 ===================== Universal scraping engine for DK/NO/DE company data. Uses multi-method bypass chain + header rotation for maximum speed. Metoder (i prioritetsordning): 1. curl_cffi chrome124 - TLS fingerprint impersonation 2. curl_cffi edge99 - alternative fingerprint 3. cloudscraper - basic CF bypass 4. requests + header rotation - direct HTTP 5. httpx async - high concurrency 6. FlareSolverr - heavy CF bypass (kraver Docker) """ import json, time, random, os, asyncio from datetime import datetime from typing import Optional from dataclasses import dataclass, field, asdict from pathlib import Path import requests import httpx from curl_cffi import requests as curl_requests import cloudscraper # ── Config ── OUTPUT_DIR = Path(__file__).parent / "output" HEADER_POOL = { "user_agents": [ "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36", "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/125.0.0.0 Safari/537.36", "Mozilla/5.0 (Macintosh; Intel Mac OS X 14_5) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.5 Safari/605.1.15", "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:127.0) Gecko/20100101 Firefox/127.0", "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36", "Mozilla/5.0 (iPhone; CPU iPhone OS 17_5 like Mac OS X) AppleWebKit/605.1.15", ], "languages": { "DK": "da-DK,da;q=0.9,en;q=0.8,de;q=0.7", "NO": "nb-NO,nb;q=0.9,no;q=0.8,en;q=0.7,da;q=0.6", "DE": "de-DE,de;q=0.9,en;q=0.8,fr;q=0.7", "SE": "sv-SE,sv;q=0.9,en;q=0.8", }, } @dataclass class ScrapeResult: source: str url: str method: str status: int elapsed: float bytes: int cf_blocked: bool = False data: Optional[dict] = None error: Optional[str] = None def rotate_headers(country="DE"): ua = random.choice(HEADER_POOL["user_agents"]) lang = HEADER_POOL["languages"].get(country, "en-US,en;q=0.9") return { "User-Agent": ua, "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,*/*;q=0.8", "Accept-Language": lang, "Accept-Encoding": "gzip, deflate, br", "Cache-Control": "no-cache", "Sec-Ch-Ua": '"Chromium";v="124", "Google Chrome";v="124", "Not-A.Brand";v="99"', "Sec-Ch-Ua-Mobile": "?0", "Sec-Fetch-Dest": "document", "Sec-Fetch-Mode": "navigate", "Sec-Fetch-Site": "none", "Sec-Fetch-User": "?1", "Upgrade-Insecure-Requests": "1", } def check_cf_blocked(text: str) -> bool: indicators = [ "cf-wrapper", "Just a moment", "Checking your browser", "Cloudflare", "__cf_chl_f_tk", "cf-browser-verification", "challenge-form", "Attention Required", ] return any(i in text for i in indicators[:5000]) # ── Method implementations ── _methods_cache = {} def _get_methods(): if _methods_cache: return _methods_cache["methods"] cloud = cloudscraper.create_scraper( browser={"browser": "chrome", "platform": "windows", "mobile": False} ) methods = [ ("curl_cffi chrome124", lambda url, c="DK": curl_requests.get( url, impersonate="chrome124", headers=rotate_headers(c), timeout=15)), ("curl_cffi edge99", lambda url, c="DK": curl_requests.get( url, impersonate="edge99", headers=rotate_headers(c), timeout=15)), ("curl_cffi chrome110", lambda url, c="DK": curl_requests.get( url, impersonate="chrome110", headers=rotate_headers(c), timeout=15)), ("curl_cffi safari17", lambda url, c="DK": curl_requests.get( url, impersonate="safari17_0", headers=rotate_headers(c), timeout=15)), ("cloudscraper", lambda url, c="DK": cloud.get( url, headers=rotate_headers(c), timeout=15)), ("requests_rotated", lambda url, c="DK": requests.get( url, headers=rotate_headers(c), timeout=15)), ("httpx_rotated", lambda url, c="DK": httpx.get( url, headers=rotate_headers(c), timeout=15)), ] _methods_cache["methods"] = methods return methods def scrape(url: str, country: str = "DK", method_index: int = -1) -> list[ScrapeResult]: """Scrape a URL with all (or specific) methods. Returns list of results.""" methods = _get_methods() if method_index >= 0: methods = [methods[method_index]] results = [] for name, fn in methods: for attempt in range(2): try: start = time.time() r = fn(url, country) elapsed = time.time() - start text_sample = str(r.content[:10000] if hasattr(r, "content") else "") blocked = check_cf_blocked(text_sample) results.append(ScrapeResult( source=url, url=url, method=name, status=r.status_code, elapsed=round(elapsed, 3), bytes=len(r.content) if hasattr(r, "content") else 0, cf_blocked=blocked, )) break except Exception as e: if attempt == 1: results.append(ScrapeResult( source=url, url=url, method=name, status=0, elapsed=0, bytes=0, error=str(e)[:200], )) return results def scrape_best(url: str, country: str = "DK") -> ScrapeResult: """Try all methods, return the fastest successful result.""" results = scrape(url, country) ok = [r for r in results if r.status == 200 and not r.cf_blocked] if ok: return min(ok, key=lambda r: r.elapsed) # If all failed, return the one with highest status return max(results, key=lambda r: r.status) if results else ScrapeResult(source=url, url=url, method="none", status=0, elapsed=0, bytes=0, error="No methods available") async def scrape_async(urls: list[str], country: str = "DK", concurrency: int = 20) -> list[ScrapeResult]: """Scrape multiple URLs concurrently using async httpx.""" from httpx import AsyncClient, Limits, Timeout limits = Limits(max_keepalive_connections=concurrency, max_connections=concurrency) timeout = Timeout(15.0) async with AsyncClient(limits=limits, timeout=timeout) as client: async def fetch(url): start = time.time() try: r = await client.get(url, headers=rotate_headers(country)) elapsed = time.time() - start text_sample = str(r.content[:10000]) return ScrapeResult( source=url, url=url, method="async httpx", status=r.status_code, elapsed=round(elapsed, 3), bytes=len(r.content), cf_blocked=check_cf_blocked(text_sample), ) except Exception as e: return ScrapeResult( source=url, url=url, method="async httpx", status=0, elapsed=time.time()-start, bytes=0, error=str(e)[:200], ) sem = asyncio.Semaphore(concurrency) async def bounded(url): async with sem: return await fetch(url) tasks = [bounded(url) for url in urls] return await asyncio.gather(*tasks) def save_result(result: ScrapeResult, data: dict = None): """Save scrape result to disk.""" OUTPUT_DIR.mkdir(parents=True, exist_ok=True) domain = result.url.split("/")[2].replace(".", "_") ts = datetime.now().strftime("%Y%m%d_%H%M%S") safe_name = f"{domain}_{result.method}_{ts}.json" path = OUTPUT_DIR / safe_name payload = { "timestamp": ts, "url": result.url, "method": result.method, "status": result.status, "elapsed": result.elapsed, "bytes": result.bytes, "cf_blocked": result.cf_blocked, "error": result.error, "data": data, } path.write_text(json.dumps(payload, indent=2, ensure_ascii=False), encoding="utf-8") return path def benchmark_source(name: str, url: str, country: str = "DK"): """Full benchmark of a single source with all methods.""" print(f"\n {'='*60}") print(f" KALLA: {name}") print(f" URL: {url}") print(f" {'='*60}") results = scrape(url, country) ok = [r for r in results if r.status == 200 and not r.cf_blocked] for r in results: status_str = "OK" if r.status == 200 else f"ERR {r.status}" cf_str = " [CF]" if r.cf_blocked else "" err_str = f" - {r.error}" if r.error else "" print(f" {r.method:30s} -> {status_str:10s} {r.elapsed:6.3f}s {r.bytes:8d}b{cf_str}{err_str}") if ok: best = min(ok, key=lambda r: r.elapsed) print(f"\n >>> BAST: {best.method} ({best.elapsed:.3f}s, {best.bytes}b)") return best return None