from __future__ import annotations

import asyncio
import ipaddress
import json
import socket
from collections.abc import Sequence
from typing import Any
from urllib.parse import urlsplit

from micro_agents import Tool


class WebPageTools:
    def __init__(self, *, timeout_seconds: float = 45, max_chars: int = 60_000):
        if timeout_seconds <= 0:
            raise ValueError("timeout_seconds must be positive")
        if max_chars < 1:
            raise ValueError("max_chars must be positive")
        self._timeout_seconds = timeout_seconds
        self._max_chars = max_chars

    def tools(self) -> list[Tool]:
        return [Tool(self.read_web_page, name="read_web_page", strict=False)]

    async def read_web_page(self, url: str) -> str:
        """Read a public web page in a Playwright browser and return cleaned Markdown.

        Use this before analyzing, summarizing, or quoting a URL. Unlike web_search,
        this loads the page itself and can render JavaScript-driven content.
        """
        safe_url = await _ensure_public_url(url)
        try:
            result = await asyncio.wait_for(
                _crawl_with_playwright(safe_url, self._timeout_seconds),
                timeout=self._timeout_seconds + 5,
            )
        except TimeoutError as exc:
            raise RuntimeError("Web page crawl timed out") from exc

        final_url = str(getattr(result, "redirected_url", None) or getattr(result, "url", safe_url))
        await _ensure_public_url(final_url)
        if not getattr(result, "success", False):
            error = str(getattr(result, "error_message", None) or "unknown crawl error")
            raise RuntimeError(f"Could not read web page: {error}")

        markdown_result = getattr(result, "markdown", None)
        markdown = str(markdown_result or "").strip()
        if not markdown:
            raise RuntimeError("Web page did not contain readable content")

        truncated = len(markdown) > self._max_chars
        if truncated:
            markdown = markdown[: self._max_chars].rstrip()

        metadata = getattr(result, "metadata", None)
        title = metadata.get("title") if isinstance(metadata, dict) else None
        return json.dumps(
            {
                "url": final_url,
                "title": title,
                "status_code": getattr(result, "status_code", None),
                "truncated": truncated,
                "markdown": markdown,
            },
            ensure_ascii=False,
        )


async def _crawl_with_playwright(url: str, timeout_seconds: float) -> Any:
    from crawl4ai import AsyncWebCrawler, BrowserConfig, CacheMode, CrawlerRunConfig

    browser_config = BrowserConfig(
        browser_type="chromium",
        headless=True,
        text_mode=True,
        light_mode=True,
        enable_stealth=True,
        verbose=False,
    )
    run_config = CrawlerRunConfig(
        cache_mode=CacheMode.BYPASS,
        page_timeout=max(1, int(timeout_seconds * 1000)),
        wait_until="domcontentloaded",
        remove_overlay_elements=True,
        remove_consent_popups=True,
        exclude_social_media_links=True,
        process_iframes=False,
        scan_full_page=False,
        verbose=False,
        log_console=False,
    )
    async with AsyncWebCrawler(config=browser_config) as crawler:
        return await crawler.arun(url=url, config=run_config)


async def _ensure_public_url(value: str) -> str:
    url = value.strip()
    if not url or len(url) > 2048:
        raise ValueError("url must contain at most 2048 characters")
    parsed = urlsplit(url)
    if parsed.scheme not in {"http", "https"}:
        raise ValueError("Only http and https URLs are allowed")
    if parsed.username is not None or parsed.password is not None:
        raise ValueError("Credentials in URLs are not allowed")
    hostname = parsed.hostname
    if not hostname:
        raise ValueError("URL must include a hostname")

    try:
        addresses = [ipaddress.ip_address(hostname)]
    except ValueError:
        addresses = await asyncio.to_thread(_resolve_addresses, hostname, parsed.port)
    if not addresses or any(not address.is_global for address in addresses):
        raise ValueError("Local and private network URLs are not allowed")
    return url


def _resolve_addresses(hostname: str, port: int | None) -> Sequence[ipaddress.IPv4Address | ipaddress.IPv6Address]:
    try:
        records = socket.getaddrinfo(hostname, port or 443, type=socket.SOCK_STREAM)
    except socket.gaierror as exc:
        raise ValueError("URL hostname could not be resolved") from exc
    return tuple({ipaddress.ip_address(record[4][0]) for record in records})