src/modules/web_browser.py

"""
霜云(Shimokumo) - 网页浏览模块

提供网页内容获取、HTML解析、内容清洗和Markdown转换功能。
"""

import re
from dataclasses import dataclass, field
from typing import Dict, List, Optional, Tuple
from urllib.parse import urlparse, urljoin

from utils.logger import get_logger

logger = get_logger("Shimokumo.WebBrowser")


@dataclass
class WebPage:
    """网页内容数据类"""
    url: str
    """网页URL"""
    title: str = ""
    """页面标题"""
    content: str = ""
    """纯文本内容（清洗后）"""
    markdown: str = ""
    """Markdown格式内容"""
    html: str = ""
    """原始HTML"""
    meta_description: str = ""
    """页面描述"""
    meta_keywords: str = ""
    """页面关键词"""
    links: List[str] = field(default_factory=list)
    """页面内链接"""
    images: List[str] = field(default_factory=list)
    """页面图片URL"""
    status_code: int = 0
    """HTTP状态码"""

    def to_dict(self) -> Dict[str, any]:
        """转为字典格式"""
        return {
            "url": self.url,
            "title": self.title,
            "content": self.content[:2000],  # 限制长度避免过大
            "markdown": self.markdown[:2000],
            "meta_description": self.meta_description,
            "meta_keywords": self.meta_keywords,
            "links_count": len(self.links),
            "images_count": len(self.images),
            "status_code": self.status_code,
        }


class WebBrowserModule:
    """网页浏览模块

    使用requests + BeautifulSoup实现网页内容获取和解析。

    功能：
    - 网页内容获取（HTTP请求）
    - HTML解析和内容提取
    - 内容清洗（移除广告、导航等无关信息）
    - HTML转Markdown
    - 网页摘要生成
    - 链接和图片提取

    用法：
        browser = WebBrowserModule()
        page = browser.fetch("https://example.com")
        print(page.title)
        print(page.markdown)
    """

    def __init__(
        self,
        timeout: int = 15,
        max_content_length: int = 500000,
        user_agent: Optional[str] = None,
    ):
        """
        初始化网页浏览器模块。

        Args:
            timeout: 请求超时时间（秒）
            max_content_length: 最大内容长度（字节）
            user_agent: 自定义User-Agent
        """
        self.timeout = timeout
        self.max_content_length = max_content_length
        self.user_agent = user_agent or (
            "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
            "AppleWebKit/537.36 (KHTML, like Gecko) "
            "Chrome/120.0.0.0 Safari/537.36"
        )

        # 需要移除的HTML标签
        self.remove_tags = {
            "script", "style", "noscript", "iframe", "svg",
            "nav", "footer", "header", "aside",
        }

        # 需要保留的HTML标签（用于Markdown转换）
        self.preserve_tags = {
            "h1", "h2", "h3", "h4", "h5", "h6",
            "p", "br", "hr",
            "strong", "b", "em", "i", "u", "s",
            "a", "img",
            "ul", "ol", "li",
            "blockquote", "pre", "code",
            "table", "thead", "tbody", "tr", "th", "td",
        }

    def fetch(self, url: str, encoding: Optional[str] = None) -> Optional[WebPage]:
        """
        获取网页内容。

        Args:
            url: 网页URL
            encoding: 手动指定编码，为None则自动检测

        Returns:
            WebPage对象，失败返回None
        """
        logger.info(f"正在获取网页: {url}")

        try:
            import requests
            resp = requests.get(
                url,
                headers={
                    "User-Agent": self.user_agent,
                    "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
                    "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
                    "Accept-Encoding": "gzip, deflate",
                },
                timeout=self.timeout,
                allow_redirects=True,
            )
            resp.raise_for_status()

            # 检查内容长度
            if len(resp.content) > self.max_content_length:
                logger.warning(f"网页内容过大 ({len(resp.content)} bytes)，将截断处理")

            # 设置编码
            if encoding:
                resp.encoding = encoding
            elif resp.apparent_encoding:
                resp.encoding = resp.apparent_encoding

            # 解析网页
            page = self.parse_html(resp.text, url)
            page.status_code = resp.status_code

            logger.info(f"网页获取成功: {page.title} ({len(page.content)} 字符)")
            return page

        except Exception as e:
            logger.error(f"获取网页失败 [{url}]: {e}")
            return None

    def parse_html(self, html: str, base_url: str = "") -> WebPage:
        """
        解析HTML内容。

        Args:
            html: HTML字符串
            base_url: 基础URL（用于解析相对链接）

        Returns:
            WebPage对象
        """
        page = WebPage(url=base_url, html=html)

        try:
            from bs4 import BeautifulSoup
            soup = BeautifulSoup(html, "html.parser")

            # 提取标题
            title_tag = soup.find("title")
            page.title = title_tag.get_text(strip=True) if title_tag else ""

            # 提取元数据
            meta_desc = soup.find("meta", attrs={"name": "description"})
            page.meta_description = meta_desc.get("content", "") if meta_desc else ""

            meta_kw = soup.find("meta", attrs={"name": "keywords"})
            page.meta_keywords = meta_kw.get("content", "") if meta_kw else ""

            # 移除不需要的标签
            for tag_name in self.remove_tags:
                for tag in soup.find_all(tag_name):
                    tag.decompose()

            # 提取纯文本内容
            body = soup.find("body")
            if body:
                page.content = self._clean_text(body.get_text(separator="\n"))
            else:
                page.content = self._clean_text(soup.get_text(separator="\n"))

            # HTML转Markdown
            page.markdown = self._html_to_markdown(soup, base_url)

            # 提取链接
            page.links = self._extract_links(soup, base_url)

            # 提取图片
            page.images = self._extract_images(soup, base_url)

        except ImportError:
            logger.error("BeautifulSoup未安装，使用纯文本回退")
            page.content = self._clean_text(html)
            page.markdown = page.content

        return page

    def _clean_text(self, text: str) -> str:
        """
        清洗文本内容。

        Args:
            text: 原始文本

        Returns:
            清洗后的文本
        """
        # 解码HTML实体
        text = text.replace(" ", " ")
        text = text.replace("&lt;", "<")
        text = text.replace("&gt;", ">")
        text = text.replace("&amp;", "&")
        text = text.replace("&quot;", '"')
        text = text.replace("&#39;", "'")

        # 移除多余空白
        lines = text.split("\n")
        cleaned_lines: List[str] = []
        for line in lines:
            line = line.strip()
            if line:  # 跳过空行
                cleaned_lines.append(line)

        # 合并连续空行为单个空行
        result: List[str] = []
        prev_empty = False
        for line in cleaned_lines:
            if not line:
                if not prev_empty:
                    result.append("")
                prev_empty = True
            else:
                result.append(line)
                prev_empty = False

        return "\n".join(result).strip()

    def _html_to_markdown(self, soup: "BeautifulSoup", base_url: str) -> str:
        """
        将BeautifulSoup对象转为Markdown格式。

        Args:
            soup: BeautifulSoup对象
            base_url: 基础URL

        Returns:
            Markdown格式文本
        """
        markdown_parts: List[str] = []

        try:
            from bs4 import BeautifulSoup, NavigableString, Tag

            def process_element(element: "Tag", depth: int = 0) -> None:
                """递归处理HTML元素"""
                if isinstance(element, NavigableString):
                    text = str(element).strip()
                    if text:
                        markdown_parts.append(text)
                    return

                if not isinstance(element, Tag):
                    return

                tag_name = element.name.lower() if element.name else ""

                # 跳过不需要的标签
                if tag_name in self.remove_tags:
                    return

                # 处理标题
                if tag_name in ("h1", "h2", "h3", "h4", "h5", "h6"):
                    level = int(tag_name[1])
                    title_text = element.get_text(strip=True)
                    markdown_parts.append(f"\n{'#' * level} {title_text}\n")
                    return

                # 处理段落
                if tag_name == "p":
                    para_text = element.get_text(strip=True)
                    if para_text:
                        markdown_parts.append(f"\n{para_text}\n")
                    return

                # 处理链接
                if tag_name == "a":
                    link_text = element.get_text(strip=True)
                    href = element.get("href", "")
                    if href:
                        # 处理相对链接
                        if not href.startswith(("http", "//")):
                            href = urljoin(base_url, href)
                        markdown_parts.append(f"[{link_text}]({href})")
                    else:
                        markdown_parts.append(link_text)
                    return

                # 处理图片
                if tag_name == "img":
                    alt = element.get("alt", "")
                    src = element.get("src", "")
                    if src and not src.startswith(("http", "//")):
                        src = urljoin(base_url, src)
                    if src:
                        markdown_parts.append(f"![{alt}]({src})")
                    return

                # 处理加粗和斜体
                if tag_name in ("strong", "b"):
                    markdown_parts.append(f"**{element.get_text(strip=True)}**")
                    return
                if tag_name in ("em", "i"):
                    markdown_parts.append(f"*{element.get_text(strip=True)}*")
                    return

                # 处理列表
                if tag_name == "ul":
                    for li in element.find_all("li", recursive=False):
                        markdown_parts.append(f"- {li.get_text(strip=True)}")
                    return
                if tag_name == "ol":
                    for i, li in enumerate(element.find_all("li", recursive=False), 1):
                        markdown_parts.append(f"{i}. {li.get_text(strip=True)}")
                    return

                # 处理代码
                if tag_name == "pre":
                    code = element.get_text()
                    markdown_parts.append(f"\n```\n{code.strip()}\n```\n")
                    return
                if tag_name == "code":
                    markdown_parts.append(f"`{element.get_text(strip=True)}`")
                    return

                # 处理换行和水平线
                if tag_name == "br":
                    markdown_parts.append("\n")
                    return
                if tag_name == "hr":
                    markdown_parts.append("\n---\n")
                    return

                # 递归处理子元素
                for child in element.children:
                    process_element(child)

            # 从body开始处理
            body = soup.find("body") or soup
            process_element(body)

        except ImportError:
            # 如果BeautifulSoup不可用，使用简单回退
            markdown_parts.append(self._clean_text(soup.get_text(separator="\n") if hasattr(soup, "get_text") else str(soup)))

        return "\n".join(markdown_parts).strip()

    def _extract_links(self, soup: "BeautifulSoup", base_url: str) -> List[str]:
        """
        提取页面中的所有链接。

        Args:
            soup: BeautifulSoup对象
            base_url: 基础URL

        Returns:
            去重后的链接列表
        """
        links: List[str] = []
        seen: set = set()

        try:
            from bs4 import BeautifulSoup
            for a_tag in soup.find_all("a", href=True):
                href = a_tag.get("href", "")
                if not href or href.startswith(("#", "javascript:", "mailto:")):
                    continue

                # 处理相对链接
                if not href.startswith(("http", "//")):
                    href = urljoin(base_url, href)

                # 去重
                if href not in seen:
                    seen.add(href)
                    links.append(href)
        except ImportError:
            pass

        return links

    def _extract_images(self, soup: "BeautifulSoup", base_url: str) -> List[str]:
        """
        提取页面中的所有图片URL。

        Args:
            soup: BeautifulSoup对象
            base_url: 基础URL

        Returns:
            去重后的图片URL列表
        """
        images: List[str] = []
        seen: set = set()

        try:
            from bs4 import BeautifulSoup
            for img_tag in soup.find_all("img", src=True):
                src = img_tag.get("src", "")
                if not src or src.startswith("data:"):
                    continue

                # 处理相对URL
                if not src.startswith(("http", "//")):
                    src = urljoin(base_url, src)

                if src not in seen:
                    seen.add(src)
                    images.append(src)
        except ImportError:
            pass

        return images

    def generate_summary(self, page: WebPage, max_length: int = 300) -> str:
        """
        生成网页摘要。

        优先使用meta description，其次使用正文前N个字符。

        Args:
            page: WebPage对象
            max_length: 摘要最大长度

        Returns:
            网页摘要文本
        """
        # 优先使用meta描述
        if page.meta_description:
            summary = page.meta_description
            if len(summary) <= max_length:
                return summary

        # 使用正文内容
        content = page.content or ""
        if not content:
            return "无法获取网页内容的说..."

        # 提取正文前几段
        paragraphs = [p.strip() for p in content.split("\n") if p.strip()]
        summary_parts: List[str] = []
        total_length = 0

        for para in paragraphs:
            if total_length + len(para) > max_length:
                remaining = max_length - total_length
                if remaining > 20:
                    summary_parts.append(para[:remaining] + "...")
                break
            summary_parts.append(para)
            total_length += len(para)

        return "\n".join(summary_parts)

    def fetch_multiple(self, urls: List[str]) -> List[WebPage]:
        """
        批量获取多个网页。

        Args:
            urls: URL列表

        Returns:
            WebPage列表（失败的为None）
        """
        results: List[WebPage] = []
        for url in urls:
            page = self.fetch(url)
            results.append(page)  # type: ignore
        return results