feat(L2): 卫生审计自动收尸——scripts/影子副本+零引用孤儿(mtime>7天)自动git mv归档并提交; dev-spec新增文件居住宪法(红线#9)

This commit is contained in:
hmo
2026-07-22 08:41:06 +08:00
parent 375f6d469a
commit 3ea7c52112
2 changed files with 86 additions and 1 deletions
@@ -62,6 +62,88 @@ def md5(p):
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 = []
@@ -292,6 +374,8 @@ def check_db_freshness():
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),