357 lines
12 KiB
Python
357 lines
12 KiB
Python
"""M2 批量正文提取入口脚本。
|
|
|
|
输入: data/raw/{source}/{YYYYMMDD}/index.jsonl (M1 产物)
|
|
输出: data/processed/{source}/{YYYYMMDD}/{url_hash}.json (Article)
|
|
data/processed/{source}/{YYYYMMDD}/index.jsonl (扁平元数据,便于检索)
|
|
|
|
用法:
|
|
uv run python -m scripts.run_extractor # 处理今日所有源
|
|
uv run python -m scripts.run_extractor --date 20260616
|
|
uv run python -m scripts.run_extractor --source sina --date 20260616
|
|
uv run python -m scripts.run_extractor --raw-root data/raw --out-root data/processed
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import sys
|
|
from datetime import date, datetime
|
|
from pathlib import Path
|
|
|
|
from loguru import logger
|
|
|
|
from extractor import Article, ExtractError, extract_article
|
|
from extractor.parser import _url_hash
|
|
|
|
|
|
def _setup_logger(level: str) -> None:
|
|
logger.remove()
|
|
logger.add(
|
|
sys.stderr,
|
|
level=level,
|
|
format="{time:YYYY-MM-DD HH:mm:ss} | {level} | {name} | {message}",
|
|
)
|
|
log_path = Path("logs") / "extractor.log"
|
|
log_path.parent.mkdir(parents=True, exist_ok=True)
|
|
logger.add(log_path, level="DEBUG", rotation="10 MB", retention=5, encoding="utf-8")
|
|
|
|
|
|
def _iter_article_records(raw_dir: Path) -> list[dict]:
|
|
"""读取 M1 产物的 index.jsonl,返回所有可处理的记录。
|
|
|
|
新闻源: stage=article + success + html_file
|
|
cninfo: 有 json_file 字段(CninfoItem 格式)
|
|
"""
|
|
index_path = raw_dir / "index.jsonl"
|
|
if not index_path.is_file():
|
|
return []
|
|
out: list[dict] = []
|
|
with index_path.open("r", encoding="utf-8") as f:
|
|
for line in f:
|
|
line = line.strip()
|
|
if not line:
|
|
continue
|
|
try:
|
|
rec = json.loads(line)
|
|
except json.JSONDecodeError as e:
|
|
logger.warning("跳过非法 jsonl 行 in {}: {}", index_path, e)
|
|
continue
|
|
# cninfo: 新格式(CninfoItem → json_file)
|
|
if rec.get("source_id") == "cninfo" or rec.get("json_file"):
|
|
if rec.get("json_file"):
|
|
out.append(rec)
|
|
continue
|
|
# 新闻源: 旧格式(CrawlResult → html_file)
|
|
if rec.get("stage") == "article" and rec.get("success") and rec.get("html_file"):
|
|
out.append(rec)
|
|
return out
|
|
|
|
|
|
def _process_one(rec: dict, raw_dir: Path, out_dir: Path,
|
|
body_xpath_map: dict[str, str] | None = None) -> Article | None:
|
|
"""处理单条记录,失败返回 None。cninfo 等结构化源跳过 GNE。"""
|
|
src_id = rec.get("source_id", "")
|
|
|
|
# cninfo: 新格式(json_file)或旧格式(html_file+meta JSON)直接解析
|
|
json_file = rec.get("json_file", "")
|
|
html_file = rec.get("html_file", "")
|
|
is_cninfo = (src_id == "cninfo" or bool(json_file))
|
|
|
|
if is_cninfo:
|
|
if json_file:
|
|
json_path = raw_dir / json_file
|
|
if json_path.is_file():
|
|
return _process_cninfo_v2(rec, json_path, out_dir)
|
|
else:
|
|
logger.warning("cninfo JSON 文件丢失: {}", json_path)
|
|
return None
|
|
# 旧格式兼容: html_file + <meta>JSON
|
|
if html_file:
|
|
html_path = raw_dir / html_file
|
|
if html_path.is_file():
|
|
return _process_cninfo(rec, html_path, out_dir)
|
|
return None
|
|
|
|
html_path = raw_dir / html_file
|
|
if not html_path or not html_path.is_file():
|
|
logger.warning("HTML 文件丢失: {}", html_path)
|
|
return None
|
|
|
|
html = html_path.read_text(encoding="utf-8", errors="ignore")
|
|
extra_config: dict[str, str] = {}
|
|
if body_xpath_map and src_id in body_xpath_map:
|
|
extra_config["body_xpath"] = body_xpath_map[src_id]
|
|
try:
|
|
article = extract_article(
|
|
html=html,
|
|
source_id=src_id,
|
|
url=rec["url"],
|
|
extra_config=extra_config or None,
|
|
)
|
|
except ExtractError as e:
|
|
logger.warning("提取失败 {} {}: {}", rec["source_id"], rec["url"], e.reason)
|
|
return None
|
|
|
|
return _save_article(article, out_dir)
|
|
|
|
|
|
def _process_cninfo(rec: dict, html_path: Path, out_dir: Path) -> Article | None:
|
|
"""处理 cninfo 公告记录:从 meta JSON 解析结构化数据。"""
|
|
import re
|
|
html = html_path.read_text(encoding="utf-8", errors="ignore")
|
|
# 提取 <meta>{...}</meta> 中的 JSON
|
|
m = re.search(r"<meta>(.+?)</meta>", html, re.DOTALL)
|
|
if not m:
|
|
logger.warning("cninfo HTML 不含 meta JSON: {}", html_path)
|
|
return None
|
|
try:
|
|
meta = json.loads(m.group(1))
|
|
except json.JSONDecodeError:
|
|
logger.warning("cninfo meta JSON 解析失败: {}", html_path)
|
|
return None
|
|
|
|
sec_code = (meta.get("secCode") or "").strip()
|
|
sec_name = (meta.get("secName") or "").strip()
|
|
ann_type = (meta.get("announcementType") or "").strip()
|
|
pdf_url = (meta.get("pdfUrl") or "").strip()
|
|
title = rec.get("title") or meta.get("title") or ""
|
|
|
|
# 构建正文:结构化摘要 + 公告类别翻译
|
|
content_parts = [f"公司: {sec_name}({sec_code})", f"公告标题: {title}"]
|
|
if ann_type:
|
|
content_parts.append(f"公告类别编码: {ann_type}")
|
|
if pdf_url:
|
|
content_parts.append(f"PDF: {pdf_url}")
|
|
content = "\n".join(content_parts)
|
|
|
|
# 时间
|
|
fetched = rec.get("fetched_at")
|
|
publish_time = None
|
|
if isinstance(fetched, str):
|
|
try:
|
|
from datetime import datetime as dt
|
|
publish_time = dt.fromisoformat(fetched)
|
|
except ValueError:
|
|
pass
|
|
|
|
article = Article(
|
|
source_id="cninfo",
|
|
url=rec.get("url") or "",
|
|
url_hash=_url_hash(rec.get("url") or ""),
|
|
title=title,
|
|
content=content,
|
|
author=sec_name,
|
|
source_name="巨潮资讯网",
|
|
publish_time=publish_time,
|
|
publish_time_raw=fetched,
|
|
word_count=len(content),
|
|
)
|
|
return _save_article(article, out_dir)
|
|
|
|
|
|
def _process_cninfo_v2(rec: dict, json_path: Path, out_dir: Path) -> Article | None:
|
|
"""处理 cninfo v2 格式: 直接读取 CninfoItem JSON 并转换为 Article。"""
|
|
try:
|
|
item_data = json.loads(json_path.read_text(encoding="utf-8"))
|
|
except (json.JSONDecodeError, OSError) as e:
|
|
logger.warning("cninfo JSON 读取失败 {}: {}", json_path, e)
|
|
return None
|
|
|
|
stock_code = item_data.get("stock_code", "")
|
|
stock_name = item_data.get("stock_name", "")
|
|
title = item_data.get("title", "")
|
|
content = item_data.get("content", "")
|
|
publish_time_str = item_data.get("publish_time", "")
|
|
item_type = item_data.get("item_type", "announcement")
|
|
url = item_data.get("url", "")
|
|
extra = item_data.get("extra", {})
|
|
|
|
# 类型中文映射
|
|
type_map = {
|
|
"announcement": "公告",
|
|
"research": "投资者调研",
|
|
"irm": "互动问答",
|
|
}
|
|
type_cn = type_map.get(item_type, item_type)
|
|
|
|
# 构建正文
|
|
content_parts = [
|
|
f"公司: {stock_name}({stock_code})",
|
|
f"类型: {type_cn}",
|
|
f"标题: {title}",
|
|
]
|
|
if extra.get("announcement_type"):
|
|
content_parts.append(f"公告类别: {extra['announcement_type']}")
|
|
if url:
|
|
content_parts.append(f"原文链接: {url}")
|
|
if content:
|
|
content_parts.append(f"\n正文:\n{content}")
|
|
full_content = "\n".join(content_parts)
|
|
|
|
# 发布时间解析
|
|
publish_time = None
|
|
publish_time_raw = publish_time_str
|
|
if publish_time_str:
|
|
try:
|
|
from datetime import datetime as dt
|
|
publish_time = dt.strptime(publish_time_str, "%Y-%m-%d")
|
|
except ValueError:
|
|
try:
|
|
publish_time = dt.fromisoformat(publish_time_str)
|
|
except ValueError:
|
|
pass
|
|
|
|
article = Article(
|
|
source_id="cninfo",
|
|
url=url,
|
|
url_hash=_url_hash(url or title),
|
|
title=title,
|
|
content=full_content,
|
|
author=stock_name,
|
|
source_name="巨潮资讯网",
|
|
publish_time=publish_time,
|
|
publish_time_raw=publish_time_raw,
|
|
word_count=len(full_content),
|
|
item_type=item_type,
|
|
)
|
|
return _save_article(article, out_dir)
|
|
|
|
|
|
def _save_article(article: Article, out_dir: Path) -> Article:
|
|
"""保存 Article JSON 并追加 index。"""
|
|
out_dir.mkdir(parents=True, exist_ok=True)
|
|
article_path = out_dir / f"{article.url_hash}.json"
|
|
article_path.write_text(article.model_dump_json(indent=2), encoding="utf-8")
|
|
flat = article.model_dump(exclude={"content", "images"}, mode="json")
|
|
flat["article_file"] = article_path.name
|
|
flat["content_preview"] = article.content[:80]
|
|
with (out_dir / "index.jsonl").open("a", encoding="utf-8") as f:
|
|
f.write(json.dumps(flat, ensure_ascii=False) + "\n")
|
|
return article
|
|
|
|
|
|
def _process_source_day(
|
|
source_id: str,
|
|
day: str,
|
|
raw_root: Path,
|
|
out_root: Path,
|
|
body_xpath_map: dict[str, str] | None = None,
|
|
) -> tuple[int, int]:
|
|
"""处理单个源单日。返回 (成功数, 总数)。"""
|
|
raw_dir = raw_root / source_id / day
|
|
out_dir = out_root / source_id / day
|
|
|
|
records = _iter_article_records(raw_dir)
|
|
if not records:
|
|
logger.info("源 {} 日期 {} 无可处理记录", source_id, day)
|
|
return 0, 0
|
|
|
|
# 清理同日旧的 index.jsonl,避免重复追加
|
|
old_index = out_dir / "index.jsonl"
|
|
if old_index.exists():
|
|
old_index.unlink()
|
|
|
|
succ = 0
|
|
for rec in records:
|
|
article = _process_one(rec, raw_dir, out_dir, body_xpath_map)
|
|
if article is not None:
|
|
succ += 1
|
|
total = len(records)
|
|
rate = succ / max(total, 1)
|
|
logger.info(
|
|
"源 {} 日期 {} 提取完成: {}/{} 成功率 {:.0%}",
|
|
source_id,
|
|
day,
|
|
succ,
|
|
total,
|
|
rate,
|
|
)
|
|
return succ, total
|
|
|
|
|
|
def _list_source_dirs(raw_root: Path) -> list[str]:
|
|
"""列出 raw_root 下所有源 id(子目录名)。"""
|
|
if not raw_root.is_dir():
|
|
return []
|
|
return sorted(p.name for p in raw_root.iterdir() if p.is_dir())
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description="A 股新闻正文提取 (M2)")
|
|
parser.add_argument("--raw-root", default="data/raw", help="M1 抓取产物根目录")
|
|
parser.add_argument("--out-root", default="data/processed", help="M2 提取结果根目录")
|
|
parser.add_argument("--source", default=None, help="只处理单个源 id,默认全部")
|
|
parser.add_argument(
|
|
"--date",
|
|
default=date.today().strftime("%Y%m%d"),
|
|
help="处理日期 YYYYMMDD,默认今日",
|
|
)
|
|
parser.add_argument("--log-level", default="INFO")
|
|
args = parser.parse_args()
|
|
|
|
_setup_logger(args.log_level)
|
|
raw_root = Path(args.raw_root)
|
|
out_root = Path(args.out_root)
|
|
|
|
sources = [args.source] if args.source else _list_source_dirs(raw_root)
|
|
if not sources:
|
|
logger.error("{} 下未发现任何源目录", raw_root)
|
|
return 2
|
|
|
|
# 加载源配置,构建 source_id → body_xpath 映射
|
|
body_xpath_map: dict[str, str] = {}
|
|
try:
|
|
from crawler.config import load_crawler_config
|
|
cfg = load_crawler_config()
|
|
for s in cfg.sources:
|
|
if s.body_xpath:
|
|
body_xpath_map[s.id] = s.body_xpath
|
|
if body_xpath_map:
|
|
logger.info("已加载 body_xpath 配置: {}", dict(body_xpath_map))
|
|
except Exception:
|
|
pass
|
|
|
|
started = datetime.now()
|
|
total_succ = 0
|
|
total_all = 0
|
|
for src in sources:
|
|
succ, total = _process_source_day(src, args.date, raw_root, out_root, body_xpath_map)
|
|
total_succ += succ
|
|
total_all += total
|
|
|
|
elapsed = (datetime.now() - started).total_seconds()
|
|
rate = total_succ / max(total_all, 1)
|
|
logger.info(
|
|
"全部完成: {}/{} 成功率 {:.0%} 用时 {:.1f}s",
|
|
total_succ,
|
|
total_all,
|
|
rate,
|
|
elapsed,
|
|
)
|
|
return 0 if rate >= 0.9 or total_all == 0 else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|