# retoor <retoor@molodetz.nl>

from __future__ import annotations

import asyncio
import hashlib
import logging
import re
import time
from dataclasses import dataclass, field
from itertools import zip_longest
from typing import Awaitable, Callable
from urllib.parse import urlparse

import httpx

from devplacepy import stealth
from devplacepy.net_guard import BlockedAddressError, guard_public_url, guarded_async_client

from .extract import extract_html, relevant_links
from .pdf import MAX_PDF_BYTES, extract_pdf_text, is_pdf

logger = logging.getLogger(__name__)

_pw_lock = asyncio.Lock()

RSEARCH_URL = "https://rsearch.app.molodetz.nl"
RSEARCH_TIMEOUT_SECONDS = 45.0
FETCH_TIMEOUT_SECONDS = 20.0
MAX_FETCH_BYTES = 2_500_000
RESULTS_PER_QUERY = 8
CRAWL_CONCURRENCY = 4
LINKS_PER_PAGE = 3
USER_AGENT = (
    "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) "
    "Chrome/131.0.0.0 Safari/537.36 DevPlaceDeepSearchBot/1.0"
)
TAG = re.compile(r"<[^>]+>")
MIN_PAGE_CHARS = 200
SNIPPET_MIN_CHARS = 120
HOSTILE_DOMAINS = (
    "x.com",
    "twitter.com",
    "mobile.twitter.com",
    "youtube.com",
    "youtu.be",
    "m.youtube.com",
    "reddit.com",
    "www.reddit.com",
    "old.reddit.com",
    "facebook.com",
    "www.facebook.com",
    "instagram.com",
    "www.instagram.com",
    "linkedin.com",
    "www.linkedin.com",
    "tiktok.com",
    "www.tiktok.com",
    "threads.net",
)
WS = re.compile(r"\s+")


def _clean_snippet(text: str) -> str:
    if not text:
        return ""
    stripped = TAG.sub(" ", text) if "<" in text and ">" in text else text
    return WS.sub(" ", stripped).strip()


def _is_hostile(url: str) -> bool:
    host = urlparse(url).netloc.lower()
    return any(host == domain or host.endswith("." + domain) for domain in HOSTILE_DOMAINS)


@dataclass
class CrawledPage:
    url: str
    title: str
    text: str
    source: str
    status: int
    depth: int = 0
    from_cache: bool = False
    links: list[tuple[str, str]] = field(default_factory=list)


@dataclass
class CrawlOutcome:
    pages: list[CrawledPage] = field(default_factory=list)
    seen_hashes: set[str] = field(default_factory=set)


def url_hash(url: str) -> str:
    return hashlib.sha256(url.strip().lower().encode("utf-8")).hexdigest()


def content_hash(text: str) -> str:
    return hashlib.sha256(text.strip().encode("utf-8")).hexdigest()


def _interleave(buckets: list[list[dict]]) -> list[dict]:
    merged: list[dict] = []
    seen: set[str] = set()
    for tier in zip_longest(*buckets):
        for item in tier:
            if not item:
                continue
            url = item["url"]
            if url in seen:
                continue
            seen.add(url)
            merged.append(item)
    return merged


async def search_queries(
    queries: list[str], emit: Callable[[dict], None] = lambda frame: None
) -> list[dict]:
    buckets: list[list[dict]] = []
    headers = {"User-Agent": USER_AGENT, "Accept": "application/json"}
    timeout = httpx.Timeout(RSEARCH_TIMEOUT_SECONDS, connect=30.0)
    async with stealth.stealth_async_client(
        base_url=RSEARCH_URL, headers=headers, follow_redirects=True, timeout=timeout
    ) as client:
        for query in queries:
            try:
                response = await client.get(
                    "/search",
                    params={"query": query, "count": RESULTS_PER_QUERY, "content": "true"},
                )
                emit({"type": "rsearch", "endpoint": "/search", "success": response.status_code < 400})
                if response.status_code >= 400:
                    continue
                data = response.json()
            except (httpx.HTTPError, ValueError) as exc:
                emit({"type": "rsearch", "endpoint": "/search", "success": False})
                logger.warning("deepsearch rsearch failed for %r: %s", query, exc)
                continue
            bucket: list[dict] = []
            for item in data.get("results") or []:
                url = (item.get("url") or "").strip()
                if not url:
                    continue
                bucket.append(
                    {
                        "url": url,
                        "title": item.get("title") or "",
                        "description": item.get("description") or "",
                        "content": item.get("content") or "",
                        "query": query,
                    }
                )
            buckets.append(bucket)
    return _interleave(buckets)


def _snippet_page(candidate: dict, depth: int) -> CrawledPage | None:
    snippet = _clean_snippet(candidate.get("content") or candidate.get("description") or "")
    if len(snippet) < SNIPPET_MIN_CHARS:
        return None
    return CrawledPage(
        url=candidate["url"],
        title=_clean_snippet(candidate.get("title") or "") or candidate["url"],
        text=snippet,
        source="search",
        status=200,
        depth=depth,
    )


async def _render_with_playwright(
    url: str, browser=None
) -> tuple[str, str, int, list[tuple[str, str]]]:
    from playwright.async_api import async_playwright

    async def _render(browser) -> tuple[str, str, int, list[tuple[str, str]]]:
        context = await browser.new_context(user_agent=USER_AGENT)
        page = await context.new_page()
        response = await page.goto(url, wait_until="load", timeout=30000)
        status = response.status if response else 0
        for hop in [response.url] if response else []:
            await guard_public_url(hop)
        content = (await page.content())[:MAX_FETCH_BYTES]
        await context.close()
        extracted = extract_html(content, base_url=url)
        return extracted.title, extracted.text, status, extracted.links

    if browser is None:
        async with async_playwright() as pw:
            browser = await pw.chromium.launch(
                headless=True, args=["--no-sandbox", "--disable-dev-shm-usage"]
            )
            try:
                return await _render(browser)
            finally:
                await browser.close()
    else:
        return await _render(browser)


async def fetch_page(url: str, depth: int, browser=None) -> CrawledPage | None:
    try:
        await guard_public_url(url)
    except BlockedAddressError:
        return None
    title = ""
    text = ""
    status = 0
    source = "httpx"
    links: list[tuple[str, str]] = []
    content_type = ""
    encoding = "utf-8"
    raw_bytes = b""
    try:
        async with guarded_async_client(
            follow_redirects=True,
            timeout=FETCH_TIMEOUT_SECONDS,
            headers={"User-Agent": USER_AGENT},
        ) as client:
            async with client.stream("GET", url) as response:
                await guard_public_url(str(response.url))
                status = response.status_code
                content_type = response.headers.get("content-type", "").lower()
                encoding = response.encoding or "utf-8"
                buffer = bytearray()
                async for chunk in response.aiter_bytes():
                    buffer.extend(chunk)
                    if len(buffer) >= MAX_PDF_BYTES:
                        break
                raw_bytes = bytes(buffer)
    except (httpx.HTTPError, BlockedAddressError) as exc:
        logger.info("deepsearch httpx fetch failed for %s: %s", url, exc)
    if is_pdf(content_type, url, raw_bytes[:5]):
        title, text = extract_pdf_text(raw_bytes[:MAX_PDF_BYTES])
        if len(text) < MIN_PAGE_CHARS:
            return None
        return CrawledPage(
            url=url, title=title or url, text=text, source="pdf", status=status, depth=depth
        )
    if raw_bytes:
        try:
            raw = raw_bytes[:MAX_FETCH_BYTES].decode(encoding, errors="replace")
            extracted = extract_html(raw, base_url=url)
            title, text, links = extracted.title, extracted.text, extracted.links
        except (LookupError, ValueError) as exc:
            logger.info("deepsearch decode failed for %s: %s", url, exc)
    if len(text) < MIN_PAGE_CHARS:
        async with _pw_lock:
            try:
                r_title, r_text, r_status, r_links = await _render_with_playwright(url, browser)
            except Exception as exc:
                logger.info("deepsearch render failed for %s: %s", url, exc)
                r_title, r_text, r_status, r_links = "", "", 0, []
        if len(r_text) > len(text):
            title, text, status, source, links = (
                r_title or title,
                r_text,
                r_status or status,
                "playwright",
                r_links,
            )
    if len(text) < MIN_PAGE_CHARS:
        return None
    return CrawledPage(
        url=url,
        title=title or url,
        text=text,
        source=source,
        status=status,
        depth=depth,
        links=links,
    )


async def _resolve_candidate(candidate: dict, depth: int, browser=None) -> CrawledPage | None:
    url = candidate["url"]
    snippet_page = _snippet_page(candidate, depth)
    if _is_hostile(url):
        return snippet_page
    page = await fetch_page(url, depth, browser)
    if page and snippet_page:
        return page if len(page.text) >= len(snippet_page.text) else snippet_page
    return page or snippet_page


async def crawl(
    candidates: list[dict],
    max_pages: int,
    emit: Callable[[dict], None],
    is_cached: Callable[[str], bool],
    should_stop: Callable[[], Awaitable[bool]],
    query: str = "",
    depth: int = 1,
    seen_hashes: set[str] | None = None,
) -> CrawlOutcome:
    from playwright.async_api import async_playwright

    outcome = CrawlOutcome(seen_hashes=seen_hashes if seen_hashes is not None else set())
    fetched = 0
    seen_urls = {candidate["url"] for candidate in candidates}
    level_candidates = list(candidates)
    total = min(len(level_candidates), max_pages)
    cancelled = False

    pw = None
    browser = None
    try:
        pw = await async_playwright().start()
        browser = await pw.chromium.launch(
            headless=True, args=["--no-sandbox", "--disable-dev-shm-usage"]
        )
    except Exception as exc:
        logger.warning("deepsearch playwright launch failed, pages will use httpx only: %s", exc)

    try:
        for level in range(max(1, depth)):
            if cancelled or fetched >= max_pages or not level_candidates:
                break
            next_candidates: list[dict] = []
            for start in range(0, len(level_candidates), CRAWL_CONCURRENCY):
                if fetched >= max_pages:
                    break
                if await should_stop():
                    emit({"type": "stage", "stage": "cancelled", "message": "Crawl cancelled"})
                    cancelled = True
                    break
                batch = level_candidates[start : start + CRAWL_CONCURRENCY][: max_pages - fetched]
                for candidate in batch:
                    emit(
                        {
                            "type": "progress",
                            "done": fetched,
                            "total": total,
                            "url": candidate["url"],
                            "depth": level,
                            "message": f"Reading {candidate['url']}",
                        }
                    )
                    if is_cached(candidate["url"]):
                        emit({"type": "page_cached", "url": candidate["url"], "reason": "seen in a prior run"})
                fetch_start = time.perf_counter()
                results = await asyncio.gather(
                    *(_resolve_candidate(candidate, level, browser) for candidate in batch),
                    return_exceptions=True,
                )
                elapsed_ms = int((time.perf_counter() - fetch_start) * 1000)
                for candidate, page in zip(batch, results):
                    url = candidate["url"]
                    if isinstance(page, BaseException):
                        logger.info("deepsearch fetch crashed for %s: %s", url, page)
                        page = None
                    if page is None:
                        emit(
                            {
                                "type": "page_skipped",
                                "url": url,
                                "reason": "no readable content",
                                "elapsed_ms": elapsed_ms,
                            }
                        )
                        continue
                    if fetched >= max_pages:
                        break
                    digest = content_hash(page.text)
                    if digest in outcome.seen_hashes:
                        emit(
                            {
                                "type": "page_duplicate",
                                "url": url,
                                "reason": "duplicate content",
                                "elapsed_ms": elapsed_ms,
                            }
                        )
                        continue
                    outcome.seen_hashes.add(digest)
                    outcome.pages.append(page)
                    fetched += 1
                    emit(
                        {
                            "type": "page_loaded",
                            "url": page.url,
                            "title": page.title,
                            "source": page.source,
                            "depth": level,
                            "render": page.source == "playwright",
                            "elapsed_ms": elapsed_ms,
                            "done": fetched,
                            "total": total,
                        }
                    )
                    if level + 1 < depth:
                        for link in relevant_links(page.links, query, LINKS_PER_PAGE):
                            if link not in seen_urls:
                                seen_urls.add(link)
                                next_candidates.append({"url": link})
            level_candidates = next_candidates
            total = min(total + len(next_candidates), max_pages)
    finally:
        if browser is not None:
            await browser.close()
        if pw is not None:
            await pw.stop()
    return outcome
