backend/core/document_intel/pipeline.py

"""
Unified document extraction pipeline for URLs / HTML / PDF bytes.

Cascade (ethical, rate-friendly):
  1. PDF URL → download + PyMuPDF/pypdf
  2. HTML URL → fetch (httpx/requests) → Trafilatura main content
  3. Optional Crawl4AI markdown if HTTP body is thin (caller may pass prefetched html)

Always attaches legal citation extraction for regulation grounding.
"""
from __future__ import annotations

import logging
import os
import re
import tempfile
from typing import Any, Dict, Optional
from urllib.parse import urlparse

from core.document_intel.html_extract import html_to_clean_text
from core.document_intel.legal_citations import extract_legal_citations
from core.document_intel.pdf_extract import extract_pdf_text

logger = logging.getLogger(__name__)

_PDF_EXT = re.compile(r"\.pdf($|\?)", re.I)


def extract_from_html(
    html: str,
    *,
    url: str = "",
    content_type: str = "",
) -> Dict[str, Any]:
    """Clean HTML + legal citations. Binary/ZIP URLs short-circuit without Trafilatura."""
    try:
        from core.document_intel.fetch_resilience import should_skip_html_extract

        if should_skip_html_extract(url, content_type=content_type or None):
            return {
                "text": "",
                "extractor": "skipped_binary",
                "chars": 0,
                "legal": extract_legal_citations(""),
                "content_type": "binary",
                "url": url,
                "ok": False,
                "skipped": True,
                "reason": "binary_or_archive",
            }
    except Exception:
        pass

    cleaned = html_to_clean_text(html or "", url=url, content_type=content_type)
    if cleaned.get("skipped"):
        return {
            "text": "",
            "extractor": cleaned.get("extractor") or "skipped_binary",
            "chars": 0,
            "legal": extract_legal_citations(""),
            "content_type": "binary",
            "url": url,
            "ok": False,
            "skipped": True,
            "reason": "binary_or_archive",
        }
    text = cleaned.get("text") or ""
    cites = extract_legal_citations(text)
    return {
        "text": text,
        "extractor": cleaned.get("extractor"),
        "chars": cleaned.get("chars") or len(text),
        "legal": cites,
        "content_type": "html",
        "url": url,
        "ok": bool(text and len(text) >= 40),
    }


def _looks_like_pdf_url(url: str) -> bool:
    if not url:
        return False
    path = urlparse(url).path or ""
    return bool(_PDF_EXT.search(path) or path.lower().endswith(".pdf"))


def _fetch_bytes(url: str, *, timeout: float = 45.0) -> Dict[str, Any]:
    """HTTP GET — prefer curl_cffi stealth on WAF domains, else requests."""
    try:
        from core.document_intel.stealth_fetch import should_use_stealth, stealth_get

        if should_use_stealth(url):
            stealth = stealth_get(url, timeout=timeout)
            if stealth.get("ok"):
                return {
                    "status_code": stealth.get("status_code") or 200,
                    "content": stealth.get("content") or b"",
                    "text": stealth.get("text") or "",
                    "headers": stealth.get("headers") or {},
                    "final_url": stealth.get("final_url") or url,
                    "via": f"stealth:{stealth.get('impersonate')}",
                }
            # fall through to plain requests
    except Exception as e:
        logger.debug("[DocIntel] stealth path skip: %s", e)

    import requests

    headers = {
        "User-Agent": (
            "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 "
            "(KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
        ),
        "Accept": "text/html,application/xhtml+xml,application/pdf,*/*;q=0.8",
        "Accept-Language": "pl-PL,pl;q=0.9,en;q=0.8",
    }
    resp = requests.get(url, headers=headers, timeout=timeout, allow_redirects=True)
    return {
        "status_code": resp.status_code,
        "content": resp.content,
        "text": resp.text if resp.encoding or resp.apparent_encoding else "",
        "headers": {k.lower(): v for k, v in resp.headers.items()},
        "final_url": str(resp.url),
        "via": "requests",
    }


def _process_fetched(
    url: str,
    fetched: Dict[str, Any],
    *,
    is_pdf: bool,
) -> Dict[str, Any]:
    status = int(fetched.get("status_code") or 0)
    if status >= 400 or status == 0:
        return {
            "ok": False,
            "reason": f"http_{status}",
            "text": "",
            "url": url,
            "status_code": status,
        }

    ctype = (fetched.get("headers") or {}).get("content-type", "")
    content: bytes = fetched.get("content") or b""
    final_url = fetched.get("final_url") or url

    if is_pdf or "application/pdf" in ctype or content[:4] == b"%PDF":
        fd, path = tempfile.mkstemp(suffix=".pdf")
        try:
            with os.fdopen(fd, "wb") as f:
                f.write(content)
            pdf = extract_pdf_text(path)
            text = pdf.get("text") or ""
            cites = extract_legal_citations(text)
            return {
                "ok": bool(text and len(text) >= 40),
                "text": text,
                "extractor": pdf.get("parser"),
                "chars": pdf.get("chars") or len(text),
                "legal": cites,
                "content_type": "pdf",
                "url": final_url,
                "status_code": status,
                "source": "pdf_bytes",
            }
        finally:
            try:
                os.unlink(path)
            except Exception:
                pass

    html = fetched.get("text") or ""
    if not html and content:
        try:
            html = content.decode("utf-8", errors="replace")
        except Exception:
            html = ""
    out = extract_from_html(html, url=final_url)
    out["status_code"] = status
    out["url"] = final_url
    out["source"] = "http_html"
    return out


def extract_document_from_url(
    url: str,
    *,
    prefer_pdf: Optional[bool] = None,
    html_hint: Optional[str] = None,
    timeout: float = 45.0,
) -> Dict[str, Any]:
    """
    Fetch URL and extract clean text + legal citations (sync).
    Soft-fails with ok=False (never raises for network issues).
    """
    if not url or not str(url).startswith(("http://", "https://")):
        return {"ok": False, "reason": "invalid_url", "text": "", "url": url}

    try:
        from core.document_intel.fetch_resilience import (
            is_binary_or_archive_url,
            path_extension,
        )

        # ZIP/archives: never fetch into Trafilatura (P2)
        if is_binary_or_archive_url(url) and path_extension(url) != ".pdf":
            return {
                "ok": False,
                "reason": "skipped_binary_archive",
                "text": "",
                "url": url,
                "skipped": True,
                "extractor": "skipped_binary",
            }
    except Exception:
        pass

    if html_hint:
        out = extract_from_html(html_hint, url=url)
        out["source"] = "html_hint"
        return out

    is_pdf = prefer_pdf if prefer_pdf is not None else _looks_like_pdf_url(url)

    try:
        fetched = _fetch_bytes(url, timeout=timeout)
    except Exception as e:
        logger.warning("[DocIntel] fetch failed %s: %s", url[:80], e)
        return {"ok": False, "reason": f"fetch_error:{e}"[:120], "text": "", "url": url}

    return _process_fetched(url, fetched, is_pdf=is_pdf)


async def extract_document_from_url_async(
    url: str,
    *,
    prefer_pdf: Optional[bool] = None,
    html_hint: Optional[str] = None,
    timeout: float = 45.0,
    use_crawl4ai_fallback: bool = True,
) -> Dict[str, Any]:
    """
    Async variant: same cascade + optional Crawl4AI when HTML is thin.
    """
    import asyncio

    if not url or not str(url).startswith(("http://", "https://")):
        return {"ok": False, "reason": "invalid_url", "text": "", "url": url}

    try:
        from core.document_intel.fetch_resilience import (
            is_binary_or_archive_url,
            path_extension,
        )

        if is_binary_or_archive_url(url) and path_extension(url) != ".pdf":
            return {
                "ok": False,
                "reason": "skipped_binary_archive",
                "text": "",
                "url": url,
                "skipped": True,
                "extractor": "skipped_binary",
            }
    except Exception:
        pass

    if html_hint:
        out = extract_from_html(html_hint, url=url)
        out["source"] = "html_hint"
        return out

    is_pdf = prefer_pdf if prefer_pdf is not None else _looks_like_pdf_url(url)

    try:
        fetched = await asyncio.to_thread(_fetch_bytes, url, timeout=timeout)
    except Exception as e:
        logger.warning("[DocIntel] async fetch failed %s: %s", url[:80], e)
        return {"ok": False, "reason": f"fetch_error:{e}"[:120], "text": "", "url": url}

    out = _process_fetched(url, fetched, is_pdf=is_pdf)
    min_chars = int(os.environ.get("DOC_INTEL_MIN_CHARS", "120"))
    if (
        use_crawl4ai_fallback
        and out.get("content_type") != "pdf"
        and (not out.get("ok") or (out.get("chars") or 0) < min_chars)
    ):
        try:
            from core.crawl4ai_client import scrape_url_to_markdown

            md = await scrape_url_to_markdown(out.get("url") or url)
            if md and len(md.strip()) > (out.get("chars") or 0):
                cites = extract_legal_citations(md)
                return {
                    "ok": True,
                    "text": md.strip(),
                    "extractor": "crawl4ai",
                    "chars": len(md.strip()),
                    "legal": cites,
                    "content_type": "markdown",
                    "url": out.get("url") or url,
                    "status_code": out.get("status_code"),
                    "source": "crawl4ai_fallback",
                }
        except Exception as e:
            logger.debug("[DocIntel] crawl4ai fallback skip: %s", e)
    return out