scrape.py

#!/usr/bin/env python3
import argparse
import json
import re
import sys
from dataclasses import dataclass
from datetime import datetime, timezone
from io import BytesIO
from typing import Any

import requests
from pypdf import PdfReader

TARGET_URL = "https://www.vermittlerregister.info/recherche?a=pdf&registernummer={regnum}"
REQUEST_TIMEOUT = 30


class ScrapeError(Exception):
    pass


class CaptchaBlocked(ScrapeError):
    pass


@dataclass
class FetchResult:
    source: str
    payload_type: str
    data: bytes | str
    status_code: int | None = None
    content_type: str | None = None


def is_pdf_content(content_type: str | None, content: bytes) -> bool:
    if content_type and "pdf" in content_type.lower():
        return True
    return content.startswith(b"%PDF")


def decode_html(content: bytes, fallback: str = "utf-8") -> str:
    try:
        return content.decode(fallback, errors="replace")
    except Exception:
        return content.decode("latin-1", errors="replace")


def looks_like_captcha(text: str) -> bool:
    lowered = text.lower()
    markers = (
        "friendlycaptcha",
        "captcha",
        "frc-captcha",
        "frc-widget",
        "verify you are human",
    )
    return any(marker in lowered for marker in markers)


def extract_pdf_text(pdf_bytes: bytes) -> str:
    reader = PdfReader(BytesIO(pdf_bytes))
    pages = []
    for page in reader.pages:
        pages.append(page.extract_text() or "")
    return "\n".join(pages)


def parse_labeled_lines(lines: list[str]) -> dict[str, str]:
    labeled: dict[str, str] = {}
    for i, line in enumerate(lines):
        match = re.match(r"^\s*([^:]{2,80}):\s*(.*?)\s*$", line)
        if not match:
            continue
        key = re.sub(r"\s+", " ", match.group(1)).strip()
        value = match.group(2).strip()
        if not value and i + 1 < len(lines):
            nxt = lines[i + 1].strip()
            if nxt and ":" not in nxt:
                value = nxt
        if key and value:
            labeled[key] = value
    return labeled


def find_first(lines: list[str], pattern: str) -> str | None:
    regex = re.compile(pattern, flags=re.IGNORECASE)
    for line in lines:
        m = regex.search(line)
        if m:
            if m.lastindex:
                return m.group(1).strip()
            return line.strip()
    return None


def extract_contact_fields(lines: list[str], labeled: dict[str, str]) -> dict[str, Any]:
    def pick(label_candidates: tuple[str, ...], pattern: str | None = None) -> str | None:
        for k, v in labeled.items():
            if any(c.lower() in k.lower() for c in label_candidates):
                return v
        if pattern:
            return find_first(lines, pattern)
        return None

    email = pick(("E-Mail", "Email"), r"([A-Z0-9._%+-]+@[A-Z0-9.-]+\.[A-Z]{2,})")
    phone = pick(("Telefon", "Tel"), r"((?:\+?\d[\d\s()/.-]{5,}\d))")
    fax = pick(("Fax",), r"(Fax[:\s]+(.+)$)")
    website = pick(("Internet", "Webseite", "Homepage"), r"(https?://\S+|www\.\S+)")
    return {"email": email, "phone": phone, "fax": fax, "website": website}


def extract_address(lines: list[str], labeled: dict[str, str]) -> dict[str, Any]:
    address_value = None
    for k, v in labeled.items():
        if "anschrift" in k.lower() or "adresse" in k.lower():
            address_value = v
            break

    address = {
        "raw": address_value,
        "street": None,
        "postal_code": None,
        "city": None,
        "country": None,
    }
    if not address_value:
        return address

    pc_city = re.search(r"\b(\d{5})\s+([^\d,;]+)$", address_value)
    if pc_city:
        address["postal_code"] = pc_city.group(1)
        address["city"] = pc_city.group(2).strip()
        street = address_value[: pc_city.start()].strip(" ,;")
        address["street"] = street or None
    else:
        address["street"] = address_value
    return address


def normalize_broker_data(text: str, registernummer: str) -> dict[str, Any]:
    lines = [re.sub(r"\s+", " ", line).strip() for line in text.splitlines()]
    lines = [line for line in lines if line]
    labeled = parse_labeled_lines(lines)

    def get_by_label(*candidates: str) -> str | None:
        for c in candidates:
            for k, v in labeled.items():
                if c.lower() in k.lower():
                    return v
        return None

    full_name = get_by_label("Name", "Firma", "Unternehmen")
    first_name = get_by_label("Vorname")
    last_name = get_by_label("Nachname")
    company = get_by_label("Firma", "Unternehmen", "Gesellschaft")
    status = get_by_label("Status", "Registerstatus")
    authority = get_by_label("Registerbehörde", "IHK")
    permission = get_by_label("Erlaubnis")

    if not first_name and not last_name and full_name:
        parts = full_name.split()
        if len(parts) >= 2:
            first_name = parts[0]
            last_name = " ".join(parts[1:])

    regnum_in_text = find_first(
        lines,
        r"(D-[A-Z0-9]{2,}-[A-Z0-9]{2,}-[A-Z0-9]{2,})",
    )

    return {
        "register_number": regnum_in_text or registernummer,
        "status": status,
        "broker": {
            "full_name": full_name,
            "first_name": first_name,
            "last_name": last_name,
            "company": company,
        },
        "address": extract_address(lines, labeled),
        "contact": extract_contact_fields(lines, labeled),
        "permissions": [permission] if permission else [],
        "register_authority": authority,
        "raw_fields": labeled,
        "raw_text": text.strip(),
    }


def parse_pdf_result(pdf_bytes: bytes, registernummer: str, source: str) -> dict[str, Any]:
    text = extract_pdf_text(pdf_bytes)
    if not text.strip():
        raise ScrapeError(f"{source}: PDF parsed but contained no text.")
    parsed = normalize_broker_data(text, registernummer)
    return {
        "ok": True,
        "source": source,
        "payload": "pdf",
        "data": parsed,
    }


def parse_html_result(html_text: str, registernummer: str, source: str) -> dict[str, Any]:
    if looks_like_captcha(html_text):
        raise CaptchaBlocked(f"{source}: blocked by captcha")
    stripped = re.sub(r"<[^>]+>", " ", html_text)
    stripped = re.sub(r"\s+", " ", stripped).strip()
    if not stripped:
        raise ScrapeError(f"{source}: empty HTML payload.")
    parsed = normalize_broker_data(stripped, registernummer)
    return {
        "ok": True,
        "source": source,
        "payload": "html",
        "data": parsed,
    }


def fetch_direct_http(registernummer: str) -> FetchResult:
    url = TARGET_URL.format(regnum=registernummer)
    resp = requests.get(url, timeout=REQUEST_TIMEOUT, allow_redirects=True)
    content_type = resp.headers.get("Content-Type", "")
    if is_pdf_content(content_type, resp.content):
        return FetchResult(
            source="direct_http",
            payload_type="pdf",
            data=resp.content,
            status_code=resp.status_code,
            content_type=content_type,
        )

    html = decode_html(resp.content)
    if looks_like_captcha(html):
        raise CaptchaBlocked("direct_http: blocked by captcha")

    return FetchResult(
        source="direct_http",
        payload_type="html",
        data=html,
        status_code=resp.status_code,
        content_type=content_type,
    )


def fetch_with_scrapling(registernummer: str) -> FetchResult:
    try:
        import scrapling
    except Exception as exc:
        raise ScrapeError(f"scrapling import failed: {exc}") from exc

    url = TARGET_URL.format(regnum=registernummer)
    response_obj = None
    last_exc: Exception | None = None

    for fetcher_name in ("StealthyFetcher", "DynamicFetcher", "Fetcher"):
        fetcher = getattr(scrapling, fetcher_name, None)
        if fetcher is None:
            continue
        try:
            if fetcher_name in ("StealthyFetcher", "DynamicFetcher"):
                response_obj = fetcher.fetch(url, headless=True, network_idle=True, timeout=45000)
            else:
                response_obj = fetcher.get(url, timeout=45)
            break
        except Exception as exc:
            last_exc = exc

    if response_obj is None:
        raise ScrapeError(f"scrapling fetch failed: {last_exc}")

    body = getattr(response_obj, "body", b"")
    headers = dict(getattr(response_obj, "headers", {}) or {})
    content_type = headers.get("Content-Type") or headers.get("content-type", "")
    status_code = getattr(response_obj, "status", None)

    if isinstance(body, str):
        body = body.encode("utf-8", errors="replace")

    if is_pdf_content(content_type, body):
        return FetchResult(
            source="scrapling",
            payload_type="pdf",
            data=body,
            status_code=status_code,
            content_type=content_type,
        )

    html = decode_html(body)
    if looks_like_captcha(html):
        raise CaptchaBlocked("scrapling: blocked by captcha")

    return FetchResult(
        source="scrapling",
        payload_type="html",
        data=html,
        status_code=status_code,
        content_type=content_type,
    )


def fetch_with_playwright(registernummer: str) -> FetchResult:
    try:
        from playwright.sync_api import sync_playwright
    except Exception as exc:
        raise ScrapeError(f"playwright import failed: {exc}") from exc

    url = TARGET_URL.format(regnum=registernummer)
    with sync_playwright() as p:
        browser = p.chromium.launch(headless=True)
        context = browser.new_context()
        page = context.new_page()
        try:
            response = page.goto(url, wait_until="networkidle", timeout=45000)
            html = page.content()
            if response is not None:
                headers = response.headers
                content_type = headers.get("content-type", "")
                body = response.body()
                if is_pdf_content(content_type, body):
                    return FetchResult(
                        source="playwright",
                        payload_type="pdf",
                        data=body,
                        status_code=response.status,
                        content_type=content_type,
                    )
            if looks_like_captcha(html):
                raise CaptchaBlocked("playwright: blocked by captcha")
            return FetchResult(
                source="playwright",
                payload_type="html",
                data=html,
                status_code=response.status if response else None,
                content_type=response.headers.get("content-type") if response else None,
            )
        finally:
            context.close()
            browser.close()


def parse_result(result: FetchResult, registernummer: str) -> dict[str, Any]:
    if result.payload_type == "pdf":
        if not isinstance(result.data, (bytes, bytearray)):
            raise ScrapeError(f"{result.source}: expected bytes for PDF data.")
        return parse_pdf_result(bytes(result.data), registernummer, result.source)

    if not isinstance(result.data, str):
        raise ScrapeError(f"{result.source}: expected string for HTML data.")
    return parse_html_result(result.data, registernummer, result.source)


def build_final_output(
    registernummer: str,
    attempts: list[dict[str, Any]],
    success_payload: dict[str, Any] | None,
    error: str | None = None,
) -> dict[str, Any]:
    return {
        "ok": success_payload is not None,
        "registernummer": registernummer,
        "timestamp_utc": datetime.now(timezone.utc).isoformat(),
        "attempts": attempts,
        "result": success_payload,
        "error": error,
    }


def run(registernummer: str) -> dict[str, Any]:
    attempts: list[dict[str, Any]] = []

    # 1) Direct HTTP
    try:
        direct = fetch_direct_http(registernummer)
        parsed = parse_result(direct, registernummer)
        attempts.append(
            {
                "source": "direct_http",
                "ok": True,
                "status_code": direct.status_code,
                "content_type": direct.content_type,
            }
        )
        return build_final_output(registernummer, attempts, parsed)
    except Exception as exc:
        attempts.append({"source": "direct_http", "ok": False, "error": str(exc)})

    # 2) Scrapling
    try:
        scrapling_result = fetch_with_scrapling(registernummer)
        parsed = parse_result(scrapling_result, registernummer)
        attempts.append(
            {
                "source": "scrapling",
                "ok": True,
                "status_code": scrapling_result.status_code,
                "content_type": scrapling_result.content_type,
            }
        )
        return build_final_output(registernummer, attempts, parsed)
    except Exception as exc:
        attempts.append({"source": "scrapling", "ok": False, "error": str(exc)})

    # 3) Playwright
    try:
        playwright_result = fetch_with_playwright(registernummer)
        parsed = parse_result(playwright_result, registernummer)
        attempts.append(
            {
                "source": "playwright",
                "ok": True,
                "status_code": playwright_result.status_code,
                "content_type": playwright_result.content_type,
            }
        )
        return build_final_output(registernummer, attempts, parsed)
    except Exception as exc:
        attempts.append({"source": "playwright", "ok": False, "error": str(exc)})
        return build_final_output(
            registernummer=registernummer,
            attempts=attempts,
            success_payload=None,
            error="All scraping strategies failed.",
        )


def main() -> int:
    parser = argparse.ArgumentParser(description="Scrape broker details from vermittlerregister.info")
    parser.add_argument("registernummer", help="IHK register number, e.g. D-21RP-R1O5O-37")
    args = parser.parse_args()

    try:
        output = run(args.registernummer)
    except Exception as exc:
        output = {
            "ok": False,
            "registernummer": args.registernummer,
            "timestamp_utc": datetime.now(timezone.utc).isoformat(),
            "attempts": [],
            "result": None,
            "error": f"Fatal error: {exc}",
        }

    print(json.dumps(output, ensure_ascii=False, indent=2))
    return 0 if output.get("ok") else 1


if __name__ == "__main__":
    sys.exit(main())