from __future__ import annotations

import asyncio
import logging
import os
import secrets
from contextlib import asynccontextmanager
from datetime import datetime, timezone
from time import struct_time
from typing import Annotated, Any, AsyncIterator
from urllib.parse import urljoin

import feedparser
import httpx
from bs4 import BeautifulSoup
from curl_cffi.requests import AsyncSession as CurlAsyncSession
from fastapi import FastAPI, Header, HTTPException
from playwright.async_api import BrowserContext, Page, TimeoutError as PlaywrightTimeoutError, async_playwright
from playwright_captcha import CaptchaType, FrameworkType, TwoCaptchaSolver
from pydantic import BaseModel, Field, HttpUrl, SecretStr
from twocaptcha import AsyncTwoCaptcha

from extractor import ConfigError, detail_from_html, extract_fields, extract_media, normalize_config

app = FastAPI(title="Site Parser Worker", version="1.0.0")
browser_lock = asyncio.Lock()
logger = logging.getLogger("site_parser")


class ParseRequest(BaseModel):
    url: HttpUrl
    config: dict[str, Any]
    rucaptcha_token: SecretStr | None = None
    browser_state: dict[str, Any] | None = None
    browser_user_agent: str | None = None


class ParsedItem(BaseModel):
    external_id: str
    url: str
    title: str
    text: str
    html: str
    published_at: str | None
    author: str | None
    media: list[dict[str, str]] = Field(default_factory=list)


class CloudflareRequired(RuntimeError):
    pass


class PermanentPageError(RuntimeError):
    pass


def require_token(worker_token: str | None) -> None:
    expected = os.getenv("WORKER_TOKEN", "")
    if not expected:
        raise HTTPException(status_code=503, detail="WORKER_TOKEN is not configured")
    if not worker_token or not secrets.compare_digest(worker_token, expected):
        raise HTTPException(status_code=401, detail="Invalid worker token")


def is_cloudflare_challenge(status: int, body: str) -> bool:
    sample = body[:20_000].lower()
    return status in {403, 429, 503} and (
        "just a moment" in sample or "cf-chl" in sample or "challenge-platform" in sample
    )


def iso_date(value: struct_time | None) -> str | None:
    if not value:
        return None
    return datetime(*value[:6], tzinfo=timezone.utc).isoformat()


def clean_feed_content(value: Any, base_url: str) -> tuple[str, str, list[dict[str, str]]]:
    html = str(value or "").strip()
    soup = BeautifulSoup(html, "html.parser")
    for element in soup.find_all(["script", "style", "noscript"]):
        element.decompose()
    text = soup.get_text("\n", strip=True)
    media = extract_media(
        soup,
        [
            {"type": "photo", "selector": "img", "attributes": ["src", "data-src", "srcset"]},
            {"type": "video", "selector": "iframe, video, source", "attributes": ["src", "data-src"]},
        ],
        base_url,
    )
    return html, text, media


def default_rss_fields() -> dict[str, Any]:
    return {
        "url": {"candidates": [{"source": "rss.link"}], "required": True},
        "external_id": {
            "candidates": [{"source": "rss.id"}, {"source": "rss.guid"}, {"source": "rss.link"}],
            "required": True,
        },
        "title": {"candidates": [{"source": "rss.title"}]},
        "published_at": {
            "candidates": [
                {"source": "rss.published"},
                {"source": "rss.updated"},
            ]
        },
        "author": {"candidates": [{"source": "rss.author"}]},
    }


def discover_rss(xml: str, config: dict[str, Any]) -> list[dict[str, Any]]:
    feed = feedparser.parse(xml, sanitize_html=False)
    if feed.bozo and not feed.entries:
        raise ValueError(f"Invalid RSS: {feed.bozo_exception}")
    discovery = config["discovery"]
    fields = discovery.get("fields") or default_rss_fields()
    items: list[dict[str, Any]] = []
    for entry in feed.entries[: discovery["limit"]]:
        raw = dict(entry)
        values, errors = extract_fields(None, fields, {"rss": raw})
        url = str(values.get("url") or raw.get("link") or "").strip()
        external_id = str(values.get("external_id") or raw.get("id") or raw.get("guid") or url).strip()
        if not url or not external_id:
            items.append({"url": url, "discovery_errors": errors or [{"field": "url", "error": "URL not found"}]})
            continue
        content = (raw.get("content") or [{}])[0].get("value") or raw.get("description") or raw.get("summary") or ""
        html, text, media = clean_feed_content(content, url)
        for enclosure in raw.get("enclosures") or []:
            enclosure_url = urljoin(url, str(enclosure.get("href") or enclosure.get("url") or "").strip())
            media_type = str(enclosure.get("type") or "")
            if enclosure_url and media_type.startswith("video/"):
                media.append({"type": "video", "url": enclosure_url})
            elif enclosure_url:
                media.append({"type": "photo", "url": enclosure_url})
        items.append(
            {
                "external_id": external_id,
                "url": url,
                "title": str(values.get("title") or BeautifulSoup(str(raw.get("title") or ""), "html.parser").get_text(" ", strip=True)),
                "text": text,
                "html": html,
                "published_at": values.get("published_at") or iso_date(raw.get("published_parsed") or raw.get("updated_parsed")),
                "author": values.get("author") or str(raw.get("author") or "").strip() or None,
                "media": list({entry["url"]: entry for entry in media}.values()),
                "source_context": {"rss": raw},
                "discovery_errors": errors,
            }
        )
    return items


def discover_html(html: str, source_url: str, config: dict[str, Any]) -> list[dict[str, Any]]:
    soup = BeautifulSoup(html, "html.parser")
    discovery = config["discovery"]
    fields = discovery.get("fields") or {}
    items: list[dict[str, Any]] = []
    for element in soup.select(str(discovery["item_selector"]))[: discovery["limit"]]:
        values, errors = extract_fields(element, fields, {"page": {"url": source_url}})
        url = urljoin(source_url, str(values.get("url") or "").strip())
        external_id = str(values.get("external_id") or url).strip()
        if not url or not external_id:
            items.append({"url": url, "discovery_errors": errors or [{"field": "url", "error": "URL not found"}]})
            continue
        items.append(
            {
                "external_id": external_id,
                "url": url,
                "title": str(values.get("title") or ""),
                "text": str(values.get("text") or ""),
                "html": "",
                "published_at": values.get("published_at"),
                "author": values.get("author"),
                "media": extract_media(element, discovery.get("media") or [], source_url),
                "source_context": {"list": values},
                "discovery_errors": errors,
            }
        )
    return items


def apply_detail(item: dict[str, Any], html: str, config: dict[str, Any]) -> tuple[dict[str, Any], list[dict[str, str]]]:
    values, errors = detail_from_html(html, item["url"], config, item.get("source_context"))
    if errors:
        return item, errors
    for key in ("title", "text", "published_at", "author", "html"):
        if values.get(key) not in {None, ""}:
            item[key] = values[key]
    item["media"] = list(
        {(entry["type"], entry["url"]): entry for entry in [*item.get("media", []), *values.get("media", [])]}.values()
    )
    return item, []


def public_item(item: dict[str, Any]) -> ParsedItem:
    return ParsedItem(
        external_id=str(item["external_id"]),
        url=str(item["url"]),
        title=str(item.get("title") or ""),
        text=str(item.get("text") or ""),
        html=str(item.get("html") or ""),
        published_at=str(item["published_at"]) if item.get("published_at") else None,
        author=str(item["author"]) if item.get("author") else None,
        media=item.get("media") or [],
    )


async def retry(operation, attempts: int, delay: float, timeout: float):
    last_error: Exception | None = None
    for attempt in range(1, attempts + 1):
        try:
            return await asyncio.wait_for(operation(), timeout=timeout), attempt
        except PermanentPageError:
            raise
        except Exception as exc:
            last_error = exc
            if attempt < attempts and delay:
                await asyncio.sleep(delay)
    raise RuntimeError(str(last_error or "operation failed")) from last_error


async def http_document(client: httpx.AsyncClient, url: str) -> str:
    response = await client.get(url)
    if is_cloudflare_challenge(response.status_code, response.text):
        raise CloudflareRequired("Cloudflare challenge received")
    response.raise_for_status()
    return response.text


async def curl_document(client: CurlAsyncSession, url: str) -> str:
    response = await client.get(url, allow_redirects=True, timeout=30)
    body = response.text
    if is_cloudflare_challenge(response.status_code, body) or "Attention Required" in body[:10_000]:
        raise CloudflareRequired("Cloudflare blocked impersonated request")
    if response.status_code < 200 or response.status_code >= 300:
        raise RuntimeError(f"detail returned HTTP {response.status_code}")
    return body


async def enrich_items(
    items: list[dict[str, Any]],
    config: dict[str, Any],
    document_loader,
) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
    attempts = config["retry"]["attempts"]
    delay = config["retry"]["delay_seconds"]
    timeout = config["retry"]["timeout_seconds"]
    errors: list[dict[str, Any]] = []
    enriched = []
    for item in items:
        if item.get("discovery_errors"):
            errors.append({"url": item.get("url"), "stage": "discovery", "attempts": 1, "errors": item["discovery_errors"]})
            continue
        if not config["detail"]["always"] and item.get("text") and item.get("media"):
            enriched.append(item)
            continue
        try:
            html, used_attempts = await retry(lambda item=item: document_loader(item["url"]), attempts, delay, timeout)
            item, field_errors = apply_detail(item, html, config)
            if field_errors:
                errors.append({"url": item["url"], "stage": "detail", "attempts": used_attempts, "errors": field_errors})
                continue
            enriched.append(item)
            logger.info("Parsed detail %s", item["url"])
        except CloudflareRequired:
            raise
        except Exception as exc:
            errors.append({"url": item.get("url"), "stage": "detail", "attempts": attempts, "error": str(exc)})
            logger.warning("Detail failed %s: %s", item.get("url"), exc)
    return enriched, errors


async def parse_over_http(source_url: str, config: dict[str, Any]) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
    attempts = config["retry"]["attempts"]
    delay = config["retry"]["delay_seconds"]
    timeout = config["retry"]["timeout_seconds"]
    async with httpx.AsyncClient(
        follow_redirects=True,
        timeout=30,
        headers={"User-Agent": "Mozilla/5.0 SiteParser/1.0"},
    ) as client:
        document, _ = await retry(lambda: http_document(client, source_url), attempts, delay, timeout)
        items = discover_rss(document, config) if config["discovery"]["type"] == "rss" else discover_html(document, source_url, config)
        if config["detail"]["enabled"]:
            return await enrich_items(items, config, lambda url: http_document(client, url))
    return items, []


@asynccontextmanager
async def captcha_solver(page: Page, token: str | None) -> AsyncIterator[TwoCaptchaSolver | None]:
    if not token:
        yield None
        return
    async with TwoCaptchaSolver(
        framework=FrameworkType.PLAYWRIGHT,
        page=page,
        async_two_captcha_client=AsyncTwoCaptcha(token),
        max_attempts=1,
    ) as solver:
        yield solver


async def browser_document(
    page: Page,
    url: str,
    solver: TwoCaptchaSolver | None,
    wait_for: str | None = None,
) -> str:
    await page.goto(url, wait_until="domcontentloaded", timeout=60_000)
    title = (await page.title()).lower()
    if "just a moment" in title or "attention required" in title:
        if solver is None:
            raise RuntimeError("Cloudflare challenge received but RuCaptcha is not configured")
        await solver.solve_captcha(
            captcha_container=page,
            captcha_type=CaptchaType.CLOUDFLARE_INTERSTITIAL,
        )
    if wait_for:
        try:
            await page.locator(wait_for).first.wait_for(state="attached", timeout=20_000)
        except PlaywrightTimeoutError as exc:
            raise PermanentPageError(
                f"selector not found: {wait_for}; url={page.url}; title={await page.title()}"
            ) from exc
    return await page.content()


async def browser_discovery_document(page: Page, source_url: str, config: dict[str, Any], solver) -> str:
    wait_for = config["access"].get("wait_for") if config["discovery"]["type"] == "html" else None
    html = await browser_document(page, source_url, solver, wait_for)
    if config["discovery"]["type"] != "rss":
        return html
    pre = page.locator("pre").first
    await pre.wait_for(state="attached", timeout=60_000)
    return await pre.inner_text()


async def browser_fetch_document(page: Page, url: str) -> str:
    result = await page.evaluate(
        """
        async (url) => {
          const response = await fetch(url, {credentials: 'include'});
          return {status: response.status, text: await response.text()};
        }
        """,
        url,
    )
    status = int(result.get("status") or 0)
    body = str(result.get("text") or "")
    if is_cloudflare_challenge(status, body) or "Attention Required" in body[:10_000]:
        raise CloudflareRequired("Cloudflare blocked browser fetch")
    if status < 200 or status >= 300:
        raise RuntimeError(f"detail returned HTTP {status}")
    return body


async def parse_in_browser(
    source_url: str,
    config: dict[str, Any],
    rucaptcha_token: str | None,
    browser_state: dict[str, Any] | None,
    browser_user_agent: str | None,
) -> tuple[list[dict[str, Any]], list[dict[str, Any]], dict[str, Any], str]:
    attempts = config["retry"]["attempts"]
    delay = config["retry"]["delay_seconds"]
    timeout = config["retry"]["timeout_seconds"]
    errors: list[dict[str, Any]] = []
    async with browser_lock:
        async with async_playwright() as playwright:
            browser = await playwright.chromium.launch(channel="chrome", headless=False)
            context: BrowserContext = await browser.new_context(storage_state=browser_state, user_agent=browser_user_agent)
            page = await context.new_page()
            async with captcha_solver(page, rucaptcha_token) as solver:
                document, _ = await retry(
                    lambda: browser_discovery_document(page, source_url, config, solver), attempts, delay, timeout
                )
                items = discover_rss(document, config) if config["discovery"]["type"] == "rss" else discover_html(document, source_url, config)
                logger.info("Discovered %s items from %s", len(items), source_url)
                if config["detail"]["enabled"]:
                    if config["detail"]["transport"] == "http":
                        original_state = browser_state if isinstance(browser_state, dict) else {}
                        original_cookies = original_state.get("cookies") or []
                        cookies = {
                            str(cookie["name"]): str(cookie["value"])
                            for cookie in (original_cookies or await context.cookies())
                            if cookie.get("name") and cookie.get("value")
                        }
                        user_agent = browser_user_agent or await page.evaluate("navigator.userAgent")
                        async with CurlAsyncSession(
                            impersonate="chrome",
                            cookies=cookies,
                            headers={"User-Agent": str(user_agent), "Referer": source_url},
                        ) as client:
                            items, errors = await enrich_items(items, config, lambda url: curl_document(client, url))
                    elif config["detail"]["transport"] == "fetch":
                        items, errors = await enrich_items(items, config, lambda url: browser_fetch_document(page, url))
                    else:
                        enriched = []
                        wait_for = str(config["detail"].get("wait_for") or config["detail"]["root_selector"])
                        for item in items:
                            if item.get("discovery_errors"):
                                errors.append({"url": item.get("url"), "stage": "discovery", "attempts": 1, "errors": item["discovery_errors"]})
                                continue
                            if not config["detail"]["always"] and item.get("text") and item.get("media"):
                                enriched.append(item)
                                continue
                            try:
                                html, used_attempts = await retry(
                                    lambda item=item: browser_document(page, item["url"], solver, wait_for), attempts, delay, timeout
                                )
                                item, field_errors = apply_detail(item, html, config)
                                if field_errors:
                                    errors.append({"url": item["url"], "stage": "detail", "attempts": used_attempts, "errors": field_errors})
                                    continue
                                enriched.append(item)
                                logger.info("Parsed detail %s", item["url"])
                            except Exception as exc:
                                logger.warning("Detail failed %s: %s", item.get("url"), exc)
                                errors.append({"url": item.get("url"), "stage": "detail", "attempts": attempts, "error": str(exc)})
                        items = enriched
                state = await context.storage_state()
                user_agent = await page.evaluate("navigator.userAgent")
            await browser.close()
    return items, errors, state, user_agent


@app.get("/health")
async def health() -> dict[str, bool]:
    return {"ok": True}


@app.post("/v1/parse")
async def parse_source(
    request: ParseRequest,
    worker_token: Annotated[str | None, Header(alias="X-Worker-Token")] = None,
) -> dict[str, Any]:
    require_token(worker_token)
    try:
        config = normalize_config(request.config)
    except ConfigError as exc:
        raise HTTPException(status_code=422, detail=str(exc)) from exc

    source_url = str(request.url)
    access = config["access"]["type"]
    state = request.browser_state
    browser_user_agent = request.browser_user_agent
    fetched_via = "http"
    logger.info("Parse request source=%s discovery=%s detail=%s", source_url, config["discovery"]["type"], config["detail"]["enabled"])
    try:
        if access in {"browser", "cloudflare"}:
            items, errors, state, browser_user_agent = await parse_in_browser(
                source_url,
                config,
                request.rucaptcha_token.get_secret_value() if request.rucaptcha_token else None,
                state,
                browser_user_agent,
            )
            fetched_via = "browser"
        else:
            try:
                items, errors = await parse_over_http(source_url, config)
            except CloudflareRequired:
                if access == "http":
                    raise
                items, errors, state, browser_user_agent = await parse_in_browser(
                    source_url,
                    config,
                    request.rucaptcha_token.get_secret_value() if request.rucaptcha_token else None,
                    state,
                    browser_user_agent,
                )
                fetched_via = "browser"
    except Exception as exc:
        raise HTTPException(status_code=502, detail=str(exc)) from exc

    if not items and not errors:
        errors = [{"url": source_url, "stage": "discovery", "attempts": 1, "error": "no items found"}]
    status = "partial" if errors and items else "failed" if errors else "ok"
    return {
        "status": status,
        "source_url": source_url,
        "fetched_via": fetched_via,
        "items": [public_item(item).model_dump() for item in items],
        "errors": errors,
        "browser_state": state,
        "browser_user_agent": browser_user_agent,
    }