homelab-configs/apps/home-security-agent/agent/pg_store.py

"""PostgreSQL — alle observaties voor dashboard :8765."""
from __future__ import annotations

import json
import logging
import os
from datetime import datetime, timezone
from typing import Any, Optional

import psycopg2
import psycopg2.extras

from agent.brain import AgentDecision
from agent.observer import ObservationReport

logger = logging.getLogger(__name__)


def _pg_enabled() -> bool:
    return os.getenv("PG_DISABLED", "").lower() not in ("1", "true", "yes")


def _connect():
    url = os.getenv("DATABASE_URL", "").strip()
    if url:
        return psycopg2.connect(url)
    return psycopg2.connect(
        host=os.getenv("PG_HOST", "192.168.1.211"),
        port=int(os.getenv("PG_PORT", "5433")),
        user=os.getenv("PG_USER", "mo"),
        password=os.getenv("PG_PASSWORD", ""),
        dbname=os.getenv("PG_DATABASE", "homelab"),
    )


def _parse_ts(ts: str) -> datetime:
    if ts.endswith("Z"):
        ts = ts[:-1] + "+00:00"
    return datetime.fromisoformat(ts)


def persist_observation(report: ObservationReport, decision: AgentDecision) -> Optional[int]:
    """Schrijf run + findings naar agent.*; retourneert run_id."""
    if not _pg_enabled():
        return None
    summary = report.to_dict().get("summary") or {}
    decision_json = {
        "alert": decision.alert,
        "severity": decision.severity,
        "title": decision.title,
        "body": decision.body,
        "fingerprint": decision.fingerprint,
        "actions": decision.actions,
    }
    try:
        conn = _connect()
        conn.autocommit = False
        with conn.cursor() as cur:
            cur.execute("SET search_path TO agent, public;")
            cur.execute(
                """
                INSERT INTO observation_runs
                  (observed_at, total_checks, failed_checks, summary, decision)
                VALUES (%s, %s, %s, %s::jsonb, %s::jsonb)
                RETURNING id
                """,
                (
                    _parse_ts(report.timestamp),
                    int(summary.get("total", len(report.findings))),
                    int(summary.get("failed", sum(1 for f in report.findings if not f.ok))),
                    json.dumps(report.to_dict(), default=str),
                    json.dumps(decision_json, default=str),
                ),
            )
            run_id = cur.fetchone()[0]
            rows = [
                (
                    run_id,
                    f.kind,
                    f.name,
                    f.ok,
                    f.detail,
                    json.dumps(f.meta or {}, default=str),
                )
                for f in report.findings
            ]
            if rows:
                psycopg2.extras.execute_batch(
                    cur,
                    """
                    INSERT INTO findings (run_id, kind, name, ok, detail, meta)
                    VALUES (%s, %s, %s, %s, %s, %s::jsonb)
                    """,
                    rows,
                )
        conn.commit()
        conn.close()
        logger.info("PostgreSQL run %s: %d findings", run_id, len(rows))
        return int(run_id)
    except Exception as e:
        logger.warning("PostgreSQL observatie mislukt: %s", e)
        return None


def persist_incident(
    run_id: Optional[int],
    fingerprint: str,
    severity: str,
    title: str,
    body: str,
    notified: bool,
) -> None:
    if not _pg_enabled():
        return
    try:
        conn = _connect()
        conn.autocommit = True
        with conn.cursor() as cur:
            cur.execute("SET search_path TO agent, public;")
            cur.execute(
                """
                INSERT INTO incidents (run_id, fingerprint, severity, title, body, notified)
                VALUES (%s, %s, %s, %s, %s, %s)
                """,
                (run_id, fingerprint, severity, title, body, notified),
            )
        conn.close()
    except Exception as e:
        logger.warning("PostgreSQL incident mislukt: %s", e)


def was_notified_recently_pg(fingerprint: str, minutes: int) -> bool:
    if not _pg_enabled():
        return False
    try:
        conn = _connect()
        with conn.cursor() as cur:
            cur.execute("SET search_path TO agent, public;")
            cur.execute(
                """
                SELECT 1 FROM incidents
                WHERE fingerprint = %s AND notified = true
                  AND created_at > NOW() - (%s || ' minutes')::interval
                LIMIT 1
                """,
                (fingerprint, str(int(minutes))),
            )
            row = cur.fetchone()
        conn.close()
        return row is not None
    except Exception:
        return False