347 lines
12 KiB
Python
347 lines
12 KiB
Python
"""
|
||
网页抓取服务
|
||
|
||
抓取策略（三级降级）：
|
||
1. crawl4ai stealth 模式 — 应对 JS 渲染 / 反爬检测
|
||
2. crawl4ai 普通模式 — stealth 失败时回退
|
||
3. httpx + trafilatura — Playwright 无法导航时（JS 跳转、协议错误等）的最终兜底
|
||
|
||
使用方式：
|
||
from services.web_crawler_service import fetch_url_content, CrawlResult
|
||
result = await fetch_url_content("https://example.com")
|
||
if result.success:
|
||
text = result.markdown
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import os
|
||
import re
|
||
from dataclasses import dataclass
|
||
from typing import Optional
|
||
|
||
from core.config import settings
|
||
from logger.logging import get_logger
|
||
|
||
logger = get_logger(__name__)
|
||
|
||
# 限制并发 Chromium 进程数，避免多用户同时提交 URL 时资源耗尽
|
||
# 每个 AsyncWebCrawler 实例对应一个 Chromium 子进程
|
||
_PLAYWRIGHT_SEMAPHORE = asyncio.Semaphore(3)
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 拦截页面关键词列表（命中即视为被反爬拦截）
|
||
# ---------------------------------------------------------------------------
|
||
_BLOCK_PATTERNS: list[re.Pattern] = [
|
||
re.compile(p, re.IGNORECASE)
|
||
for p in [
|
||
r"环境异常",
|
||
r"完成验证后即可继续访问",
|
||
r"百度安全验证",
|
||
r"robot.*check",
|
||
r"security check",
|
||
r"access denied",
|
||
r"403 forbidden",
|
||
r"captcha",
|
||
r"请输入验证码",
|
||
r"人机验证",
|
||
r"滑动验证",
|
||
r"点击验证",
|
||
r"请证明.*不是机器人",
|
||
]
|
||
]
|
||
|
||
# 内容最短有效长度（字符数）
|
||
_MIN_CONTENT_LENGTH = 50
|
||
|
||
# 通用浏览器 UA
|
||
_CHROME_UA = (
|
||
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
|
||
"AppleWebKit/537.36 (KHTML, like Gecko) "
|
||
"Chrome/124.0.0.0 Safari/537.36"
|
||
)
|
||
|
||
|
||
@dataclass
|
||
class CrawlResult:
|
||
"""网页抓取结果"""
|
||
success: bool
|
||
markdown: str = ""
|
||
error_message: str = ""
|
||
is_blocked: bool = False
|
||
url: str = ""
|
||
title: str = ""
|
||
|
||
|
||
def _is_blocked_content(text: str) -> bool:
|
||
"""判断内容是否为反爬拦截页面。
|
||
|
||
只扫描前 500 字符：拦截页面的特征词（验证码提示、安全验证等）通常出现在
|
||
页面最开头，扫全文会增加不必要的计算开销。
|
||
已知权衡：若 WAF 在前 500 字符放置正常内容、拦截词出现在后面，会漏判；
|
||
实际观察中此场景极少见，当前策略是合理的工程取舍。
|
||
"""
|
||
if not text:
|
||
return False
|
||
return any(p.search(text[:500]) for p in _BLOCK_PATTERNS)
|
||
|
||
|
||
def _validate(markdown: str, url: str, stealth: bool) -> CrawlResult | None:
|
||
"""校验内容有效性，返回失败的 CrawlResult；内容有效时返回 None。"""
|
||
text = markdown.strip()
|
||
if len(text) < _MIN_CONTENT_LENGTH:
|
||
return CrawlResult(
|
||
success=False, url=url,
|
||
error_message=f"页面内容过短（{len(text)} 字符），可能是空白页或加载失败",
|
||
)
|
||
if _is_blocked_content(text):
|
||
mode_label = "stealth" if stealth else "普通"
|
||
return CrawlResult(
|
||
success=False, url=url, is_blocked=True,
|
||
error_message=(
|
||
f"该网页拒绝了自动抓取请求（{mode_label}模式）。"
|
||
"常见原因：微信公众号、百度百科等平台对非浏览器访问设有验证墙，"
|
||
"暂时无法自动获取内容。建议手动复制文本后通过「上传文本文件」导入知识库。"
|
||
),
|
||
)
|
||
return None
|
||
|
||
|
||
async def fetch_url_content(
|
||
url: str,
|
||
*,
|
||
stealth: Optional[bool] = None,
|
||
timeout_ms: Optional[int] = None,
|
||
wait_for_js_ms: Optional[int] = None,
|
||
) -> CrawlResult:
|
||
"""
|
||
抓取指定 URL 的正文内容，返回干净的 Markdown 文本。
|
||
|
||
三级降级策略：
|
||
1. crawl4ai stealth 模式
|
||
2. crawl4ai 普通模式
|
||
3. httpx 直接 HTTP 抓取 + trafilatura 正文提取
|
||
"""
|
||
stealth = settings.crawler_stealth_enabled if stealth is None else stealth
|
||
timeout_ms = timeout_ms or settings.crawler_timeout_ms
|
||
wait_for_js_ms = wait_for_js_ms or settings.crawler_wait_for_js_ms
|
||
|
||
# --- 级别 1：crawl4ai（stealth 由配置决定）---
|
||
ran_stealth = stealth # 记录本次 level 1 实际运行的模式
|
||
result = await _crawl_with_playwright(
|
||
url, stealth=stealth, timeout_ms=timeout_ms, wait_for_js_ms=wait_for_js_ms
|
||
)
|
||
if result.success:
|
||
return result
|
||
|
||
# --- 级别 2：crawl4ai 普通模式（仅当 level 1 以 stealth 运行但失败时才有意义）---
|
||
if ran_stealth:
|
||
logger.info(f"[WebCrawler] stealth 模式失败，尝试普通模式: {url}")
|
||
result = await _crawl_with_playwright(
|
||
url, stealth=False, timeout_ms=timeout_ms, wait_for_js_ms=wait_for_js_ms
|
||
)
|
||
if result.success:
|
||
return result
|
||
|
||
# --- 级别 3：httpx + trafilatura 兜底 ---
|
||
logger.info(f"[WebCrawler] Playwright 全部失败，尝试 httpx 兜底: {url}")
|
||
return await _crawl_with_httpx(url, timeout_ms=timeout_ms)
|
||
|
||
|
||
async def _crawl_with_playwright(
|
||
url: str,
|
||
*,
|
||
stealth: bool,
|
||
timeout_ms: int,
|
||
wait_for_js_ms: int,
|
||
) -> CrawlResult:
|
||
"""用 crawl4ai (Playwright) 抓取。"""
|
||
try:
|
||
from crawl4ai import AsyncWebCrawler, BrowserConfig, CrawlerRunConfig
|
||
from crawl4ai.markdown_generation_strategy import DefaultMarkdownGenerator
|
||
except ImportError:
|
||
msg = "crawl4ai 未安装，请执行 `uv add crawl4ai` 并运行 `crawl4ai-setup` 安装浏览器"
|
||
logger.error(f"[WebCrawler] {msg}")
|
||
return CrawlResult(success=False, url=url, error_message=msg)
|
||
|
||
# 透传系统代理给 Chromium 子进程（用新版 proxy_config）
|
||
_sys_proxy = (
|
||
os.environ.get("https_proxy") or os.environ.get("HTTPS_PROXY")
|
||
or os.environ.get("http_proxy") or os.environ.get("HTTP_PROXY")
|
||
) or None
|
||
|
||
from crawl4ai.async_configs import ProxyConfig
|
||
proxy_cfg = ProxyConfig(server=_sys_proxy) if _sys_proxy else None
|
||
|
||
browser_cfg = BrowserConfig(
|
||
headless=True,
|
||
enable_stealth=stealth,
|
||
proxy_config=proxy_cfg,
|
||
user_agent=_CHROME_UA,
|
||
)
|
||
|
||
# 统一等待 JS 渲染，不依赖页面特定 selector
|
||
# delay_before_return_html 在 domcontentloaded 后额外等待，保证动态内容渲染完成
|
||
run_cfg = CrawlerRunConfig(
|
||
markdown_generator=DefaultMarkdownGenerator(
|
||
options={"ignore_links": False, "body_width": 0},
|
||
),
|
||
page_timeout=timeout_ms,
|
||
wait_until="domcontentloaded",
|
||
delay_before_return_html=wait_for_js_ms / 1000,
|
||
word_count_threshold=5,
|
||
remove_overlay_elements=True,
|
||
excluded_tags=["nav", "footer", "aside", "script", "style"],
|
||
)
|
||
|
||
try:
|
||
async with _PLAYWRIGHT_SEMAPHORE:
|
||
async with AsyncWebCrawler(config=browser_cfg) as crawler:
|
||
crawl_res = await crawler.arun(url=url, config=run_cfg)
|
||
|
||
if not crawl_res.success:
|
||
err = getattr(crawl_res, "error_message", "页面加载失败")
|
||
logger.warning(f"[WebCrawler] Playwright 抓取失败 (stealth={stealth}): {url} | {err}")
|
||
return CrawlResult(success=False, url=url, error_message=err)
|
||
|
||
fit = (crawl_res.markdown.fit_markdown or "").strip() if crawl_res.markdown else ""
|
||
raw = (crawl_res.markdown.raw_markdown or "").strip() if crawl_res.markdown else ""
|
||
|
||
# 优先用 trafilatura 从原始 HTML 提取正文（适用于任意网站结构）
|
||
# 回退顺序：trafilatura(html) > fit_markdown > raw_markdown
|
||
markdown = ""
|
||
if crawl_res.html:
|
||
markdown = _extract_text_from_html(crawl_res.html, url)
|
||
if len(markdown) < _MIN_CONTENT_LENGTH:
|
||
markdown = fit if len(fit) >= _MIN_CONTENT_LENGTH else raw
|
||
# title 在 crawl_res.metadata 字典里
|
||
meta = crawl_res.metadata or {}
|
||
title = (meta.get("title") or meta.get("og:title") or "").strip()
|
||
# metadata 没有时，从原始 HTML 提取
|
||
if not title and crawl_res.html:
|
||
title = _extract_title_from_html(crawl_res.html)
|
||
|
||
failed = _validate(markdown, url, stealth)
|
||
if failed:
|
||
logger.warning(f"[WebCrawler] 内容验证失败 (stealth={stealth}): {url}")
|
||
return failed
|
||
|
||
logger.info(f"[WebCrawler] Playwright 抓取成功 (stealth={stealth}): {url} | {len(markdown)} 字符")
|
||
return CrawlResult(success=True, url=url, markdown=markdown, title=title)
|
||
|
||
except Exception as exc:
|
||
logger.warning(f"[WebCrawler] Playwright 异常 (stealth={stealth}): {url} | {exc}")
|
||
return CrawlResult(success=False, url=url, error_message=str(exc))
|
||
|
||
|
||
async def _crawl_with_httpx(url: str, *, timeout_ms: int) -> CrawlResult:
|
||
"""
|
||
httpx 直接 HTTP 请求 + trafilatura 正文提取。
|
||
适用于：含 JS 跳转导致 Playwright 导航失败、但 HTML 本身已包含正文的页面。
|
||
"""
|
||
try:
|
||
import httpx
|
||
except ImportError:
|
||
return CrawlResult(success=False, url=url, error_message="httpx 未安装")
|
||
|
||
# 读取系统代理（httpx 新版用 proxy 单数形式）
|
||
_sys_proxy = (
|
||
os.environ.get("https_proxy") or os.environ.get("HTTPS_PROXY")
|
||
or os.environ.get("http_proxy") or os.environ.get("HTTP_PROXY")
|
||
) or None
|
||
|
||
try:
|
||
client_kwargs: dict = dict(
|
||
follow_redirects=True,
|
||
timeout=timeout_ms / 1000,
|
||
headers={
|
||
"User-Agent": _CHROME_UA,
|
||
"Accept": "text/html,application/xhtml+xml,*/*;q=0.9",
|
||
"Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
|
||
},
|
||
)
|
||
if _sys_proxy:
|
||
client_kwargs["proxy"] = _sys_proxy
|
||
|
||
async with httpx.AsyncClient(**client_kwargs) as client:
|
||
resp = await client.get(url)
|
||
resp.raise_for_status()
|
||
html = resp.text
|
||
except Exception as exc:
|
||
error_msg = f"httpx 抓取失败: {exc}"
|
||
logger.warning(f"[WebCrawler] {error_msg} | url={url}")
|
||
return CrawlResult(success=False, url=url, error_message=error_msg)
|
||
|
||
# 用 trafilatura 提取正文（如未安装则退回 BeautifulSoup 简单提取）
|
||
markdown = _extract_text_from_html(html, url)
|
||
title = _extract_title_from_html(html)
|
||
|
||
if not markdown:
|
||
error_msg = "httpx 抓取成功但未能提取到正文内容"
|
||
logger.warning(f"[WebCrawler] {error_msg}: {url}")
|
||
return CrawlResult(success=False, url=url, error_message=error_msg)
|
||
|
||
failed = _validate(markdown, url, stealth=False)
|
||
if failed:
|
||
return failed
|
||
|
||
logger.info(f"[WebCrawler] httpx 兜底抓取成功: {url} | {len(markdown)} 字符")
|
||
return CrawlResult(success=True, url=url, markdown=markdown, title=title)
|
||
|
||
|
||
def _extract_text_from_html(html: str, url: str) -> str:
|
||
"""从 HTML 提取正文，优先用 trafilatura，降级用 BeautifulSoup。"""
|
||
# 优先：trafilatura（专为正文提取设计，过滤导航/广告效果好）
|
||
try:
|
||
import trafilatura
|
||
text = trafilatura.extract(
|
||
html,
|
||
include_comments=False,
|
||
include_tables=True,
|
||
no_fallback=False,
|
||
url=url,
|
||
)
|
||
if text and len(text.strip()) >= _MIN_CONTENT_LENGTH:
|
||
return text.strip()
|
||
except ImportError:
|
||
pass
|
||
except Exception as e:
|
||
logger.debug(f"[WebCrawler] trafilatura 提取失败，降级到 bs4: {e}")
|
||
|
||
# 降级：BeautifulSoup 去除脚本/样式后取全文
|
||
try:
|
||
from bs4 import BeautifulSoup
|
||
soup = BeautifulSoup(html, "html.parser")
|
||
for tag in soup(["script", "style", "nav", "footer", "header", "aside"]):
|
||
tag.decompose()
|
||
return soup.get_text(separator="\n", strip=True)
|
||
except Exception:
|
||
return ""
|
||
|
||
|
||
def _extract_title_from_html(html: str) -> str:
|
||
"""从 HTML 提取标题，依次尝试 <title>、og:title、h1。"""
|
||
try:
|
||
from bs4 import BeautifulSoup
|
||
soup = BeautifulSoup(html, "html.parser")
|
||
# 1. <title> 标签
|
||
tag = soup.find("title")
|
||
if tag:
|
||
t = tag.get_text(strip=True)
|
||
if t:
|
||
return t
|
||
# 2. Open Graph og:title
|
||
og = soup.find("meta", property="og:title")
|
||
if og and og.get("content"):
|
||
return og["content"].strip()
|
||
# 3. 第一个 <h1>
|
||
h1 = soup.find("h1")
|
||
if h1:
|
||
t = h1.get_text(strip=True)
|
||
if t:
|
||
return t
|
||
except Exception:
|
||
pass
|
||
return ""
|