Your Name
b19975c369
- AI Agent Prompt 移除不存在的 etcdutl 命令引用 - Prompt 增加 containerd(ctr/nerdctl) 和 Linux 基础调优命令示例 - processing_runs 表新增 context_json 字段，保存 Agent 执行上下文 - 每次命令执行后自动保存上下文到 DB
212 lines
9.1 KiB
Python
212 lines
9.1 KiB
Python
"""SQLite 持久化诊断历史、AI 分析和 Agent 处理记录。"""
|
|
import json
|
|
import os
|
|
import sqlite3
|
|
import uuid
|
|
from datetime import datetime
|
|
from typing import Optional
|
|
|
|
|
|
def _now() -> str:
|
|
return datetime.now().isoformat(timespec="seconds")
|
|
|
|
|
|
class HistoryStore:
|
|
def __init__(self, path: str):
|
|
self.path = path
|
|
parent = os.path.dirname(os.path.abspath(path))
|
|
os.makedirs(parent, exist_ok=True)
|
|
self._init_db()
|
|
|
|
def _connect(self):
|
|
connection = sqlite3.connect(self.path, timeout=30)
|
|
connection.row_factory = sqlite3.Row
|
|
connection.execute("PRAGMA journal_mode=WAL")
|
|
connection.execute("PRAGMA foreign_keys=ON")
|
|
return connection
|
|
|
|
def _init_db(self):
|
|
with self._connect() as db:
|
|
db.executescript("""
|
|
CREATE TABLE IF NOT EXISTS diagnoses (
|
|
id TEXT PRIMARY KEY,
|
|
created_at TEXT NOT NULL,
|
|
namespace TEXT NOT NULL,
|
|
score INTEGER NOT NULL,
|
|
status TEXT NOT NULL,
|
|
critical_count INTEGER NOT NULL DEFAULT 0,
|
|
warning_count INTEGER NOT NULL DEFAULT 0,
|
|
elapsed_seconds REAL NOT NULL DEFAULT 0,
|
|
diagnosis_json TEXT NOT NULL,
|
|
report TEXT NOT NULL
|
|
);
|
|
CREATE INDEX IF NOT EXISTS idx_diagnoses_created_at ON diagnoses(created_at DESC);
|
|
CREATE INDEX IF NOT EXISTS idx_diagnoses_status ON diagnoses(status);
|
|
|
|
CREATE TABLE IF NOT EXISTS processing_runs (
|
|
id TEXT PRIMARY KEY,
|
|
diagnosis_id TEXT NOT NULL,
|
|
kind TEXT NOT NULL,
|
|
mode TEXT NOT NULL DEFAULT '',
|
|
question TEXT NOT NULL DEFAULT '',
|
|
status TEXT NOT NULL,
|
|
summary TEXT NOT NULL DEFAULT '',
|
|
started_at TEXT NOT NULL,
|
|
finished_at TEXT,
|
|
context_json TEXT NOT NULL DEFAULT '',
|
|
FOREIGN KEY(diagnosis_id) REFERENCES diagnoses(id) ON DELETE CASCADE
|
|
);
|
|
CREATE INDEX IF NOT EXISTS idx_runs_diagnosis ON processing_runs(diagnosis_id, started_at);
|
|
|
|
CREATE TABLE IF NOT EXISTS processing_events (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
run_id TEXT NOT NULL,
|
|
sequence INTEGER NOT NULL,
|
|
type TEXT NOT NULL,
|
|
created_at TEXT NOT NULL,
|
|
payload_json TEXT NOT NULL,
|
|
FOREIGN KEY(run_id) REFERENCES processing_runs(id) ON DELETE CASCADE,
|
|
UNIQUE(run_id, sequence)
|
|
);
|
|
""")
|
|
# 兼容旧表：确保 context_json 列存在
|
|
self._ensure_column(db, "processing_runs", "context_json", "TEXT NOT NULL DEFAULT ''")
|
|
|
|
def _ensure_column(self, db, table: str, column: str, col_def: str):
|
|
try:
|
|
db.execute(f"ALTER TABLE {table} ADD COLUMN {column} {col_def}")
|
|
except sqlite3.OperationalError as e:
|
|
if "duplicate column" not in str(e).lower():
|
|
raise
|
|
|
|
def save_diagnosis(self, diagnosis: dict, report: str) -> str:
|
|
record_id = uuid.uuid4().hex
|
|
created_at = diagnosis.get("timestamp") or _now()
|
|
with self._connect() as db:
|
|
db.execute(
|
|
"""INSERT INTO diagnoses
|
|
(id, created_at, namespace, score, status, critical_count, warning_count,
|
|
elapsed_seconds, diagnosis_json, report)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
|
|
(
|
|
record_id, created_at, diagnosis.get("namespace", "all"),
|
|
int(diagnosis.get("score", 0)), diagnosis.get("status", "unknown"),
|
|
int(diagnosis.get("critical_count", 0)), int(diagnosis.get("warning_count", 0)),
|
|
float(diagnosis.get("elapsed_seconds", 0)),
|
|
json.dumps(diagnosis, ensure_ascii=False), report,
|
|
),
|
|
)
|
|
return record_id
|
|
|
|
def list_diagnoses(self, status: str = "", namespace: str = "", limit: int = 100, offset: int = 0) -> list[dict]:
|
|
conditions = []
|
|
params: list = []
|
|
if status:
|
|
conditions.append("status = ?")
|
|
params.append(status)
|
|
if namespace:
|
|
conditions.append("namespace = ?")
|
|
params.append(namespace)
|
|
where = " WHERE " + " AND ".join(conditions) if conditions else ""
|
|
params.extend([max(1, min(limit, 500)), max(0, offset)])
|
|
with self._connect() as db:
|
|
rows = db.execute(
|
|
f"""SELECT id, created_at, namespace, score, status, critical_count,
|
|
warning_count, elapsed_seconds,
|
|
(SELECT COUNT(*) FROM processing_runs r WHERE r.diagnosis_id = diagnoses.id) AS run_count
|
|
FROM diagnoses{where}
|
|
ORDER BY created_at DESC, rowid DESC LIMIT ? OFFSET ?""",
|
|
params,
|
|
).fetchall()
|
|
return [dict(row) for row in rows]
|
|
|
|
def get_diagnosis(self, record_id: str) -> Optional[dict]:
|
|
with self._connect() as db:
|
|
row = db.execute("SELECT * FROM diagnoses WHERE id = ?", (record_id,)).fetchone()
|
|
if row is None:
|
|
return None
|
|
run_rows = db.execute(
|
|
"SELECT * FROM processing_runs WHERE diagnosis_id = ? ORDER BY started_at, rowid",
|
|
(record_id,),
|
|
).fetchall()
|
|
runs = []
|
|
for run_row in run_rows:
|
|
run = dict(run_row)
|
|
event_rows = db.execute(
|
|
"SELECT type, created_at, payload_json FROM processing_events WHERE run_id = ? ORDER BY sequence",
|
|
(run["id"],),
|
|
).fetchall()
|
|
run["events"] = [
|
|
{"type": event["type"], "created_at": event["created_at"], **json.loads(event["payload_json"])}
|
|
for event in event_rows
|
|
]
|
|
run["context_json"] = run.get("context_json") or ""
|
|
if run["context_json"]:
|
|
run["context"] = json.loads(run["context_json"])
|
|
else:
|
|
run["context"] = None
|
|
runs.append(run)
|
|
return {
|
|
"id": row["id"], "created_at": row["created_at"],
|
|
"diagnosis": json.loads(row["diagnosis_json"]),
|
|
"report": row["report"], "runs": runs,
|
|
}
|
|
|
|
def latest(self) -> Optional[dict]:
|
|
records = self.list_diagnoses(limit=1)
|
|
return self.get_diagnosis(records[0]["id"]) if records else None
|
|
|
|
def create_run(self, diagnosis_id: str, kind: str, mode: str = "", question: str = "") -> str:
|
|
run_id = uuid.uuid4().hex
|
|
with self._connect() as db:
|
|
db.execute(
|
|
"""INSERT INTO processing_runs
|
|
(id, diagnosis_id, kind, mode, question, status, started_at)
|
|
VALUES (?, ?, ?, ?, ?, 'running', ?)""",
|
|
(run_id, diagnosis_id, kind, mode, question, _now()),
|
|
)
|
|
return run_id
|
|
|
|
def update_context(self, run_id: str, context: dict):
|
|
"""保存/更新 Agent 执行上下文到 processing_runs，支持断点续传。"""
|
|
with self._connect() as db:
|
|
db.execute(
|
|
"UPDATE processing_runs SET context_json = ? WHERE id = ?",
|
|
(json.dumps(context, ensure_ascii=False), run_id),
|
|
)
|
|
|
|
def get_context(self, run_id: str) -> Optional[dict]:
|
|
"""读取 Agent 执行上下文，如果不存在或为空则返回 None。"""
|
|
with self._connect() as db:
|
|
row = db.execute(
|
|
"SELECT context_json, diagnosis_id, report FROM processing_runs pr JOIN diagnoses d ON pr.diagnosis_id = d.id WHERE pr.id = ?",
|
|
(run_id,),
|
|
).fetchone()
|
|
if row is None:
|
|
return None
|
|
ctx_str = row["context_json"] or ""
|
|
if not ctx_str:
|
|
return None
|
|
ctx = json.loads(ctx_str)
|
|
ctx["report"] = row["report"]
|
|
return ctx
|
|
|
|
def append_event(self, run_id: str, event_type: str, payload: dict):
|
|
with self._connect() as db:
|
|
sequence = db.execute(
|
|
"SELECT COALESCE(MAX(sequence), 0) + 1 FROM processing_events WHERE run_id = ?",
|
|
(run_id,),
|
|
).fetchone()[0]
|
|
db.execute(
|
|
"""INSERT INTO processing_events (run_id, sequence, type, created_at, payload_json)
|
|
VALUES (?, ?, ?, ?, ?)""",
|
|
(run_id, sequence, event_type, _now(), json.dumps(payload, ensure_ascii=False)),
|
|
)
|
|
|
|
def finish_run(self, run_id: str, status: str, summary: str = ""):
|
|
with self._connect() as db:
|
|
db.execute(
|
|
"UPDATE processing_runs SET status = ?, summary = ?, finished_at = ? WHERE id = ?",
|
|
(status, summary, _now(), run_id),
|
|
)
|