"""中文新闻正文提取(M2)。
核心流程:
1. GNE 主提取(title / content / publish_time / author / images);
2. title 校正:若 GNE 抓到
标签内容明显短于 ,优先 ;
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 比 gne_title 短,则保留 gne_title。
经验:GNE 偶尔会抓 标签,而 常包含站名后缀
(如 "宁德时代 - 新华网"),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 更短(可能就是页面 简称),用 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