app.py

import asyncio
import time
import json
import hashlib
import re
from typing import List, Dict, Optional, Any
from datetime import datetime
from collections import OrderedDict
from contextlib import asynccontextmanager
from urllib.parse import urlparse, parse_qs, unquote

import psutil
import trafilatura
from bs4 import BeautifulSoup
import httpx
from fake_useragent import UserAgent

# Crawl4AI 0.8.0 imports
from crawl4ai import (
    AsyncWebCrawler,
    BrowserConfig,
    CacheMode,
    CrawlerRunConfig,
    MemoryAdaptiveDispatcher,
)

from fastapi import FastAPI, HTTPException, BackgroundTasks, Depends
from fastapi.responses import JSONResponse
from starlette.responses import StreamingResponse as StarletteStreamingResponse
from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.gzip import GZipMiddleware
from pydantic import BaseModel, Field, field_validator
from prometheus_client import Counter, Histogram, Gauge, generate_latest, CONTENT_TYPE_LATEST, CollectorRegistry
from starlette.responses import Response
import structlog
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type

# ─── Structured Logging ──────────────────────────────────────────────────────
logger = structlog.get_logger()

# ─── Prometheus Metrics ──────────────────────────────────────────────────────
METRICS_REGISTRY = CollectorRegistry()
REQUEST_COUNT = Counter('scraper_requests_total', 'Total scraping requests', ['method', 'endpoint'], registry=METRICS_REGISTRY)
REQUEST_LATENCY = Histogram('scraper_request_duration_seconds', 'Request latency in seconds', ['endpoint'], registry=METRICS_REGISTRY)
ACTIVE_SCRAPERS = Gauge('scraper_active_tasks', 'Number of active scraping tasks', registry=METRICS_REGISTRY)
CACHE_HITS = Counter('scraper_cache_hits_total', 'Total cache hits', registry=METRICS_REGISTRY)
CACHE_MISSES = Counter('scraper_cache_misses_total', 'Total cache misses', registry=METRICS_REGISTRY)
SCRAPE_ERRORS = Counter('scraper_errors_total', 'Total scraping errors', ['error_type'], registry=METRICS_REGISTRY)

# ─── Domains where httpx works better than a browser ─────────────────────────
HTTPX_PREFERRED_DOMAINS = {
    'github.com', 'raw.githubusercontent.com', 'gist.github.com',
    'huggingface.co', 'hf.co',
    'arxiv.org', 'abs.arxiv.org',
    'pypi.org', 'readthedocs.io',
    'stackoverflow.com', 'stackexchange.com',
    'wikipedia.org', 'en.wikipedia.org',
    'docs.python.org',
    'npmjs.com',
}


# ─── LRU Cache ───────────────────────────────────────────────────────────────
class LRUCache:
    def __init__(self, max_size: int = 1000, ttl_seconds: int = 3600):
        self.cache: OrderedDict = OrderedDict()
        self.max_size = max_size
        self.ttl = ttl_seconds
        self._lock = asyncio.Lock()

    async def get(self, key: str) -> Optional[Any]:
        async with self._lock:
            if key not in self.cache:
                CACHE_MISSES.inc()
                return None
            item = self.cache[key]
            if time.time() - item['timestamp'] > self.ttl:
                del self.cache[key]
                CACHE_MISSES.inc()
                return None
            self.cache.move_to_end(key)
            CACHE_HITS.inc()
            return item['data']

    async def set(self, key: str, data: Any):
        async with self._lock:
            if key in self.cache:
                self.cache.move_to_end(key)
                self.cache[key] = {'data': data, 'timestamp': time.time()}
            else:
                if len(self.cache) >= self.max_size:
                    self.cache.popitem(last=False)
                self.cache[key] = {'data': data, 'timestamp': time.time()}

    async def clear_expired(self):
        async with self._lock:
            current_time = time.time()
            expired_keys = [
                key for key, item in self.cache.items()
                if current_time - item['timestamp'] > self.ttl
            ]
            for key in expired_keys:
                del self.cache[key]
            if expired_keys:
                logger.info("cleared_expired_cache_items", count=len(expired_keys))


cache = LRUCache(max_size=1000, ttl_seconds=1800)


# ─── Pydantic Models ─────────────────────────────────────────────────────────
class ScraperRequest(BaseModel):
    query: str = Field(..., min_length=1, max_length=500, description="Search query")
    max_results: int = Field(5, ge=1, le=20, description="Maximum number of results")
    extract_full_content: bool = Field(True, description="Extract full article content")
    use_stealth_mode: bool = Field(True, description="Use Crawl4AI stealth mode")
    timeout: int = Field(30, ge=10, le=120, description="Request timeout in seconds")

    @field_validator('query')
    @classmethod
    def validate_query(cls, v: str) -> str:
        if not v.strip():
            raise ValueError('Query cannot be empty')
        return v.strip()


class ScrapedResult(BaseModel):
    url: str
    title: str
    snippet: Optional[str] = None
    full_content: Optional[str] = None
    published_date: Optional[str] = None
    author: Optional[str] = None
    domain: str
    scraped_at: datetime
    metadata: Optional[Dict[str, Any]] = None


class ScraperResponse(BaseModel):
    query: str
    results: List[ScrapedResult]
    total_results: int
    processing_time_ms: float
    cached: bool
    timestamp: datetime


class HealthResponse(BaseModel):
    status: str
    version: str
    uptime_seconds: float
    memory_usage_mb: float
    cpu_percent: float
    active_scrapers: int
    cache_size: int


# ─── Rate Limiter ─────────────────────────────────────────────────────────────
class RateLimiter:
    def __init__(self, max_requests: int = 10, window_seconds: int = 60):
        self.max_requests = max_requests
        self.window_seconds = window_seconds
        self.requests: Dict[str, List[float]] = {}
        self._lock = asyncio.Lock()

    async def is_allowed(self, client_id: str) -> bool:
        async with self._lock:
            now = time.time()
            if client_id in self.requests:
                self.requests[client_id] = [
                    req_time for req_time in self.requests[client_id]
                    if now - req_time < self.window_seconds
                ]
            else:
                self.requests[client_id] = []
            if len(self.requests[client_id]) >= self.max_requests:
                return False
            self.requests[client_id].append(now)
            return True


rate_limiter = RateLimiter(max_requests=10, window_seconds=60)


# ─── Utility Functions ────────────────────────────────────────────────────────

def _extract_domain(url: str) -> str:
    try:
        return urlparse(url).netloc or ''
    except Exception:
        return ''


def _strip_text_fragment(url: str) -> str:
    """Remove Google #:~:text=... fragments that cause service-worker conflicts."""
    if '#:~:text=' in url:
        return url.split('#:~:text=')[0]
    return url


def _clean_url(url: str) -> str:
    return _strip_text_fragment(url.strip())


def _unwrap_google_redirect(href: str) -> Optional[str]:
    if '/url?' in href:
        parsed = urlparse(href)
        qs = parse_qs(parsed.query)
        real_url = qs.get('q', [None])[0] or qs.get('url', [None])[0]
        if real_url and real_url.startswith('http'):
            return unquote(real_url)
    return None


def _is_valid_result_url(href: str) -> bool:
    if not href or not href.startswith('http'):
        return False
    skip_patterns = [
        'google.com/search', 'google.com/url', 'google.com/preferences',
        'google.com/advanced_search', 'google.com/intl',
        'accounts.google.com', 'support.google.com', 'maps.google.com',
        'policies.google.com', 'play.google.com',
        'youtube.com', 'youtu.be',
        'webcache.googleusercontent.com', 'translate.google.com',
    ]
    href_lower = href.lower()
    return not any(p in href_lower for p in skip_patterns)


def _should_use_httpx(url: str) -> bool:
    domain = _extract_domain(url).lower()
    return any(preferred in domain for preferred in HTTPX_PREFERRED_DOMAINS)


def generate_cache_key(query: str, max_results: int, extract_full_content: bool) -> str:
    content = f"{query}_{max_results}_{extract_full_content}"
    return hashlib.md5(content.encode()).hexdigest()


# ─── Content Extraction Pipeline ─────────────────────────────────────────────

def _extract_content_from_html(html: str, url: str) -> Dict[str, Any]:
    """Multi-method content extraction: Trafilatura → BeautifulSoup."""
    extracted = {}
    domain = _extract_domain(url)

    # Method 1: Trafilatura
    try:
        traf = trafilatura.extract(
            html, include_comments=False, include_tables=True, output_format='json'
        )
        if traf:
            extracted = json.loads(traf)
    except Exception:
        pass

    # Method 2: BeautifulSoup for site-specific + general fallback
    if not extracted.get('text'):
        try:
            soup = BeautifulSoup(html, 'html.parser')
            title_elem = soup.find('title')
            title = title_elem.get_text(strip=True) if title_elem else ''

            # GitHub README
            if 'github.com' in domain:
                readme = soup.find('article', class_='markdown-body')
                if readme:
                    return {'text': readme.get_text(separator=' ', strip=True), 'title': title}

            # HuggingFace model card
            if 'huggingface.co' in domain:
                card = soup.find('div', class_='prose') or soup.find('article')
                if card:
                    return {'text': card.get_text(separator=' ', strip=True), 'title': title}

            # General
            main = (
                soup.find('main') or soup.find('article') or
                soup.find('div', {'role': 'main'}) or soup.find('div', class_='content')
            )
            full_content = main.get_text(separator=' ', strip=True) if main else ' '.join(
                p.get_text(strip=True) for p in soup.find_all('p')
            )
            meta_desc = soup.find('meta', attrs={'name': 'description'})
            desc = meta_desc.get('content', '') if meta_desc else full_content[:300]

            extracted.setdefault('title', title)
            extracted.setdefault('text', full_content)
            extracted.setdefault('excerpt', desc)
        except Exception:
            pass

    return extracted


# ═══════════════════════════════════════════════════════════════════════════════
# DUAL CRAWLER ARCHITECTURE
# ═══════════════════════════════════════════════════════════════════════════════
#
#  ┌─────────────────────────────────────────────────────────────────────┐
#  │  CRAWLER A — Google Search (Persistent Context, Serialized)        │
#  │  • use_persistent_context=True → cookies saved (Google consent)    │
#  │  • headless=False → visible window (Google trusts it more)         │
#  │  • Serialized via _search_lock (one search at a time)              │
#  └─────────────────────────────────────────────────────────────────────┘
#
#  ┌─────────────────────────────────────────────────────────────────────┐
#  │  CRAWLER B — URL Scraping (Non-Persistent, arun_many Parallel)     │
#  │  • headless=True → fast, no visual rendering overhead              │
#  │  • ignore_https_errors=True → doesn't hang on SSL                  │
#  │  • viewport 800x600 → less memory per tab                          │
#  │  • arun_many() + MemoryAdaptiveDispatcher → native parallel tabs   │
#  │  • Each URL gets its own tab, managed by crawl4ai internally       │
#  └─────────────────────────────────────────────────────────────────────┘
#
#  ┌─────────────────────────────────────────────────────────────────────┐
#  │  FALLBACK C — httpx + Trafilatura (for GitHub, HuggingFace, etc.)  │
#  │  • Parallel asyncio.gather for pure HTTP requests                  │
#  │  • No browser overhead — fastest for static content sites          │
#  └─────────────────────────────────────────────────────────────────────┘

_search_lock = asyncio.Lock()


async def _get_search_crawler() -> AsyncWebCrawler:
    """Persistent-context crawler for Google Search."""
    if 'search' not in app.state.crawler_pool:
        config = BrowserConfig(
            headless=False,
            enable_stealth=True,
            user_agent_mode="random",
            use_persistent_context=True,
            user_data_dir="./browser_data",
            extra_args=[
                "--disable-blink-features=AutomationControlled",
                "--disable-gpu",
                "--disable-gpu-compositing",
                "--disable-software-rasterizer",
                "--no-sandbox",
                "--disable-dev-shm-usage",
                "--no-first-run",
                "--no-default-browser-check",
                "--disable-infobars",
                "--ignore-certificate-errors",
                "--ignore-certificate-errors-spki-list",
                "--disable-ipc-flooding-protection",
                "--force-color-profile=srgb",
                "--mute-audio",
                "--disable-background-timer-throttling",
                "--disable-renderer-backgrounding",
            ],
        )
        crawler = AsyncWebCrawler(config=config)
        await crawler.start()
        logger.info("SEARCH crawler created (persistent context).")
        app.state.crawler_pool['search'] = crawler
    return app.state.crawler_pool['search']


async def _get_scrape_crawler() -> AsyncWebCrawler:
    """High-speed non-persistent crawler for parallel URL scraping via arun_many."""
    if 'scrape' not in app.state.crawler_pool:
        config = BrowserConfig(
            headless=False,
            enable_stealth=True,
            user_agent_mode="random",
            ignore_https_errors=True,
            viewport_width=800,
            viewport_height=600,
            extra_args=[
                "--disable-blink-features=AutomationControlled",
                "--disable-gpu",
                "--no-sandbox",
                "--disable-dev-shm-usage",
                "--no-first-run",
                "--disable-infobars",
                "--ignore-certificate-errors",
                "--disable-ipc-flooding-protection",
                "--mute-audio",
                "--disable-background-timer-throttling",
                "--disable-renderer-backgrounding",
                "--disable-features=IsolateOrigins,site-per-process",
            ],
        )
        crawler = AsyncWebCrawler(config=config)
        await crawler.start()
        logger.info("SCRAPE crawler created (headless parallel via arun_many).")
        app.state.crawler_pool['scrape'] = crawler
    return app.state.crawler_pool['scrape']


# ─── Google SERP Extraction ──────────────────────────────────────────────────

async def extract_search_results(query: str, max_results: int, timeout: int) -> List[Dict[str, str]]:
    """Extract URLs from Google SERP using persistent search crawler."""
    search_url = f"https://www.google.com/search?q={query.replace(' ', '+')}"

    try:
        crawler = await _get_search_crawler()
        run_config = CrawlerRunConfig(
            cache_mode=CacheMode.BYPASS,
            js_code="""
            Object.defineProperty(navigator, 'webdriver', {get: () => undefined});
            window.scrollBy(0, 600);
            await new Promise(r => setTimeout(r, 1000));
            window.scrollBy(0, 600);
            await new Promise(r => setTimeout(r, 500));
            """,
            wait_for="css:div.g a[href], a[data-ved]",
        )

        async with _search_lock:
            try:
                result = await asyncio.wait_for(
                    crawler.arun(url=search_url, config=run_config),
                    timeout=15
                )
            except asyncio.TimeoutError:
                logger.error("search_hard_timeout", query=query)
                return []
            except Exception as e:
                logger.error("search_crawl_exception", query=query, error=str(e))
                return []

        if not result or not result.success:
            err = getattr(result, 'error_message', 'Unknown') if result else 'None'
            logger.warning("search_crawl_failed", url=search_url, error=err)
            return []

        html = result.html or ''
        soup = BeautifulSoup(html, 'html.parser')
        title_text = soup.title.string if soup.title else 'None'

        # ── Debug: understand what Google is actually serving ──
        all_a_tags = soup.find_all('a', href=True)
        has_rso = soup.find('div', id='rso') is not None
        sample_hrefs = [a.get('href', '')[:80] for a in all_a_tags[:8]]
        logger.info(
            "serp_debug",
            html_len=len(html),
            title=str(title_text)[:60],
            total_a_tags=len(all_a_tags),
            has_rso=has_rso,
            sample_hrefs=sample_hrefs,
        )

        urls, seen = [], set()

        def _add(href, title):
            if not href:
                return
            unwrapped = _unwrap_google_redirect(href)
            if unwrapped:
                href = unwrapped
            href = _clean_url(href)
            if not _is_valid_result_url(href) or href in seen:
                return
            seen.add(href)
            urls.append({'url': href, 'title': (title or 'Untitled').strip()[:100]})

        selectors = [
            'div.g a[href]', 'h3 a[href]', 'div#rso div.g a[href]',
            'div[data-ocrs] a[href]', 'g-more-link a[href]',
            'div#search div.g a[href]', 'a[href][data-ved][href^=http]',
            'div.yuRUbf a[href]', 'a[jsname] h3',
            'div[data-snhf] a[href]', 'div[data-sokoban] a[href]',
            'a[href^="/url?q="]',
        ]
        for sel in selectors:
            try:
                matches = soup.select(sel)
                if matches:
                    logger.info(f"Selector '{sel}': {len(matches)} matches")
                for link in matches:
                    if link.name == 'h3':
                        pa = link.find_parent('a')
                        if pa:
                            _add(pa.get('href', ''), link.get_text(strip=True))
                    else:
                        _add(link.get('href', ''), link.get_text(strip=True))
            except Exception:
                pass
            if len(urls) >= max_results * 3:
                break

        if len(urls) < max_results:
            for m in re.finditer(r'href="(https?://[^"]+)"', html, re.I):
                _add(m.group(1), 'Untitled')
                if len(urls) >= max_results * 3:
                    break

        logger.info(f"Extracted {len(urls)} unique URLs from Google SERP")
        return urls[:max_results * 3]

    except Exception as e:
        logger.error("search_extraction_error", query=query, error=str(e))
        SCRAPE_ERRORS.labels(error_type='search_error').inc()
        return []


# ─── httpx-based Scraping (GitHub, HuggingFace, fallback) ────────────────────

@retry(
    stop=stop_after_attempt(2),
    wait=wait_exponential(multiplier=1, min=1, max=5),
    retry=retry_if_exception_type((httpx.RequestError, asyncio.TimeoutError))
)
async def _fetch_with_retry(client: httpx.AsyncClient, url: str, timeout: int, headers: dict):
    resp = await client.get(url, headers=headers, timeout=timeout, follow_redirects=True)
    resp.raise_for_status()
    return resp


async def scrape_url_httpx(url: str, timeout: int) -> Dict[str, Any]:
    """Fast HTTP-based scraping. No browser overhead."""
    t0 = time.time()
    domain = _extract_domain(url)
    try:
        async with httpx.AsyncClient(verify=False) as client:
            headers = {
                'User-Agent': app.state.user_agent.random,
                'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8',
                'Accept-Language': 'en-US,en;q=0.5',
                'Accept-Encoding': 'gzip, deflate, br',
                'DNT': '1',
                'Connection': 'keep-alive',
                'Upgrade-Insecure-Requests': '1',
            }
            resp = await _fetch_with_retry(client, url, timeout, headers)
            extracted = _extract_content_from_html(resp.text, url)
            if extracted.get('text'):
                return {
                    'url': url, 'title': extracted.get('title', ''),
                    'snippet': extracted.get('excerpt', '')[:300],
                    'full_content': extracted.get('text', ''),
                    'published_date': extracted.get('date'),
                    'author': extracted.get('author'),
                    'domain': domain,
                    'extraction_time_ms': (time.time() - t0) * 1000,
                    'extraction_method': 'httpx',
                }
            return {'url': url, 'error': 'No content extracted', 'extraction_method': 'failed'}
    except Exception as e:
        logger.debug("httpx_scraping_error", url=url, error=str(e))
        SCRAPE_ERRORS.labels(error_type='httpx_failed').inc()
        return {'url': url, 'error': str(e), 'extraction_method': 'failed'}


# ─── arun_many Parallel Browser Scraping ─────────────────────────────────────

async def stream_urls_browser_parallel(urls: List[str], timeout: int):
    """
    Scrape multiple URLs in TRUE PARALLEL using crawl4ai's native arun_many()
    with stream=True to yield results as soon as they finish.
    """
    if not urls:
        return

    t0 = time.time()
    crawler = await _get_scrape_crawler()

    # Config for actual content extraction
    scrape_config = CrawlerRunConfig(
        cache_mode=CacheMode.BYPASS,
        page_timeout=min(timeout, 8) * 1000,
        word_count_threshold=0,
        process_iframes=False,
        stream=True,  # Crucial for 0.8.0: yields results as they finish
    )

    dispatcher = MemoryAdaptiveDispatcher(
        max_session_permit=5,
        memory_threshold_percent=80.0,
        check_interval=0.5,
    )

    try:
        # arun_many with stream=True returns an async generator
        streamer = await crawler.arun_many(
            urls=urls,
            config=scrape_config,
            dispatcher=dispatcher,
        )
        
        async for cr in streamer:
            # Enforce overall timeout manual bailout
            if time.time() - t0 > min(timeout, 30):
                logger.warning("arun_many_timeout", urls=len(urls), timeout_s=min(timeout, 30))
                SCRAPE_ERRORS.labels(error_type='arun_many_timeout').inc()
                break

            url = cr.url
            domain = _extract_domain(url)
            if cr.success:
                extracted = {}
                if cr.html:
                    extracted = _extract_content_from_html(cr.html, url)
                if not extracted.get('text') and cr.markdown:
                    extracted['text'] = cr.markdown
                    extracted.setdefault('title', '')

                if extracted.get('text'):
                    yield {
                        'url': url,
                        'title': extracted.get('title', ''),
                        'snippet': extracted.get('excerpt', '')[:300],
                        'full_content': extracted.get('text', ''),
                        'published_date': extracted.get('date'),
                        'author': extracted.get('author'),
                        'domain': domain,
                        'extraction_time_ms': (time.time() - t0) * 1000,
                        'extraction_method': 'crawl4ai_parallel',
                    }
                    logger.info("browser_parallel_ok", url=url)
                else:
                    logger.debug("browser_no_content", url=url)
            else:
                err = getattr(cr, 'error_message', 'Unknown')
                logger.warning("browser_parallel_failed", url=url, error=err)

    except Exception as e:
        logger.error("arun_many_error", error=str(e))
        SCRAPE_ERRORS.labels(error_type='arun_many_error').inc()

    logger.info(f"arun_many stream completed in {(time.time()-t0)*1000:.0f}ms")


async def scrape_urls_browser_parallel(urls: List[str], timeout: int) -> List[Dict[str, Any]]:
    """Helper to collect streaming browser results into a list for batch use."""
    results = []
    async for r in stream_urls_browser_parallel(urls, timeout):
        results.append(r)
    return results


# ─── Smart Scraping Orchestrator ─────────────────────────────────────────────

async def scrape_urls_smart(
    urls: List[str],
    timeout: int,
    max_results: int,
) -> List[Dict[str, Any]]:
    """
    Orchestrate scraping across three parallel channels:
      1. httpx for GitHub/HuggingFace/arxiv (instant, no browser)
      2. arun_many for all other URLs (parallel browser tabs)
      3. httpx fallback for any browser failures
    All channels run simultaneously.
    """
    ACTIVE_SCRAPERS.inc()
    try:
        # Split URLs by routing strategy
        httpx_urls = []
        browser_urls = []
        for url in urls:
            url = _clean_url(url)
            if _should_use_httpx(url):
                httpx_urls.append(url)
            else:
                browser_urls.append(url)

        logger.info("routing_urls", httpx=len(httpx_urls), browser=len(browser_urls), total=len(urls))

        async def _empty():
            return []

        # Fire BOTH channels in parallel
        httpx_coro = asyncio.gather(
            *[scrape_url_httpx(u, timeout) for u in httpx_urls],
            return_exceptions=True
        ) if httpx_urls else _empty()

        # Only send max_results+2 URLs to the browser — no need to wait for extras
        browser_coro = scrape_urls_browser_parallel(
            browser_urls[:max_results + 2], timeout
        ) if browser_urls else _empty()

        httpx_results_raw, browser_results = await asyncio.gather(
            httpx_coro, browser_coro, return_exceptions=True
        )

        # Collect valid results
        valid = []

        # Process httpx results
        if isinstance(httpx_results_raw, list):
            for r in httpx_results_raw:
                if isinstance(r, dict) and 'error' not in r:
                    valid.append(r)
                    logger.info("httpx_ok", url=r.get('url', '?'))

        # Process browser results
        if isinstance(browser_results, list):
            valid.extend(browser_results)

        # If we still need more results, try httpx as fallback for failed browser URLs
        if len(valid) < max_results:
            succeeded_urls = {r['url'] for r in valid}
            failed_browser_urls = [u for u in browser_urls if u not in succeeded_urls]

            if failed_browser_urls:
                logger.info("httpx_fallback", count=len(failed_browser_urls))
                fallback_tasks = [scrape_url_httpx(u, timeout) for u in failed_browser_urls]
                fallback_results = await asyncio.gather(*fallback_tasks, return_exceptions=True)
                for r in fallback_results:
                    if isinstance(r, dict) and 'error' not in r:
                        valid.append(r)

        return valid[:max_results]

    finally:
        ACTIVE_SCRAPERS.dec()


# ─── Background Tasks ────────────────────────────────────────────────────────

async def periodic_cache_cleanup():
    while True:
        try:
            await asyncio.sleep(300)
            await cache.clear_expired()
        except asyncio.CancelledError:
            break
        except Exception as e:
            logger.error("cache_cleanup_error", error=str(e))


# ─── App Lifespan ────────────────────────────────────────────────────────────

@asynccontextmanager
async def lifespan(app: FastAPI):
    logger.info("application_starting")
    app.state.start_time = time.time()
    app.state.user_agent = UserAgent()
    app.state.crawler_pool = {}
    app.state.cache_cleanup_task = asyncio.create_task(periodic_cache_cleanup())

    yield

    logger.info("application_shutting_down")
    if hasattr(app.state, 'crawler_pool'):
        for name, crawler in app.state.crawler_pool.items():
            try:
                await crawler.close()
            except Exception as e:
                logger.warning("crawler_close_error", name=name, error=str(e))
        logger.info(f"All {len(app.state.crawler_pool)} crawlers closed.")
        app.state.crawler_pool.clear()
    if hasattr(app.state, 'cache_cleanup_task'):
        app.state.cache_cleanup_task.cancel()
    logger.info("Shutdown complete.")


# ─── FastAPI App ─────────────────────────────────────────────────────────────

app = FastAPI(
    title="NeuralStream Scraping API",
    description="Production-ready async web scraping API with Crawl4AI 0.8 stealth",
    version="2.1.0",
    lifespan=lifespan,
)

app.add_middleware(GZipMiddleware, minimum_size=1000)
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)


async def check_rate_limit(client_ip: str = "127.0.0.1"):
    if not await rate_limiter.is_allowed(client_ip):
        raise HTTPException(status_code=429, detail="Rate limit exceeded. Try again later.")
    return True


# ─── API Endpoints ───────────────────────────────────────────────────────────

@app.get("/")
async def root():
    return {
        "message": "NeuralStream Scraping API — Production Ready",
        "version": "2.1.0",
        "features": [
            "Dual Crawler (Persistent Search + Parallel arun_many Scrape)",
            "MemoryAdaptiveDispatcher (auto-scales tabs by RAM)",
            "Smart Routing (httpx for GitHub/HF, Browser for JS sites)",
            "SSE Real-Time Streaming (POST /scrape/stream)",
            "Three-Channel Parallel Scraping",
            "Graceful Fallback Chain",
        ],
    }


@app.post("/scrape", response_model=ScraperResponse)
async def scrape_data(
    request: ScraperRequest,
    background_tasks: BackgroundTasks,
    _: bool = Depends(check_rate_limit),
):
    """Main scraping endpoint (batch — returns all results at once)."""
    start_time = time.time()
    REQUEST_COUNT.labels(method='POST', endpoint='/scrape').inc()

    cache_key = generate_cache_key(request.query, request.max_results, request.extract_full_content)
    cached_result = await cache.get(cache_key)
    if cached_result:
        logger.info("serving_from_cache", query=request.query)
        return JSONResponse(content=cached_result)

    logger.info("scraping_request",
                query=request.query,
                max_results=request.max_results,
                use_stealth=request.use_stealth_mode)

    with REQUEST_LATENCY.labels(endpoint='/scrape').time():
        try:
            # Phase 1: Google SERP → candidate URLs
            search_results = await extract_search_results(
                request.query, request.max_results, request.timeout
            )
            urls = [r['url'] for r in search_results if r.get('url')]
            if not urls:
                raise HTTPException(status_code=404, detail="No relevant URLs found")

            # Phase 2: Scrape all URLs in parallel (browser tabs + httpx)
            scraped = await scrape_urls_smart(urls, request.timeout, request.max_results)
            if not scraped:
                raise HTTPException(status_code=404, detail="Failed to scrape any URLs")

            # Phase 3: Format response
            results = []
            for d in scraped:
                results.append(ScrapedResult(
                    url=d['url'],
                    title=d.get('title', 'Untitled'),
                    snippet=d.get('snippet', ''),
                    full_content=d.get('full_content', '') if request.extract_full_content else None,
                    published_date=d.get('published_date'),
                    author=d.get('author'),
                    domain=d.get('domain', _extract_domain(d['url'])),
                    scraped_at=datetime.utcnow(),
                    metadata={
                        'extraction_time_ms': d.get('extraction_time_ms', 0),
                        'content_length': len(d.get('full_content', '')),
                        'extraction_method': d.get('extraction_method', 'unknown'),
                    },
                ))

            proc_time = (time.time() - start_time) * 1000
            response = ScraperResponse(
                query=request.query, results=results,
                total_results=len(results), processing_time_ms=proc_time,
                cached=False, timestamp=datetime.utcnow(),
            )

            response_dict = json.loads(response.model_dump_json())
            background_tasks.add_task(cache.set, cache_key, response_dict)

            logger.info("scraping_completed",
                        query=request.query, results=len(results),
                        processing_time_ms=proc_time,
                        stealth_enabled=request.use_stealth_mode)
            return response

        except HTTPException:
            raise
        except Exception as e:
            logger.error("scraping_failed", query=request.query, error=str(e))
            raise HTTPException(status_code=500, detail=f"Scraping failed: {str(e)}")


# ─── SSE Streaming Endpoint ──────────────────────────────────────────────────

@app.post("/scrape/stream")
async def scrape_stream(
    request: ScraperRequest,
    _: bool = Depends(check_rate_limit),
):
    """
    Real-time streaming scrape — sends results via Server-Sent Events
    as each URL finishes scraping. Frontend receives cards immediately.
    """
    async def event_generator():
        start_time = time.time()
        total_results = 0

        try:
            # Phase 1: Google SERP search
            yield f"event: status\ndata: {json.dumps({'message': 'Searching Google...'})}\n\n"

            search_results = await extract_search_results(
                request.query, request.max_results, request.timeout
            )
            urls = [r['url'] for r in search_results if r.get('url')]

            if not urls:
                yield f"event: error\ndata: {json.dumps({'message': 'No relevant URLs found'})}\n\n"
                return

            yield f"event: search_complete\ndata: {json.dumps({'urls_found': len(urls), 'urls': urls[:10]})}\n\n"

            # Phase 2: Route URLs to channels
            httpx_urls = []
            browser_urls = []
            for url in urls:
                url = _clean_url(url)
                if _should_use_httpx(url):
                    httpx_urls.append(url)
                else:
                    browser_urls.append(url)

            # Multiplex streaming via asyncio.Queue
            queue = asyncio.Queue()
            active_tasks = []

            async def _run_httpx(u):
                try:
                    res = await scrape_url_httpx(u, request.timeout)
                    await queue.put(res)
                except Exception as e:
                    await queue.put({'url': u, 'error': str(e)})

            async def _run_browser():
                try:
                    async for res in stream_urls_browser_parallel(browser_urls[:request.max_results + 2], request.timeout):
                        await queue.put(res)
                except Exception as e:
                    logger.error("stream_browser_error", error=str(e))

            for u in httpx_urls:
                active_tasks.append(asyncio.create_task(_run_httpx(u)))
            
            if browser_urls:
                active_tasks.append(asyncio.create_task(_run_browser()))

            # Stream results as soon as they hit the queue
            def _all_done():
                return all(t.done() for t in active_tasks)

            while not _all_done() or not queue.empty():
                try:
                    res = await asyncio.wait_for(queue.get(), timeout=0.5)
                    if isinstance(res, dict) and 'error' not in res:
                        total_results += 1
                        yield f"event: result\ndata: {json.dumps(res, default=str)}\n\n"
                    else:
                        yield f"event: skipped\ndata: {json.dumps({'url': res.get('url', '?'), 'reason': res.get('error', 'unknown')})}\n\n"
                except asyncio.TimeoutError:
                    continue  # Just check _all_done() again
                except Exception as e:
                    logger.error("queue_consumer_error", error=str(e))

            # Final complete event
            proc_time = (time.time() - start_time) * 1000
            yield f"event: complete\ndata: {json.dumps({'total_results': total_results, 'processing_time_ms': round(proc_time)})}\n\n"

        except Exception as e:
            logger.error("stream_error", error=str(e))
            yield f"event: error\ndata: {json.dumps({'message': str(e)})}\n\n"

    return StarletteStreamingResponse(
        event_generator(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "Connection": "keep-alive",
            "X-Accel-Buffering": "no",
        },
    )


# ─── Health, Metrics, Cache ──────────────────────────────────────────────────

@app.get("/health", response_model=HealthResponse)
async def health_check():
    process = psutil.Process()
    return HealthResponse(
        status="healthy", version="2.1.0",
        uptime_seconds=time.time() - app.state.start_time,
        memory_usage_mb=process.memory_info().rss / 1024 / 1024,
        cpu_percent=process.cpu_percent(interval=0.1),
        active_scrapers=ACTIVE_SCRAPERS._value.get(),
        cache_size=len(cache.cache),
    )


@app.get("/metrics")
async def metrics():
    return Response(content=generate_latest(METRICS_REGISTRY), media_type=CONTENT_TYPE_LATEST)


@app.post("/clear-cache")
async def clear_cache():
    cache.cache.clear()
    logger.info("cache_cleared")
    return {"message": "Cache cleared successfully"}


if __name__ == "__main__":
    import uvicorn
    uvicorn.run(
        "app:app", host="0.0.0.0", port=8000, workers=1,
        loop="asyncio", http="httptools",
        access_log=False, server_header=False, date_header=False,
        forwarded_allow_ips="*",
    )