simon
85ba218f2f
- run_step: 补跑保护——历史日期跳过 crawler(首页只含当天内容)并告警,复用已有 raw 数据 - run_step: report 步骤改用传入的 date_str,不再硬编码 date.today() - 新增 4 个补跑相关测试,修正 3 个既有测试适配新行为 - 全量 259 passed(3 个 crawler 基线失败与本次无关)
349 lines
13 KiB
Python
349 lines
13 KiB
Python
"""增量处理与 pipeline 断点续跑测试。
|
|
|
|
覆盖:
|
|
- M2/M4/M5 脚本的「产物存在即跳过」过滤逻辑
|
|
- scheduler.pipeline 断点状态记录与 --resume 续跑逻辑
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# M4 / M5: 产物存在即跳过
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
def test_llm_filter_existing_skips_done(tmp_path: Path) -> None:
|
|
"""M4:输出目录已有 {url_hash}.json 的输入被过滤,不重复调用 LLM API。"""
|
|
from scripts.run_event_extraction import _filter_existing
|
|
|
|
out_dir = tmp_path / "out"
|
|
out_dir.mkdir()
|
|
# 已处理
|
|
(out_dir / "aaa.json").write_text("{}", encoding="utf-8")
|
|
(out_dir / "ccc.json").write_text("{}", encoding="utf-8")
|
|
files = [
|
|
tmp_path / "in" / "aaa.json", # 已处理 → 跳过
|
|
tmp_path / "in" / "bbb.json", # 未处理 → 待处理
|
|
tmp_path / "in" / "ccc.json", # 已处理 → 跳过
|
|
]
|
|
pending, skipped = _filter_existing(files, out_dir)
|
|
assert skipped == 2
|
|
assert [p.stem for p in pending] == ["bbb"]
|
|
|
|
|
|
def test_llm_filter_existing_force_keeps_all(tmp_path: Path) -> None:
|
|
"""M4:--force 时不做过滤(全量重抽由调用方控制)。"""
|
|
from scripts.run_event_extraction import _filter_existing
|
|
|
|
out_dir = tmp_path / "out"
|
|
out_dir.mkdir()
|
|
(out_dir / "aaa.json").write_text("{}", encoding="utf-8")
|
|
files = [tmp_path / "in" / "aaa.json"]
|
|
# _filter_existing 本身不含 force 逻辑,验证在 force 下不会被调用:
|
|
# 直接验证「已存在也被返回」需由上层跳过调用,这里仅确认过滤函数行为。
|
|
pending, skipped = _filter_existing(files, out_dir)
|
|
assert skipped == 1
|
|
assert pending == []
|
|
|
|
|
|
def test_embedding_filter_existing_skips_done(tmp_path: Path) -> None:
|
|
"""M5:输出目录已有 {url_hash}.json 的输入被过滤,不重复调用 embed API。"""
|
|
from scripts.run_embedding import _filter_existing
|
|
|
|
out_dir = tmp_path / "emb"
|
|
out_dir.mkdir()
|
|
(out_dir / "h1.json").write_text("{}", encoding="utf-8")
|
|
files = [
|
|
(tmp_path / "in" / "h1.json", "event"), # 已处理 → 跳过
|
|
(tmp_path / "in" / "h2.json", "event"), # 未处理 → 待处理
|
|
]
|
|
pending, skipped = _filter_existing(files, out_dir)
|
|
assert skipped == 1
|
|
assert [p.stem for p, _ in pending] == ["h2"]
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# M2: 已提取文章跳过
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
def test_extractor_process_source_day_skips_existing(tmp_path: Path) -> None:
|
|
"""M2:输出目录已有产物的记录被跳过提取,且 index 回补完整。"""
|
|
from scripts.run_extractor import _process_source_day
|
|
|
|
raw_dir = tmp_path / "raw" / "cls" / "20260616"
|
|
raw_dir.mkdir(parents=True)
|
|
# 两条 raw 记录(url_hash 与产物文件名一致)
|
|
rec1 = {"source_id": "cls", "url": "https://a/1", "url_hash": "aaa1111111111111",
|
|
"stage": "article", "success": True, "html_file": "aaa1111111111111.html"}
|
|
rec2 = {"source_id": "cls", "url": "https://b/2", "url_hash": "bbb2222222222222",
|
|
"stage": "article", "success": True, "html_file": "bbb2222222222222.html"}
|
|
with (raw_dir / "index.jsonl").open("a", encoding="utf-8") as f:
|
|
f.write(json.dumps(rec1, ensure_ascii=False) + "\n")
|
|
f.write(json.dumps(rec2, ensure_ascii=False) + "\n")
|
|
|
|
out_dir = tmp_path / "proc" / "cls" / "20260616"
|
|
out_dir.mkdir(parents=True)
|
|
# 预置一条已有产物(视为已提取)
|
|
article = {
|
|
"source_id": "cls", "url": "https://a/1", "url_hash": "aaa1111111111111",
|
|
"title": "已有", "content": "内容", "word_count": 2,
|
|
}
|
|
(out_dir / "aaa1111111111111.json").write_text(
|
|
json.dumps(article, ensure_ascii=False), encoding="utf-8"
|
|
)
|
|
|
|
# 另一条无 html 文件 → 提取失败(但不影响跳过逻辑断言)
|
|
succ, total, skipped = _process_source_day(
|
|
"cls", "20260616", tmp_path / "raw", tmp_path / "proc"
|
|
)
|
|
assert total == 2
|
|
assert skipped == 1 # 已有产物被跳过
|
|
assert succ == 0 # 另一条因 html 缺失提取失败
|
|
# index 回补了被跳过条目的行
|
|
idx = (out_dir / "index.jsonl").read_text(encoding="utf-8").strip()
|
|
assert "aaa1111111111111" in idx
|
|
|
|
|
|
def test_extractor_process_source_day_force_rebuilds(tmp_path: Path) -> None:
|
|
"""M2:--force 时不做跳过,并重建 index。"""
|
|
from scripts.run_extractor import _process_source_day
|
|
|
|
raw_dir = tmp_path / "raw" / "cls" / "20260616"
|
|
raw_dir.mkdir(parents=True)
|
|
rec = {"source_id": "cls", "url": "https://a/1", "url_hash": "aaa1111111111111",
|
|
"stage": "article", "success": True, "html_file": "aaa1111111111111.html"}
|
|
with (raw_dir / "index.jsonl").open("a", encoding="utf-8") as f:
|
|
f.write(json.dumps(rec, ensure_ascii=False) + "\n")
|
|
|
|
out_dir = tmp_path / "proc" / "cls" / "20260616"
|
|
out_dir.mkdir(parents=True)
|
|
(out_dir / "aaa1111111111111.json").write_text("{}", encoding="utf-8")
|
|
(out_dir / "index.jsonl").write_text("旧内容", encoding="utf-8")
|
|
|
|
succ, total, skipped = _process_source_day(
|
|
"cls", "20260616", tmp_path / "raw", tmp_path / "proc", force=True
|
|
)
|
|
assert skipped == 0
|
|
# force 模式重建 index(旧内容被清掉;此处无 html 提取失败,index 为空或不存在)
|
|
idx = out_dir / "index.jsonl"
|
|
assert not idx.exists() or idx.read_text(encoding="utf-8") == ""
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# pipeline 断点状态与 --resume
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
@pytest.fixture
|
|
def fake_subprocess(monkeypatch: pytest.MonkeyPatch):
|
|
"""mock subprocess.run,按步骤名返回 returncode,并记录调用顺序。"""
|
|
from scheduler import pipeline
|
|
|
|
calls: list[str] = []
|
|
|
|
def _step_name(cmd: list[str]) -> str:
|
|
"""从命令中提取脚本名,如 scripts.run_extractor → run_extractor。"""
|
|
return next(c.split(".")[-1] for c in cmd if "scripts.run_" in c)
|
|
|
|
def _fake_run(cmd, timeout=None): # noqa: ARG001
|
|
calls.append(_step_name(cmd))
|
|
return SimpleNamespace(returncode=0)
|
|
|
|
monkeypatch.setattr(pipeline.subprocess, "run", _fake_run)
|
|
return calls
|
|
|
|
|
|
def _run_with_steps(monkeypatch: pytest.MonkeyPatch, failures: set[str]):
|
|
"""构造 run_step:指定步骤(如 'extractor')返回失败。"""
|
|
from scheduler import pipeline
|
|
|
|
def _fake_run(cmd, timeout=None): # noqa: ARG001
|
|
name = next(c.split(".")[-1] for c in cmd if "scripts.run_" in c)
|
|
name = name.replace("run_", "") # run_extractor → extractor
|
|
return SimpleNamespace(returncode=1 if name in failures else 0)
|
|
|
|
monkeypatch.setattr(pipeline.subprocess, "run", _fake_run)
|
|
|
|
|
|
def test_pipeline_records_state(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""全量运行后状态文件按日期记录每个步骤的 ok/failed。"""
|
|
from scheduler import pipeline
|
|
|
|
_run_with_steps(monkeypatch, failures={"extractor"})
|
|
state_path = tmp_path / "state.json"
|
|
steps = ["extractor", "dedup", "llm"]
|
|
pipeline.run_pipeline("20260616", steps=steps, state_path=state_path)
|
|
|
|
state = pipeline._load_pipeline_state(state_path)
|
|
day = state["20260616"]
|
|
assert day["extractor"]["status"] == "failed"
|
|
assert day["dedup"]["status"] == "ok"
|
|
assert day["llm"]["status"] == "ok"
|
|
# dedup 返回 1 被特判为成功,故用 extractor 制造失败
|
|
assert day["extractor"]["exit_code"] == 1
|
|
|
|
|
|
def test_pipeline_resume_skips_success_prefix(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
"""resume 跳过连续成功步骤,从失败步骤继续。"""
|
|
from scheduler import pipeline
|
|
|
|
state_path = tmp_path / "state.json"
|
|
# 预置状态:extractor 失败,dedup/llm 成功(模拟上次运行)
|
|
state = {"20260616": {
|
|
"extractor": {"status": "failed", "exit_code": 1},
|
|
"dedup": {"status": "ok", "exit_code": 0},
|
|
"llm": {"status": "ok", "exit_code": 0},
|
|
}}
|
|
pipeline._save_pipeline_state(state, state_path)
|
|
|
|
calls: list[str] = []
|
|
|
|
def _fake_run(cmd, timeout=None): # noqa: ARG001
|
|
name = next(c.split(".")[-1] for c in cmd if "scripts.run_" in c)
|
|
calls.append(name)
|
|
return SimpleNamespace(returncode=0)
|
|
|
|
monkeypatch.setattr(pipeline.subprocess, "run", _fake_run)
|
|
steps = ["extractor", "dedup", "llm"]
|
|
result = pipeline.run_pipeline(
|
|
"20260616", steps=steps, resume=True, state_path=state_path
|
|
)
|
|
# 从 extractor 开始重跑全部(extractor 之后的 dedup/llm 需重跑以覆盖降级数据)
|
|
assert calls == ["run_extractor", "run_dedup", "run_event_extraction"]
|
|
assert all(s.success for s in result.steps)
|
|
|
|
|
|
def test_pipeline_resume_all_done_noop(tmp_path: Path) -> None:
|
|
"""resume 且所有步骤均已成功时,不执行任何步骤。"""
|
|
from scheduler import pipeline
|
|
|
|
state_path = tmp_path / "state.json"
|
|
state = {"20260616": {
|
|
"extractor": {"status": "ok", "exit_code": 0},
|
|
"dedup": {"status": "ok", "exit_code": 0},
|
|
"llm": {"status": "ok", "exit_code": 0},
|
|
}}
|
|
pipeline._save_pipeline_state(state, state_path)
|
|
|
|
steps = ["extractor", "dedup", "llm"]
|
|
result = pipeline.run_pipeline(
|
|
"20260616", steps=steps, resume=True, state_path=state_path
|
|
)
|
|
assert result.steps == []
|
|
assert result.all_success # 空步骤视为成功
|
|
|
|
|
|
def test_pipeline_resume_missing_step_starts_from_first_missing(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""resume:部分步骤无历史记录时,从首个缺失步骤开始。"""
|
|
from scheduler import pipeline
|
|
|
|
state_path = tmp_path / "state.json"
|
|
state = {"20260616": {
|
|
"extractor": {"status": "ok", "exit_code": 0},
|
|
}}
|
|
pipeline._save_pipeline_state(state, state_path)
|
|
|
|
idx = pipeline._resume_start_index(
|
|
["extractor", "dedup", "llm"], "20260616",
|
|
pipeline._load_pipeline_state(state_path),
|
|
)
|
|
assert idx == 1 # dedup 缺失 → 从它开始
|
|
|
|
|
|
def test_once_rejects_resume_with_steps(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""--resume 与 --steps 同时使用时报错。"""
|
|
import scripts.run_scheduler as rs
|
|
|
|
args = SimpleNamespace(steps="crawler,extractor", resume=True,
|
|
date="20260616")
|
|
rc = rs._once(args)
|
|
assert rc == 2
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# P0-3: 补跑模式跳过抓取 / P0-4: report 使用传入日期
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
def test_crawler_backfill_skips_past_date(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""补跑历史日期时跳过 crawler(首页只含当天内容),不执行子进程。"""
|
|
from scheduler import pipeline
|
|
|
|
def _fail_run(cmd, timeout=None): # noqa: ARG001
|
|
raise AssertionError("补跑模式不应执行子进程")
|
|
|
|
monkeypatch.setattr(pipeline.subprocess, "run", _fail_run)
|
|
sr = pipeline.run_step("crawler", "20260101")
|
|
assert sr.success is True
|
|
assert "跳过" in sr.tail_msg
|
|
assert sr.elapsed_sec == 0.0
|
|
|
|
|
|
def test_crawler_today_runs_normally(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""当天日期正常执行 crawler(不受补跑保护影响)。"""
|
|
from datetime import date
|
|
|
|
from scheduler import pipeline
|
|
|
|
called: list[str] = []
|
|
|
|
def _fake_run(cmd, timeout=None): # noqa: ARG001
|
|
called.append(" ".join(cmd))
|
|
return SimpleNamespace(returncode=0)
|
|
|
|
monkeypatch.setattr(pipeline.subprocess, "run", _fake_run)
|
|
sr = pipeline.run_step("crawler", date.today().strftime("%Y%m%d"))
|
|
assert sr.success is True
|
|
assert called and "run_crawler" in called[0]
|
|
|
|
|
|
def test_report_step_uses_date_str(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""report 步骤必须使用传入的 date_str,而非 date.today()(P0-4)。"""
|
|
from scheduler import pipeline
|
|
from scheduler import reporter
|
|
|
|
received: list[str] = []
|
|
|
|
def _fake_generate(day_str, upload=True): # noqa: ARG001
|
|
received.append(day_str)
|
|
return 12345
|
|
|
|
monkeypatch.setattr(reporter, "generate_report", _fake_generate)
|
|
sr = pipeline.run_step("report", "20260101")
|
|
assert sr.success is True
|
|
assert received == ["20260101"]
|
|
|
|
|
|
def test_pipeline_backfill_skips_crawler_keeps_rest(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
"""补跑历史日期:全链路中仅 crawler 跳过,其余步骤照常执行。"""
|
|
from scheduler import pipeline
|
|
|
|
calls: list[str] = []
|
|
|
|
def _fake_run(cmd, timeout=None): # noqa: ARG001
|
|
name = next(c.split(".")[-1] for c in cmd if "scripts.run_" in c)
|
|
calls.append(name)
|
|
return SimpleNamespace(returncode=0)
|
|
|
|
monkeypatch.setattr(pipeline.subprocess, "run", _fake_run)
|
|
state_path = tmp_path / "state.json"
|
|
result = pipeline.run_pipeline(
|
|
"20260101",
|
|
steps=["crawler", "extractor", "dedup"],
|
|
state_path=state_path,
|
|
)
|
|
# crawler 跳过(未执行子进程),extractor/dedup 正常
|
|
assert calls == ["run_extractor", "run_dedup"]
|
|
assert all(s.success for s in result.steps)
|
|
state = pipeline._load_pipeline_state(state_path)
|
|
assert state["20260101"]["crawler"]["status"] == "ok"
|