356 lines
10 KiB
Python
356 lines
10 KiB
Python
"""
|
||
文档内容获取服务（轻量版）
|
||
|
||
只做一件事：从 URL/GitHub 获取文档内容，转换成 Markdown reference 文件。
|
||
不涉及复杂的爬取、代码分析、脚本执行等功能。
|
||
|
||
适用于：
|
||
- 用户输入文档 URL，自动获取内容作为 references
|
||
- 简单的文档解析和清理
|
||
- 与现有 LLM 生成流程配合
|
||
"""
|
||
|
||
import asyncio
|
||
import re
|
||
from typing import Optional, List, Dict, Any, TYPE_CHECKING
|
||
from pathlib import Path
|
||
import logging
|
||
|
||
try:
|
||
import httpx
|
||
from bs4 import BeautifulSoup
|
||
HTTPX_AVAILABLE = True
|
||
except ImportError:
|
||
HTTPX_AVAILABLE = False
|
||
# TYPE_CHECKING is always False at runtime, so BeautifulSoup won't be imported
|
||
if TYPE_CHECKING:
|
||
from bs4 import BeautifulSoup
|
||
|
||
from app.utils.logger import get_logger
|
||
|
||
logger = get_logger(__name__)
|
||
|
||
|
||
class DocumentationFetcher:
|
||
"""
|
||
轻量级文档获取器
|
||
|
||
功能：
|
||
1. 从 URL 获取网页内容
|
||
2. 提取主要内容（去除导航、广告等）
|
||
3. 转换为 Markdown 格式
|
||
4. 清理和格式化
|
||
|
||
不做：
|
||
- 复杂的爬虫（不递归抓取）
|
||
- 代码分析
|
||
- 脚本执行
|
||
"""
|
||
|
||
def __init__(self, timeout: int = 30):
|
||
"""
|
||
初始化获取器
|
||
|
||
Args:
|
||
timeout: 请求超时时间（秒）
|
||
"""
|
||
self.timeout = timeout
|
||
|
||
if not HTTPX_AVAILABLE:
|
||
logger.warning("httpx 或 beautifulsoup4 未安装，部分功能不可用")
|
||
logger.warning("请安装: pip install httpx beautifulsoup4")
|
||
|
||
async def fetch_from_url(
|
||
self,
|
||
url: str,
|
||
selector: Optional[str] = None
|
||
) -> Dict[str, Any]:
|
||
"""
|
||
从 URL 获取文档内容
|
||
|
||
Args:
|
||
url: 文档 URL
|
||
selector: CSS 选择器（可选，用于定位主内容）
|
||
|
||
Returns:
|
||
包含获取结果的字典
|
||
"""
|
||
if not HTTPX_AVAILABLE:
|
||
return {
|
||
"success": False,
|
||
"error": "httpx 未安装，请运行: pip install httpx"
|
||
}
|
||
|
||
logger.info(f"获取文档: {url}")
|
||
|
||
try:
|
||
async with httpx.AsyncClient(timeout=self.timeout) as client:
|
||
response = await client.get(url, follow_redirects=True)
|
||
response.raise_for_status()
|
||
|
||
# 解析 HTML
|
||
soup = BeautifulSoup(response.text, "html.parser")
|
||
|
||
# 提取内容
|
||
title = self._extract_title(soup)
|
||
content = self._extract_main_content(soup, selector)
|
||
|
||
# 转换为 Markdown
|
||
markdown = self._html_to_markdown(title, content, url)
|
||
|
||
# 清理
|
||
markdown = self._clean_markdown(markdown)
|
||
|
||
return {
|
||
"success": True,
|
||
"url": url,
|
||
"title": title,
|
||
"content": markdown,
|
||
"word_count": len(markdown.split()),
|
||
"char_count": len(markdown)
|
||
}
|
||
|
||
except httpx.TimeoutException:
|
||
return {
|
||
"success": False,
|
||
"error": f"请求超时: {url}"
|
||
}
|
||
except httpx.HTTPStatusError as e:
|
||
return {
|
||
"success": False,
|
||
"error": f"HTTP 错误: {e.response.status_code}"
|
||
}
|
||
except Exception as e:
|
||
logger.error(f"获取文档失败: {str(e)}")
|
||
return {
|
||
"success": False,
|
||
"error": str(e)
|
||
}
|
||
|
||
async def fetch_from_github(
|
||
self,
|
||
repo_url: str,
|
||
docs_path: str = "README.md"
|
||
) -> Dict[str, Any]:
|
||
"""
|
||
从 GitHub 获取文档
|
||
|
||
Args:
|
||
repo_url: 仓库 URL (如 https://github.com/owner/repo)
|
||
docs_path: 文档路径 (如 README.md, docs/README.md)
|
||
|
||
Returns:
|
||
包含获取结果的字典
|
||
"""
|
||
if not HTTPX_AVAILABLE:
|
||
return {
|
||
"success": False,
|
||
"error": "httpx 未安装"
|
||
}
|
||
|
||
logger.info(f"获取 GitHub 文档: {repo_url}/{docs_path}")
|
||
|
||
try:
|
||
# 解析 repo 信息
|
||
match = re.match(r'https://github\.com/([^/]+)/([^/]+)', repo_url)
|
||
if not match:
|
||
return {
|
||
"success": False,
|
||
"error": "无效的 GitHub URL"
|
||
}
|
||
|
||
owner, repo = match.groups()
|
||
|
||
# 使用 raw.githubusercontent.com 获取文件
|
||
raw_url = f"https://raw.githubusercontent.com/{owner}/{repo}/main/{docs_path}"
|
||
|
||
async with httpx.AsyncClient(timeout=self.timeout) as client:
|
||
response = await client.get(raw_url, follow_redirects=True)
|
||
|
||
# 如果 main 分支不存在，尝试 master
|
||
if response.status_code == 404:
|
||
raw_url = raw_url.replace("/main/", "/master/")
|
||
response = await client.get(raw_url, follow_redirects=True)
|
||
|
||
response.raise_for_status()
|
||
|
||
content = response.text
|
||
|
||
return {
|
||
"success": True,
|
||
"repo_url": repo_url,
|
||
"docs_path": docs_path,
|
||
"content": content,
|
||
"word_count": len(content.split()),
|
||
"char_count": len(content)
|
||
}
|
||
|
||
except Exception as e:
|
||
logger.error(f"获取 GitHub 文档失败: {str(e)}")
|
||
return {
|
||
"success": False,
|
||
"error": str(e)
|
||
}
|
||
|
||
def _extract_title(self, soup: Any) -> str:
|
||
"""提取页面标题"""
|
||
# 尝试从 h1 获取
|
||
h1 = soup.find("h1")
|
||
if h1:
|
||
return h1.get_text().strip()
|
||
|
||
# 尝试从 title 标签获取
|
||
title_tag = soup.find("title")
|
||
if title_tag:
|
||
return title_tag.get_text().strip()
|
||
|
||
return "Untitled"
|
||
|
||
def _extract_main_content(
|
||
self,
|
||
soup: Any,
|
||
selector: Optional[str]
|
||
) -> Any:
|
||
"""提取主要内容"""
|
||
|
||
# 如果指定了选择器，使用它
|
||
if selector:
|
||
main = soup.select_one(selector)
|
||
if main:
|
||
return main
|
||
|
||
# 尝试常见的内容区域选择器
|
||
content_selectors = [
|
||
"article",
|
||
"main",
|
||
'[role="main"]',
|
||
".content",
|
||
"#content",
|
||
".documentation",
|
||
".docs-content",
|
||
"main .content"
|
||
]
|
||
|
||
for sel in content_selectors:
|
||
main = soup.select_one(sel)
|
||
if main:
|
||
return main
|
||
|
||
# 如果都找不到，返回 body
|
||
body = soup.find("body")
|
||
return body if body else soup
|
||
|
||
def _html_to_markdown(
|
||
self,
|
||
title: str,
|
||
content: Any,
|
||
url: str
|
||
) -> str:
|
||
"""将 HTML 转换为 Markdown"""
|
||
|
||
lines = [f"# {title}\n"]
|
||
lines.append(f"Source: {url}\n")
|
||
|
||
# 提取标题
|
||
for i, heading in enumerate(content.find_all(["h1", "h2", "h3", "h4", "h5", "h6"])):
|
||
if heading.name == "h1":
|
||
continue # 跳过第一个 h1（已经用作标题）
|
||
level = int(heading.name[1])
|
||
text = heading.get_text().strip()
|
||
lines.append(f"\n{'#' * level} {text}\n")
|
||
|
||
# 提取段落
|
||
paragraphs = []
|
||
for p in content.find_all("p"):
|
||
text = p.get_text().strip()
|
||
if len(text) > 20: # 只保留有意义的段落
|
||
paragraphs.append(text)
|
||
|
||
if paragraphs:
|
||
lines.append("\n## Content\n")
|
||
lines.extend(paragraphs)
|
||
|
||
# 提取代码块
|
||
code_blocks = []
|
||
for pre in content.find_all("pre"):
|
||
code = pre.get_text()
|
||
if len(code) > 10:
|
||
code_blocks.append(f"```\n{code}\n```")
|
||
|
||
if code_blocks:
|
||
lines.append("\n## Code Examples\n")
|
||
lines.extend(code_blocks[:5]) # 最多 5 个代码块
|
||
|
||
# 提取列表
|
||
for ul in content.find_all(["ul", "ol"]):
|
||
items = [f"- {li.get_text().strip()}" for li in ul.find_all("li")]
|
||
if items:
|
||
lines.append("\n")
|
||
lines.extend(items)
|
||
|
||
return "\n".join(lines)
|
||
|
||
def _clean_markdown(self, markdown: str) -> str:
|
||
"""清理 Markdown 内容"""
|
||
|
||
# 移除过多的空行
|
||
markdown = re.sub(r'\n{3,}', '\n\n', markdown)
|
||
|
||
# 移除导航类文本
|
||
noise_patterns = [
|
||
r'Table of Contents.*?(?=\n##)',
|
||
r'Navigation.*?(?=\n##)',
|
||
r'Menu.*?(?=\n##)',
|
||
r'Skip to content',
|
||
r'© \d{4}.*',
|
||
]
|
||
|
||
for pattern in noise_patterns:
|
||
markdown = re.sub(pattern, '', markdown, flags=re.IGNORECASE | re.DOTALL)
|
||
|
||
return markdown.strip()
|
||
|
||
async def fetch_multiple_urls(
|
||
self,
|
||
urls: List[str],
|
||
selector: Optional[str] = None
|
||
) -> List[Dict[str, Any]]:
|
||
"""
|
||
批量获取多个 URL 的内容
|
||
|
||
Args:
|
||
urls: URL 列表
|
||
selector: CSS 选择器
|
||
|
||
Returns:
|
||
获取结果列表
|
||
"""
|
||
tasks = [self.fetch_from_url(url, selector) for url in urls]
|
||
results = await asyncio.gather(*tasks, return_exceptions=True)
|
||
|
||
# 处理异常
|
||
processed_results = []
|
||
for i, result in enumerate(results):
|
||
if isinstance(result, Exception):
|
||
processed_results.append({
|
||
"success": False,
|
||
"url": urls[i],
|
||
"error": str(result)
|
||
})
|
||
else:
|
||
processed_results.append(result)
|
||
|
||
return processed_results
|
||
|
||
|
||
# 全局单例
|
||
_doc_fetcher: Optional[DocumentationFetcher] = None
|
||
|
||
|
||
def get_documentation_fetcher() -> DocumentationFetcher:
|
||
"""获取文档获取器单例"""
|
||
global _doc_fetcher
|
||
if _doc_fetcher is None:
|
||
_doc_fetcher = DocumentationFetcher()
|
||
return _doc_fetcher
|