simon
60f8c263f2
从 CCTV 主页抓取《新闻联播》，下载 → 转 MP3/WAV → 静音切分 → ASR 识别 → LLM 校对 → 切分为单条新闻 → 入库 MySQL。 主要内容： - 全链路：getVideo5 抓取下载、audioRead 转写、deepseek 校对与切分、newsProcess 入库 - 可移植化：配置分层，.env 只放密钥、config.yml 放模型/接入点/路由/参数 - 可换供应商：endpoints（kind/base_url/api_key_env/extra_body）+ routes 按环节选路 - 数据保真：数值事实守卫，校对改动数字/年份/届次则整片回退 ASR 原文； 识别不完整不发布该日精编，避免半天内容被当成完整一天 - 定时任务：systemd 每天 21:00，失败 21:30 / 22:00 重试； 只缺切分时只重跑切分（省掉全部 ASR），用 state/.asr_complete_* 标记判定阶段 - 隧道自愈：13306 不通时自动执行 autossh.sh（所有入口共用，systemd 托管时只等待） - 中间产物每日清理；97 项离线自检（配置/清理/事实守卫/解析/隧道）
106 lines
3.9 KiB
Python
106 lines
3.9 KiB
Python
"""清理逻辑回归测试（不依赖 pytest，且**不触碰真实目录**）
|
||
|
||
python tests/test_cleanup.py
|
||
|
||
做法：用 XWLB_CONFIG_FILE 指向临时 config.yml，把 paths 指向临时目录，
|
||
验证 cleanup 段各开关的组合行为。
|
||
"""
|
||
import os
|
||
import sys
|
||
import tempfile
|
||
from pathlib import Path
|
||
|
||
ROOT = Path(__file__).resolve().parent.parent
|
||
sys.path.insert(0, str(ROOT))
|
||
|
||
|
||
def build_env(tmp, after_daily_run=1, only_on_success=1, remove_video=1, remove_audio=1):
|
||
"""在临时目录生成 config.yml，并让 config 模块加载它；返回 (video_dir, audio_dir)"""
|
||
video_dir = Path(tmp) / 'video'
|
||
audio_dir = Path(tmp) / 'audio'
|
||
video_dir.mkdir(parents=True, exist_ok=True)
|
||
audio_dir.mkdir(parents=True, exist_ok=True)
|
||
for name in ('a.mp3', 'b.mp4', 'keep.txt'):
|
||
(video_dir / name).write_bytes(b'x' * 10)
|
||
for name in ('c.wav', 'd.wav', 'keep.txt'):
|
||
(audio_dir / name).write_bytes(b'x' * 10)
|
||
|
||
cfg = Path(tmp) / 'config.yml'
|
||
cfg.write_text(f"""
|
||
paths:
|
||
video_dir: {video_dir}
|
||
audio_dir: {audio_dir}
|
||
cleanup:
|
||
after_daily_run: {after_daily_run}
|
||
only_on_success: {only_on_success}
|
||
remove_video: {remove_video}
|
||
remove_audio: {remove_audio}
|
||
""", encoding='utf-8')
|
||
os.environ['XWLB_CONFIG_FILE'] = str(cfg)
|
||
import config
|
||
config.reload_config()
|
||
return video_dir, audio_dir
|
||
|
||
|
||
def snapshot(video_dir, audio_dir):
|
||
return sorted(p.name for p in list(video_dir.iterdir()) + list(audio_dir.iterdir()))
|
||
|
||
|
||
def main():
|
||
failed = 0
|
||
|
||
def check(desc, got, want):
|
||
nonlocal failed
|
||
if got == want:
|
||
print(f" ✓ {desc}: {got}")
|
||
else:
|
||
failed += 1
|
||
print(f" ✗ {desc}: 期望 {want}，实际 {got}")
|
||
|
||
import cleanup
|
||
|
||
# 1) 成功 + 开关全开 → 删除 mp3/mp4/wav，保留其他文件
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
video_dir, audio_dir = build_env(tmp)
|
||
result = cleanup.maybe_cleanup_after_run('2026-09-04', day_ok=True)
|
||
check('成功时删除 4 个中间产物', result['removed'], 4)
|
||
check('非中间产物保留', snapshot(video_dir, audio_dir), ['keep.txt', 'keep.txt'])
|
||
check('清理字节数>0', result['bytes'] > 0, True)
|
||
|
||
# 2) 失败 + only_on_success=1（默认）→ 保留文件便于重跑
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
video_dir, audio_dir = build_env(tmp, only_on_success=1)
|
||
result = cleanup.maybe_cleanup_after_run('2026-09-04', day_ok=False)
|
||
check('失败时保留文件', result['skipped'], True)
|
||
check('文件未被删除', len(snapshot(video_dir, audio_dir)), 6)
|
||
|
||
# 3) 失败 + only_on_success=0 → 仍然清理
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
video_dir, audio_dir = build_env(tmp, only_on_success=0)
|
||
result = cleanup.maybe_cleanup_after_run('2026-09-04', day_ok=False)
|
||
check('only_on_success=0 时失败也清理', result['removed'], 4)
|
||
|
||
# 4) after_daily_run=0 → 完全不动
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
video_dir, audio_dir = build_env(tmp, after_daily_run=0)
|
||
result = cleanup.maybe_cleanup_after_run('2026-09-04', day_ok=True)
|
||
check('after_daily_run=0 时跳过', result['skipped'], True)
|
||
check('文件未被删除', len(snapshot(video_dir, audio_dir)), 6)
|
||
|
||
# 5) 只关 wav 清理 → 只删 mp3/mp4
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
video_dir, audio_dir = build_env(tmp, remove_audio=0)
|
||
result = cleanup.maybe_cleanup_after_run('2026-09-04', day_ok=True)
|
||
check('remove_audio=0 时只删视频', result['removed'], 2)
|
||
check('wav 保留、mp3/mp4 已删', snapshot(video_dir, audio_dir),
|
||
['c.wav', 'd.wav', 'keep.txt', 'keep.txt'])
|
||
|
||
os.environ.pop('XWLB_CONFIG_FILE', None)
|
||
import config
|
||
config.reload_config()
|
||
print(f"\n结果: {'全部通过' if failed == 0 else f'{failed} 项失败'}")
|
||
return 1 if failed else 0
|
||
|
||
|
||
if __name__ == '__main__':
|
||
sys.exit(main()) |