PYTHON database.py

import sqlite3
import os
from contextlib import contextmanager

DB_PATH = os.environ.get("DB_PATH", "/data/analytics.db")


def get_connection():
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row
    conn.execute("PRAGMA journal_mode=WAL")
    conn.execute("PRAGMA synchronous=NORMAL")
    return conn


@contextmanager
def get_db():
    conn = get_connection()
    try:
        yield conn
        conn.commit()
    finally:
        conn.close()


def init_db():
    """Initialize the database schema."""
    with get_db() as conn:
        conn.executescript("""
            CREATE TABLE IF NOT EXISTS domains (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                domain TEXT UNIQUE NOT NULL,
                created_at DATETIME DEFAULT CURRENT_TIMESTAMP
            );

            CREATE TABLE IF NOT EXISTS visits (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                domain_id INTEGER NOT NULL,
                page TEXT,
                referrer TEXT,
                country TEXT,
                city TEXT,
                browser TEXT,
                os TEXT,
                device TEXT,
                is_bot INTEGER DEFAULT 0,
                ip_hash TEXT,
                timestamp DATETIME DEFAULT CURRENT_TIMESTAMP,
                FOREIGN KEY (domain_id) REFERENCES domains(id)
            );

            CREATE INDEX IF NOT EXISTS idx_visits_domain_timestamp
                ON visits(domain_id, timestamp);
            CREATE INDEX IF NOT EXISTS idx_visits_timestamp
                ON visits(timestamp);
            CREATE INDEX IF NOT EXISTS idx_visits_page
                ON visits(domain_id, page);
            CREATE INDEX IF NOT EXISTS idx_visits_referrer
                ON visits(domain_id, referrer);
            CREATE INDEX IF NOT EXISTS idx_visits_country
                ON visits(domain_id, country);

            -- Pre-computed daily stats for fast dashboard queries
            CREATE TABLE IF NOT EXISTS daily_stats (
                domain_id INTEGER NOT NULL,
                day DATE NOT NULL,
                total_visits INTEGER DEFAULT 0,
                unique_ips INTEGER DEFAULT 0,
                PRIMARY KEY (domain_id, day),
                FOREIGN KEY (domain_id) REFERENCES domains(id)
            );

            -- Hourly stats for time-series charts
            CREATE TABLE IF NOT EXISTS hourly_stats (
                domain_id INTEGER NOT NULL,
                hour DATETIME NOT NULL,
                visits INTEGER DEFAULT 0,
                PRIMARY KEY (domain_id, hour),
                FOREIGN KEY (domain_id) REFERENCES domains(id)
            );
        """)


def get_domain_id(conn, domain: str):
    """Get or create a domain entry."""
    row = conn.execute("SELECT id FROM domains WHERE domain = ?", (domain,)).fetchone()
    if row:
        return row["id"]
    cursor = conn.execute("INSERT INTO domains (domain) VALUES (?)", (domain,))
    return cursor.lastrowid