""" SQLite-backed cache for articles and narrative snapshots. Prevents re-processing the same URLs and stores historical trend data. """ import sqlite3 import json import logging from datetime import datetime, timezone, timedelta from pathlib import Path from config import DB_PATH, MAX_ARTICLE_AGE_HOURS logger = logging.getLogger(__name__) _SCHEMA = """ CREATE TABLE IF NOT EXISTS articles ( url TEXT PRIMARY KEY, source TEXT, title TEXT, summary TEXT, published TEXT, type TEXT, score REAL, text TEXT, sentiment TEXT, -- JSON narratives TEXT, -- JSON list entities TEXT, -- JSON fetched_at TEXT ); CREATE TABLE IF NOT EXISTS narrative_snapshots ( id INTEGER PRIMARY KEY AUTOINCREMENT, snapshot_at TEXT, data TEXT -- JSON list of scored narratives ); CREATE TABLE IF NOT EXISTS fear_greed ( timestamp TEXT PRIMARY KEY, value INTEGER, label TEXT ); CREATE TABLE IF NOT EXISTS coin_prices ( id TEXT, snapshot_at TEXT, data TEXT, -- JSON PRIMARY KEY (id, snapshot_at) ); CREATE TABLE IF NOT EXISTS sentiment_history ( snapshot_at TEXT PRIMARY KEY, net_score REAL, index_val INTEGER, label TEXT, bullish INTEGER, bearish INTEGER, neutral INTEGER, per_narrative TEXT, -- JSON: {narrative: net_score} bull_prob REAL, -- #7 probabilistic outputs bear_prob REAL, vol_prob REAL, confidence REAL ); -- #1 Market reaction labeling: forward BTC/ETH moves after each story CREATE TABLE IF NOT EXISTS market_reactions ( url TEXT PRIMARY KEY, published TEXT, sentiment TEXT, -- label at publish time (for later analysis) btc_at_pub REAL, eth_at_pub REAL, btc_1h REAL, btc_4h REAL, btc_24h REAL, eth_1h REAL, eth_4h REAL, eth_24h REAL, label_24h TEXT, -- 'bullish'|'bearish'|'flat' from BTC 24h move updated_at TEXT ); -- #5 On-chain + derivatives signals CREATE TABLE IF NOT EXISTS onchain_signals ( snapshot_at TEXT PRIMARY KEY, funding_btc REAL, funding_eth REAL, oi_btc REAL, oi_btc_change REAL, liq_proxy REAL, stablecoin_flow REAL, data TEXT -- JSON raw ); -- #3 Per-story consensus across duplicate sources CREATE TABLE IF NOT EXISTS story_consensus ( url TEXT PRIMARY KEY, cluster_size INTEGER, sentiment_variance REAL, consensus_strength REAL, snapshot_at TEXT ); -- #6 Social attention velocity: mention counts over time CREATE TABLE IF NOT EXISTS mention_history ( snapshot_at TEXT, entity TEXT, kind TEXT, -- 'coin' | 'narrative' count INTEGER, PRIMARY KEY (snapshot_at, entity, kind) ); -- Magnet track record: snapshot the liquidation magnets, evaluate how price -- actually behaved toward them after a horizon. CREATE TABLE IF NOT EXISTS magnet_snapshots ( snapshot_at TEXT PRIMARY KEY, price REAL, near_price REAL, near_strength INTEGER, near_side TEXT, near_dist REAL, strong_price REAL, strong_strength INTEGER, strong_side TEXT, strong_dist REAL, reached_near INTEGER, reached_strong INTEGER, end_price REAL, evaluated INTEGER DEFAULT 0, updated_at TEXT ); -- Magnet target registry: each distinct magnet level is registered once and -- tracked for 24h; we record if/when price hit it (never re-added while active). CREATE TABLE IF NOT EXISTS magnet_targets ( id INTEGER PRIMARY KEY AUTOINCREMENT, first_seen TEXT, ref_price REAL, price_level REAL, side TEXT, -- above | below strength INTEGER, -- 0-100 at registration dist_pct REAL, -- distance from price at registration is_strongest INTEGER DEFAULT 0, is_nearest_strong INTEGER DEFAULT 0, hit INTEGER DEFAULT 0, hit_at TEXT, time_to_hit REAL, -- hours from first_seen to first touch expired INTEGER DEFAULT 0 ); -- #8 System health dataset CREATE TABLE IF NOT EXISTS system_health ( id INTEGER PRIMARY KEY AUTOINCREMENT, ts TEXT, component TEXT, -- 'source' | 'model' | 'pipeline' event_type TEXT, -- 'source_failure'|'model_fallback'|'disagreement'|'component_failure'|'ok' name TEXT, -- source/model/component name detail TEXT, value REAL ); CREATE TABLE IF NOT EXISTS clients ( id INTEGER PRIMARY KEY AUTOINCREMENT, email TEXT UNIQUE, name TEXT, active INTEGER DEFAULT 1, added_at TEXT ); CREATE TABLE IF NOT EXISTS email_settings ( id INTEGER PRIMARY KEY CHECK (id = 1), enabled INTEGER DEFAULT 0, send_time_1 TEXT DEFAULT '09:00', -- UTC HH:MM send_time_2 TEXT DEFAULT '21:00', subject_prefix TEXT DEFAULT 'Crypto Narrative Brief', last_sent_slot TEXT DEFAULT '' -- 'YYYY-MM-DD#1' to avoid double-send ); CREATE TABLE IF NOT EXISTS email_log ( id INTEGER PRIMARY KEY AUTOINCREMENT, sent_at TEXT, slot TEXT, recipients INTEGER, status TEXT, detail TEXT ); """ class Database: def __init__(self, path: str = DB_PATH): self.path = path try: # Make sure the parent directory exists and is writable parent = Path(path).resolve().parent parent.mkdir(parents=True, exist_ok=True) self._conn = sqlite3.connect(path, check_same_thread=False) except (sqlite3.OperationalError, OSError) as exc: # Fall back to a guaranteed-writable temp location (e.g. read-only # working dir on some cloud hosts). History won't persist across # restarts, but the app stays up. import tempfile fallback = str(Path(tempfile.gettempdir()) / "narrative_analysis.db") logger.warning(f"[DB] Could not open {path} ({exc}); using {fallback}") self.path = fallback self._conn = sqlite3.connect(fallback, check_same_thread=False) self._conn.row_factory = sqlite3.Row self._conn.executescript(_SCHEMA) self._migrate() self._conn.commit() def _migrate(self) -> None: """Add columns introduced after a table was first created. CREATE TABLE IF NOT EXISTS won't alter an existing table, so we add any missing columns here (safe, idempotent).""" wanted = { "sentiment_history": [ ("bull_prob", "REAL"), ("bear_prob", "REAL"), ("vol_prob", "REAL"), ("confidence", "REAL"), ], "clients": [ ("confirmed", "INTEGER DEFAULT 0"), # double opt-in state ("token", "TEXT"), # confirm / unsubscribe token ("source", "TEXT"), # where they signed up ("ref_code", "TEXT"), # this subscriber's own referral code ("referred_by", "TEXT"), # ref_code of whoever referred them ("referrals", "INTEGER DEFAULT 0"), # how many they've referred (confirmed) ], } for table, cols in wanted.items(): try: existing = {r["name"] for r in self._conn.execute(f"PRAGMA table_info({table})")} except Exception: continue for name, typ in cols: if name not in existing: try: self._conn.execute(f"ALTER TABLE {table} ADD COLUMN {name} {typ}") except Exception as exc: logger.debug(f"[DB] migrate {table}.{name}: {exc}") # ------------------------------------------------------------------ # Articles # ------------------------------------------------------------------ def save_articles(self, articles: list[dict]) -> int: """Upserts articles. Returns number inserted.""" inserted = 0 now = datetime.now(timezone.utc).isoformat() for a in articles: url = a.get("url", "") if not url: continue try: self._conn.execute( """ INSERT OR REPLACE INTO articles (url, source, title, summary, published, type, score, text, sentiment, narratives, entities, fetched_at) VALUES (?,?,?,?,?,?,?,?,?,?,?,?) """, ( url, a.get("source", ""), a.get("title", ""), a.get("summary", ""), a.get("published", ""), a.get("type", "news"), a.get("score", 0), a.get("text", ""), json.dumps(a.get("sentiment", {})), json.dumps(a.get("narratives", [])), json.dumps(a.get("entities", {})), now, ), ) inserted += 1 except Exception as exc: logger.debug(f"DB insert failed for {url}: {exc}") self._conn.commit() return inserted def get_recent_articles(self, hours: int = MAX_ARTICLE_AGE_HOURS) -> list[dict]: cutoff = (datetime.now(timezone.utc) - timedelta(hours=hours)).isoformat() rows = self._conn.execute( "SELECT * FROM articles WHERE published >= ? ORDER BY published DESC", (cutoff,), ).fetchall() return [self._row_to_article(r) for r in rows] def article_count(self) -> int: return self._conn.execute("SELECT COUNT(*) FROM articles").fetchone()[0] def get_article(self, url: str) -> dict | None: row = self._conn.execute("SELECT * FROM articles WHERE url=?", (url,)).fetchone() return self._row_to_article(row) if row else None def snapshot_to(self, path: str) -> bool: """Write a clean, consistent copy of the DB to `path` (for backup). Uses SQLite's online backup API — safe on a live connection and, unlike VACUUM, does NOT touch the main connection's transaction state.""" import os try: if os.path.exists(path): os.remove(path) dest = sqlite3.connect(path) try: self._conn.backup(dest) # read-only on source; no COMMIT issued finally: dest.close() return True except Exception as exc: logger.warning(f"[DB] snapshot failed: {exc}") return False @staticmethod def _row_to_article(row: sqlite3.Row) -> dict: d = dict(row) for key in ("sentiment", "narratives", "entities"): if d.get(key): try: d[key] = json.loads(d[key]) except Exception: pass # Re-derive the macro flag from the source (not stored as a column) d["is_macro"] = str(d.get("source", "")).startswith("Macro") return d # ------------------------------------------------------------------ # #1 Market reactions # ------------------------------------------------------------------ def upsert_reaction_seed(self, url: str, published: str, sentiment: str, btc: float | None, eth: float | None) -> None: """Record the publish-time price reference for a story (once).""" if not url: return now = datetime.now(timezone.utc).isoformat() self._conn.execute( """INSERT OR IGNORE INTO market_reactions (url, published, sentiment, btc_at_pub, eth_at_pub, updated_at) VALUES (?,?,?,?,?,?)""", (url, published, sentiment, btc, eth, now), ) self._conn.commit() def reactions_needing_fill(self) -> list[dict]: rows = self._conn.execute( "SELECT * FROM market_reactions WHERE btc_24h IS NULL" ).fetchall() return [dict(r) for r in rows] def update_reaction(self, url: str, fields: dict) -> None: if not fields: return fields["updated_at"] = datetime.now(timezone.utc).isoformat() cols = ", ".join(f"{k}=?" for k in fields) self._conn.execute(f"UPDATE market_reactions SET {cols} WHERE url=?", (*fields.values(), url)) self._conn.commit() def labeled_reactions(self, limit: int = 5000) -> list[dict]: """Stories that now have a 24h BTC label — training data for #2.""" rows = self._conn.execute( "SELECT * FROM market_reactions WHERE label_24h IS NOT NULL ORDER BY published DESC LIMIT ?", (limit,), ).fetchall() return [dict(r) for r in rows] # ------------------------------------------------------------------ # #5 On-chain / derivatives # ------------------------------------------------------------------ def save_onchain(self, sig: dict) -> None: now = datetime.now(timezone.utc).isoformat() self._conn.execute( """INSERT OR REPLACE INTO onchain_signals (snapshot_at, funding_btc, funding_eth, oi_btc, oi_btc_change, liq_proxy, stablecoin_flow, data) VALUES (?,?,?,?,?,?,?,?)""", (now, sig.get("funding_btc"), sig.get("funding_eth"), sig.get("oi_btc"), sig.get("oi_btc_change"), sig.get("liq_proxy"), sig.get("stablecoin_flow"), json.dumps(sig)), ) self._conn.commit() def last_onchain(self) -> dict | None: row = self._conn.execute( "SELECT * FROM onchain_signals ORDER BY snapshot_at DESC LIMIT 1" ).fetchone() return dict(row) if row else None def prev_oi_btc(self) -> float | None: row = self._conn.execute( "SELECT oi_btc FROM onchain_signals WHERE oi_btc IS NOT NULL ORDER BY snapshot_at DESC LIMIT 1" ).fetchone() return row["oi_btc"] if row else None # ------------------------------------------------------------------ # #3 Story consensus # ------------------------------------------------------------------ def save_consensus(self, url: str, cluster_size: int, variance: float, strength: float) -> None: if not url: return self._conn.execute( """INSERT OR REPLACE INTO story_consensus (url, cluster_size, sentiment_variance, consensus_strength, snapshot_at) VALUES (?,?,?,?,?)""", (url, cluster_size, variance, strength, datetime.now(timezone.utc).isoformat()), ) self._conn.commit() # ------------------------------------------------------------------ # #6 Mention history (velocity) # ------------------------------------------------------------------ def save_mentions(self, mentions: dict[str, int], kind: str = "coin") -> None: now = datetime.now(timezone.utc).isoformat() for entity, count in mentions.items(): self._conn.execute( "INSERT OR REPLACE INTO mention_history (snapshot_at, entity, kind, count) VALUES (?,?,?,?)", (now, entity, kind, count), ) self._conn.commit() def mention_series(self, kind: str = "coin", limit: int = 400) -> list[dict]: rows = self._conn.execute( "SELECT snapshot_at, entity, count FROM mention_history WHERE kind=? ORDER BY snapshot_at DESC LIMIT ?", (kind, limit), ).fetchall() return [dict(r) for r in rows] # ------------------------------------------------------------------ # Magnet track record # ------------------------------------------------------------------ def save_magnet_seed(self, liq: dict) -> None: near = liq.get("nearest_magnet") strong = liq.get("strongest_magnet") price = liq.get("current_price") if not price or not near or not strong: return now = datetime.now(timezone.utc).isoformat() self._conn.execute( """INSERT OR IGNORE INTO magnet_snapshots (snapshot_at, price, near_price, near_strength, near_side, near_dist, strong_price, strong_strength, strong_side, strong_dist, updated_at) VALUES (?,?,?,?,?,?,?,?,?,?,?)""", (now, price, near["price"], near["strength"], near["side"], near["dist_pct"], strong["price"], strong["strength"], strong["side"], strong["dist_pct"], now), ) self._conn.commit() def get_magnet_strength_history(self, limit: int = 2000) -> list[dict]: """[{snapshot_at, strength}] — strongest liquidation magnet (0-100) per cycle, chronological. Used to mark high-liquidation points on the chart.""" rows = self._conn.execute( "SELECT snapshot_at, strong_strength FROM magnet_snapshots " "ORDER BY snapshot_at DESC LIMIT ?", (limit,) ).fetchall() return [{"snapshot_at": r["snapshot_at"], "strength": r["strong_strength"]} for r in reversed(rows) if r["strong_strength"] is not None] def magnets_needing_eval(self) -> list[dict]: return [dict(r) for r in self._conn.execute( "SELECT * FROM magnet_snapshots WHERE evaluated=0" ).fetchall()] def update_magnet_eval(self, snapshot_at: str, fields: dict) -> None: fields["evaluated"] = 1 fields["updated_at"] = datetime.now(timezone.utc).isoformat() cols = ", ".join(f"{k}=?" for k in fields) self._conn.execute(f"UPDATE magnet_snapshots SET {cols} WHERE snapshot_at=?", (*fields.values(), snapshot_at)) self._conn.commit() def evaluated_magnets(self, limit: int = 5000) -> list[dict]: return [dict(r) for r in self._conn.execute( "SELECT * FROM magnet_snapshots WHERE evaluated=1 ORDER BY snapshot_at ASC LIMIT ?", (limit,), ).fetchall()] # -- magnet target registry (24h tracking) ------------------------- def add_magnet_target(self, t: dict) -> None: self._conn.execute( """INSERT INTO magnet_targets (first_seen, ref_price, price_level, side, strength, dist_pct, is_strongest, is_nearest_strong) VALUES (?,?,?,?,?,?,?,?)""", (t["first_seen"], t["ref_price"], t["price_level"], t["side"], t["strength"], t["dist_pct"], t.get("is_strongest", 0), t.get("is_nearest_strong", 0)), ) self._conn.commit() def open_magnet_targets(self) -> list[dict]: """Targets still within their 24h window (not expired).""" return [dict(r) for r in self._conn.execute( "SELECT * FROM magnet_targets WHERE expired=0" ).fetchall()] def mark_magnet_hit(self, tid: int, hit_at: str, tth: float) -> None: self._conn.execute( "UPDATE magnet_targets SET hit=1, hit_at=?, time_to_hit=? WHERE id=?", (hit_at, tth, tid)) self._conn.commit() def expire_magnet_target(self, tid: int) -> None: self._conn.execute("UPDATE magnet_targets SET expired=1 WHERE id=?", (tid,)) self._conn.commit() def set_magnet_flags(self, tid: int, strongest: int, nearest_strong: int) -> None: # only ever set to 1 (a level that was ever strongest stays flagged) self._conn.execute( """UPDATE magnet_targets SET is_strongest = MAX(is_strongest, ?), is_nearest_strong = MAX(is_nearest_strong, ?) WHERE id=?""", (strongest, nearest_strong, tid)) self._conn.commit() def resolved_magnet_targets(self, limit: int = 8000) -> list[dict]: """Targets whose outcome is known: hit, or expired without a hit.""" return [dict(r) for r in self._conn.execute( "SELECT * FROM magnet_targets WHERE hit=1 OR expired=1 ORDER BY id DESC LIMIT ?", (limit,)).fetchall()] def magnet_target_active_count(self) -> int: return self._conn.execute( "SELECT COUNT(*) FROM magnet_targets WHERE expired=0 AND hit=0").fetchone()[0] # ------------------------------------------------------------------ # #8 System health # ------------------------------------------------------------------ def log_health(self, component: str, event_type: str, name: str = "", detail: str = "", value: float | None = None) -> None: try: self._conn.execute( "INSERT INTO system_health (ts, component, event_type, name, detail, value) VALUES (?,?,?,?,?,?)", (datetime.now(timezone.utc).isoformat(), component, event_type, name, detail, value), ) self._conn.commit() except Exception: pass def health_summary(self, hours: int = 24) -> dict: cutoff = (datetime.now(timezone.utc) - timedelta(hours=hours)).isoformat() rows = self._conn.execute( "SELECT event_type, COUNT(*) c FROM system_health WHERE ts >= ? GROUP BY event_type", (cutoff,), ).fetchall() by_type = {r["event_type"]: r["c"] for r in rows} recent = self._conn.execute( "SELECT * FROM system_health ORDER BY id DESC LIMIT 30" ).fetchall() return {"by_type": by_type, "recent": [dict(r) for r in recent]} # ------------------------------------------------------------------ # Narrative snapshots # ------------------------------------------------------------------ def save_snapshot(self, scored_narratives: list[dict]) -> None: now = datetime.now(timezone.utc).isoformat() self._conn.execute( "INSERT INTO narrative_snapshots (snapshot_at, data) VALUES (?, ?)", (now, json.dumps(scored_narratives)), ) self._conn.commit() def get_snapshots(self, limit: int = 48) -> list[dict]: rows = self._conn.execute( "SELECT snapshot_at, data FROM narrative_snapshots ORDER BY id DESC LIMIT ?", (limit,), ).fetchall() return [{"snapshot_at": r["snapshot_at"], "data": json.loads(r["data"])} for r in rows] # ------------------------------------------------------------------ # Sentiment history (for charting + spike detection) # ------------------------------------------------------------------ def save_sentiment_history(self, sentiment_24h: dict) -> None: o = sentiment_24h.get("overall", {}) per = {n["name"]: n["net_score"] for n in sentiment_24h.get("per_narrative", [])} now = datetime.now(timezone.utc).isoformat() self._conn.execute( """INSERT OR REPLACE INTO sentiment_history (snapshot_at, net_score, index_val, label, bullish, bearish, neutral, per_narrative, bull_prob, bear_prob, vol_prob, confidence) VALUES (?,?,?,?,?,?,?,?,?,?,?,?)""", (now, o.get("net_score", 0), o.get("index", 50), o.get("label", "NEUTRAL"), o.get("bullish", 0), o.get("bearish", 0), o.get("neutral", 0), json.dumps(per), o.get("bull_prob"), o.get("bear_prob"), o.get("vol_prob"), o.get("confidence")), ) self._conn.commit() def get_sentiment_history(self, limit: int = 96) -> list[dict]: rows = self._conn.execute( "SELECT * FROM sentiment_history ORDER BY snapshot_at DESC LIMIT ?", (limit,) ).fetchall() out = [] for r in reversed(rows): # chronological order for charting d = dict(r) try: d["per_narrative"] = json.loads(d["per_narrative"]) except Exception: d["per_narrative"] = {} out.append(d) return out def sentiment_span(self) -> dict: """How much sentiment history we hold: count + first/last + days span. Used to show how many days the relative-index baseline is built on.""" row = self._conn.execute( "SELECT COUNT(*) AS n, MIN(snapshot_at) AS first, MAX(snapshot_at) AS last " "FROM sentiment_history" ).fetchone() n = int(row["n"]) if row and row["n"] else 0 first, last = (row["first"], row["last"]) if row else (None, None) days = None if first and last: try: f = datetime.fromisoformat(first) l = datetime.fromisoformat(last) days = round((l - f).total_seconds() / 86400, 1) except Exception: days = None return {"count": n, "first": first, "last": last, "days": days} def get_index_near(self, hours_ago: float, tolerance_hours: float = 8.0) -> dict | None: """ Return the recorded sentiment snapshot closest to `hours_ago` hours in the past, if one exists within `tolerance_hours`. Used for a reliable 'vs ~24h ago' comparison instead of recomputing from stale articles. """ target = datetime.now(timezone.utc) - timedelta(hours=hours_ago) target_iso = target.isoformat() row = self._conn.execute( """SELECT *, ABS(strftime('%s', snapshot_at) - strftime('%s', ?)) AS diff FROM sentiment_history ORDER BY diff ASC LIMIT 1""", (target_iso,), ).fetchone() if not row: return None d = dict(row) if d.get("diff") is not None and d["diff"] > tolerance_hours * 3600: return None return d # ------------------------------------------------------------------ # Fear & Greed # ------------------------------------------------------------------ def save_fear_greed(self, entries: list[dict]) -> None: for e in entries: self._conn.execute( "INSERT OR REPLACE INTO fear_greed (timestamp, value, label) VALUES (?,?,?)", (e["timestamp"], e["value"], e["label"]), ) self._conn.commit() def get_fear_greed(self, limit: int = 7) -> list[dict]: rows = self._conn.execute( "SELECT * FROM fear_greed ORDER BY timestamp DESC LIMIT ?", (limit,) ).fetchall() return [dict(r) for r in rows] # ------------------------------------------------------------------ # Coin prices # ------------------------------------------------------------------ def save_prices(self, prices: list[dict]) -> None: now = datetime.now(timezone.utc).isoformat() for p in prices: self._conn.execute( "INSERT OR REPLACE INTO coin_prices (id, snapshot_at, data) VALUES (?,?,?)", (p.get("id", ""), now, json.dumps(p)), ) self._conn.commit() def get_price_path(self, coin_id: str, t0_iso: str, t1_iso: str) -> list[float]: """All recorded prices for a coin between two timestamps (for extremes).""" rows = self._conn.execute( "SELECT data FROM coin_prices WHERE id=? AND snapshot_at>=? AND snapshot_at<=?", (coin_id, t0_iso, t1_iso), ).fetchall() out = [] for r in rows: try: p = json.loads(r["data"]).get("price_usd") if p: out.append(p) except Exception: pass return out def get_price_history(self, coin_id: str = "bitcoin", limit: int = 200) -> list[dict]: """Returns [{snapshot_at, price}] for a coin, chronological order.""" rows = self._conn.execute( "SELECT snapshot_at, data FROM coin_prices WHERE id = ? ORDER BY snapshot_at DESC LIMIT ?", (coin_id, limit), ).fetchall() out = [] for r in reversed(rows): try: price = json.loads(r["data"]).get("price_usd") except Exception: price = None if price is not None: out.append({"snapshot_at": r["snapshot_at"], "price": price}) return out # ------------------------------------------------------------------ # Clients & email settings (admin panel) # ------------------------------------------------------------------ def _ensure_email_settings(self) -> None: self._conn.execute("INSERT OR IGNORE INTO email_settings (id) VALUES (1)") self._conn.commit() def get_email_settings(self) -> dict: self._ensure_email_settings() row = self._conn.execute("SELECT * FROM email_settings WHERE id = 1").fetchone() return dict(row) if row else {} def update_email_settings(self, **fields) -> None: self._ensure_email_settings() allowed = {"enabled", "send_time_1", "send_time_2", "subject_prefix", "last_sent_slot"} sets = {k: v for k, v in fields.items() if k in allowed} if not sets: return cols = ", ".join(f"{k} = ?" for k in sets) self._conn.execute(f"UPDATE email_settings SET {cols} WHERE id = 1", tuple(sets.values())) self._conn.commit() def add_client(self, email: str, name: str = "") -> bool: """Admin-added client: trusted, so active + confirmed immediately.""" try: self._conn.execute( "INSERT INTO clients (email, name, active, confirmed, added_at, source) " "VALUES (?,?,1,1,?,?)", (email.strip().lower(), name.strip(), datetime.now(timezone.utc).isoformat(), "admin"), ) self._conn.commit() return True except sqlite3.IntegrityError: return False def add_pending_subscriber(self, email: str, token: str, source: str = "web", ref_code: str = "", referred_by: str = "") -> str: """Self-serve signup (double opt-in). Returns: 'pending' -> new pending row (send confirmation) 'resend' -> existed but unconfirmed; token refreshed (resend confirmation) 'exists' -> already confirmed & active (no email needed) """ email = email.strip().lower() now = datetime.now(timezone.utc).isoformat() row = self._conn.execute( "SELECT id, confirmed, active FROM clients WHERE email = ?", (email,) ).fetchone() if row is None: self._conn.execute( "INSERT INTO clients (email, name, active, confirmed, added_at, token, " "source, ref_code, referred_by, referrals) VALUES (?,?,0,0,?,?,?,?,?,0)", (email, "", now, token, source, ref_code, referred_by or None), ) self._conn.commit() return "pending" if row["confirmed"] and row["active"]: return "exists" # existed but not confirmed (or unsubscribed) -> refresh token, resend self._conn.execute( "UPDATE clients SET token = ?, confirmed = 0, active = 0 WHERE id = ?", (token, row["id"]), ) self._conn.commit() return "resend" def confirm_subscriber(self, token: str) -> dict | None: """Activate a subscriber by their confirm token. Credits the referrer on the FIRST confirmation only. Returns {email, ref_code, referrals} or None.""" if not token: return None row = self._conn.execute( "SELECT id, email, confirmed, ref_code, referred_by FROM clients WHERE token = ?", (token,), ).fetchone() if row is None: return None first_time = not row["confirmed"] self._conn.execute( "UPDATE clients SET active = 1, confirmed = 1 WHERE id = ?", (row["id"],) ) if first_time and row["referred_by"]: self._conn.execute( "UPDATE clients SET referrals = referrals + 1 WHERE ref_code = ?", (row["referred_by"],), ) self._conn.commit() refs = self._conn.execute( "SELECT referrals FROM clients WHERE id = ?", (row["id"],) ).fetchone() return {"email": row["email"], "ref_code": row["ref_code"] or "", "referrals": int(refs["referrals"]) if refs else 0} def top_referrers(self, limit: int = 3) -> list[dict]: """Confirmed subscribers with the most referrals (referrals > 0).""" rows = self._conn.execute( "SELECT email, referrals FROM clients " "WHERE confirmed = 1 AND referrals > 0 " "ORDER BY referrals DESC, id ASC LIMIT ?", (limit,), ).fetchall() return [{"email": r["email"], "referrals": int(r["referrals"])} for r in rows] def referral_count(self, ref_code: str) -> int | None: """Live referral count for a ref_code, or None if the code is unknown.""" if not ref_code: return None row = self._conn.execute( "SELECT referrals FROM clients WHERE ref_code = ?", (ref_code,) ).fetchone() return int(row["referrals"]) if row else None def unsubscribe_email(self, email: str) -> bool: """Deactivate a subscriber by email (keeps the row). Returns True if found.""" cur = self._conn.execute( "UPDATE clients SET active = 0 WHERE email = ?", (email.strip().lower(),) ) self._conn.commit() return cur.rowcount > 0 def remove_client(self, client_id: int) -> None: self._conn.execute("DELETE FROM clients WHERE id = ?", (client_id,)) self._conn.commit() def set_client_active(self, client_id: int, active: bool) -> None: self._conn.execute("UPDATE clients SET active = ? WHERE id = ?", (1 if active else 0, client_id)) self._conn.commit() def subscriber_count(self, confirmed_only: bool = True) -> int: q = "SELECT COUNT(*) AS n FROM clients" if confirmed_only: q += " WHERE confirmed = 1" row = self._conn.execute(q).fetchone() return int(row["n"]) if row else 0 def get_clients(self, active_only: bool = False) -> list[dict]: q = "SELECT * FROM clients" if active_only: q += " WHERE active = 1" q += " ORDER BY added_at DESC" return [dict(r) for r in self._conn.execute(q).fetchall()] def log_email(self, slot: str, recipients: int, status: str, detail: str = "") -> None: self._conn.execute( "INSERT INTO email_log (sent_at, slot, recipients, status, detail) VALUES (?,?,?,?,?)", (datetime.now(timezone.utc).isoformat(), slot, recipients, status, detail), ) self._conn.commit() def get_email_log(self, limit: int = 30) -> list[dict]: return [dict(r) for r in self._conn.execute( "SELECT * FROM email_log ORDER BY id DESC LIMIT ?", (limit,) ).fetchall()] # ------------------------------------------------------------------ # Maintenance # ------------------------------------------------------------------ def purge_old(self) -> None: """Remove articles older than MAX_ARTICLE_AGE_HOURS, and cap the growth of the high-volume time-series tables (they were never purged, so the DB grew unbounded and eventually strained the small Space).""" now = datetime.now(timezone.utc) art_cutoff = (now - timedelta(hours=MAX_ARTICLE_AGE_HOURS)).isoformat() self._conn.execute("DELETE FROM articles WHERE published < ?", (art_cutoff,)) # Keep ~60 days of time-series so the track records stay intact. ts_cutoff = (now - timedelta(days=60)).isoformat() for tbl, col in (("coin_prices", "snapshot_at"), ("mention_history", "snapshot_at"), ("sentiment_history", "snapshot_at"), ("onchain_signals", "snapshot_at")): try: self._conn.execute(f"DELETE FROM {tbl} WHERE {col} < ?", (ts_cutoff,)) except Exception: pass # Drop resolved magnet targets older than 60 days (stats already counted) try: self._conn.execute( "DELETE FROM magnet_targets WHERE (hit=1 OR expired=1) AND first_seen < ?", (ts_cutoff,)) except Exception: pass self._conn.commit() def close(self) -> None: self._conn.close()