""" 霜云(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("<", "<") text = text.replace(">", ">") text = text.replace("&", "&") text = text.replace(""", '"') text = text.replace("'", "'") # 移除多余空白 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