Download src/modules/web_browser.py from Shimokumo/shimokumo: direct link, hf CLI and curl.
- Browser
- Download file 15.9 kB
-
https://huggingface.co/Shimokumo/shimokumo/resolve/main/src/modules/web_browser.py
- Command line
-
hf download hf://Shimokumo/shimokumo/src/modules/web_browser.py
-
curl -L -o web_browser.py https://huggingface.co/Shimokumo/shimokumo/resolve/main/src/modules/web_browser.py
15.9 kB
| """ | |
| 霜云(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") | |
| 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"") | |
| 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 | |