MoFin/scripts/audit_deadcode.py

#!/usr/bin/env python3
"""audit_deadcode.py — 废弃模块/死代码审计：小果生态、旧bot、疑似废弃脚本"""
import os, subprocess, json
from datetime import datetime

def sh(cmd, timeout=10):
    try:
        r = subprocess.run(cmd, shell=True, capture_output=True, text=True, timeout=timeout)
        return r.stdout.strip()
    except Exception as e:
        return f"ERR: {e}"

print("=== 1. 小果(xiaoguo)生态残留 ===")
print("--- 运行中的 xiaoguo 进程 ---")
print(sh("ps aux | grep -i xiaoguo | grep -v grep"))
print("\n--- xiaoguo 相关文件 ---")
print(sh("ls -la /home/hmo/MoFin/deploy/profile-scripts/ 2>/dev/null | grep -i xiaoguo; ls /home/hmo/MoFin/ | grep -i xiaoguo; ls /home/hmo/MoFin/scripts/ 2>/dev/null | grep -i xiaoguo; ls /home/hmo/AgentsMeeting/ 2>/dev/null | grep -i xiaoguo | head -5"))
print("\n--- xiaoguo 相关数据文件 ---")
print(sh("ls -la /home/hmo/web-dashboard/data/ | grep -i xiaoguo; ls -la /home/hmo/MoFin/data/ 2>/dev/null | grep -i xiaoguo"))
print("\n--- xiaoguo 相关 cron ---")
for jf, prof in [('/home/hmo/.hermes/profiles/position-analyst/cron/jobs.json', 'pa'), ('/home/hmo/.hermes/cron/jobs.json', 'default')]:
    try:
        d = json.load(open(jf))
        jobs = d if isinstance(d, list) else d.get('jobs', [])
        for j in jobs:
            if 'xiaoguo' in str(j.get('script', '')).lower() or '小果' in str(j.get('name', '')):
                print(f"  [{prof}] {j.get('name')} script={j.get('script')} enabled={j.get('enabled')} last={str(j.get('last_run_at'))[:10]} status={j.get('last_status')}")
    except Exception as e:
        print(f"  {prof}: {e}")

print("\n=== 2. 旧 bot 文件 ===")
for f in ['/home/hmo/xmpp_zhiwei_bot.py.bak', '/home/hmo/run_zhiwei_bot.py', '/home/hmo/xmpp_bot_rest.py',
          '/home/hmo/xmpp_xiaoguo_bot.py', '/home/hmo/xmpp_mohe_bot.py', '/home/hmo/xmpp_zhiwei_bot.py']:
    if os.path.exists(f):
        p = f
        mt = datetime.fromtimestamp(os.path.getmtime(p))
        link = os.path.islink(p)
        tgt = os.readlink(p) if link else ''
        print(f"  {f} {'symlink->'+tgt if link else str(os.path.getsize(p)//1024)+'KB'} {mt.strftime('%m-%d')}")

print("\n=== 3. AgentsMeeting 目录与 MoFin 的关系（是否重复项目）===")
print(sh("du -sh /home/hmo/AgentsMeeting /home/hmo/MoFin /home/hmo/web-dashboard /home/hmo/projects/AgentsMeeting /home/hmo/projects/MoFin 2>/dev/null"))
print("\n--- AgentsMeeting 里和 MoFin 同名的脚本 ---")
am = '/home/hmo/AgentsMeeting'
if os.path.isdir(am):
    am_scripts = set()
    for r, _, fs in os.walk(am):
        if 'venv' in r or 'node_modules' in r or '.git' in r:
            continue
        for f in fs:
            if f.endswith('.py'):
                am_scripts.add(f)
    mofin_scripts = set()
    for d in ['/home/hmo/MoFin', '/home/hmo/MoFin/scripts', '/home/hmo/MoFin/deploy/profile-scripts']:
        if os.path.isdir(d):
            mofin_scripts.update(f for f in os.listdir(d) if f.endswith('.py'))
    common = sorted(am_scripts & mofin_scripts)
    print(f"  同名 .py 文件: {len(common)} 个")
    for f in common[:20]:
        print(f"    {f}")

print("\n=== 4. 超过30天未修改的可疑废弃脚本（deploy/profile-scripts 中）===")
now = datetime.now()
dd = '/home/hmo/MoFin/deploy/profile-scripts'
for f in sorted(os.listdir(dd)):
    if not f.endswith('.py'):
        continue
    p = os.path.join(dd, f)
    mt = datetime.fromtimestamp(os.path.getmtime(p))
    age = (now - mt).total_seconds() / 86400
    if age > 30:
        print(f"  {f:45s} {mt.strftime('%m-%d')} ({age:.0f}d)")

print("\n=== 5. scripts/ 下的工具脚本（非 cron 调度，疑似一次性）===")
# 哪些脚本不在 cron jobs 里
cron_scripts = set()
for jf in ['/home/hmo/.hermes/profiles/position-analyst/cron/jobs.json', '/home/hmo/.hermes/cron/jobs.json']:
    try:
        d = json.load(open(jf))
        jobs = d if isinstance(d, list) else d.get('jobs', [])
        for j in jobs:
            if j.get('script'):
                cron_scripts.add(j['script'])
    except Exception:
        pass
for d in ['/home/hmo/MoFin/deploy/profile-scripts']:
    for f in sorted(os.listdir(d)):
        if f.endswith('.py') and f not in cron_scripts:
            p = os.path.join(d, f)
            mt = datetime.fromtimestamp(os.path.getmtime(p))
            age = (now - mt).total_seconds() / 86400
            print(f"  {f:45s} 不在cron中 ({age:.0f}d)")