#!/usr/bin/env python3 """system_hygiene_audit.py — 系统卫生审计（防冗余复发） 每周运行。检查六类问题，输出 hygiene_report.json，有问题时推 XMPP。 红线6/7/8/9/10 的自动化 enforcement。 检查项： 1. 分叉副本：同名 .py 在不同权威位置内容不一致 2. 断裂硬链接：deploy/profile-scripts vs profile scripts 内容不一致（cron 会跑旧代码！） 3. 僵尸进程：>3 天的 python/node 进程不在白名单 4. 孤儿数据文件：生产数据目录 >14 天未修改且不在活文件注册表 5. 死 cron：jobs.json 中 script 不存在 6. DB 表新鲜度：核心表 >24h 无新记录（交易时间） """ import os, sys, json, glob, hashlib, sqlite3, subprocess from datetime import datetime, timedelta from pathlib import Path sys.path.insert(0, '/home/hmo/MoFin') DEPLOY = '/home/hmo/MoFin/deploy/profile-scripts' PA_SCRIPTS = '/home/hmo/.hermes/profiles/position-analyst/scripts' MOFIN_ROOT = '/home/hmo/MoFin' DATA_DIR = '/home/hmo/MoFin/data' REPORT = '/home/hmo/MoFin/gateway/logs/hygiene_report.json' # 活文件注册表（生产数据目录允许存在的非数据文件） LIVE_DATA_FILES = { 'mofin.db', 'mofin.db-shm', 'mofin.db-wal', 'mofin_health.json', 'portfolio.json', 'preflight_result.json', 'growth_registry.json', 'hardcode_audit.json', 'health_checklist.json', 'macro_divergence_state.json', 'macro_risk_state.json', 'market.json', 'scanner_state.json', 'state.db', 'strategy_staleness_report.json', 'system_audit_report.json', 'price_history.json', 'evaluation.json', 'accuracy_stats.json', 'format_error_library.json', 'candidate_pool.json', 'pipeline_registry.json', 'analyst-knowledge-log.md', 'mofin_health.html', 'evaluation_input.json', 'push_cooldown.json', 'system_audit.json', 'system_inventory.json', 'stocks', } PROCESS_WHITELIST = [ 'hermes_cli.main', 'server.py', 'xmpp_zhiwei_bot.py', 'xmpp_mohe_bot.py', 'shadowsocks', 'unattended-upgrades', 'dashboard.py', 'kanban_api.py', 'main_dsa.py', 'vc-webhook.py', 'obsidian-api.py', 'http.server', 'wechat_webhook.py', 'qq-poll', 'afw', 'mcp_server', 'uvicorn', 'agentmemory', 'kimi_collect', 'mohe_knowledge_relay', 'wechat_watchdog', 'todo_scanner', 'miner.py', ] CORE_TABLES = [ ('live_prices', 'updated_at', '实时价格'), ('market_snapshots', 'created_at', '市场快照'), ('mtf_cache', 'updated_at', '多周期缓存'), ('macro_context_log', 'created_at', '宏观上下文'), ('price_events', 'created_at', '价格事件'), ] def md5(p): try: return hashlib.md5(open(p, 'rb').read()).hexdigest() except Exception: return 'ERR' def check_diverged(): """检查 deploy vs MoFin/scripts vs MoFin根 的分叉副本""" issues = [] compare_dirs = [f'{MOFIN_ROOT}/scripts', MOFIN_ROOT, '/home/hmo/web-dashboard'] for f in os.listdir(DEPLOY): if not f.endswith('.py'): continue dp = os.path.join(DEPLOY, f) d_md5 = md5(dp) for d in compare_dirs: p = os.path.join(d, f) if os.path.exists(p) and not os.path.islink(p): try: if os.path.samefile(p, dp): continue except Exception: pass if md5(p) != d_md5: issues.append({ 'type': 'diverged_copy', 'file': f, 'canonical': dp, 'stale_copy': p, 'action': f'归档 {p} 或硬链接到权威版', }) return issues def check_broken_hardlinks(): """deploy vs pa/scripts 内容不一致 = cron 跑旧代码""" issues = [] for f in os.listdir(DEPLOY): if not f.endswith('.py'): continue dp = os.path.join(DEPLOY, f) pp = os.path.join(PA_SCRIPTS, f) if os.path.exists(pp): try: if os.path.samefile(dp, pp): continue except Exception: pass if md5(dp) != md5(pp): issues.append({ 'type': 'broken_hardlink', 'file': f, 'action': 'bash deploy/profile-scripts/sync_profile_scripts.sh', }) else: issues.append({ 'type': 'missing_profile_link', 'file': f, 'action': 'bash deploy/profile-scripts/sync_profile_scripts.sh', }) return issues def check_zombies(): """>3 天的 python/node 进程不在白名单（docker 容器内进程豁免）""" issues = [] try: r = subprocess.run(['ps', '-eo', 'pid,etime,args'], capture_output=True, text=True, timeout=10) for line in r.stdout.splitlines()[1:]: parts = line.split(None, 2) if len(parts) < 3: continue pid, etime, cmd = parts if 'python' not in cmd and 'node' not in cmd: continue # docker 容器内进程豁免（cgroup 含 docker） try: cg = open(f'/proc/{pid}/cgroup').read() if 'docker' in cg: continue except Exception: pass # etime 格式: dd-hh:mm:ss 或 hh:mm:ss days = 0 if '-' in etime: days = int(etime.split('-')[0]) if days >= 3: if not any(w in cmd for w in PROCESS_WHITELIST): issues.append({ 'type': 'zombie_process', 'pid': pid, 'days': days, 'cmd': cmd[:120], 'action': f'确认后 kill {pid}（红线9 收尸流程）', }) except Exception: pass return issues def check_orphan_files(): """生产数据目录的孤儿文件""" issues = [] now = datetime.now() for f in os.listdir(DATA_DIR): p = os.path.join(DATA_DIR, f) if not os.path.isfile(p) or f.startswith('.'): continue if f in LIVE_DATA_FILES: continue age_d = (now.timestamp() - os.path.getmtime(p)) / 86400 if age_d > 14: issues.append({ 'type': 'orphan_data_file', 'file': f, 'age_days': round(age_d), 'action': f'归档到 archive/（红线8）', }) return issues def check_dead_cron(): """cron job 指向不存在的脚本""" issues = [] for jf, sdir in [('/home/hmo/.hermes/profiles/position-analyst/cron/jobs.json', PA_SCRIPTS), ('/home/hmo/.hermes/cron/jobs.json', '/home/hmo/.hermes/scripts')]: try: d = json.load(open(jf)) jobs = d if isinstance(d, list) else d.get('jobs', []) for j in jobs: s = j.get('script') if s and j.get('enabled', True) and not os.path.exists(os.path.join(sdir, s)): issues.append({ 'type': 'dead_cron_script', 'job': j.get('name'), 'script': s, 'action': '删除 job 或补齐脚本', }) except Exception: pass return issues def check_db_freshness(): """核心表新鲜度（红线10）""" issues = [] try: c = sqlite3.connect(os.path.join(DATA_DIR, 'mofin.db'), timeout=10) now = datetime.now() is_trading_time = now.weekday() < 5 and 9 <= now.hour <= 16 for table, col, label in CORE_TABLES: try: row = c.execute(f"SELECT MAX({col}) FROM {table}").fetchone() if row and row[0]: last = datetime.fromisoformat(str(row[0]).replace('Z', '')) age_h = (now - last).total_seconds() / 3600 threshold = 4 if is_trading_time else 48 if age_h > threshold: issues.append({ 'type': 'stale_table', 'table': table, 'label': label, 'age_hours': round(age_h, 1), 'threshold': threshold, 'action': '查对应采集脚本的 cron 状态', }) else: issues.append({'type': 'empty_table', 'table': table, 'label': label, 'action': '查采集链路'}) except Exception: pass c.close() except Exception as e: issues.append({'type': 'db_error', 'error': str(e)[:100]}) return issues def main(): print('🧹 系统卫生审计', datetime.now().strftime('%Y-%m-%d %H:%M')) all_issues = [] for name, fn in [('分叉副本', check_diverged), ('断裂硬链接', check_broken_hardlinks), ('僵尸进程', check_zombies), ('孤儿文件', check_orphan_files), ('死cron', check_dead_cron), ('DB新鲜度', check_db_freshness)]: found = fn() status = f'❌ {len(found)}' if found else '✅' print(f' {status} {name}') all_issues.extend(found) report = { 'generated_at': datetime.now().strftime('%Y-%m-%d %H:%M:%S'), 'issue_count': len(all_issues), 'issues': all_issues, 'status': 'warn' if all_issues else 'ok', } os.makedirs(os.path.dirname(REPORT), exist_ok=True) with open(REPORT, 'w', encoding='utf-8') as f: json.dump(report, f, ensure_ascii=False, indent=2) if all_issues: # 推 XMPP try: import urllib.request lines = [f"🧹 系统卫生审计发现 {len(all_issues)} 个问题："] for i in all_issues[:8]: lines.append(f"• [{i['type']}] {i.get('file') or i.get('job') or i.get('table') or i.get('pid')}: {i.get('action','')[:60]}") if len(all_issues) > 8: lines.append(f'… 共 {len(all_issues)} 个，详见 hygiene_report.json') payload = json.dumps({'to': 'hmo@yoin.fun', 'body': '\n'.join(lines), 'type': 'chat'}).encode() req = urllib.request.Request('http://127.0.0.1:5805/', data=payload, headers={'Content-Type': 'application/json'}) urllib.request.urlopen(req, timeout=5) print(' 📨 已推 XMPP') except Exception as e: print(f' XMPP 推送失败: {e}') else: print(' ✅ 系统卫生良好') if __name__ == '__main__': main()