from __future__ import annotations
from dataclasses import dataclass, field
from datetime import date
import re
from urllib.parse import urljoin
from parsel import Selector
from govcrawler.parser.cleaner import html_to_text
from govcrawler.parser.extractor import gne_extract, trafilatura_extract
def _robust_css(sel: Selector, css: str):
"""CSS query with two sister-features against parsel's defaults:
1. Priority fallback chain (NOT union). yaml authors write
`div.zw, div.TRS_Editor, div.content` meaning "use div.zw if
present, else div.TRS_Editor, else div.content". parsel's
sel.css(comma-list) instead returns ALL matches in DOM order,
which on TRS-CMS pages picks the OUTER wrapper (div.content)
before the INNER body (div.zw) — pulling in toolbars + sidebars.
We split on commas and return the first sub-selector that hits.
2. // xpath fallback for the libxml2 axis bug. On gov.cn
/zhengce/content/* the body sits inside double-nested tables and
descendant-or-self::div[has-class('foo')] from finds
nothing, while //div[has-class('foo')] does. Each sub-selector
gets the // retry before we move on to the next.
Returns a SelectorList. Empty list if nothing matched either pass.
"""
if not css:
return sel.css("") # empty SelectorList
try:
from cssselect import HTMLTranslator
translator = HTMLTranslator()
except Exception:
translator = None
last_empty = None # remember an empty SelectorList to return on miss
for sub in css.split(","):
sub = sub.strip()
if not sub:
continue
# Pass 1: vanilla parsel CSS for this sub-selector alone.
out = sel.css(sub)
if out:
return out
last_empty = out
# Pass 2: // xpath fallback for the libxml2 axis bug.
if translator is not None:
try:
xp = translator.css_to_xpath(sub, prefix="//")
except Exception:
continue
try:
found = sel.xpath(xp)
except Exception:
found = None
if found:
return found
return last_empty if last_empty is not None else sel.css("")
MIN_CONTENT_CHARS = 100 # 主 XPath content_html 短于此值 → 触发 GNE
MIN_FALLBACK_CHARS = 50 # GNE/trafilatura 正文短于此值 → 认为失败
@dataclass
class DetailFields:
title: str
publish_time_raw: str
source: str
content_html: str
content_text: str
attachment_urls: list[str]
used_fallback: bool
fallback_engine: str | None # "gne" | "trafilatura" | None
# Count of inline with non-empty src inside content_html. Lets the
# pipeline mark image-only articles (gov 通知 with photos, no body text)
# as ready instead of content_text_too_short.
inline_image_count: int = 0
index_no: str | None = None
publisher: str | None = None
doc_no: str | None = None
publish_date: date | None = None
effective_date: date | None = None
is_effective: bool | None = None
expiry_date: date | None = None
topic_words: str | None = None
open_category: str | None = None
content_category: str | None = None
content_subcategory: str | None = None
public_meta: dict[str, str] = field(default_factory=dict)
def _clean_meta_text(v: str | None) -> str | None:
if v is None:
return None
cleaned = re.sub(r"\s+", " ", v).strip()
return cleaned or None
def _clean_source_text(v: str | None) -> str | None:
cleaned = _clean_meta_text(v)
if not cleaned:
return None
for prefix in ("文章来源：", "信息来源：", "来源：", "来源:", "发布机构："):
if cleaned.startswith(prefix):
return _clean_meta_text(cleaned[len(prefix):])
return cleaned
def _date_from_meta_text(v: str | None) -> date | None:
cleaned = _clean_meta_text(v)
if not cleaned:
return None
m = re.search(r"(\d{4})[-/.年](\d{1,2})[-/.月](\d{1,2})", cleaned)
if not m:
return None
return date(int(m.group(1)), int(m.group(2)), int(m.group(3)))
def _dates_from_meta_text(v: str | None) -> list[date]:
cleaned = _clean_meta_text(v)
if not cleaned:
return []
out: list[date] = []
for m in re.finditer(r"(\d{4})[-/.年](\d{1,2})[-/.月](\d{1,2})", cleaned):
try:
out.append(date(int(m.group(1)), int(m.group(2)), int(m.group(3))))
except ValueError:
continue
return out
def _is_effective_from_meta_text(v: str | None) -> bool | None:
cleaned = _clean_meta_text(v)
if not cleaned:
return None
compact = cleaned.replace(" ", "")
if any(word in compact for word in ("已废止", "废止", "已失效", "失效", "无效", "停止执行", "不再有效")):
return False
if any(word in compact for word in ("现行有效", "继续有效", "有效")):
return True
return None
def _split_open_category(v: str | None) -> tuple[str | None, str | None, str | None]:
cleaned = _clean_meta_text(v)
if not cleaned:
return None, None, None
cleaned = cleaned.replace("\\", "、")
parts = [p.strip() for p in cleaned.split("、") if p.strip()]
if len(parts) <= 1:
return cleaned, None, cleaned
return "、".join(parts[:-1]), parts[-1], cleaned
_PUBLIC_META_LABELS = {
"索引号": "索引号",
"主题分类": "分类",
"分类": "分类",
"发布机构": "发布机构",
"发文机关": "发布机构",
"文号": "文号",
"发文字号": "文号",
"成文日期": "成文日期",
"生效日期": "生效日期",
"施行日期": "生效日期",
"实施日期": "生效日期",
"执行日期": "生效日期",
"效力状态": "效力状态",
"时效状态": "效力状态",
"有效性": "效力状态",
"失效日期": "失效日期",
"废止日期": "失效日期",
"有效期至": "失效日期",
"有效期": "有效期",
"发布日期": "发布日期",
"主题词": "主题词",
}
def _canonical_public_label(v: str | None) -> str | None:
cleaned = _clean_meta_text(v)
if not cleaned:
return None
compact = cleaned.rstrip(":：").replace(" ", "")
return _PUBLIC_META_LABELS.get(compact)
def _extract_gov_public_meta(sel: Selector) -> dict[str, object]:
"""Extract common government public-document metadata tables.
gd.gov.cn/wjk renders these as `...`;
gkmlpt renders a similar table with label/value `
` pairs. Keep this
parser generic so YAML-based targets can enrich normalized Article fields
without writing a bespoke adapter for every static TRS template.
"""
pairs: dict[str, str] = {}
# Shape A: ...
for label in sel.xpath("//label"):
key = _canonical_public_label(" ".join(label.xpath(".//text()").getall()))
if not key:
continue
value = _clean_meta_text(" ".join(label.xpath("following-sibling::span[1]//text()").getall()))
if key and value:
pairs[key] = value
# Shape B:
索引号：
...
for tr in sel.xpath("//tr"):
cells = [_clean_meta_text(" ".join(td.xpath(".//text()").getall())) for td in tr.xpath("./td")]
cells = [c for c in cells if c]
for idx in range(0, len(cells) - 1, 2):
key = _canonical_public_label(cells[idx])
value = cells[idx + 1]
if key and value and _canonical_public_label(value) is None:
pairs.setdefault(key, value)
# Shape C: mobile gov.cn table clone:
#