257 lines
12 KiB
Python
257 lines
12 KiB
Python
"""
|
|
0pt Scraper Engine v2 - Production
|
|
====================================
|
|
Körs som Docker container på SIAX Cl0ud Server 2.
|
|
Data lagras i MinIO S3 (s3.siax.io) för obegränsad lagring.
|
|
|
|
Stödjer:
|
|
- Alla DK/NO/DE källor med auto-detection av bästa metod
|
|
- Header rotation + curl_cffi TLS fingerprint bypass
|
|
- Async concurrency för maxhastighet
|
|
- Resume-kapabilitet (kraschtålig)
|
|
- Batch processing med checkpointing
|
|
"""
|
|
import json, time, random, os, asyncio, sys, gzip
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
from dataclasses import dataclass, field, asdict
|
|
from concurrent.futures import ThreadPoolExecutor, as_completed
|
|
|
|
import requests
|
|
import httpx
|
|
from curl_cffi import requests as curl_requests
|
|
import cloudscraper
|
|
|
|
# ── Config from env ──
|
|
S3_ENDPOINT = os.getenv("S3_ENDPOINT", "https://s3.siax.io")
|
|
S3_ACCESS_KEY = os.getenv("S3_ACCESS_KEY", "")
|
|
S3_SECRET_KEY = os.getenv("S3_SECRET_KEY", "")
|
|
S3_BUCKET = os.getenv("S3_BUCKET", "opt-scraped-data")
|
|
OUTPUT_DIR = Path(os.getenv("OUTPUT_DIR", "/data/output"))
|
|
CHECKPOINT_DIR = Path(os.getenv("CHECKPOINT_DIR", "/data/checkpoints"))
|
|
MAX_CONCURRENCY = int(os.getenv("MAX_CONCURRENCY", "50"))
|
|
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",
|
|
},
|
|
}
|
|
# Pre-computed method rankings per source from benchmarks
|
|
METHOD_RANKING = {
|
|
"no-brreg-api": ["cloudscraper", "curl_cffi safari17", "curl_cffi chrome124"],
|
|
"no-proff": ["curl_cffi safari17", "curl_cffi edge99", "curl_cffi chrome110"],
|
|
"no-1881": ["curl_cffi chrome124", "curl_cffi safari17", "cloudscraper"],
|
|
"no-gulesider": ["curl_cffi safari17", "curl_cffi chrome124", "cloudscraper"],
|
|
"no-flowfirma": ["curl_cffi chrome124", "curl_cffi safari17", "curl_cffi chrome110"],
|
|
"dk-cvrapi": ["curl_cffi safari17", "curl_cffi chrome110", "cloudscraper"],
|
|
"dk-proff": ["curl_cffi safari17", "curl_cffi edge99", "curl_cffi chrome110"],
|
|
"dk-krak": ["curl_cffi safari17", "curl_cffi chrome124", "cloudscraper"],
|
|
"dk-118": ["curl_cffi safari17", "curl_cffi edge99", "curl_cffi chrome124"],
|
|
"dk-grundigt": ["curl_cffi safari17", "curl_cffi edge99", "curl_cffi chrome110"],
|
|
"de-northdata": ["curl_cffi edge99", "curl_cffi chrome110", "curl_cffi safari17"],
|
|
"de-handelsregister": ["curl_cffi edge99", "curl_cffi chrome110", "cloudscraper"],
|
|
"de-moneyhouse": ["curl_cffi chrome110", "curl_cffi edge99", "curl_cffi safari17"],
|
|
"de-unternehmensregister": ["curl_cffi safari17", "curl_cffi chrome110", "curl_cffi chrome124"],
|
|
"de-creditreform": ["curl_cffi chrome110", "curl_cffi safari17", "curl_cffi edge99"],
|
|
"default": ["curl_cffi chrome124", "curl_cffi safari17", "cloudscraper", "curl_cffi edge99", "requests"],
|
|
}
|
|
|
|
|
|
@dataclass
|
|
class ScrapeResult:
|
|
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", "challenge-form"]
|
|
return any(i in text[:5000] for i in indicators)
|
|
|
|
|
|
class ScraperEngine:
|
|
"""Main scraper engine - production ready."""
|
|
|
|
def __init__(self):
|
|
self.cloud = cloudscraper.create_scraper(
|
|
browser={"browser": "chrome", "platform": "windows", "mobile": False}
|
|
)
|
|
self.cache = {}
|
|
self.stats = {"total": 0, "ok": 0, "cf_blocked": 0, "errors": 0, "total_bytes": 0}
|
|
OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
|
|
CHECKPOINT_DIR.mkdir(parents=True, exist_ok=True)
|
|
|
|
def _get_method(self, name: str):
|
|
"""Get a scraping method by name."""
|
|
if name == "curl_cffi chrome124":
|
|
return lambda url, c="DE": curl_requests.get(url, impersonate="chrome124", headers=rotate_headers(c), timeout=15)
|
|
elif name == "curl_cffi edge99":
|
|
return lambda url, c="DE": curl_requests.get(url, impersonate="edge99", headers=rotate_headers(c), timeout=15)
|
|
elif name == "curl_cffi chrome110":
|
|
return lambda url, c="DE": curl_requests.get(url, impersonate="chrome110", headers=rotate_headers(c), timeout=15)
|
|
elif name == "curl_cffi safari17":
|
|
return lambda url, c="DE": curl_requests.get(url, impersonate="safari17_0", headers=rotate_headers(c), timeout=15)
|
|
elif name == "cloudscraper":
|
|
return lambda url, c="DE": self.cloud.get(url, headers=rotate_headers(c), timeout=15)
|
|
elif name == "requests":
|
|
return lambda url, c="DE": requests.get(url, headers=rotate_headers(c), timeout=15)
|
|
return None
|
|
|
|
def scrape_url(self, url: str, country: str = "DE", method_rank_key: str = "default") -> ScrapeResult:
|
|
"""Try methods in ranking order until one succeeds."""
|
|
methods = METHOD_RANKING.get(method_rank_key, METHOD_RANKING["default"])
|
|
|
|
for method_name in methods:
|
|
fn = self._get_method(method_name)
|
|
if not fn:
|
|
continue
|
|
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)
|
|
|
|
self.stats["total"] += 1
|
|
if r.status_code == 200 and not blocked:
|
|
self.stats["ok"] += 1
|
|
self.stats["total_bytes"] += len(r.content)
|
|
return ScrapeResult(url=url, method=method_name, status=r.status_code,
|
|
elapsed=round(elapsed, 3), bytes=len(r.content), cf_blocked=blocked)
|
|
elif blocked:
|
|
self.stats["cf_blocked"] += 1
|
|
else:
|
|
self.stats["errors"] += 1
|
|
except Exception as e:
|
|
self.stats["errors"] += 1
|
|
continue
|
|
|
|
return ScrapeResult(url=url, method="all_failed", status=0, elapsed=0, bytes=0, error="All methods failed")
|
|
|
|
def scrape_batch(self, urls: list[tuple], country: str = "DE", method_key: str = "default", max_workers: int = 10) -> list[ScrapeResult]:
|
|
"""Scrape a batch of URLs concurrently using thread pool."""
|
|
results = []
|
|
with ThreadPoolExecutor(max_workers=max_workers) as ex:
|
|
futures = {ex.submit(self.scrape_url, url, country, method_key): url for url, _ in urls}
|
|
for f in as_completed(futures):
|
|
results.append(f.result())
|
|
return results
|
|
|
|
async def scrape_async_batch(self, urls: list[str], country: str = "DE", concurrency: int = None) -> list[ScrapeResult]:
|
|
"""Massively parallel async scraping."""
|
|
concurrency = concurrency or MAX_CONCURRENCY
|
|
from httpx import AsyncClient, Limits, Timeout
|
|
limits = Limits(max_keepalive_connections=concurrency, max_connections=concurrency)
|
|
timeout = Timeout(30.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), follow_redirects=True)
|
|
elapsed = time.time() - start
|
|
return ScrapeResult(url=url, method="async_httpx", status=r.status_code,
|
|
elapsed=round(elapsed, 3), bytes=len(r.content),
|
|
cf_blocked=check_cf_blocked(str(r.content[:10000])))
|
|
except Exception as e:
|
|
return ScrapeResult(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]
|
|
results = await asyncio.gather(*tasks)
|
|
for r in results:
|
|
self.stats["total"] += 1
|
|
if r.status == 200:
|
|
self.stats["ok"] += 1
|
|
self.stats["total_bytes"] += r.bytes
|
|
return results
|
|
|
|
def save_result(self, result: ScrapeResult, raw_data: str = None):
|
|
"""Save result with compression to disk."""
|
|
domain = result.url.split("/")[2].replace(".", "_")
|
|
ts = datetime.now().strftime("%Y%m%d_%H%M%S")
|
|
safe_name = f"{domain}_{result.method}_{ts}.json.gz"
|
|
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_sample": raw_data[:10000] if raw_data else None,
|
|
}
|
|
with gzip.open(path, "wt", encoding="utf-8") as f:
|
|
json.dump(payload, f, ensure_ascii=False)
|
|
return path
|
|
|
|
def save_checkpoint(self, batch_id: str, completed: list, total: int):
|
|
"""Save progress checkpoint for resume."""
|
|
path = CHECKPOINT_DIR / f"checkpoint_{batch_id}.json"
|
|
data = {"batch_id": batch_id, "completed": len(completed), "total": total,
|
|
"timestamp": datetime.now().isoformat(), "urls": completed}
|
|
path.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8")
|
|
|
|
def load_checkpoint(self, batch_id: str) -> set:
|
|
"""Load completed URLs from checkpoint."""
|
|
path = CHECKPOINT_DIR / f"checkpoint_{batch_id}.json"
|
|
if path.exists():
|
|
data = json.loads(path.read_text(encoding="utf-8"))
|
|
return set(data.get("urls", []))
|
|
return set()
|
|
|
|
def print_stats(self):
|
|
"""Print current scraping statistics."""
|
|
total = self.stats["total"]
|
|
ok = self.stats["ok"]
|
|
rate = ok / (self.stats.get("start_time", 1) and
|
|
max(1, (time.time() - self.stats.get("start_time", time.time()))))
|
|
print(f" Stats: {ok}/{total} OK ({self.stats['cf_blocked']} CF, {self.stats['errors']} err)")
|
|
print(f" Rate: {rate:.0f} req/s | Data: {self.stats['total_bytes']/1024/1024:.1f} MB")
|
|
|
|
|
|
def print_header():
|
|
print("""
|
|
╔══════════════════════════════════════════════════════════════╗
|
|
║ 0PT SCRAPER ENGINE v2 - PRODUCTION ║
|
|
║ DK/NO/DE Company Data Pipeline → MinIO S3 ║
|
|
╚══════════════════════════════════════════════════════════════╝
|
|
""")
|