simon
366e60e8a9
- 新增 report_db 包: MySQL 连接/建表/幂等写入 (news_report/news_event, myquant 库) - 新增 report_import 包: 历史 178 份日报 HTML 解析入库, 表头驱动列映射 - reporter.py 完全切换: generate_report 结构化入库, 不再生成/上传 HTML - CLI: 新增 report-import 子命令 - 依赖: uv add pymysql; 配置: NEWS_DB_* / REPORT_HISTORY_DIR - 文档: docs/report_db_design.md(实现逻辑), docs/db_schema.md(表结构供 API/前端) - 测试: 24 个单测通过 (parser/builder/models/importer)
90 lines
3.1 KiB
Python
90 lines
3.1 KiB
Python
"""历史日报批量导入（解析 → 入库，幂等）。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
from dataclasses import dataclass, field
|
|
from pathlib import Path
|
|
|
|
from loguru import logger
|
|
|
|
from report_db import connect, exists_report, save_report
|
|
from report_import.parser import ReportParseError, parse_report
|
|
|
|
_REPORT_FILE_RE = re.compile(r"([a-z]+)_news_daily_\d{8}(?:_\d{4,6})?\.html$")
|
|
|
|
|
|
@dataclass
|
|
class ImportStats:
|
|
"""一次导入的统计结果。"""
|
|
|
|
scanned: int = 0 # 扫描到的日报文件数
|
|
imported: int = 0 # 新入库
|
|
skipped: int = 0 # 已存在（幂等跳过）
|
|
failed: int = 0 # 解析失败
|
|
errors: list[str] = field(default_factory=list)
|
|
|
|
|
|
def _match_file(path: Path, date_str: str | None, report_type: str | None) -> bool:
|
|
"""按文件名判断是否属于本次导入范围。"""
|
|
m = _REPORT_FILE_RE.search(path.name)
|
|
if not m:
|
|
return False
|
|
if report_type is not None and m.group(1) != report_type:
|
|
return False
|
|
return date_str is None or date_str in path.name
|
|
|
|
|
|
def import_history(
|
|
report_dir: str | Path,
|
|
date_str: str | None = None,
|
|
report_type: str | None = None,
|
|
*,
|
|
force: bool = False,
|
|
) -> ImportStats:
|
|
"""扫描 {report_dir}/{YYYYMMDD}/ 下全部 `*_news_daily_*.html` 并入库。
|
|
|
|
- 幂等：主表唯一键 (report_date, report_type, file_name) 已存在则跳过；
|
|
- `force=True` 时跳过存在性检查，直接覆盖重导；
|
|
- 单文件解析失败不影响其他文件。
|
|
"""
|
|
stats = ImportStats()
|
|
root = Path(report_dir)
|
|
if not root.is_dir():
|
|
logger.error("日报目录不存在: {}", root)
|
|
raise FileNotFoundError(f"日报目录不存在: {root}")
|
|
|
|
files = sorted(p for p in root.glob("*/[a-z]*_news_daily_*.html") if _match_file(p, date_str, report_type))
|
|
stats.scanned = len(files)
|
|
logger.info("扫描到日报文件 {} 份: {}", stats.scanned, root)
|
|
|
|
conn = connect()
|
|
try:
|
|
for path in files:
|
|
try:
|
|
html = path.read_text(encoding="utf-8")
|
|
report = parse_report(html, path.name)
|
|
except ReportParseError as e:
|
|
stats.failed += 1
|
|
stats.errors.append(f"{path.name}: {e}")
|
|
logger.warning("解析失败: {} ({})", path.name, e)
|
|
continue
|
|
except Exception as e: # 防御未知异常，不中断批量
|
|
stats.failed += 1
|
|
stats.errors.append(f"{path.name}: {type(e).__name__}: {e}")
|
|
logger.exception("读取/解析异常: {}", path.name)
|
|
continue
|
|
|
|
if not force and exists_report(conn, report.report_date, report.report_type, report.file_name):
|
|
stats.skipped += 1
|
|
logger.debug("已存在, 跳过: {}", path.name)
|
|
continue
|
|
save_report(conn, report)
|
|
stats.imported += 1
|
|
finally:
|
|
conn.close()
|
|
|
|
logger.info("导入完成: scanned={} imported={} skipped={} failed={}",
|
|
stats.scanned, stats.imported, stats.skipped, stats.failed)
|
|
return stats
|