File size: 7,369 Bytes
494c9e4 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 | """URL 文本提取 API"""
import json
import re
from urllib.parse import urlparse
import trafilatura
import requests
from backend.api.utils import handle_api_error
# 单次提取的最大字符数上限（防止异常大页面影响性能）
MAX_EXTRACTED_TEXT_LENGTH = 20000
def _is_valid_url(url: str) -> bool:
"""验证 URL 格式"""
try:
result = urlparse(url)
return all([result.scheme in ['http', 'https'], result.netloc])
except Exception:
return False
def _is_local_or_private(url: str) -> bool:
"""检查是否为本地或私有网络地址（防止 SSRF 攻击）"""
try:
parsed = urlparse(url)
hostname = parsed.hostname
if not hostname:
return True
# 检查是否为 localhost
if hostname in ['localhost', '127.0.0.1', '::1']:
return True
# 检查是否为私有 IP 地址
private_patterns = [
r'^10\.', # 10.0.0.0/8
r'^172\.(1[6-9]|2[0-9]|3[0-1])\.', # 172.16.0.0/12
r'^192\.168\.', # 192.168.0.0/16
r'^169\.254\.', # 169.254.0.0/16 (link-local)
]
for pattern in private_patterns:
if re.match(pattern, hostname):
return True
return False
except Exception:
return True # 解析失败时保守处理，拒绝访问
def _format_article_text(metadata: dict) -> str:
"""
将元数据和正文格式化为类似网页显示的纯文本
Args:
metadata: trafilatura 提取的 JSON 数据（已解析为字典）
Returns:
格式化后的文章文本
"""
lines = []
# 标题
if metadata.get('title'):
lines.append(metadata['title'])
lines.append('')
# 元数据信息（无标签，直接显示内容）
meta_parts = []
if metadata.get('author'):
meta_parts.append(metadata['author'])
if metadata.get('date'):
meta_parts.append(metadata['date'])
# if metadata.get('hostname'):
# meta_parts.append(metadata['hostname'])
if metadata.get('source-hostname'):
meta_parts.append(metadata['source-hostname'])
# if metadata.get('filedate'):
# meta_parts.append(metadata['filedate'])
if meta_parts:
lines.append(' | '.join(meta_parts))
lines.append('')
# 正文
if metadata.get('text'):
lines.append(metadata['text'])
return '\n'.join(lines)
def fetch_url(fetch_request):
"""
从 URL 提取文本内容
Args:
fetch_request: 包含 url 字段的字典
Returns:
(响应字典, 状态码) 元组
"""
url = fetch_request.get('url', '').strip()
# 验证 URL
if not url:
return {
'success': False,
'message': '缺少 URL 参数，请提供 url 字段'
}, 400
if not _is_valid_url(url):
return {
'success': False,
'message': f'无效的 URL 格式: {url}'
}, 400
# 安全检查：防止 SSRF 攻击
if _is_local_or_private(url):
return {
'success': False,
'message': '不允许访问本地或私有网络地址'
}, 400
# 提取文本和元数据
try:
from backend.access_log import log_fetch_url
log_fetch_url(url)
# 使用 requests 下载网页，设置浏览器 User-Agent 和请求头
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8',
'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8',
'Accept-Encoding': 'gzip, deflate, br',
'Connection': 'keep-alive',
'Upgrade-Insecure-Requests': '1',
}
# 下载网页内容（设置超时和请求头）
response = requests.get(url, headers=headers, timeout=10, allow_redirects=True)
response.raise_for_status()
# 检查响应内容类型
content_type = response.headers.get('Content-Type', '').lower()
if 'text/html' not in content_type and 'text/xml' not in content_type:
return {
'success': False,
'message': f'不支持的内容类型: {content_type}，仅支持 HTML/XML 页面'
}, 400
# 使用 trafilatura 提取结构化数据（包含元数据和正文）
result_json = trafilatura.extract(
response.text,
url=url,
with_metadata=True,
output_format='json'
)
if not result_json:
print("⚠️ 无法提取页面内容")
return {
'success': False,
'message': '无法从网页中提取文本内容，可能不是文章页面或页面需要验证'
}, 400
# 解析 JSON 数据
metadata = json.loads(result_json)
# 检查是否有正文内容
if not metadata.get('text') or not metadata['text'].strip():
print("⚠️ 提取到元数据但无正文内容")
print("元数据:", json.dumps(metadata, ensure_ascii=False, indent=2))
return {
'success': False,
'message': '无法从网页中提取正文内容'
}, 400
# 格式化文本（元数据 + 正文）
formatted_text = _format_article_text(metadata)
original_char_count = len(formatted_text)
# 构建返回消息（如果截断了，添加提示）
message = None
# 检查并截断超长文本
if original_char_count > MAX_EXTRACTED_TEXT_LENGTH:
formatted_text = formatted_text[:MAX_EXTRACTED_TEXT_LENGTH]
message = f'内容较长，已截断为前 {MAX_EXTRACTED_TEXT_LENGTH} 字符（原始长度: {original_char_count} 字符）'
char_count = len(formatted_text)
# 打印提取结果
# print(formatted_text.split('\n')[:4])
# print(f"✓ 提取成功: {char_count} 字符" + (f" (截断前: {original_char_count} 字符)" if original_char_count > char_count else ""))
# 打印除正文外的metadata内容
metadata_less = metadata.copy()
metadata_less['raw_text'] = ''
metadata_less['text'] = ''
# print(json.dumps(metadata_less, ensure_ascii=False, indent=2))
return {
'success': True,
'text': formatted_text,
'url': url,
'char_count': char_count,
'message': message
}, 200
except requests.exceptions.Timeout:
return {
'success': False,
'message': '请求超时，请检查网络连接或稍后重试'
}, 400
except requests.exceptions.RequestException as e:
return {
'success': False,
'message': f'无法访问 URL: {str(e)}'
}, 400
except Exception as e: # noqa: BLE001
error_response = handle_api_error('URL 文本提取失败', e)
return error_response, 500
|