"""中文新闻正文提取(M2)。 核心流程: 1. GNE 主提取(title / content / publish_time / author / images); 2. title 校正:若 GNE 抓到 标签内容明显短于 <h1>,优先 <h1>; 3. content 清理:剥离 GNE 习惯性附加在正文头部的 title/time/author 行, 去除连续空行,去重前导空格; 4. publish_time 标准化为 datetime,加合理性检查与 HTML 中文日期兜底; 5. 中文字符数统计; 6. 校验最小长度,过短抛 ExtractError。 """ from __future__ import annotations import hashlib import re from datetime import datetime, timedelta from typing import Any from bs4 import BeautifulSoup from dateutil import parser as date_parser from gne import GeneralNewsExtractor from loguru import logger from .models import Article, ExtractError # 全局复用单例,GeneralNewsExtractor 内部加载规则文件,避免重复初始化 _EXTRACTOR = GeneralNewsExtractor() # 验收门槛:正文太短视为提取失败 MIN_CONTENT_LENGTH = 50 # 模板兜底检测:关键词组(必须全部出现) + 长度上限(超过则视为合法长文)。 # 命中说明 GNE 提取失败,落到了网站固定模板/广告/版权声明文本,实际文章正文未抓到。 # 长度上限避免真实长篇文章中偶尔提到这些词被误判。 _BOILERPLATE_PATTERNS: list[tuple[tuple[str, ...], int]] = [ # eastmoney:页面底部"郑重声明...证券法...东方财富社区管理规定" (("郑重声明", "证券法"), 800), # yicai:"第一财经广告合作 ... 著作权归第一财经所有" (("第一财经广告合作", "著作权"), 800), # yicai 变体:"未经第一财经书面授权 不得以任何方式加以使用" (("第一财经", "未经", "书面授权"), 800), # sina:嵌入式个人专栏推送(同一篇被多个新闻页复用) (("北京红竹", "跷跷板"), 800), # 通用版权页兜底 (("未经", "授权", "禁止转载"), 600), ] def _is_boilerplate(content: str) -> tuple[bool, str | None]: """检查内容是否为已知模板兜底文本。 返回 (is_boilerplate, matched_pattern_summary)。 """ if not content: return False, None for keywords, max_len in _BOILERPLATE_PATTERNS: if len(content) > max_len: continue if all(kw in content for kw in keywords): return True, "+".join(keywords) return False, None # 站点中文名映射(便于在 Article.source_name 标注) SOURCE_NAME_MAP = { "cls": "财联社", "eastmoney": "东方财富", "sina": "新浪财经", "stcn": "证券时报", "yicai": "第一财经", } # --------------------------------------------------------------------------- # # 工具 # --------------------------------------------------------------------------- # def _url_hash(url: str) -> str: return hashlib.sha1(url.encode("utf-8")).hexdigest()[:16] _CHINESE_CHAR_RE = re.compile(r"[一-鿿]") def _count_chinese(text: str) -> int: return len(_CHINESE_CHAR_RE.findall(text)) # --------------------------------------------------------------------------- # # title 校正 # --------------------------------------------------------------------------- # def _refine_title(html: str, gne_title: str) -> str: """优先使用 <h1>;若无 h1 或 h1 比 gne_title 短,则保留 gne_title。 经验:GNE 偶尔会抓 <title> 标签,而 <title> 常包含站名后缀 (如 "宁德时代 - 新华网"),H1 通常更纯净。 """ soup = BeautifulSoup(html, "html.parser") h1 = soup.find("h1") h1_text = (h1.get_text(strip=True) if h1 else "").strip() g = (gne_title or "").strip() # 两个都为空 -> 失败由调用方处理 if not h1_text and not g: return "" # 只有一个非空 if not h1_text: return g if not g: return h1_text # H1 是 gne_title 的子串(说明 gne 带了网站后缀),用 H1 if h1_text in g and len(h1_text) < len(g): return h1_text # gne_title 是 H1 子串,用 H1 if g in h1_text: return h1_text # 两者差异大,gne_title 更短(可能就是页面 <title> 简称),用 H1 if len(h1_text) > len(g) * 1.2: return h1_text return g # --------------------------------------------------------------------------- # # content 清理 # --------------------------------------------------------------------------- # _MULTI_BLANK_RE = re.compile(r"\n{3,}") _TRAILING_SPACE_RE = re.compile(r"[ \t]+\n") def _clean_content(content: str, title: str, author: str | None, time_raw: str | None) -> str: """去除 GNE 输出 content 头部混入的 title/time/author 行,以及多余空行。""" if not content: return "" lines = [ln.rstrip() for ln in content.splitlines()] # 跳过头部若干行,只要它们与 title/author/time_raw 有明显重合 drop_targets: list[str] = [s for s in (title, author, time_raw) if s] drop_targets_norm = {t.strip() for t in drop_targets if t and t.strip()} cleaned: list[str] = [] head_skipping = True for ln in lines: s = ln.strip() if head_skipping: if not s: # 头部空行直接跳 continue if s in drop_targets_norm: continue # 仅由 author 字符串前缀(如 "记者: 张三" vs "张三") if any(s.startswith(t) or t.startswith(s) for t in drop_targets_norm if len(t) >= 4): continue head_skipping = False cleaned.append(ln) # 去尾部空行 while cleaned and not cleaned[-1].strip(): cleaned.pop() text = "\n".join(cleaned) text = _TRAILING_SPACE_RE.sub("\n", text) text = _MULTI_BLANK_RE.sub("\n\n", text) return text.strip() # --------------------------------------------------------------------------- # # 时间标准化 # --------------------------------------------------------------------------- # # 中文时间常见模式预清理(GNE 可能给出 "2026年6月15日 14:30") _CN_DATE_RE = re.compile(r"(\d{4})年(\d{1,2})月(\d{1,2})日") _CN_TIME_RE = re.compile(r"(\d{1,2})时(\d{1,2})分(?:(\d{1,2})秒)?") # 时间合理性边界:超过当前 +1 天为未来时间;早于 -365 天视为页脚等噪声 _TIME_FUTURE_TOLERANCE = timedelta(days=1) _TIME_PAST_TOLERANCE = timedelta(days=365) # HTML 中文日期兜底正则(在 HTML 全文中找首个看似发布时间的字符串) # 命中形如 "2026年06月16日 17:54" / "2026年6月16日 17:54:30" / "2026-06-16 17:54" _HTML_DATE_FALLBACK_RE = re.compile( r"(\d{4}[-年/]\s*\d{1,2}[-月/]\s*\d{1,2}日?" # 日期 r"(?:\s+\d{1,2}[:时]\d{1,2}(?:[:分]\d{1,2}秒?)?)?)" # 可选时分秒 ) def _normalize_time(raw: str | None) -> datetime | None: """把抽到的时间字符串解析为 datetime。失败返回 None。""" if not raw: return None text = raw.strip() if not text: return None # 拒绝明显不是具体日期的模式(日期范围/部分日期) if re.search(r"\d{4}年\d{1,2}\s*[-~至到]", text): return None text = _CN_DATE_RE.sub(r"\1-\2-\3", text) text = _CN_TIME_RE.sub( lambda m: f"{m.group(1)}:{m.group(2)}" + (f":{m.group(3)}" if m.group(3) else ""), text, ) # 必须有完整的年月日才解析(过滤只有年月的片段) if not re.search(r"\d{4}-\d{1,2}-\d{1,2}", text): return None try: return date_parser.parse(text, fuzzy=True) except (ValueError, OverflowError) as e: logger.debug("时间解析失败 raw={!r} err={}", raw, e) return None def _is_reasonable_time(dt: datetime | None, ref: datetime | None = None) -> bool: """判断 dt 是否在 ref 附近合理范围内。 - 未来超过 1 天 -> 不合理; - 早于 365 天 -> 不合理(GNE 偶尔抓到页脚备案/版权时间)。 """ if dt is None: return False ref = ref or datetime.now() # 同时移除时区信息以便比较(GNE 给出的时间多为 naive) if dt.tzinfo is not None: dt = dt.replace(tzinfo=None) if dt > ref + _TIME_FUTURE_TOLERANCE: return False return dt >= ref - _TIME_PAST_TOLERANCE def _fallback_time_from_html(html: str) -> tuple[datetime | None, str | None]: """从 HTML 全文中正则搜索中文日期模式,返回 (datetime, raw_str)。 搜索顺序对所有匹配做合理性过滤,选择第一个合理时间。这是站点无关的 通用兜底,适用于 GNE 误抓页脚备案时间(如 eastmoney 的 2019-01-16)的场景。 """ if not html: return None, None ref = datetime.now() for match in _HTML_DATE_FALLBACK_RE.finditer(html): raw = match.group(1).strip() dt = _normalize_time(raw) if dt is not None and _is_reasonable_time(dt, ref): logger.debug("HTML 兜底时间命中: {!r} -> {}", raw, dt) return dt, raw return None, None def _resolve_publish_time( html: str, gne_time_raw: str | None ) -> tuple[datetime | None, str | None]: """两阶段时间解析:先 GNE,合理性失败则 HTML 兜底。 返回 (publish_time, publish_time_raw)。 """ primary = _normalize_time(gne_time_raw) if _is_reasonable_time(primary): return primary, gne_time_raw # GNE 时间不可用或不合理,尝试从 HTML 兜底 fallback_dt, fallback_raw = _fallback_time_from_html(html) if fallback_dt is not None: if gne_time_raw and primary is not None: logger.info( "GNE 时间 {!r} 与当前差距过大,改用 HTML 兜底 {!r}", gne_time_raw, fallback_raw ) return fallback_dt, fallback_raw # 都失败,保留 GNE 原始字符串供人工核对 return None, gne_time_raw # --------------------------------------------------------------------------- # # 主入口 # --------------------------------------------------------------------------- # def extract_article( html: str, source_id: str, url: str, *, extra_noise_xpath: list[str] | None = None, extra_config: dict[str, str] | None = None, ) -> Article: """从 HTML 中提取 Article。 参数: html: 原始 HTML 字符串。 source_id: M1 sources.yaml 中的 source id。 url: 文章 URL。 extra_noise_xpath: 额外的噪声节点 XPath(如评论区/相关阅读容器)。 返回: Article。 异常: ExtractError: GNE 提取失败 / 正文过短。 """ if not html or not html.strip(): raise ExtractError("空 HTML", url=url) try: # 从 url 推 host(GNE 用它解析图片相对路径) from urllib.parse import urlparse parsed = urlparse(url) host = f"{parsed.scheme}://{parsed.netloc}" if parsed.scheme else "" result: dict[str, Any] = _EXTRACTOR.extract( html, host=host, body_xpath=(extra_config or {}).get("body_xpath", ""), noise_node_list=extra_noise_xpath or [], ) except Exception as e: raise ExtractError(f"GNE 提取异常: {type(e).__name__}: {e}", url=url) from e gne_title = result.get("title", "") or "" gne_content = result.get("content", "") or "" gne_time = result.get("publish_time", "") or None gne_author = (result.get("author", "") or "").strip() or None images = list(result.get("images", []) or []) title = _refine_title(html, gne_title) if not title: raise ExtractError("未能提取到标题", url=url) content = _clean_content(gne_content, title, gne_author, gne_time) if len(content) < MIN_CONTENT_LENGTH: raise ExtractError( f"正文过短(长度 {len(content)} < {MIN_CONTENT_LENGTH})", url=url, ) is_bp, bp_reason = _is_boilerplate(content) if is_bp: raise ExtractError( f"内容疑似模板兜底({bp_reason}),长度 {len(content)}", url=url, ) publish_time, publish_time_raw = _resolve_publish_time(html, gne_time) article = Article( source_id=source_id, url=url, url_hash=_url_hash(url), title=title, content=content, author=gne_author, source_name=SOURCE_NAME_MAP.get(source_id), publish_time=publish_time, publish_time_raw=publish_time_raw, images=images, word_count=_count_chinese(content), ) logger.debug("提取成功: {}", article.short_summary()) return article