74 lines
2.3 KiB
Python
74 lines
2.3 KiB
Python
import sqlite3
|
|
from pathlib import Path
|
|
from typing import Any, Dict
|
|
|
|
from app_state import StateStore
|
|
|
|
|
|
def health_payload(
|
|
data_dir: Path,
|
|
session_dir: Path,
|
|
state_store: StateStore,
|
|
continuous_snapshot: Dict[str, Any],
|
|
job_queue_size: int,
|
|
) -> Dict[str, Any]:
|
|
checks = {
|
|
"data_dir": _dir_check(data_dir, writable=True),
|
|
"session_dir": _dir_check(session_dir, writable=True),
|
|
"state_file": _state_check(state_store),
|
|
"sqlite": _sqlite_check(),
|
|
"continuous": _continuous_check(continuous_snapshot),
|
|
"job_queue": {"ok": True, "size": job_queue_size},
|
|
}
|
|
ok = all(item.get("ok", False) for item in checks.values())
|
|
return {"ok": ok, "status": "ok" if ok else "degraded", "checks": checks}
|
|
|
|
|
|
def _dir_check(path: Path, writable: bool = False) -> Dict[str, Any]:
|
|
path.mkdir(parents=True, exist_ok=True)
|
|
ok = path.exists() and path.is_dir()
|
|
payload: Dict[str, Any] = {"ok": ok, "path": str(path)}
|
|
if writable:
|
|
probe = path / ".healthcheck"
|
|
try:
|
|
probe.write_text("ok", encoding="utf-8")
|
|
probe.unlink(missing_ok=True)
|
|
payload["writable"] = True
|
|
except OSError as exc:
|
|
payload.update({"ok": False, "writable": False, "error": str(exc)})
|
|
return payload
|
|
|
|
|
|
def _state_check(state_store: StateStore) -> Dict[str, Any]:
|
|
try:
|
|
state = state_store.load()
|
|
return {
|
|
"ok": True,
|
|
"path": str(state_store.path),
|
|
"has_api_credentials": bool(state.get("api_id") and state.get("api_hash")),
|
|
"tracked_channels": len(state.get("channels", {})),
|
|
}
|
|
except Exception as exc:
|
|
return {"ok": False, "path": str(state_store.path), "error": str(exc)}
|
|
|
|
|
|
def _sqlite_check() -> Dict[str, Any]:
|
|
try:
|
|
conn = sqlite3.connect(":memory:")
|
|
conn.execute("SELECT 1")
|
|
conn.close()
|
|
return {"ok": True}
|
|
except sqlite3.Error as exc:
|
|
return {"ok": False, "error": str(exc)}
|
|
|
|
|
|
def _continuous_check(snapshot: Dict[str, Any]) -> Dict[str, Any]:
|
|
status = snapshot.get("status", {})
|
|
config = snapshot.get("config", {})
|
|
return {
|
|
"ok": True,
|
|
"enabled": bool(config.get("enabled")),
|
|
"running": bool(status.get("running")),
|
|
"last_error": status.get("last_error"),
|
|
}
|