95 lines
2.6 KiB
Python
95 lines
2.6 KiB
Python
#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 |