"""Bounded site crawler with robots.txt, SSRF-safe fetching and pagination."""

from __future__ import annotations

import asyncio
import logging
import time
from collections.abc import Awaitable, Callable
from dataclasses import dataclass, field
from pathlib import PurePosixPath
from urllib.parse import unquote, urlsplit
from urllib.robotparser import RobotFileParser

from app.adapters.base import ParsedPage, SiteAdapter
from app.adapters.browser_adapter import BrowserRenderer
from app.adapters.registry import AdapterRegistry
from app.config import Settings
from app.database.models import (
    Job,
    PageStatus,
    Video,
    VideoStatus,
)
from app.database.repository import Repository
from app.errors import (
    AccessDeniedError,
    AppError,
    BrowserUnavailableError,
    ContentTooLargeError,
    DownloadCancelledError,
    MediaNotFoundError,
    NetworkError,
    NotFoundError,
    PermanentSiteError,
    RateLimitedError,
    RobotsDeniedError,
    SsrfBlockedError,
)
from app.services.caption_parser import normalize_text
from app.services.extractor import Extractor
from app.services.http_client import FetchedPage, SafeHttpClient
from app.services.url_validator import NetworkGuard
from app.utils.files import ALLOWED_MEDIA_EXTENSIONS
from app.utils.retry import retry_async

logger = logging.getLogger(__name__)

PLAYLIST_EXTENSIONS = {".m3u8", ".mpd"}
_ALL_DIRECT_EXTENSIONS = ALLOWED_MEDIA_EXTENSIONS | PLAYLIST_EXTENSIONS
Checkpoint = Callable[[], Awaitable[None]]


@dataclass(frozen=True, slots=True)
class CrawlResult:
    pagination_pages: int
    detail_pages: int
    videos_found: int
    duplicate_videos: int
    stop_reason: str
    notes: list[str] = field(default_factory=list)


@dataclass(slots=True)
class _RobotsEntry:
    expires_at: float
    parser: RobotFileParser | None
    disallow_all: bool = False
    delay: float = 0.0


class RobotsPolicy:
    def __init__(self, http: SafeHttpClient, *, enabled: bool = True) -> None:
        self.http = http
        self.enabled = enabled
        self._cache: dict[str, _RobotsEntry] = {}

    async def _entry(self, url: str) -> _RobotsEntry:
        parts = urlsplit(url)
        origin = f"{parts.scheme}://{parts.netloc}"
        now = time.monotonic()
        cached = self._cache.get(origin)
        if cached and cached.expires_at > now:
            return cached
        robots_url = f"{origin}/robots.txt"
        entry: _RobotsEntry
        try:
            raw = await self.http.fetch_bytes(robots_url, max_bytes=512 * 1024)
            text = raw.decode("utf-8", errors="replace")
            parser = RobotFileParser()
            parser.set_url(robots_url)
            parser.parse(text.splitlines())
            delay = parser.crawl_delay("TelegramVideoBot") or parser.crawl_delay("*") or 0.0
            entry = _RobotsEntry(now + 3600, parser, False, float(delay))
        except NotFoundError:
            entry = _RobotsEntry(now + 3600, None, False, 0.0)
        except (AccessDeniedError, RateLimitedError):
            # Access restriction: fail closed for this crawler run.
            entry = _RobotsEntry(now + 300, None, True, 0.0)
        except ContentTooLargeError:
            entry = _RobotsEntry(now + 300, None, True, 0.0)
        except NetworkError:
            # robots.txt being unreachable is not proof of permission. Be
            # conservative for this origin until the cache expires.
            entry = _RobotsEntry(now + 300, None, True, 0.0)
        except AppError:
            entry = _RobotsEntry(now + 300, None, True, 0.0)
        self._cache[origin] = entry
        return entry

    async def allowed(self, url: str) -> bool:
        if not self.enabled:
            return True
        entry = await self._entry(url)
        if entry.disallow_all:
            return False
        if entry.parser is None:
            return True
        return entry.parser.can_fetch("TelegramVideoBot", url)

    async def delay(self, url: str) -> float:
        if not self.enabled:
            return 0.0
        entry = await self._entry(url)
        return max(0.0, entry.delay)


class Crawler:
    def __init__(
        self,
        settings: Settings,
        guard: NetworkGuard,
        http: SafeHttpClient,
        browser: BrowserRenderer,
        extractor: Extractor,
        registry: AdapterRegistry,
        repo: Repository,
    ) -> None:
        self.settings = settings
        self.guard = guard
        self.http = http
        self.browser = browser
        self.extractor = extractor
        self.registry = registry
        self.repo = repo
        self.robots = RobotsPolicy(http, enabled=settings.respect_robots_txt)
        self._last_request_at: float = 0.0

    @staticmethod
    def _display(url: str) -> str:
        parts = urlsplit(url)
        return f"{parts.scheme}://{parts.hostname or ''}{parts.path[:120]}"

    @staticmethod
    def _direct_extension(url: str) -> str:
        suffix = PurePosixPath(urlsplit(url).path).suffix.lower()
        return suffix if suffix in _ALL_DIRECT_EXTENSIONS else ""

    async def crawl(
        self,
        job: Job,
        cancel_event: asyncio.Event,
        checkpoint: Checkpoint,
    ) -> CrawlResult:
        if cancel_event.is_set():
            raise DownloadCancelledError()
        await checkpoint()
        deadline = time.monotonic() + self.settings.max_job_seconds
        validated = await self.guard.validate(job.source_url, resolve_dns=True)
        adapter = self.registry.resolve(validated.url)
        existing_videos = await self.repo.list_videos(job.id)
        videos_found = 0
        duplicates = 0
        detail_pages = 0
        browser_renders = 0
        stop_reason = "exhausted"
        seen_pages: set[str] = {validated.url}
        seen_detail_pages: set[str] = set()
        notes: list[str] = []
        logger.info(
            "crawl started: %s (adapter=%s)",
            self._display(validated.url),
            adapter.name,
            extra={"job_id": job.id[:8]},
        )

        direct_extension = self._direct_extension(validated.url)
        if direct_extension:
            if not await self.robots.allowed(validated.url):
                raise RobotsDeniedError("robots.txt disallows the direct media URL")
            page, _ = await self.repo.add_page(job.id, validated.url)
            await self.repo.update_page(page.id, PageStatus.PROCESSED)
            source_path = unquote(PurePosixPath(urlsplit(validated.url).path).stem)
            title = normalize_text(source_path.replace("-", " ").replace("_", " ")) or "ویدئو"
            created = await self._store_video(
                job,
                page_url=validated.url,
                source_url=validated.url,
                media_url=validated.url,
                title=title,
                caption=title,
                media_type="video",
                source="direct_source_url",
            )
            await self.repo.refresh_job_counts(job.id)
            logger.info(
                "direct media source URL accepted without page crawl",
                extra={"job_id": job.id[:8]},
            )
            return CrawlResult(1, 0, 1 if created else 0, 0 if created else 1, "direct_media")

        current: str | None = validated.url
        pagination_pages = 0
        while current is not None:
            await checkpoint()
            if cancel_event.is_set():
                raise DownloadCancelledError()
            if time.monotonic() > deadline:
                stop_reason = "time_limit"
                notes.append("سقف زمان اجرای Job به پایان رسید.")
                break
            if pagination_pages >= self.settings.max_pages_per_job:
                stop_reason = "max_pages"
                notes.append("سقف تعداد صفحات بررسی شد.")
                break
            if len(existing_videos) + videos_found >= self.settings.max_videos_per_job:
                stop_reason = "max_videos"
                notes.append("سقف تعداد ویدئوها شناسایی شد.")
                break
            if stop_reason.startswith("access"):
                break
            if current in seen_pages and current != validated.url:
                stop_reason = "duplicate_page"
                break

            page_adapter = self.registry.resolve(current)
            try:
                page_result = await self._fetch_and_parse(
                    job,
                    page_url=current,
                    adapter=page_adapter,
                    cancel_event=cancel_event,
                    checkpoint=checkpoint,
                    allow_browser_fallback=True,
                    browser_renders=browser_renders,
                )
            except BrowserUnavailableError:
                raise
            except (AccessDeniedError, RateLimitedError, RobotsDeniedError, SsrfBlockedError) as exc:
                if pagination_pages == 0 and not videos_found:
                    raise
                logger.warning(
                    "site denied continuation; processing already discovered videos: %s",
                    type(exc).__name__,
                    extra={"job_id": job.id[:8]},
                )
                stop_reason = "access_denied"
                notes.append("سایت اجازه ادامه بررسی را نداد؛ فقط موارد کشف‌شده پردازش می‌شوند.")
                break
            except (NotFoundError, PermanentSiteError, NetworkError, ContentTooLargeError) as exc:
                if pagination_pages == 0:
                    raise
                logger.warning(
                    "pagination stopped after page error: %s",
                    type(exc).__name__,
                    extra={"job_id": job.id[:8]},
                )
                stop_reason = "permanent_page_error"
                notes.append("خطای صفحه بعد؛ پردازش با ویدئوهای کشف‌شده ادامه یافت.")
                break

            browser_renders += page_result.browser_renders
            parsed = page_result.parsed
            current = page_result.final_url
            seen_pages.add(current)
            pagination_pages += 1

            for candidate in parsed.videos:
                if len(existing_videos) + videos_found >= self.settings.max_videos_per_job:
                    stop_reason = "max_videos"
                    break
                created = await self._store_video(
                    job,
                    page_url=candidate.page_url,
                    source_url=candidate.source_url,
                    media_url=candidate.media_url,
                    title=candidate.title,
                    caption=candidate.caption or candidate.title,
                    media_type=candidate.media_type,
                    thumbnail_url=candidate.thumbnail_url,
                    duration_seconds=candidate.duration_seconds,
                    file_size=candidate.file_size,
                    source=candidate.source,
                )
                if created:
                    videos_found += 1
                else:
                    duplicates += 1
            await self.repo.refresh_job_counts(job.id)
            logger.info(
                "page processed: strategy=%s videos=%s next=%s",
                parsed.next_page_strategy,
                len(parsed.videos),
                bool(parsed.next_page_url),
                extra={"job_id": job.id[:8]},
            )

            # Explicit video-detail pages discovered by structured data or a
            # configured adapter. One level only, never recursive.
            if parsed.video_pages and len(existing_videos) + videos_found < self.settings.max_videos_per_job:
                for link in parsed.video_pages:
                    await checkpoint()
                    if cancel_event.is_set():
                        raise DownloadCancelledError()
                    if time.monotonic() > deadline:
                        stop_reason = "time_limit"
                        break
                    if detail_pages >= self.settings.max_videos_per_job:
                        break
                    if len(existing_videos) + videos_found >= self.settings.max_videos_per_job:
                        stop_reason = "max_videos"
                        break
                    if link.url in seen_detail_pages or link.url in seen_pages:
                        continue
                    seen_detail_pages.add(link.url)
                    try:
                        link_adapter = self.registry.resolve(link.url)
                        detail = await self._fetch_and_parse(
                            job,
                            page_url=link.url,
                            adapter=link_adapter,
                            cancel_event=cancel_event,
                            checkpoint=checkpoint,
                            allow_browser_fallback=True,
                            browser_renders=browser_renders,
                            link_title=link.title,
                            link_caption=link.caption,
                            link_thumbnail=link.thumbnail_url,
                        )
                        browser_renders += detail.browser_renders
                        detail_pages += 1
                        detail_parsed = detail.parsed
                        for candidate in detail_parsed.videos:
                            if len(existing_videos) + videos_found >= self.settings.max_videos_per_job:
                                break
                            created = await self._store_video(
                                job,
                                page_url=candidate.page_url,
                                source_url=candidate.source_url,
                                media_url=candidate.media_url,
                                title=candidate.title,
                                caption=candidate.caption or candidate.title,
                                media_type=candidate.media_type,
                                thumbnail_url=candidate.thumbnail_url,
                                duration_seconds=candidate.duration_seconds,
                                file_size=candidate.file_size,
                                source=candidate.source,
                            )
                            if created:
                                videos_found += 1
                            else:
                                duplicates += 1
                        await self.repo.refresh_job_counts(job.id)
                    except (AccessDeniedError, RateLimitedError, RobotsDeniedError):
                        logger.warning(
                            "site denied a video-detail page; stopping further details",
                            extra={"job_id": job.id[:8]},
                        )
                        stop_reason = "access_denied"
                        notes.append("دسترسی به یکی از صفحات ویدئو رد شد.")
                        break
                    except (NotFoundError, NetworkError, PermanentSiteError) as exc:
                        logger.warning(
                            "video-detail page failed: %s",
                            type(exc).__name__,
                            extra={"job_id": job.id[:8]},
                        )
                        notes.append("یک صفحه ویدئو قابل پردازش نبود.")
                    if await self._page_delay(current, cancel_event):
                        raise DownloadCancelledError()

            if stop_reason in {"max_videos", "time_limit", "access_denied"}:
                break
            if parsed.next_page_url:
                if parsed.next_page_url in seen_pages:
                    stop_reason = "duplicate_page"
                    break
                if await self._page_delay(current, cancel_event):
                    raise DownloadCancelledError()
                current = parsed.next_page_url
                await checkpoint()
            else:
                current = None
                stop_reason = "no_next_page"
                break

        await self.repo.refresh_job_counts(job.id)
        logger.info(
            "crawl finished: pages=%s details=%s videos=%s duplicates=%s reason=%s",
            pagination_pages,
            detail_pages,
            videos_found,
            duplicates,
            stop_reason,
            extra={"job_id": job.id[:8]},
        )
        if not await self.repo.list_videos(job.id):
            raise MediaNotFoundError("no videos were found within the configured limits")
        return CrawlResult(
            pagination_pages=pagination_pages,
            detail_pages=detail_pages,
            videos_found=videos_found,
            duplicate_videos=duplicates,
            stop_reason=stop_reason,
            notes=notes,
        )

    async def _fetch_and_parse(
        self,
        job: Job,
        *,
        page_url: str,
        adapter: SiteAdapter,
        cancel_event: asyncio.Event,
        checkpoint: Checkpoint,
        allow_browser_fallback: bool,
        browser_renders: int,
        link_title: str = "",
        link_caption: str = "",
        link_thumbnail: str | None = None,
    ) -> _FetchedParsed:
        if not await self.robots.allowed(page_url):
            raise RobotsDeniedError(f"robots.txt disallows {self._display(page_url)}")
        await checkpoint()
        page, _ = await self.repo.add_page(job.id, page_url)
        await self.repo.update_page(page.id, PageStatus.PROCESSING)
        try:
            final_url = page_url
            html = ""
            browser_used = 0
            if adapter.render_with_browser:
                rendered = await self.browser.render(page_url, adapter)
                final_url, html = rendered.final_url, rendered.html
                browser_used = 1
            else:
                fetched = await self._fetch_html_with_retry(page_url, cancel_event)
                final_url, html = fetched.final_url, fetched.html

            if final_url != page_url:
                await self.repo.update_page(
                    page.id,
                    PageStatus.SKIPPED,
                    error_message="redirected",
                )
                final_page, _ = await self.repo.add_page(job.id, final_url)
                page = final_page
                await self.repo.update_page(page.id, PageStatus.PROCESSING)
                if not await self.robots.allowed(final_url):
                    raise RobotsDeniedError("robots.txt disallows the redirect target")
            parsed = self._parse(adapter, html, final_url, link_title, link_caption, link_thumbnail)

            if (
                allow_browser_fallback
                and not adapter.render_with_browser
                and self.settings.browser_fallback_enabled
                and browser_renders < 10
                and not parsed.videos
                and not parsed.video_pages
            ):
                try:
                    await checkpoint()
                    rendered = await self.browser.render(final_url, adapter)
                    await self.guard.validate(rendered.final_url, resolve_dns=True)
                    parsed = self._parse(
                        adapter,
                        rendered.html,
                        rendered.final_url,
                        link_title,
                        link_caption,
                        link_thumbnail,
                    )
                    final_url = rendered.final_url
                    browser_used = 1
                    if final_url == page_url:
                        await self.repo.update_page(page.id, PageStatus.PROCESSED)
                    else:
                        await self.repo.update_page(
                            page.id, PageStatus.SKIPPED, error_message="browser redirect"
                        )
                        final_page, _ = await self.repo.add_page(job.id, final_url)
                        page = final_page
                        await self.repo.update_page(page.id, PageStatus.PROCESSED)
                except BrowserUnavailableError as exc:
                    logger.warning(
                        "browser fallback unavailable; keeping HTTP result: %s", exc
                    )
                except (AccessDeniedError, RateLimitedError, RobotsDeniedError):
                    raise
                except Exception as exc:  # noqa: BLE001 - fallback is best-effort
                    logger.warning(
                        "browser fallback failed: %s", type(exc).__name__
                    )
            else:
                await self.repo.update_page(page.id, PageStatus.PROCESSED)
            return _FetchedParsed(parsed=parsed, final_url=final_url, browser_renders=browser_used)
        except Exception as exc:
            await self.repo.update_page(
                page.id,
                PageStatus.FAILED,
                error_message=type(exc).__name__,
            )
            raise

    def _parse(
        self,
        adapter: SiteAdapter,
        html: str,
        url: str,
        link_title: str,
        link_caption: str,
        link_thumbnail: str | None,
    ) -> ParsedPage:
        parsed = self.extractor.extract(adapter, html, url)
        if link_title and (parsed.title in {"ویدئو", ""} or len(parsed.title) < 3):
            parsed.title = normalize_text(link_title)[:500]
        if link_caption and (not parsed.caption or parsed.caption == parsed.title):
            parsed.caption = normalize_text(link_caption)[:20_000] or parsed.caption
        if link_thumbnail:
            for candidate in parsed.videos:
                if not candidate.thumbnail_url:
                    candidate.thumbnail_url = link_thumbnail
        return parsed

    async def _fetch_html_with_retry(
        self, page_url: str, cancel_event: asyncio.Event
    ) -> FetchedPage:
        async def operation() -> FetchedPage:
            return await self.http.fetch_html(
                page_url, max_bytes=self.settings.max_html_bytes
            )

        return await retry_async(
            operation,
            attempts=self.settings.retry_count + 1,
            base_delay=1.5,
            max_delay=20,
            retry_on=(NetworkError, RateLimitedError),
            cancel_event=cancel_event,
        )

    async def _page_delay(self, url: str, cancel_event: asyncio.Event) -> bool:
        """Sleep between requests; return True if cancelled."""

        try:
            robots_delay = await self.robots.delay(url)
        except AppError:
            robots_delay = 0.0
        delay = max(self.settings.page_delay_seconds, robots_delay)
        if delay <= 0:
            return cancel_event.is_set()
        try:
            await asyncio.wait_for(cancel_event.wait(), timeout=delay)
        except TimeoutError:
            return False
        return True

    async def _store_video(
        self,
        job: Job,
        *,
        page_url: str,
        source_url: str,
        media_url: str | None,
        title: str,
        caption: str,
        media_type: str,
        thumbnail_url: str | None = None,
        duration_seconds: float | None = None,
        file_size: int | None = None,
        source: str = "generic",
    ) -> bool:
        video = Video.create(
            job_id=job.id,
            source_url=source_url,
            page_url=page_url,
            title=normalize_text(title)[:500] or "ویدئو",
            caption=normalize_text(caption)[:20_000] or normalize_text(title),
            media_type=media_type or "video/unknown",
            media_url=media_url,
            thumbnail_url=thumbnail_url,
            duration_seconds=duration_seconds,
            file_size=file_size,
        )
        stored, created = await self.repo.add_video(video)
        if not created:
            created = await self._revive_for_retry(stored, job)
        if created:
            logger.info(
                "video discovered (%s)",
                source,
                extra={"job_id": job.id[:8]},
            )
        return created

    async def _revive_for_retry(self, existing: Video, job: Job) -> bool:
        """Move a previously failed/cancelled video into this job for a retry.

        ``videos.source_url`` is globally unique, so a URL that once failed can
        never be re-added as a new row. Reusing the row keeps deduplication for
        published items while still allowing an operator to retry a failure.
        """

        if existing.job_id == job.id:
            return False
        if existing.status not in {VideoStatus.FAILED, VideoStatus.CANCELLED}:
            return False
        revived = await self.repo.update_video_fields(
            existing.id,
            job_id=job.id,
            status=VideoStatus.PENDING,
            error_message=None,
            local_path=None,
            telegram_message_id=None,
            uploaded_at=None,
            attempts=0,
        )
        if not revived:
            return False
        await self.repo.refresh_job_counts(existing.job_id)
        logger.info(
            "retrying a previously incomplete video in a new job",
            extra={"job_id": job.id[:8]},
        )
        return True


@dataclass(frozen=True, slots=True)
class _FetchedParsed:
    parsed: ParsedPage
    final_url: str
    browser_renders: int