Bot Version 1.0.0

app/db.py

@@ -0,0 +1,95 @@
+#app.db
+
+import sqlite3
+from datetime import datetime, timezone, timedelta
+
+from .config import settings
+DB_PATH = settings.database_path
+
+def init_db() -> None:
+    conn = sqlite3.connect(DB_PATH)
+    cur = conn.cursor()
+
+    cur.execute("""
+    CREATE TABLE IF NOT EXISTS site_checks (
+        id INTEGER PRIMARY KEY AUTOINCREMENT,
+        site_name TEXT NOT NULL,
+        checked_at TEXT NOT NULL,
+        http_status INTEGER,
+        latency_ms INTEGER,
+        result TEXT NOT NULL,
+        error_type TEXT,
+        notes TEXT
+    )
+    """)
+
+    cur.execute("""
+    CREATE TABLE IF NOT EXISTS monthly_reports (
+        id INTEGER PRIMARY KEY AUTOINCREMENT,
+        month_key TEXT NOT NULL,
+        site_name TEXT NOT NULL,
+        total_checks INTEGER NOT NULL,
+        up_checks INTEGER NOT NULL,
+        degraded_checks INTEGER NOT NULL,
+        down_checks INTEGER NOT NULL,
+        uptime_percent REAL NOT NULL,
+        created_at TEXT NOT NULL
+    )
+    """)
+
+    conn.commit()
+    conn.close()
+
+
+def insert_check(
+    site_name: str,
+    checked_at: str,
+    http_status: int | None,
+    latency_ms: int | None,
+    result: str,
+    error_type: str | None,
+    notes: str | None,
+) -> None:
+    conn = sqlite3.connect(DB_PATH)
+    cur = conn.cursor()
+    cur.execute("""
+    INSERT INTO site_checks (
+        site_name, checked_at, http_status, latency_ms, result, error_type, notes
+    ) VALUES (?, ?, ?, ?, ?, ?, ?)
+    """, (site_name, checked_at, http_status, latency_ms, result, error_type, notes))
+    conn.commit()
+    conn.close()
+
+
+def fetch_checks_since(site_name: str, since_dt: datetime) -> list[tuple]:
+    conn = sqlite3.connect(DB_PATH)
+    cur = conn.cursor()
+    cur.execute("""
+    SELECT checked_at, result, http_status, latency_ms, error_type
+    FROM site_checks
+    WHERE site_name = ? AND checked_at >= ?
+    ORDER BY checked_at ASC
+    """, (site_name, since_dt.isoformat()))
+    rows = cur.fetchall()
+    conn.close()
+    return rows
+
+
+def fetch_month_checks(site_name: str, year: int, month: int) -> list[tuple]:
+    start = datetime(year, month, 1, tzinfo=timezone.utc)
+    if month == 12:
+        end = datetime(year + 1, 1, 1, tzinfo=timezone.utc)
+    else:
+        end = datetime(year, month + 1, 1, tzinfo=timezone.utc)
+
+    conn = sqlite3.connect(DB_PATH)
+    cur = conn.cursor()
+    cur.execute("""
+    SELECT checked_at, result, http_status, latency_ms, error_type
+    FROM site_checks
+    WHERE site_name = ? AND checked_at >= ? AND checked_at < ?
+    ORDER BY checked_at ASC
+    """, (site_name, start.isoformat(), end.isoformat()))
+    rows = cur.fetchall()
+    conn.close()
+    return rows