"""英文正文提取引擎 策略： 1. 优先 trafilatura 从原始 HTML 提取（除导航/广告/页脚） 2. 回退 Crawl4AI Markdown（如果 HTML 不可用） 3. 元数据提取（发布时间、作者） 所有业务参数从 configs/system.yaml 的 extractor 节读取。 """ import logging from pathlib import Path import trafilatura import yaml from htmldate import find_date from extractor.models import ProcessedArticle logger = logging.getLogger(__name__) def _load_extractor_config() -> dict: """从 system.yaml 读取 extractor 配置节""" config_path = Path("configs/system.yaml") if config_path.exists(): try: with open(config_path, encoding="utf-8") as f: raw = yaml.safe_load(f) return raw.get("extractor", {}) except Exception: logger.warning("读取 system.yaml 失败，使用默认值") return {} _cfg = _load_extractor_config() # 最小正文字数阈值 MIN_CONTENT_WORDS = int(_cfg.get("min_content_words", 50)) def extract_article( source_id: str, source_name: str, url: str, url_hash: str, html_path: str, md_path: str, crawl_publish_time: str = "", ) -> ProcessedArticle: """提取单篇文章的正文 Args: source_id: 新闻源 ID source_name: 新闻源名称 url: 文章 URL url_hash: URL SHA256 前 16 位 html_path: 原始 HTML 文件路径 md_path: Crawl4AI 生成的 Markdown 文件路径 crawl_publish_time: RSS/Crawler 提取的发布时间（回退用） Returns: ProcessedArticle """ article = ProcessedArticle( source_id=source_id, source_name=source_name, url=url, url_hash=url_hash, html_path=html_path, md_path=md_path, ) html_p = Path(html_path) if html_path else None md_p = Path(md_path) if md_path else None # ── 策略 1: trafilatura 从 HTML 提取 ────────────── if html_p and html_p.exists(): try: html_raw = html_p.read_text(encoding="utf-8") content = trafilatura.extract( html_raw, include_comments=False, include_tables=False, no_fallback=False, favor_precision=True, ) if content and _count_words(content) >= MIN_CONTENT_WORDS: article.content = content.strip() article.extractor = "trafilatura" article.word_count = _count_words(article.content) # 元数据 article.title = _extract_title(html_raw) or "" article.publish_time = ( _extract_publish_time(html_raw, url=url) or crawl_publish_time or "" ) article.author = _extract_author(html_raw) or "" article.status = "success" return article except Exception as e: logger.warning("[%s] trafilatura 提取失败: %s — %s", source_id, url_hash, e) # ── 策略 2: Crawl4AI Markdown 回退 ───────────────── if md_p and md_p.exists(): try: md_content = md_p.read_text(encoding="utf-8") if _count_words(md_content) >= MIN_CONTENT_WORDS: article.content = _clean_markdown(md_content) article.extractor = "crawl4ai_md" article.word_count = _count_words(article.content) # 从 Markdown 中提取元数据（RSS 源会写入） article.title = _extract_from_md( md_content, r"^#\s+(.+)$", article.title ) article.publish_time = ( _extract_from_md( md_content, r"\*\*发布时间\*\*:?\s*([^\n]+)", "" ) or _extract_publish_time("", url=url) or crawl_publish_time or "" ) article.status = "success" return article except Exception as e: logger.warning("[%s] Markdown 回退失败: %s — %s", source_id, url_hash, e) # ── 失败 ───────────────────────────────────────── article.status = "no_content" article.error = "No extractable content" return article # ── 辅助函数 ──────────────────────────────────────── def _count_words(text: str) -> int: """英文词数统计""" return len(text.split()) if text else 0 def _extract_title(html: str) -> str: """从 HTML 标签提取标题""" import re m = re.search(r"<title[^>]*>(.*?)", html, re.IGNORECASE | re.DOTALL) if m: title = re.sub(r"<[^>]+>", "", m.group(1)) return title.strip() return "" def _extract_publish_time(html: str, url: str = "") -> str: """从 HTML 提取发布时间（含回退策略）。 优先级: 1. htmldate 提取（元数据/标签/正文） 2. 等常见标签 3. URL 中的日期（/YYYY/MM/DD/ 或 /YYYYMMDD/） """ # 策略 1: htmldate try: result = find_date(html) if result: return result except Exception: pass # 策略 2: 常见 meta 标签 if html: import re meta_patterns = [ r']+property=["\']article:published_time["\'][^>]+content=["\']([^"\']+)', r']+content=["\']([^"\']+)["\'][^>]+property=["\']article:published_time', r']+name=["\']pubdate["\'][^>]+content=["\']([^"\']+)', r']+name=["\']publish_date["\'][^>]+content=["\']([^"\']+)', r']+name=["\']date["\'][^>]+content=["\']([^"\']+)', r']+datetime=["\']([^"\']+)', ] for pat in meta_patterns: m = re.search(pat, html, re.IGNORECASE) if m: date_str = m.group(1).strip() if date_str: return date_str # 策略 3: URL 中的日期 if url: from datetime import datetime as dt import re url_patterns = [ r"/(\d{4})/(\d{2})/(\d{2})/", r"/(\d{4})(\d{2})(\d{2})/", r"-(\d{4})(\d{2})(\d{2})(?:[/-]|$)", r"-(\d{4})-(\d{2})-(\d{2})[/-]", ] for pat in url_patterns: m = re.search(pat, url) if m: try: y, mo, d = int(m.group(1)), int(m.group(2)), int(m.group(3)) return dt(y, mo, d).isoformat() except ValueError: continue return "" def _extract_author(html: str) -> str: """从 HTML meta 标签提取作者""" import re for pat in [ r']+name=["\']author["\'][^>]+content=["\']([^"\']+)', r']+content=["\']([^"\']+)["\'][^>]+name=["\']author', ]: m = re.search(pat, html, re.IGNORECASE) if m: return m.group(1).strip() return "" def _extract_from_md(md: str, pattern: str, default: str) -> str: """从 Markdown 内容中按正则提取第一个捕获组，失败返回 default。""" import re m = re.search(pattern, md) return m.group(1).strip() if m else default def _clean_markdown(md: str) -> str: """清理 Crawl4AI MD 中的导航/广告噪音 规则：移除明显非正文行（短链接行、纯导航行等） """ lines = md.split("\n") cleaned: list[str] = [] skip_patterns = [ "ADVERTISEMENT", "Continue Reading Below", "Sign in", "Log in", "Subscribe", "Advertisement", ] for line in lines: stripped = line.strip() # 跳过明显广告/导航行 if any(p in stripped for p in skip_patterns): continue # 跳过纯导航链接行（[text](url) 且整行只有链接） if stripped.startswith("[") and stripped.endswith(")") and stripped.count("[") <= 3: continue cleaned.append(line) return "\n".join(cleaned)