fix: 管道审计改为动态读jobs.json，无需手动维护脚本列表

scripts/morning_health_check.py

@@ -619,26 +619,30 @@ def run_check(item):
                 if not p.get("verified"):
                     gaps.append(p["name"])
             
-            # 自动发现：对比cron脚本列表和注册表，漏登的报警
+            # 自动发现：从cron jobs.json读所有脚本，交叉比对注册表
             known_sources = set()
             for p in reg.get("pipelines", []):
                 src = p.get("source", "")
-                for s in ["morning_health_check", "intraday_health_check", "strategy_review",
+                # 从source描述中提取脚本名
-                          "xiaoguo_scanner", "xiaoguo_signal_consumer", "data_governance",
+                import re as rr
-                          "price_monitor", "stale_push_wlin", "per_stock_reassess",
+                for m in rr.findall(r'[\w_-]+\.py', src):
-                          "macro_context_collector", "divergence_detector", "macro_signal_consumer",
+                    known_sources.add(m.replace('.py', ''))
-                          "self_todo_executor", "strategy_evaluator", "hk_rate"]:
+            
-                    if s in src:
+            # 从jobs.json动态读取所有启用脚本
-                        known_sources.add(s)
+            unregistered = []
-            # 检查是否有重要脚本未出现在任何管道定义中
+            try:
-            all_scripts = ["morning_health_check", "intraday_health_check", "strategy_review",
+                jobs_data = j2.loads(open(str(HERMES_CRON_DIR / "jobs.json")).read())
-                          "xiaoguo_scanner", "xiaoguo_signal_consumer", "data_governance",
+                active_scripts = set()
-                          "price_monitor", "stale_push_wlin", "per_stock_reassess",
+                for job in jobs_data.get("jobs", []):
-                          "macro_context_collector", "divergence_detector", "macro_signal_consumer",
+                    script = job.get("script", "") or ""
-                          "self_todo_executor", "strategy_evaluator"]
+                    if script and job.get("enabled", True):
-            for s in all_scripts:
+                        name = script.replace('.py', '')
+                        active_scripts.add(name)
+                for s in sorted(active_scripts):
                     if s not in known_sources:
                         unregistered.append(s)
+            except Exception:
+                pass
             
             if unregistered:
                 ok = False