426 lines
18 KiB
Python
426 lines
18 KiB
Python
#!/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
|
||
|
||
# ── 消息通道统一路由(broadcast/xmpp by delivery) ──
|
||
try:
|
||
from messenger import install_stdio_hook as _msh
|
||
_msh()
|
||
except Exception:
|
||
pass
|
||
|
||
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 _collect_live_refs():
|
||
"""收集活引用：cron jobs.json 脚本名 + 活代码 import/字符串路径引用"""
|
||
import re as _re
|
||
refs = set()
|
||
for pj in glob.glob('/home/hmo/.hermes/profiles/*/cron/jobs.json'):
|
||
try:
|
||
with open(pj, encoding='utf-8') as f:
|
||
jobs = json.load(f)
|
||
jobs = jobs if isinstance(jobs, list) else jobs.get('jobs', [])
|
||
for j in jobs:
|
||
if j.get('enabled', True) and j.get('script'):
|
||
refs.add(os.path.basename(j['script']))
|
||
except Exception:
|
||
pass
|
||
scan = [f'{MOFIN_ROOT}/server.py', f'{MOFIN_ROOT}/xmpp_logger.py']
|
||
scan += glob.glob(f'{DEPLOY}/*.py')
|
||
scan += glob.glob(f'{MOFIN_ROOT}/deploy/bot/*.py')
|
||
scan += glob.glob(f'{MOFIN_ROOT}/prompt_manager/*.py')
|
||
for sf in scan:
|
||
try:
|
||
with open(sf, encoding='utf-8', errors='replace') as f:
|
||
c = f.read()
|
||
for m in _re.finditer(r'(?:from|import)\s+([a-zA-Z0-9_]+)', c):
|
||
refs.add(m.group(1) + '.py')
|
||
for m in _re.finditer(r'["\']([a-zA-Z0-9_\-]+\.py)["\']', c):
|
||
refs.add(m.group(1))
|
||
except Exception:
|
||
pass
|
||
return refs
|
||
|
||
|
||
def auto_archive_orphans(now):
|
||
"""源头自动化收尸（2026-07-22 老爸批准）：
|
||
scripts/ 下的 ①影子副本(deploy同名) ②零引用孤儿，且 mtime>7天 →
|
||
自动 git mv 到 archive/YYYYMM-auto/ 并自动提交（GIT_ALLOW_COMMIT）。
|
||
返回归档记录列表。7天内新文件不动（防误收在途工作）。"""
|
||
archived = []
|
||
scripts_dir = f'{MOFIN_ROOT}/scripts'
|
||
if not os.path.isdir(scripts_dir):
|
||
return archived
|
||
cutoff = (now - timedelta(days=7)).timestamp()
|
||
refs = _collect_live_refs()
|
||
dest_rel = f'archive/{now.strftime("%Y%m")}-auto'
|
||
to_move = []
|
||
for f in sorted(os.listdir(scripts_dir)):
|
||
if not f.endswith('.py'):
|
||
continue
|
||
p = os.path.join(scripts_dir, f)
|
||
if os.path.getmtime(p) > cutoff:
|
||
continue
|
||
if os.path.exists(os.path.join(DEPLOY, f)):
|
||
to_move.append((f, 'shadow'))
|
||
elif f not in refs:
|
||
to_move.append((f, 'orphan'))
|
||
if not to_move:
|
||
return archived
|
||
os.makedirs(f'{MOFIN_ROOT}/{dest_rel}', exist_ok=True)
|
||
env = dict(os.environ)
|
||
env['GIT_ALLOW_COMMIT'] = '1'
|
||
for f, why in to_move:
|
||
r = subprocess.run(['git', '-C', MOFIN_ROOT, 'mv', f'scripts/{f}', f'{dest_rel}/{f}'],
|
||
capture_output=True, text=True, env=env, timeout=30)
|
||
if r.returncode != 0:
|
||
try:
|
||
os.rename(f'{scripts_dir}/{f}', f'{MOFIN_ROOT}/{dest_rel}/{f}')
|
||
r = subprocess.run(['git', '-C', MOFIN_ROOT, 'add', f'{dest_rel}/{f}'],
|
||
capture_output=True, env=env, timeout=30)
|
||
except Exception as e:
|
||
print(f' ⚠️ 归档失败 {f}: {e}', flush=True)
|
||
continue
|
||
archived.append({'type': 'auto_archive', 'file': f'scripts/{f}',
|
||
'action': f'已自动归档({why}) → {dest_rel}/'})
|
||
if archived:
|
||
subprocess.run(['git', '-C', MOFIN_ROOT, 'add', '-A', 'archive/', 'scripts/'],
|
||
capture_output=True, env=env, timeout=30)
|
||
subprocess.run(['git', '-C', MOFIN_ROOT, 'commit', '-m',
|
||
f'chore: L2卫生自动归档 {len(archived)} 个影子/孤儿文件（{dest_rel}）'],
|
||
capture_output=True, env=env, timeout=60)
|
||
print(f' 🧹 自动归档 {len(archived)} 个文件 → {dest_rel}/', flush=True)
|
||
return archived
|
||
|
||
|
||
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_commit_hook():
|
||
"""提交白名单钩子存在性检查（2026-07-21 起：知微 git 写权限封锁的载体，
|
||
被删/被改 = 告警）"""
|
||
hook = f'{MOFIN_ROOT}/.git/hooks/pre-commit'
|
||
if not os.path.isfile(hook):
|
||
return [{'type': '安全机制缺失', 'file': hook,
|
||
'action': 'pre-commit 提交白名单钩子丢失！知微封锁失效，立即恢复'}]
|
||
with open(hook, encoding='utf-8', errors='replace') as f:
|
||
content = f.read()
|
||
if 'GIT_ALLOW_COMMIT' not in content:
|
||
return [{'type': '安全机制异常', 'file': hook,
|
||
'action': 'pre-commit 钩子内容被篡改（不含白名单令牌校验）'}]
|
||
return []
|
||
|
||
|
||
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_stale_sessions():
|
||
"""盲区③：常驻 agent session 指令冻结检测（2026-07-21 发现：常驻 session 的
|
||
system_prompt 冻结于创建时，SOUL.md 更新后旧 session 仍按旧指令行动）。
|
||
判定：未关闭的 api_server/cli session（msgs>50）启动时间 < 对应 profile SOUL.md
|
||
的 mtime → 指令过期，建议 bump session id / 重启对应 bot。"""
|
||
issues = []
|
||
import sqlite3 as _sq
|
||
for soul in glob.glob('/home/hmo/.hermes/profiles/*/SOUL.md'):
|
||
profile = soul.split('/')[-2]
|
||
state_db = os.path.join(os.path.dirname(soul), 'state.db')
|
||
if not os.path.exists(state_db):
|
||
continue
|
||
soul_mtime = os.path.getmtime(soul)
|
||
try:
|
||
conn = _sq.connect(f"file:{state_db}?mode=ro", uri=True, timeout=10)
|
||
conn.row_factory = _sq.Row
|
||
rows = conn.execute("""
|
||
SELECT id, source, started_at, message_count FROM sessions
|
||
WHERE (ended_at IS NULL OR ended_at = 0)
|
||
AND message_count > 50
|
||
ORDER BY message_count DESC LIMIT 10
|
||
""").fetchall()
|
||
conn.close()
|
||
except Exception:
|
||
continue
|
||
for r in rows:
|
||
raw_sa = r['started_at'] or 0
|
||
sa = (raw_sa if raw_sa > 1e12 else raw_sa * 1000) / 1000 # 单位自适应
|
||
if not (0 < sa < soul_mtime):
|
||
continue
|
||
# 只报"仍活跃"的：最近 6 小时内有消息（已被 bump 遗弃的旧 session 不报）
|
||
try:
|
||
conn2 = _sq.connect(f"file:{state_db}?mode=ro", uri=True, timeout=10)
|
||
last_msg = conn2.execute(
|
||
"SELECT MAX(created_at) FROM messages WHERE session_id=?",
|
||
(r['id'],)).fetchone()[0]
|
||
conn2.close()
|
||
lm = last_msg or 0
|
||
last_ts = (lm if lm > 1e12 else lm * 1000) / 1000 # 单位自适应
|
||
if last_ts and (datetime.now().timestamp() - last_ts) > 6 * 3600:
|
||
continue
|
||
except Exception:
|
||
pass # 无法判活跃度时宁报不漏
|
||
issues.append({
|
||
'type': '指令冻结session', 'file': f"[{profile}] {r['id']}",
|
||
'action': f"常驻session({r['message_count']}条消息)启动于"
|
||
f"{datetime.fromtimestamp(sa).strftime('%m-%d %H:%M')}，"
|
||
f"早于SOUL.md最后修改({datetime.fromtimestamp(soul_mtime).strftime('%m-%d %H:%M')})"
|
||
f"→按旧指令运行中，需bump session id/重启bot",
|
||
})
|
||
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 = []
|
||
# 源头自动化：先把影子/孤儿归档，再跑检查（检查看到的应是归档后的干净状态）
|
||
all_issues.extend(auto_archive_orphans(datetime.now()))
|
||
for name, fn in [('分叉副本', check_diverged), ('断裂硬链接', check_broken_hardlinks),
|
||
('僵尸进程', check_zombies), ('孤儿文件', check_orphan_files),
|
||
('死cron', check_dead_cron), ('DB新鲜度', check_db_freshness),
|
||
('指令冻结session', check_stale_sessions), ('提交白名单钩子', check_commit_hook)]:
|
||
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（经 alert_helper：30min限速+8行截断+24h内容去重——同一批问题不会每天重复轰炸）
|
||
try:
|
||
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||
from alert_helper import notify, INFO
|
||
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')
|
||
if notify("卫生审计", '\n'.join(lines), INFO):
|
||
print(' 📨 已推 XMPP')
|
||
else:
|
||
print(' 📨 XMPP 已按频率/去重策略静默')
|
||
except Exception as e:
|
||
print(f' XMPP 推送失败: {e}')
|
||
else:
|
||
print(' ✅ 系统卫生良好')
|
||
|
||
|
||
if __name__ == '__main__':
|
||
main() |