的父元素 + 3. 移除导航、广告等干扰元素 + 4. 过滤掉过短的段落（可能是元数据） + """ + # 策略1: 尝试常见选择器 + for selector in CONTENT_SELECTORS: + element = soup.select_one(selector) + if element and len(element.get_text(strip=True)) > 100: + # 移除干扰元素 + for remove_sel in REMOVE_SELECTORS: + for bad in element.select(remove_sel): + bad.decompose() + return element + + # 策略2: 查找包含最多段落的容器 + body = soup.find('body') + if not body: + body = soup + + # 找到包含最多
标签的元素 + best_element = None + max_p_count = 0 + + for elem in body.find_all(True): + p_count = len(elem.find_all('p')) + if p_count > max_p_count: + max_p_count = p_count + best_element = elem + + if best_element and max_p_count >= 2: + # 移除干扰元素 + for remove_sel in REMOVE_SELECTORS: + for bad in best_element.select(remove_sel): + bad.decompose() + return best_element + + # 策略3: 直接使用 body + if body and len(body.get_text(strip=True)) > 100: + for remove_sel in REMOVE_SELECTORS: + for bad in body.select(remove_sel): + bad.decompose() + return body + + return None + + +def _clean_content(element) -> str: + """清理提取的正文内容""" + # 过滤掉过短的段落（可能是元数据） + paragraphs = element.find_all('p') + valid_paragraphs = [] + + for p in paragraphs: + text = p.get_text(strip=True) + if _is_likely_body_text(text): + valid_paragraphs.append(p) + + # 如果过滤后还有有效段落，使用它们 + if valid_paragraphs: + # 用有效段落替换原内容 + new_div = element.find_parent().new_tag('div') + for p in valid_paragraphs: + new_div.append(p) + return str(new_div) + + # 如果没有有效段落（可能段落都很短或者是其他结构），保留原始内容 + return str(element) + + +def extract_text_from_html(html: str) -> str: + """从HTML中提取纯文本正文""" + if not html or len(html.strip()) < 50: + return "" + + try: + soup = BeautifulSoup(html, 'html.parser') + + # 移除所有链接的 href（避免干扰） + for a in soup.find_all('a'): + a.unwrap() + + # 移除图片 + for img in soup.find_all('img'): + img.decompose() + + # 保留段落结构，转换为文本 + text_parts = [] + for elem in soup.find_all(['p', 'h1', 'h2', 'h3', 'h4', 'h5', 'h6', 'li', 'blockquote']): + text = elem.get_text(strip=True) + if text and len(text) > 10: + text_parts.append(text) + + if not text_parts: + # 如果没有找到结构化内容，尝试获取所有文本 + text_parts = [soup.get_text(separator='\n', strip=True)] + + return '\n\n'.join(text_parts) + + except Exception as e: + print(f" ⚠️ 内容提取失败: {e}") + return "" + + +async def fetch_article_content( + session: aiohttp.ClientSession, + url: str, + timeout: int = FETCH_TIMEOUT +) -> Optional[str]: + """异步抓取单篇文章的完整内容 + + Returns: + 提取后的纯文本正文，失败返回 None + """ + try: + client_timeout = aiohttp.ClientTimeout(total=timeout) + headers = { + **DEFAULT_HEADERS, + "Referer": url, + } + + async with session.get(url, headers=headers, timeout=client_timeout) as resp: + if resp.status != 200: + print(f" ⚠️ HTTP {resp.status}: {url}") + return None + + content_type = resp.headers.get('Content-Type', '') + if 'text/html' not in content_type and 'application/xhtml' not in content_type: + # 不是HTML内容（可能是PDF等） + return None + + html = await resp.text() + + # 提取正文 + soup = BeautifulSoup(html, 'html.parser') + main_content = _extract_main_content(soup, url) + + if not main_content: + return None + + # 清理并提取文本 + cleaned = _clean_content(main_content) if hasattr(main_content, 'find_all') else str(main_content) + text = extract_text_from_html(cleaned) + + return text if text else None + + except asyncio.TimeoutError: + print(f" ⚠️ 抓取超时: {url}") + return None + except Exception as e: + print(f" ⚠️ 抓取异常: {url} - {e}") + return None + + +async def fetch_contents_batch( + entries: List[Dict], + max_workers: int = 5, + timeout: int = FETCH_TIMEOUT +) -> List[Dict]: + """批量抓取文章完整内容 + + Args: + entries: RSS条目列表，每个条目需包含 'link' 字段 + max_workers: 最大并发数 + timeout: 单篇文章超时时间 + + Returns: + 更新后的条目列表，新增 'full_content' 字段 + """ + if not entries: + return entries + + print(f" 📄 开始抓取全文内容: {len(entries)} 篇文章 (并发={max_workers})") + + semaphore = asyncio.Semaphore(max_workers) + + async def fetch_one(entry, session): + async with semaphore: + link = entry.get("link", "") + if not link: + return entry + + content = await fetch_article_content(session, link, timeout) + if content: + entry["full_content"] = content + # 如果原文太短（摘要），用全文替换 + original_content = entry.get("content", "") + if len(original_content) < 200 and len(content) > 500: + entry["content"] = content + print(f" ✅ {entry.get('title', '')[:40]}... (摘要→全文)") + else: + print(f" ✅ {entry.get('title', '')[:40]}...") + else: + print(f" ⚠️ {entry.get('title', '')[:40]}... (抓取失败)") + return entry + + # 创建共享的 session + async with aiohttp.ClientSession() as session: + tasks = [asyncio.create_task(fetch_one(e, session)) for e in entries] + results = await asyncio.gather(*tasks, return_exceptions=True) + + # 处理结果 + success_count = sum(1 for e in entries if e.get("full_content")) + fail_count = len(entries) - success_count + print(f" 全文抓取完成: 成功 {success_count} 条 | 失败 {fail_count} 条") + + return entries diff --git a/my-daily/src/fetch_pipeline.py b/my-daily/src/fetch_pipeline.py index a725780..8199a66 100644 --- a/my-daily/src/fetch_pipeline.py +++ b/my-daily/src/fetch_pipeline.py @@ -8,6 +8,7 @@ from typing import Dict, List, Tuple from src.cache import get_cache from src.config import get_merged_sources, get_timezone +from src.content_fetcher import fetch_contents_batch from src.fetcher import fetch_all_feeds from src.llm import ( generate_immediate_push, @@ -127,6 +128,42 @@ async def fetch_rss_entries( return entries, lookback +async def fetch_full_contents( + entries: List[Dict], + config: Dict, + monitor, +) -> List[Dict]: + """阶段1.5: 全文内容抓取 + + 从RSS条目的链接地址获取完整文章内容。 + 仅当配置启用且条目内容较短（摘要）时触发。 + + Args: + entries: RSS条目列表 + config: 配置字典 + monitor: 监控器实例 + + Returns: + 更新后的条目列表 + """ + content_config = config.get("content_fetch", {}) + if not content_config.get("enabled", False): + return entries + + # 检查是否有需要抓取全文的条目（内容较短的视为摘要） + needs_fetch = [e for e in entries if len(e.get("content", "")) < 200 and e.get("link")] + if not needs_fetch: + print(f" ℹ️ 所有条目内容完整，跳过全文抓取") + return entries + + with monitor.stage("content_fetch"): + max_workers = content_config.get("max_workers", 5) + timeout = content_config.get("timeout", 15) + entries = await fetch_contents_batch(entries, max_workers=max_workers, timeout=timeout) + + return entries + + def deduplicate_entries( entries: List[Dict], config: Dict, @@ -542,6 +579,9 @@ async def run_fetch_pipeline( if not entries: return + # 阶段1.5: 全文抓取（如启用） + entries = await fetch_full_contents(entries, config, monitor) + # 阶段2: 去重 new_entries, cache_hits = deduplicate_entries(entries, config, monitor, lookback) if not new_entries: