Files
MoFin/scripts/add_hygiene_cron.py
T
hmo 4f83ee8a01 feat(hygiene): anti-redundancy enforcement — spec rules + weekly audit
Root cause analysis of the 2026-07-20 redundancy incident:
1. No single-source-of-truth rule -> same file legitimately lived in 4+
   locations, diverging silently
2. Relative path resolution (Path(__file__).parent/'data') -> each
   hardlinked copy of mofin_db.py pointed to a DIFFERENT database
3. 'Backup habit' left .bak/legacy files in production dirs, which
   monitoring then scanned and reported as false alarms
4. Half-done migrations: DB tables created but old JSON writers/readers
   stayed (price_events), old files stayed
5. Dead modules never got buried: xiaoguo 'dead' but bot ran 8 days
   as root eating 2.5GB
6. Monitoring checked 'does it exist' not 'is it alive' -> stale file
   mtime reported as 'pipeline stalled 14 days' (false alarm)
7. No 'system hygiene' as a check category at all

Prevention implemented:
- dev-spec.md v2.0: 五条红线 -> 十条红线
  #6 single source of truth (hardlink only, no independent copies)
  #7 absolute data paths only (no __file__-relative data resolution)
  #8 no backups/legacy in production data dirs (archive immediately)
  #9 dead module burial checklist (6 mandatory steps)
  #10 monitor liveness (DB table freshness) not existence
- File Location Constitution: canonical location per content type
- NEW system_hygiene_audit.py: weekly Monday 07:30 cron checking
  diverged copies / broken hardlinks / zombie processes / orphan data
  files / dead cron scripts / DB freshness -> hygiene_report.json + XMPP
- specs/hygiene.json: module spec per red line #1
- Verified: audit found 5 real issues on first run, all fixed, re-run clean
2026-07-20 19:04:05 +08:00

45 lines
1.4 KiB
Python

import json, shutil, uuid
from datetime import datetime
jf = '/home/hmo/.hermes/profiles/position-analyst/cron/jobs.json'
shutil.copy(jf, jf + '.bak-20260720-hygiene')
d = json.load(open(jf))
is_list = isinstance(d, list)
jobs = d if is_list else d.get('jobs', [])
# 防重复
if any(j.get('script') == 'system_hygiene_audit.py' for j in jobs):
print('job already exists')
else:
job = {
"id": uuid.uuid4().hex[:12],
"name": "系统卫生审计-每周",
"prompt": "",
"skills": [],
"skill": None,
"model": None,
"provider": None,
"base_url": None,
"script": "system_hygiene_audit.py",
"no_agent": True,
"context_from": None,
"schedule": {"kind": "cron", "expr": "30 7 * * 1", "display": "30 7 * * 1"},
"schedule_display": "30 7 * * 1",
"repeat": {"times": None, "completed": 0},
"enabled": True,
"state": "scheduled",
"paused_at": None,
"paused_reason": None,
"created_at": datetime.now().isoformat(),
"next_run_at": "2026-07-21T07:30:00+08:00",
"last_run_at": None,
"last_status": None,
}
jobs.append(job)
if is_list:
json.dump(jobs, open(jf, 'w'), ensure_ascii=False, indent=2)
else:
d['jobs'] = jobs
json.dump(d, open(jf, 'w'), ensure_ascii=False, indent=2)
print('added job: 系统卫生审计-每周 (Mon 07:30)')
print('total jobs:', len(jobs))