《新闻联播》每日抓取入库：全链路 + 可移植化 + 定时任务

从 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 项离线自检（配置/清理/事实守卫/解析/隧道）

tests/test_cleanup.py

@@ -0,0 +1,106 @@
+"""清理逻辑回归测试（不依赖 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())

tests/test_config.py

@@ -0,0 +1,173 @@
+"""配置层回归测试（不依赖 pytest）
+
+    python tests/test_config.py
+
+覆盖：
+- config.yml 能被加载，关键配置项存在且类型正确；
+- 点号取值 / 布尔与整数转换；
+- 目录解析（相对路径基于项目根）；
+- 环境变量覆盖 config.yml（临时试验用）；
+- 敏感项不在 config.yml 里（只能在 .env）。
+"""
+import os
+import sys
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parent.parent
+sys.path.insert(0, str(ROOT))
+
+import yaml  # noqa: E402
+
+import config  # noqa: E402
+
+# 测试不硬编码"当前配的是哪个模型/哪条路由"——那是用户随时会改的。
+# 这里直接用 yaml 独立读一遍 config.yml，与 config.py 的解析结果对比：
+# 校验的是"解析机制正确"，而不是"值恰好等于某个名字"。
+with open(ROOT / 'config.yml', encoding='utf-8') as _f:
+    RAW = yaml.safe_load(_f)
+
+def raw(dotted, default=None):
+    node = RAW
+    for key in dotted.split('.'):
+        if not isinstance(node, dict) or key not in node:
+            return default
+        node = node[key]
+    return node
+
+# 按 config.yml 自己的 routes/endpoints 推导"应要求哪些密钥"（独立算法）
+_expected_secrets = {'MYSQL_PASSWORD'}
+for _role in ('asr', 'correct', 'split'):
+    _ep = (RAW.get('endpoints') or {}).get((RAW.get('routes') or {}).get(_role), {})
+    if _ep.get('api_key_env'):
+        _expected_secrets.add(_ep['api_key_env'])
+
+
+def main():
+    failed = 0
+
+    def check(desc, got, want):
+        nonlocal failed
+        if got == want:
+            print(f"  ✓ {desc}: {got!r}")
+        else:
+            failed += 1
+            print(f"  ✗ {desc}: 期望 {want!r}，实际 {got!r}")
+
+    print("config.yml 加载:")
+    check('配置文件存在', config.LOADED_CONFIG_FILE is not None, True)
+    check('数据库端口', config.get_int('mysql.port'), 13306)
+    check('数据库库名', config.get('mysql.database'), 'myquant')
+    check('ASR 模型', config.model('asr_model'), 'paraformer-realtime-v2')
+    check('校对模型（与 config.yml 一致）', config.model('correct_model'), raw('models.correct_model'))
+    check('切分模型（与 config.yml 一致）', config.model('split_model'), raw('models.split_model'))
+    check('校对开关(默认1)', config.get_bool('llm_correct.enabled'), True)
+    check('清理开关(默认1)', config.get_bool('cleanup.after_daily_run'), True)
+    check('仅成功时清理', config.get_bool('cleanup.only_on_success'), True)
+    check('语言提示', config.get_list('asr.language_hints'), ['zh', 'en'])
+
+    print("\n目录解析:")
+    check('video_dir 基于项目根', config.video_dir(), ROOT / 'xwlb_video')
+    check('audio_dir 基于项目根', config.audio_dir(), ROOT / 'audio_processing')
+
+    print("\n环境变量覆盖优先于 config.yml:")
+    os.environ['DASHSCOPE_LLM_MODEL'] = 'qwen-max-test'
+    os.environ['LLM_CORRECT_ENABLED'] = '0'
+    cfg = config.reload_config()
+    check('模型被环境变量覆盖', cfg['models']['correct_model'], 'qwen-max-test')
+    check('开关被环境变量覆盖', config.get_bool('llm_correct.enabled'), False)
+    del os.environ['DASHSCOPE_LLM_MODEL'], os.environ['LLM_CORRECT_ENABLED']
+    config.reload_config()
+    check('恢复 config.yml 值', config.model('correct_model'), raw('models.correct_model'))
+
+    print("\n接入点与路由（换供应商的核心）:")
+    check('asr 路由（与 config.yml 一致）', config.route('asr'), raw('routes.asr'))
+    check('correct 路由（与 config.yml 一致）', config.route('correct'), raw('routes.correct'))
+    check('correct 接入点类型（与 config.yml 一致）',
+          config.endpoint_for('correct')[1]['kind'], raw(f"endpoints.{raw('routes.correct')}.kind"))
+    check('split 接入点类型', config.endpoint_for('split')[1]['kind'], 'openai')
+    check('split 请求地址拼接', config.openai_url('split'),
+          'https://api.deepseek.com/v1/chat/completions')
+    check('密钥变量名来自接入点', config.api_key_env('split'), 'DEEPSEEK_API_KEY')
+    check('必需敏感项 = 路由用到的接入点密钥 + MySQL 口令',
+          sorted(config.required_secrets()), sorted(_expected_secrets))
+    check('默认配置无问题', config.validate_endpoints(), [])
+    import dashscope
+    check('dashscope SDK 地址已按配置生效', dashscope.base_http_api_url,
+          config.endpoint('dashscope')['http_base_url'])
+    check('dashscope WS 地址已按配置生效', dashscope.base_websocket_api_url,
+          config.endpoint('dashscope')['websocket_base_url'])
+
+    print("\n切换供应商（把校对接管到 OpenAI 兼容接入点）:")
+    os.environ['XWLB_ROUTE_CORRECT'] = 'deepseek'
+    os.environ['DASHSCOPE_LLM_MODEL'] = 'deepseek-chat'
+    config.reload_config()
+    check('correct 已切到 openai 类型', config.endpoint_for('correct')[1]['kind'], 'openai')
+    check('correct 请求地址随之改变', config.openai_url('correct'),
+          'https://api.deepseek.com/v1/chat/completions')
+    check('correct 模型名可独立覆盖', config.model('correct_model'), 'deepseek-chat')
+    check('密钥变量名随供应商变化', config.api_key_env('correct'), 'DEEPSEEK_API_KEY')
+
+    # 把 ASR 也切到非 dashscope 接入点：应被校验拦下（实时识别协议不支持）
+    os.environ['XWLB_ROUTE_ASR'] = 'deepseek'
+    config.reload_config()
+    check('非 dashscope 的 ASR 被校验提示', any('仅支持 kind=dashscope' in p
+                                              for p in config.validate_endpoints()), True)
+    check('不需要 dashscope 密钥了', config.required_secrets(), ['DEEPSEEK_API_KEY', 'MYSQL_PASSWORD'])
+
+    # 指向不存在的接入点：必须给出可读错误
+    os.environ['XWLB_ROUTE_SPLIT'] = 'not-exist'
+    config.reload_config()
+    try:
+        config.endpoint_for('split')
+        check('未知接入点应报错', '未报错', 'RuntimeError')
+    except RuntimeError as e:
+        check('未知接入点报错可读', '没有该定义' in str(e), True)
+
+    # 地址也可用环境变量覆盖
+    os.environ['XWLB_ROUTE_CORRECT'] = 'deepseek'
+    os.environ['DEEPSEEK_BASE_URL'] = 'https://my-gateway.example.com/llm'
+    config.reload_config()
+    check('base_url 可被环境变量覆盖', config.openai_url('correct'),
+          'https://my-gateway.example.com/llm/chat/completions')
+
+    for key in ('XWLB_ROUTE_CORRECT', 'XWLB_ROUTE_ASR', 'XWLB_ROUTE_SPLIT',
+                'DASHSCOPE_LLM_MODEL', 'DEEPSEEK_BASE_URL'):
+        os.environ.pop(key, None)
+    config.reload_config()
+    check('恢复 config.yml 路由', [config.route(r) for r in ('asr', 'correct', 'split')],
+          [raw('routes.asr'), raw('routes.correct'), raw('routes.split')])
+
+    print("\n敏感项与 config.yml 隔离:")
+    import yaml
+    yml = yaml.safe_load((ROOT / 'config.yml').read_text(encoding='utf-8'))
+
+    def walk(node, path=''):
+        for key, value in (node or {}).items():
+            here = f'{path}.{key}' if path else str(key)
+            if isinstance(value, dict):
+                yield from walk(value, here)
+            else:
+                yield here, value
+
+    leaves = list(walk(yml))
+    # 1) 配置项名不得是口令/密钥类（注意排除 max_tokens 这类合法项）
+    risky_names = {'password', 'passwd', 'secret', 'secret_key', 'api_key', 'apikey',
+                   'token', 'access_token', 'auth_token'}
+    risky = [p for p, _ in leaves
+             if (lambda last: last in risky_names or last.endswith(('_password', '_secret', '_api_key')))(
+                 p.split('.')[-1].lower())]
+    check('config.yml 无口令/密钥类配置项', risky, [])
+    # 2) 配置值不得等于 .env 中的真实密钥（防误粘贴）
+    real_secrets = {v for v in (os.getenv('MYSQL_PASSWORD'), os.getenv('DASHSCOPE_API_KEY'),
+                                os.getenv('DEEPSEEK_API_KEY')) if v}
+    leaked = [p for p, v in leaves if isinstance(v, str) and v in real_secrets]
+    check('config.yml 未泄漏 .env 中的密钥值', leaked, [])
+    check('db_config 口令来自环境变量',
+          config.db_config()['password'] == os.getenv('MYSQL_PASSWORD'), True)
+
+    print(f"\n结果: {'全部通过' if failed == 0 else f'{failed} 项失败'}")
+    return 1 if failed else 0
+
+
+if __name__ == '__main__':
+    sys.exit(main())

tests/test_fidelity_guard.py

@@ -0,0 +1,70 @@
+"""数值事实守卫回归测试（不依赖 pytest）
+
+    python tests/test_fidelity_guard.py
+
+背景：LLM 校对（qwen）曾把 ASR 原文的事实改错——实测样例：
+    `自2027年1月1日起施行` -> `自2024年1月1日起施行`
+    `第十一届东方经济论坛` -> `第九届东方经济论坛`
+    `十五五` -> `十四五`
+因此校对结果在落库前必须通过数值事实守卫：数值被改动则回退 ASR 原文。
+
+注意守卫的取向是「宁可保留原文，也不接受 LLM 改数」：
+- 书写形式差异（7 ↔ 七、2026 ↔ 二零二六）必须放行，否则会误伤大量正常校对；
+- 任何真实数值变化必须拦下。
+"""
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from audioRead import _number_drift, _number_signature  # noqa: E402
+
+# (原文, 校对后, 说明, 是否应拦下)
+CASES = [
+    # 应当放行：仅书写形式/切分差异
+    ('7月23号', '七月二十三号', '阿拉伯数字 ↔ 中文数字', False),
+    ('2026年', '二零二六年', '阿拉伯 ↔ 逐年读法', False),
+    ('2027年1月1日', '2027年1月1日', '完全相同', False),
+    ('第十五届', '第十五届', '序数未变', False),
+    ('增长55%', '增长百分之五十五', '百分比写法差异', False),
+    ('他3日在莫斯科', '他三日在莫斯科', '日期写法差异', False),
+
+    # 应当拦下：真实数值改动
+    ('自2027年1月1日起施行', '自2024年1月1日起施行', '年份篡改（实测样例）', True),
+    ('第十一届东方经济论坛', '第九届东方经济论坛', '届次篡改（实测样例）', True),
+    ('十五五规划', '十四五规划', '规划期篡改（实测样例）', True),
+    ('增长5.3%', '增长5.31%', '小数点篡改', True),
+    ('55个', '60个', '数量篡改', True),
+    ('2025年', '2023年', '年份篡改', True),
+    ('300亿元', '30亿元', '量级篡改', True),
+]
+
+
+def main():
+    failed = 0
+    for raw, corrected, desc, expect_flag in CASES:
+        only_raw, only_new = _number_drift(raw, corrected)
+        flagged = bool(only_raw or only_new)
+        if flagged == expect_flag:
+            tag = '拦下' if flagged else '放行'
+            print(f"  ✓ {desc:20} -> {tag} {only_raw}{only_new}")
+        else:
+            failed += 1
+            expect = '拦下' if expect_flag else '放行'
+            print(f"  ✗ {desc:20} -> 期望{expect}，实际{'拦下' if flagged else '放行'} {only_raw}{only_new}")
+
+    # 签名函数自检：十五五/十四五 必须解析为不同的数值序列
+    c1, s1 = _number_signature('十五五规划')
+    c2, s2 = _number_signature('十四五规划')
+    if c1 != c2:
+        print(f"  ✓ 缩写解析 十五五={c1} 十四五={c2}（不同）")
+    else:
+        failed += 1
+        print(f"  ✗ 缩写解析失败：十五五与十四五都被解析为 {c1}")
+
+    print(f"\n结果: {'全部通过' if failed == 0 else f'{failed} 项失败'}")
+    return 1 if failed else 0
+
+
+if __name__ == '__main__':
+    sys.exit(main())

tests/test_news_parse.py

@@ -0,0 +1,84 @@
+"""newsProcess 解析逻辑回归测试（不依赖 pytest，直接运行）
+
+    python tests/test_news_parse.py
+
+覆盖 docs/BUGS.md B3 在生产日志中出现过的真实失败形态：
+- ```json 围栏包裹
+- 前后夹杂解释文字
+- {"news": [...]} / {"1": {...}, ...} 等对象包装
+- 数组里混入字符串元素（原实现报 `string indices must be integers`）
+- 输出被截断（无法修复时必须显式失败，交由重试，而不是静默写坏数据）
+"""
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from newsProcess import extract_news_rows, normalize_date  # noqa: E402
+
+DATE = '2026-09-23'
+GOOD = '{"news_id": 1, "news_title": "标题一", "news_content": "正文一"}'
+GOOD2 = '{"news_id": 2, "news_title": "标题二", "news_content": "正文二"}'
+
+CASES = [
+    # (名称, 原始响应, 期望条数)
+    ("纯数组", f'[{GOOD}, {GOOD2}]', 2),
+    ("```json 围栏（生产实测）", f'```json\n[{GOOD}, {GOOD2}]\n```', 2),
+    ("``` 无语言标记", f'```\n[{GOOD}]\n```', 1),
+    ("前后夹解释文字", f'好的，结果如下：\n[{GOOD}, {GOOD2}]\n以上。', 2),
+    ("对象包装 news 键", f'{{"news": [{GOOD}, {GOOD2}]}}', 2),
+    ("对象包装 数字键", f'{{"1": {GOOD}, "2": {GOOD2}}}', 2),
+    ("数组混入字符串（原 string indices 报错）", f'["垃圾", {GOOD}, 123, {GOOD2}]', 2),
+    ("条目缺 news_content 被跳过", f'[{GOOD}, {{"news_id": 9, "news_title": "空"}}]', 1),
+    ("news_id 非数字回退序号", '[{"news_title": "t", "news_content": "c"}]', 1),
+]
+
+FAIL_CASES = [
+    ("空响应", ""),
+    ("非 JSON", "今天的新闻联播主要内容有……"),
+    ("被截断（生产实测 Unterminated string）", '{"news": [{"news_id": 1, "news_title": "标题", "news_content": "正文未结束'),
+    ("单条新闻对象（无列表）", GOOD),
+    ("所有条目都无正文", f'[{{"news_id": 1, "news_title": "只有标题"}}]'),
+    ("空数组", "[]"),
+]
+
+
+def main():
+    failed = 0
+
+    for name, raw, expected in CASES:
+        try:
+            rows = extract_news_rows(raw, DATE)
+            assert len(rows) == expected, f"期望 {expected} 条，实际 {len(rows)} 条"
+            for row in rows:
+                assert row['news_date'] == DATE
+                assert isinstance(row['sub_id'], int)
+                assert row['news_content'].strip()
+                assert len(row['news_title']) <= 256
+            print(f"  ✓ {name} -> {len(rows)} 条")
+        except Exception as e:
+            failed += 1
+            print(f"  ✗ {name}: {type(e).__name__}: {e}")
+
+    for name, raw in FAIL_CASES:
+        try:
+            rows = extract_news_rows(raw, DATE)
+            failed += 1
+            print(f"  ✗ {name}: 本应失败，却解析出 {len(rows)} 条")
+        except ValueError as e:
+            print(f"  ✓ {name} -> 按预期抛 ValueError（触发重试）: {str(e)[:48]}")
+        except Exception as e:
+            failed += 1
+            print(f"  ✗ {name}: 抛出了非 ValueError: {type(e).__name__}: {e}")
+
+    # 日期归一化
+    assert normalize_date('20260923') == DATE
+    assert normalize_date(DATE) == DATE
+    print("  ✓ 日期归一化 20260923 / 2026-09-23")
+
+    print(f"\n结果: {'全部通过' if failed == 0 else f'{failed} 项失败'}")
+    return 1 if failed else 0
+
+
+if __name__ == '__main__':
+    sys.exit(main())

tests/test_tunnel.py

@@ -0,0 +1,183 @@
+"""隧道自愈回归测试（不依赖 pytest，不碰真实隧道）
+
+    python tests/test_tunnel.py
+
+覆盖：
+- 端口探测 / 本机地址判定；
+- 各分支状态：禁用、非本机、已可连、systemd 托管中等待、执行脚本后恢复、脚本不存在；
+- **端口已通时绝不执行 autossh.sh**（避免在正常运行时多起进程抢端口）；
+- systemd 单元 active 时只等待、不抢端口；
+- dry-run 不执行脚本。
+
+测试用临时 config.yml（`XWLB_CONFIG_FILE`）+ 临时端口的假脚本，
+不会触碰真实的 13306 与项目里的 autossh.sh。
+"""
+import os
+import socket
+import subprocess
+import sys
+import tempfile
+import time
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parent.parent
+sys.path.insert(0, str(ROOT))
+
+
+def free_port():
+    with socket.socket() as s:
+        s.bind(('127.0.0.1', 0))
+        return s.getsockname()[1]
+
+
+def write_cfg(path, port, script, unit='xwlb-tunnel.service', enabled=1):
+    path.write_text(f"""mysql:
+  tunnel:
+    enabled: {enabled}
+    script: {script}
+    systemd_unit: {unit}
+    wait_seconds: 4
+    connect_timeout: 1
+  host: 127.0.0.1
+  port: {port}
+  user: myquant
+  database: myquant
+paths:
+  video_dir: xwlb_video
+  audio_dir: audio_processing
+models:
+  asr_model: paraformer-realtime-v2
+  correct_model: qwen3.8-flash
+  split_model: deepseek-flash
+""", encoding='utf-8')
+
+
+def make_fake_script(dirpath, port, marker):
+    """假 autossh.sh：起一个 TCP 监听并把"被执行过"记录到 marker 文件"""
+    p = dirpath / 'fake_tunnel.sh'
+    p.write_text(f"""#!/bin/bash
+touch "{marker}"
+setsid .venv/bin/python -m http.server {port} --bind 127.0.0.1 >/dev/null 2>&1 &
+""", encoding='utf-8')
+    p.chmod(0o755)
+    return p
+
+
+def kill_listener(port):
+    subprocess.run(['pkill', '-f', f'http[.]server {port}'], capture_output=True)
+    time.sleep(0.3)
+
+
+def main():
+    failed = 0
+    tmp = Path(tempfile.mkdtemp(prefix='xwlb_tunnel_test_'))
+
+    def check(desc, got, want):
+        nonlocal failed
+        if got == want:
+            print(f"  ✓ {desc}: {got!r}")
+        else:
+            failed += 1
+            print(f"  ✗ {desc}: 期望 {want!r}，实际 {got!r}")
+
+    # 临时配置 + 重新加载（用环境变量指定，不污染真实 config.yml）
+    port = free_port()
+    script = make_fake_script(tmp, port, tmp / 'executed')
+    cfg = tmp / 'config.yml'
+    write_cfg(cfg, port, str(script))
+    os.environ['XWLB_CONFIG_FILE'] = str(cfg)
+
+    import config
+    config.reload_config()
+    import tunnel
+
+    try:
+        print("端口探测:")
+        check('关闭的端口判定为不通', tunnel.port_open('127.0.0.1', free_port(), timeout=0.5), False)
+        srv = subprocess.Popen([sys.executable, '-m', 'http.server', str(port), '--bind', '127.0.0.1'],
+                               stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
+        time.sleep(1.0)
+        check('已监听端口判定为可连', tunnel.port_open('127.0.0.1', port, timeout=1), True)
+        srv.terminate()
+        srv.wait(timeout=5)
+
+        print("\n本机地址判定:")
+        for h in ('localhost', '127.0.0.1', '::1'):
+            check(f'{h} 视为本机', tunnel.is_local(h), True)
+        check('远端地址不视为本机', tunnel.is_local('db.example.com'), False)
+
+        print("\n分支：端口已通（不得执行脚本）:")
+        if (tmp / 'executed').exists():
+            os.remove(tmp / 'executed')
+        srv = subprocess.Popen([sys.executable, '-m', 'http.server', str(port), '--bind', '127.0.0.1'],
+                               stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
+        time.sleep(1.0)
+        r = tunnel.ensure_tunnel()
+        check('状态 = 已可连', r['status'], tunnel.ST_ALREADY_OPEN)
+        check('未执行隧道脚本', (tmp / 'executed').exists(), False)
+        srv.terminate()
+        srv.wait(timeout=5)
+
+        print("\n分支：端口不通 + systemd 托管中（只等待，不抢端口）:")
+        if (tmp / 'executed').exists():
+            os.remove(tmp / 'executed')
+        tunnel.systemd_unit_active = lambda unit, timeout=5: True
+        r = tunnel.ensure_tunnel()
+        tunnel.systemd_unit_active = lambda unit, timeout=5: False  # 还原
+        check('状态 = 等待 systemd', r['status'], tunnel.ST_UNAVAILABLE)
+        check('未执行隧道脚本（不抢 systemd 的端口）', (tmp / 'executed').exists(), False)
+
+        print("\n分支：端口不通 + 执行脚本后恢复:")
+        if (tmp / 'executed').exists():
+            os.remove(tmp / 'executed')
+        r = tunnel.ensure_tunnel(wait_seconds=6)
+        check('状态 = 执行脚本后端口就绪', r['status'], tunnel.ST_SCRIPT_STARTED)
+        check('脚本确实被执行', (tmp / 'executed').exists(), True)
+        check('判定为已恢复', r['healed'], True)
+        check('端口现在可连', tunnel.port_open('127.0.0.1', port), True)
+        kill_listener(port)
+
+        print("\n分支：脚本不存在 / 禁用 / 非本机:")
+        write_cfg(cfg, free_port(), str(tmp / '不存在.sh'))
+        config.reload_config()
+        r = tunnel.ensure_tunnel()
+        check('脚本不存在 → 不可用', r['status'], tunnel.ST_UNAVAILABLE)
+        check('且未判定为已恢复', r['healed'], False)
+
+        write_cfg(cfg, free_port(), str(script), enabled=0)
+        config.reload_config()
+        check('enabled=0 → 跳过探测', tunnel.ensure_tunnel()['status'], tunnel.ST_DISABLED)
+
+        write_cfg(cfg, free_port(), str(script))
+        config.reload_config()
+        real_db_config = config.db_config
+        config.db_config = lambda *a, **k: {'host': 'db.example.com', 'port': 3306,
+                                            'username': 'u', 'password': 'p', 'database': 'd'}
+        try:
+            check('远端地址 → 不适用隧道', tunnel.ensure_tunnel()['status'], tunnel.ST_NOT_LOCAL)
+        finally:
+            config.db_config = real_db_config  # 必须还原，否则污染后续用例
+
+        print("\n分支：dry-run 不执行脚本:")
+        port2 = free_port()
+        write_cfg(cfg, port2, str(script))
+        config.reload_config()
+        if (tmp / 'executed').exists():
+            os.remove(tmp / 'executed')
+        r = tunnel.ensure_tunnel(dry_run=True)
+        check('dry-run 报告将执行脚本', 'dry-run' in r['detail'], True)
+        check('dry-run 未真的执行', (tmp / 'executed').exists(), False)
+
+    finally:
+        os.environ.pop('XWLB_CONFIG_FILE', None)
+        kill_listener(port)
+        kill_listener(port)
+        import shutil
+        shutil.rmtree(tmp, ignore_errors=True)
+
+    print(f"\n结果: {'全部通过' if failed == 0 else f'{failed} 项失败'}")
+    return 1 if failed else 0
+
+
+if __name__ == '__main__':
+    sys.exit(main())