bwstudio
ceed63fcb0
High priority: - Fix concurrent race condition for view_count/like_count (atomic update) - Add route request ID tracking to prevent race conditions - Filter get_joke by status=approved (no pending content leak) - Add error feedback for like button Performance: - Optimize random joke query (avoid full table sort) - Limit page_size max to 100 (DoS prevention) Medium: - Add localStorage quota error handling - Handle empty AI response gracefully - Fix generate content title extraction Low: - Add rejected_jokes to stats API - Update dashboard to show rejected count
439 lines
18 KiB
Python
439 lines
18 KiB
Python
"""统一网页获取：搜索笑话站点 + 深度翻页抓取。"""
|
||
|
||
import asyncio
|
||
import os
|
||
import re
|
||
import sys
|
||
from urllib.parse import quote, urljoin, urlparse
|
||
|
||
# Windows 下禁用 rich 控制台输出，避免 GBK 编码错误
|
||
if sys.platform == "win32":
|
||
os.environ["PYTHONIOENCODING"] = "utf-8"
|
||
os.environ["TERM"] = "dumb"
|
||
|
||
try:
|
||
from crawl4ai import AsyncWebCrawler
|
||
from crawl4ai.async_configs import CrawlerRunConfig, BrowserConfig
|
||
HAS_CRAWL4AI = True
|
||
except ImportError:
|
||
HAS_CRAWL4AI = False
|
||
print("[!] crawl4ai 未安装，将使用 requests 替代（不支持 JS 渲染）")
|
||
|
||
import httpx
|
||
|
||
|
||
class CrawlerService:
|
||
def __init__(self, headless: bool = True):
|
||
self.headless = headless
|
||
self.session = None
|
||
self._crawler = None
|
||
# 站点级别容错跟踪
|
||
self.site_failures: dict[str, int] = {}
|
||
# 最近一次抓取的原始 HTML（供翻页链接发现使用）
|
||
self._last_raw_html: str | None = None
|
||
|
||
def _get_session(self) -> httpx.Client:
|
||
if self.session is None:
|
||
self.session = httpx.Client(
|
||
headers={
|
||
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 Chrome/125.0.0.0 Safari/537.36",
|
||
"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",
|
||
"Referer": "https://www.baidu.com/",
|
||
},
|
||
timeout=30,
|
||
follow_redirects=True,
|
||
)
|
||
return self.session
|
||
|
||
async def _get_crawler(self):
|
||
"""获取或创建 crawl4ai 实例"""
|
||
if HAS_CRAWL4AI and self._crawler is None:
|
||
browser_cfg = BrowserConfig(headless=self.headless, verbose=False)
|
||
self._crawler = AsyncWebCrawler(config=browser_cfg)
|
||
await self._crawler.__aenter__()
|
||
if not self.headless:
|
||
print(" [*] 浏览器窗口已打开（crawl4ai 控制）")
|
||
return self._crawler
|
||
|
||
async def close(self):
|
||
if self._crawler:
|
||
await self._crawler.__aexit__(None, None, None)
|
||
self._crawler = None
|
||
if self.session:
|
||
self.session.close()
|
||
self.session = None
|
||
|
||
# ============================================================
|
||
# 页面抓取（带重试）
|
||
# ============================================================
|
||
|
||
async def crawl_page_with_retry(self, url: str, max_retries: int = 2, timeout: int = 30000) -> tuple[str, bool]:
|
||
"""抓取页面，返回 (html内容, 是否成功)，失败自动重试"""
|
||
for attempt in range(max_retries + 1):
|
||
print(f" [~] 正在获取页面... (尝试 {attempt+1}/{max_retries+1})")
|
||
try:
|
||
html = await self._crawl_one(url, timeout)
|
||
if html:
|
||
print(f" [~] 页面获取成功，内容长度: {len(html)} 字符")
|
||
return html, True
|
||
print(f" [!] 页面内容为空")
|
||
except Exception as e:
|
||
print(f" [!] 抓取失败 (尝试 {attempt+1}/{max_retries+1}): {url[:60]} - {e}")
|
||
|
||
if attempt < max_retries:
|
||
wait = 3 * (attempt + 1)
|
||
print(f" [~] 等待 {wait} 秒后重试...")
|
||
await asyncio.sleep(wait)
|
||
return "", False
|
||
|
||
async def _crawl_one(self, url: str, timeout: int) -> str:
|
||
"""单次页面抓取，返回纯文本内容（同时保存原始 HTML 供翻页发现）"""
|
||
if HAS_CRAWL4AI:
|
||
# crawl4ai 统一处理 headless / visible 两种模式
|
||
crawler = await self._get_crawler()
|
||
result = await crawler.arun(url, config=CrawlerRunConfig(verbose=False))
|
||
if result and result.success:
|
||
# fit_html 是 AI 清洗后的正文（翻页链接常被清洗掉）
|
||
# cleaned_html 保留完整结构（用于翻页发现）
|
||
self._last_raw_html = result.cleaned_html or result.html or ""
|
||
html = result.fit_html or result.cleaned_html or ""
|
||
return self._html_to_text(html)
|
||
return ""
|
||
else:
|
||
# 无 crawl4ai 时 fallback 到 requests
|
||
resp = self._get_session().get(url, timeout=timeout / 1000)
|
||
if resp.status_code == 200:
|
||
self._last_raw_html = resp.text
|
||
return self._html_to_text(resp.text)
|
||
return ""
|
||
|
||
@staticmethod
|
||
def _html_to_text(html: str) -> str:
|
||
"""简易 HTML 转纯文本"""
|
||
# 移除 script/style 标签内容
|
||
text = re.sub(r'<script[^>]*>.*?</script>', '', html, flags=re.DOTALL | re.IGNORECASE)
|
||
text = re.sub(r'<style[^>]*>.*?</style>', '', text, flags=re.DOTALL | re.IGNORECASE)
|
||
# 移除 HTML 标签
|
||
text = re.sub(r'<[^>]+>', ' ', text)
|
||
# 合并空白
|
||
text = re.sub(r'\s+', ' ', text).strip()
|
||
return text
|
||
|
||
# ============================================================
|
||
# 搜索笑话站点
|
||
# ============================================================
|
||
|
||
# 已知的笑话聚合站域名黑/白名单
|
||
JOKE_SITE_KEYWORDS = [
|
||
"笑话大全", "冷笑话", "搞笑段子", "笑话集锦",
|
||
"幽默笑话", "爆笑笑话", "成人笑话", "小笑话",
|
||
]
|
||
|
||
async def search_joke_sites(self, keywords: list[str], max_results: int = 10) -> list[dict]:
|
||
"""搜索笑话站点，返回 [{url, title, domain}]"""
|
||
all_results = []
|
||
|
||
for keyword in keywords:
|
||
results = await self._search_and_filter(keyword, max_results=max_results)
|
||
all_results.extend(results)
|
||
|
||
# 去重（按域名）
|
||
seen_domains = set()
|
||
unique = []
|
||
for r in all_results:
|
||
domain = r["domain"]
|
||
if domain not in seen_domains:
|
||
seen_domains.add(domain)
|
||
unique.append(r)
|
||
|
||
print(f" [*] 搜索到 {len(unique)} 个唯一站点")
|
||
for s in unique:
|
||
print(f" - {s['domain']}: {s['title'][:40]}")
|
||
return unique[:max_results]
|
||
|
||
async def _search_and_filter(self, keyword: str, max_results: int = 10) -> list[dict]:
|
||
"""搜索并过滤出疑似笑话聚合站的结果"""
|
||
# 用多个搜索词提高覆盖率
|
||
search_queries = [
|
||
f"{keyword} 网站",
|
||
f"{keyword} 大全",
|
||
f"{keyword} 列表",
|
||
]
|
||
|
||
seen = set()
|
||
sites = []
|
||
|
||
for q in search_queries:
|
||
if len(sites) >= max_results:
|
||
break
|
||
|
||
search_results = await self._search_bing(q, max_pages=2)
|
||
|
||
for r in search_results:
|
||
if len(sites) >= max_results:
|
||
break
|
||
|
||
url = r["url"]
|
||
domain = urlparse(url).netloc.lower()
|
||
|
||
if domain in seen:
|
||
continue
|
||
seen.add(domain)
|
||
|
||
# 过滤：排除已知的单篇文章站点和搜索引擎
|
||
if self._is_joke_collection_site(url, r.get("title", "")):
|
||
r["domain"] = domain
|
||
sites.append(r)
|
||
|
||
return sites
|
||
|
||
def _is_joke_collection_site(self, url: str, title: str) -> bool:
|
||
"""判断URL是否疑似笑话聚合站（不是单篇文章）"""
|
||
domain = urlparse(url).netloc.lower()
|
||
path = urlparse(url).path.lower()
|
||
|
||
# 排除项
|
||
exclude_domains = [
|
||
"bing.com", "microsoft.com", "baidu.com", "google.com",
|
||
"sohu.com", "sina.com", "163.com", "qq.com", "toutiao.com",
|
||
"weibo.com", "zhihu.com", "bilibili.com", "douban.com",
|
||
]
|
||
if any(d in domain for d in exclude_domains):
|
||
return False
|
||
|
||
# 排除明显的单篇文章模式
|
||
single_article_patterns = [
|
||
r'/p/\d+', r'/article/\d+', r'/post/\d+', r'/archives/\d+',
|
||
r'/a/\d+', r'/\d{5,}', r'/detail/\d+', r'/read/\d+',
|
||
r'\.html$', # 静态 html 文章页
|
||
]
|
||
# 但如果域名本身含 joke 特征，不排除
|
||
is_joke_domain = any(kw in domain or kw in title for kw in
|
||
["joke", "xiaohua", "笑话", "段子", "幽默", "搞笑"])
|
||
|
||
for p in single_article_patterns:
|
||
if re.search(p, path) and not is_joke_domain:
|
||
return False
|
||
|
||
# 聚合站特征：域名或标题含特定词，或URL有分类/列表模式
|
||
collection_patterns = [
|
||
"joke", "xiaohua", "笑话", "段子", "幽默", "搞笑",
|
||
"/page/", "/list/", "/category/", "/tag/", "joke",
|
||
]
|
||
for p in collection_patterns:
|
||
if p in domain or p in path or p.lower() in title:
|
||
return True
|
||
|
||
# 有列表/目录模式的也认为是聚合站
|
||
if re.search(r'(page|list|category|tag|index)', path):
|
||
return True
|
||
|
||
return False
|
||
|
||
# ============================================================
|
||
# 翻页链接发现
|
||
# ============================================================
|
||
|
||
def discover_page_links(self, html: str, base_url: str) -> list[str]:
|
||
"""从页面 HTML 中发现翻页链接，返回去重排序后的 URL 列表"""
|
||
# 优先使用 _last_raw_html（完整 HTML 而非纯文本），兜底用传入的 html
|
||
raw = self._last_raw_html or html
|
||
base_parsed = urlparse(base_url)
|
||
base_domain = f"{base_parsed.scheme}://{base_parsed.netloc}"
|
||
|
||
links = set()
|
||
|
||
# 1. <link rel="next">
|
||
for m in re.finditer(r'<link[^>]*rel="next"[^>]*href="([^"]+)"', raw, re.IGNORECASE):
|
||
links.add(urljoin(base_domain, m.group(1)))
|
||
|
||
# 2. 翻页文字链接（下一页、下页、>、» 等）
|
||
page_text_patterns = [
|
||
r'<a[^>]*href="([^"]*page[^"]*)"[^>]*>\s*(?:下一页|下页|下一页»|»|›|>|Next|last)\s*</a>',
|
||
r'<a[^>]*>\s*(?:下一页|下页|»|›|>)\s*</a>\s*<a[^>]*href="([^"]*)"',
|
||
]
|
||
for pattern in page_text_patterns:
|
||
for m in re.finditer(pattern, raw, re.IGNORECASE):
|
||
href = m.group(1).strip()
|
||
if href and href not in ("#", "javascript:void(0)"):
|
||
links.add(urljoin(base_url, href))
|
||
|
||
# 3. 提取所有带数字的翻页 link（?page=N, /page/N/, index_N.html, page_N.html, &page=N）
|
||
page_link_patterns = [
|
||
r'href="([^"]*[?&]page=(\d+)[^"]*)"', # ?page=2 &page=2
|
||
r'href="([^"]*/page/(\d+)[^"]*)"', # /page/2/
|
||
r'href="([^"]*[?&]p=(\d+)[^"]*)"', # ?p=2
|
||
r'href="([^"]*[?&]pn=(\d+)[^"]*)"', # ?pn=2
|
||
r'href="([^"]*[?&]offset=(\d+)[^"]*)"', # ?offset=10
|
||
r'href="([^"]*[?&]start=(\d+)[^"]*)"', # ?start=10
|
||
r'href="([^"]*[?&]page_index=(\d+)[^"]*)"', # ?page_index=2
|
||
# 匹配 xxx_N.html / page_N.html / list_2.html
|
||
r'href="([^"]*(?:page|list|index)[-_]?(\d+)\.html?)"',
|
||
# 匹配 /page_N/ 格式
|
||
r'href="([^"]*/page[-_]?(\d+)/?)"',
|
||
]
|
||
for pattern in page_link_patterns:
|
||
for m in re.finditer(pattern, raw, re.IGNORECASE):
|
||
full_url = urljoin(base_url, m.group(1))
|
||
links.add(full_url)
|
||
|
||
# 4. 翻页数字链接
|
||
for m in re.finditer(r'<a[^>]*href="([^"]*page=(\d+)[^"]*)"[^>]*>\s*\d+\s*</a>', raw, re.IGNORECASE):
|
||
links.add(urljoin(base_url, m.group(1)))
|
||
|
||
# 过滤：只保留同一域名下的链接
|
||
result = []
|
||
for link in links:
|
||
parsed = urlparse(link)
|
||
if parsed.netloc and parsed.netloc != base_parsed.netloc:
|
||
continue # 跨域排除
|
||
if parsed.path == base_parsed.path and parsed.query == base_parsed.query:
|
||
continue # 排除自身
|
||
result.append(link)
|
||
|
||
# 去重排序
|
||
return sorted(set(result))
|
||
|
||
def _extract_page_number(self, url: str) -> int:
|
||
"""从 URL 中提取页码，用于排序"""
|
||
nums = re.findall(r'page[=/](\d+)|[?&]p=(\d+)|index[-_]?(\d+)|/page[-_]?(\d+)', url, re.IGNORECASE)
|
||
for n in nums:
|
||
for g in n:
|
||
if g:
|
||
return int(g)
|
||
return 99 # 没识别到页码的排最后
|
||
|
||
# ============================================================
|
||
# 分类/标签链接发现
|
||
# ============================================================
|
||
|
||
def discover_category_links(self, html: str, base_url: str) -> list[str]:
|
||
"""从页面 HTML 中发现分类/标签链接，返回去重排序后的 URL 列表"""
|
||
raw = self._last_raw_html or html
|
||
base_parsed = urlparse(base_url)
|
||
base_domain = f"{base_parsed.scheme}://{base_parsed.netloc}"
|
||
|
||
links = set()
|
||
|
||
# 1. 匹配分类链接（category-N.html, category-N_M.html, tag-N.html 等）
|
||
cat_patterns = [
|
||
r'href="([^"]*/(?:category|cat|sort|type)[-_]?\d+(?:[-_]\d+)?\.html?)"',
|
||
r'href="([^"]*/(?:tag|tags)/?[-_]?\d*)["\s>]',
|
||
r'href="([^"]*/tags?[-_]?\d+\.html?)"',
|
||
]
|
||
for pattern in cat_patterns:
|
||
for m in re.finditer(pattern, raw, re.IGNORECASE):
|
||
full_url = urljoin(base_domain, m.group(1))
|
||
parsed = urlparse(full_url)
|
||
# 只保留同域链接
|
||
if parsed.netloc and parsed.netloc != base_parsed.netloc:
|
||
continue
|
||
links.add(full_url)
|
||
|
||
# 2. 过滤：排除单篇文章、首页、搜索页
|
||
single_article = re.compile(
|
||
r'/(?:p|post|article|archives|detail|read|xiaohua)/\d+',
|
||
re.IGNORECASE,
|
||
)
|
||
result = []
|
||
for link in links:
|
||
parsed = urlparse(link)
|
||
path = parsed.path.rstrip("/")
|
||
# 排除自身
|
||
if path == base_parsed.path.rstrip("/") and parsed.query == base_parsed.query:
|
||
continue
|
||
# 排除单篇文章
|
||
if single_article.search(path):
|
||
continue
|
||
# 排除明显的非分类路径（首页翻页）
|
||
if re.search(r'/page[-_]?\d+\.html?$', path) and 'category' not in path and 'tag' not in path:
|
||
continue
|
||
result.append(link)
|
||
|
||
return sorted(set(result))
|
||
|
||
# ============================================================
|
||
# Bing 搜索（复用旧逻辑）
|
||
# ============================================================
|
||
|
||
async def _search_bing(self, keyword: str, max_pages: int = 3) -> list[dict]:
|
||
"""搜索 Bing，返回结果 URL 列表"""
|
||
results = []
|
||
session = self._get_session()
|
||
|
||
for page in range(max_pages):
|
||
first = page * 10
|
||
url = f"https://www.bing.com/search?q={quote(keyword)}&first={first}"
|
||
print(f" [*] Bing 搜索: {keyword[:20]}")
|
||
|
||
html = None
|
||
try:
|
||
if HAS_CRAWL4AI:
|
||
crawler = await self._get_crawler()
|
||
result = await crawler.arun(url, config=CrawlerRunConfig(verbose=False))
|
||
if result.success:
|
||
html = result.html if result.html else result.cleaned_html
|
||
else:
|
||
resp = session.get(url)
|
||
html = resp.text if resp.status_code == 200 else None
|
||
|
||
if html:
|
||
urls = self._extract_bing_urls(html)
|
||
for item in urls:
|
||
item["keyword"] = keyword
|
||
results.append(item)
|
||
else:
|
||
print(f" [!] 获取搜索页面失败")
|
||
|
||
await asyncio.sleep(2)
|
||
|
||
except Exception as e:
|
||
print(f" [!] 搜索异常: {e}")
|
||
continue
|
||
|
||
return results
|
||
|
||
def _extract_bing_urls(self, html: str) -> list[dict]:
|
||
"""从 Bing 搜索结果 HTML 中提取链接和标题"""
|
||
results = []
|
||
pattern = re.compile(
|
||
r'<h2[^>]*>\s*<a[^>]*href="(https?[^"]+)"[^>]*>(.*?)</a>',
|
||
re.DOTALL,
|
||
)
|
||
for match in pattern.finditer(html):
|
||
href = match.group(1).strip()
|
||
title = re.sub(r'<[^>]+>', '', match.group(2)).strip()
|
||
if href and title and len(title) > 5 and "bing.com" not in href and "microsoft.com" not in href:
|
||
results.append({"url": href, "title": title})
|
||
return results
|
||
|
||
# 别名兼容
|
||
async def search_bing(self, keyword: str, max_pages: int = 3) -> list[dict]:
|
||
return await self._search_bing(keyword, max_pages)
|
||
|
||
async def search(self, keyword: str, max_pages: int = 3) -> list[dict]:
|
||
return await self._search_bing(keyword, max_pages)
|
||
|
||
async def search_baidu(self, keyword: str, max_pages: int = 3) -> list[dict]:
|
||
return await self._search_bing(keyword, max_pages)
|
||
|
||
# ============================================================
|
||
# 旧接口兼容（单页抓取）
|
||
# ============================================================
|
||
|
||
async def crawl_page(self, url: str) -> str:
|
||
"""抓取单个页面（兼容旧接口）"""
|
||
content, ok = await self.crawl_page_with_retry(url)
|
||
if ok:
|
||
return content[:8000]
|
||
return ""
|
||
|
||
async def crawl_batch(self, urls: list[str]) -> list[tuple[str, str]]:
|
||
"""批量抓取（兼容旧接口）"""
|
||
contents = []
|
||
for url in urls:
|
||
content, ok = await self.crawl_page_with_retry(url)
|
||
contents.append((url, content[:8000] if ok else ""))
|
||
await asyncio.sleep(1.5)
|
||
return contents |