26 lines
1.7 KiB
Python
26 lines
1.7 KiB
Python
#!/usr/bin/env python3
|
|
from pathlib import Path
|
|
import os, sqlite3, json, time, uuid
|
|
root=Path(os.getenv('AIOS_WEBOPS_ROOT', Path(__file__).resolve().parents[1]/'active-runtime'))
|
|
db_path=Path(os.getenv('AIOS_WEBOPS_DB', root/'data'/'webops.sqlite3'))
|
|
if not db_path.exists():
|
|
db_path=root/'data'/'webops_v5.sqlite3'
|
|
db_path.parent.mkdir(parents=True, exist_ok=True)
|
|
con=sqlite3.connect(db_path); con.row_factory=sqlite3.Row
|
|
now=time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime())
|
|
def uid(p): return f'{p}-{int(time.time()*1000)}-{uuid.uuid4().hex[:8]}'
|
|
findings=[]
|
|
try:
|
|
roles=[r['role'] for r in con.execute('SELECT role FROM routing_policy ORDER BY failover_priority')]
|
|
if roles!=['owner','admin','user']:
|
|
findings.append(('routing','high','角色迁移顺序异常',str(roles),'修复为 owner/admin/user'))
|
|
bad=list(con.execute("SELECT username,role FROM users WHERE role NOT IN ('owner','admin','user')"))
|
|
if bad:
|
|
findings.append(('security','critical','存在非法角色',json.dumps([dict(x) for x in bad],ensure_ascii=False),'迁移非法角色为 user'))
|
|
except Exception as e:
|
|
findings.append(('schema','medium','诊断失败',str(e),'运行迁移后复测'))
|
|
for cat,sev,title,detail,fix in findings or [('health','low','未发现阻塞问题','静态策略检查通过','无须处理')]:
|
|
con.execute('INSERT INTO self_optimization_suggestions(id,category,severity,title,detail,proposed_fix,safe_auto_fix,status,created_at) VALUES(?,?,?,?,?,?,?,?,?)',(uid('suggest'),cat,sev,title,detail,fix,1 if sev in ('low','medium') else 0,'open',now))
|
|
con.commit(); con.close()
|
|
print(json.dumps({'diagnosis':'ok','findings':len(findings)},ensure_ascii=False))
|