"""Database layer — Step 1 of the enterprise re-platform. Defaults to zero-setup SQLite (data.db next to the app). To move to Postgres, set the DATABASE_URL environment variable, e.g. export DATABASE_URL=postgresql://user:pass@host:5432/seo and `pip install psycopg2-binary`. The SQL below is written to run unchanged on both engines. """ from __future__ import annotations import json import os import sqlite3 import threading from datetime import datetime from pathlib import Path _LOCK = threading.Lock() _DB_PATH = Path(__file__).parent.parent / "data.db" SCHEMA = """ CREATE TABLE IF NOT EXISTS crawls ( id INTEGER PRIMARY KEY, site TEXT NOT NULL, started_at TEXT NOT NULL, overall_score REAL, pages INTEGER ); CREATE TABLE IF NOT EXISTS page_fingerprints ( id INTEGER PRIMARY KEY, crawl_id INTEGER NOT NULL, url TEXT NOT NULL, fingerprint TEXT NOT NULL -- JSON of element values ); CREATE TABLE IF NOT EXISTS element_changes ( id INTEGER PRIMARY KEY, site TEXT NOT NULL, detected_at TEXT NOT NULL, url TEXT NOT NULL, element TEXT NOT NULL, -- e.g. 'canonical', 'title', 'schema_types' old_value TEXT, new_value TEXT, severity TEXT NOT NULL, -- 'critical' | 'warning' | 'info' deploy_id INTEGER -- correlated deploy, if any ); CREATE TABLE IF NOT EXISTS deploys ( id INTEGER PRIMARY KEY, site TEXT NOT NULL, received_at TEXT NOT NULL, ref TEXT, -- deploy id / git sha / tag from CI source TEXT, -- 'github', 'gitlab', 'manual', ... note TEXT ); CREATE TABLE IF NOT EXISTS citation_checks ( id INTEGER PRIMARY KEY, site TEXT NOT NULL, checked_at TEXT NOT NULL, engine TEXT NOT NULL, -- 'openai' | 'anthropic' | 'perplexity' question TEXT NOT NULL, mentioned INTEGER NOT NULL, -- brand named in the answer (0/1) cited INTEGER NOT NULL, -- site URL cited/linked (0/1) competitors_mentioned TEXT, -- JSON list answer_snippet TEXT ); CREATE TABLE IF NOT EXISTS approvals ( id INTEGER PRIMARY KEY, site TEXT NOT NULL, created_at TEXT NOT NULL, status TEXT NOT NULL, -- pending | approved | rejected | applied | failed summary TEXT NOT NULL, -- JSON: {file: [fixes]} diff TEXT, decided_at TEXT, note TEXT, kind TEXT NOT NULL DEFAULT 'remediation' -- 'remediation' | 'content' ); CREATE INDEX IF NOT EXISTS ix_fp_crawl ON page_fingerprints(crawl_id); CREATE INDEX IF NOT EXISTS ix_changes_site ON element_changes(site, detected_at); CREATE INDEX IF NOT EXISTS ix_cit_site ON citation_checks(site, checked_at); """ def _connect(): url = os.environ.get("DATABASE_URL", "") if url.startswith("postgres"): import psycopg2 # optional, installed only when scaling up conn = psycopg2.connect(url) conn.autocommit = True return conn, "%s" conn = sqlite3.connect(_DB_PATH, check_same_thread=False) conn.row_factory = sqlite3.Row return conn, "?" class DB: """Tiny DAL usable from Flask threads; one shared connection + lock.""" def __init__(self): self.conn, self.ph = _connect() with _LOCK: cur = self.conn.cursor() for stmt in SCHEMA.split(";"): if stmt.strip(): cur.execute(stmt) self.conn.commit() # migration: older data.db files predate the 'kind' column try: cur.execute("ALTER TABLE approvals ADD COLUMN kind TEXT " "NOT NULL DEFAULT 'remediation'") self.conn.commit() except Exception: pass # column already exists def q(self, sql: str, args: tuple = ()): # query with _LOCK: cur = self.conn.cursor() cur.execute(sql.replace("?", self.ph) if self.ph != "?" else sql, args) try: rows = cur.fetchall() except Exception: rows = [] self.conn.commit() return rows def insert(self, sql: str, args: tuple = ()) -> int: with _LOCK: cur = self.conn.cursor() cur.execute(sql.replace("?", self.ph) if self.ph != "?" else sql, args) self.conn.commit() return cur.lastrowid # ---------- convenience helpers ---------- def new_crawl(self, site: str) -> int: return self.insert( "INSERT INTO crawls (site, started_at) VALUES (?, ?)", (site, datetime.now().isoformat(timespec="seconds"))) def finish_crawl(self, crawl_id: int, score: float, pages: int): self.q("UPDATE crawls SET overall_score=?, pages=? WHERE id=?", (score, pages, crawl_id)) def save_fingerprint(self, crawl_id: int, url: str, fp: dict): self.insert( "INSERT INTO page_fingerprints (crawl_id, url, fingerprint) VALUES (?,?,?)", (crawl_id, url, json.dumps(fp, sort_keys=True))) def previous_crawl_id(self, site: str, before_crawl: int) -> int | None: rows = self.q("SELECT id FROM crawls WHERE site=? AND id dict[str, dict]: rows = self.q("SELECT url, fingerprint FROM page_fingerprints " "WHERE crawl_id=?", (crawl_id,)) return {r[0]: json.loads(r[1]) for r in rows} def record_deploy(self, site: str, ref: str, source: str, note: str = "") -> int: return self.insert( "INSERT INTO deploys (site, received_at, ref, source, note) " "VALUES (?,?,?,?,?)", (site, datetime.now().isoformat(timespec="seconds"), ref, source, note)) def latest_deploy_since(self, site: str, since_iso: str): rows = self.q("SELECT id, received_at, ref, source FROM deploys " "WHERE site=? AND received_at>=? ORDER BY id DESC LIMIT 1", (site, since_iso)) return rows[0] if rows else None def record_change(self, site: str, url: str, element: str, old, new, severity: str, deploy_id: int | None): self.insert( "INSERT INTO element_changes (site, detected_at, url, element, " "old_value, new_value, severity, deploy_id) VALUES (?,?,?,?,?,?,?,?)", (site, datetime.now().isoformat(timespec="seconds"), url, element, json.dumps(old), json.dumps(new), severity, deploy_id)) def recent_changes(self, site: str, limit: int = 50): rows = self.q( "SELECT c.detected_at, c.url, c.element, c.old_value, c.new_value, " "c.severity, d.ref, d.source, d.received_at " "FROM element_changes c LEFT JOIN deploys d ON c.deploy_id = d.id " "WHERE c.site=? ORDER BY c.id DESC LIMIT ?", (site, limit)) return [{"detected_at": r[0], "url": r[1], "element": r[2], "old": json.loads(r[3]) if r[3] else None, "new": json.loads(r[4]) if r[4] else None, "severity": r[5], "deploy": ({"ref": r[6], "source": r[7], "at": r[8]} if r[6] else None)} for r in rows] def record_citation(self, site: str, engine: str, question: str, mentioned: bool, cited: bool, competitors: list[str], snippet: str): self.insert( "INSERT INTO citation_checks (site, checked_at, engine, question, " "mentioned, cited, competitors_mentioned, answer_snippet) " "VALUES (?,?,?,?,?,?,?,?)", (site, datetime.now().isoformat(timespec="seconds"), engine, question, int(mentioned), int(cited), json.dumps(competitors), snippet[:400])) def citation_summary(self, site: str): rows = self.q( "SELECT engine, COUNT(*), SUM(mentioned), SUM(cited) " "FROM citation_checks WHERE site=? GROUP BY engine", (site,)) per_engine = [{"engine": r[0], "checks": r[1], "mention_rate": round(100 * (r[2] or 0) / r[1], 1), "citation_rate": round(100 * (r[3] or 0) / r[1], 1)} for r in rows] latest = self.q( "SELECT checked_at, engine, question, mentioned, cited, " "competitors_mentioned, answer_snippet FROM citation_checks " "WHERE site=? ORDER BY id DESC LIMIT 30", (site,)) return { "per_engine": per_engine, "latest": [{"at": r[0], "engine": r[1], "question": r[2], "mentioned": bool(r[3]), "cited": bool(r[4]), "competitors": json.loads(r[5] or "[]"), "snippet": r[6]} for r in latest], } _db: DB | None = None def get_db() -> DB: global _db if _db is None: _db = DB() return _db