import sqlite3 import os from datetime import datetime, timezone from typing import Dict, Any, Optional import pandas as pd from config.loader import CONFIG DB_FILE = CONFIG["storage"]["db_file"] def get_db_connection(db_path: Optional[str] = None) -> sqlite3.Connection: target_path = db_path or DB_FILE os.makedirs(os.path.dirname(os.path.abspath(target_path)), exist_ok=True) conn = sqlite3.connect(target_path, timeout=30.0) conn.execute("PRAGMA journal_mode=WAL;") conn.execute("PRAGMA synchronous=NORMAL;") return conn def init_db(db_path: Optional[str] = None): with get_db_connection(db_path) as conn: cursor = conn.cursor() cursor.execute(""" CREATE TABLE IF NOT EXISTS hashtag_snapshots ( id INTEGER PRIMARY KEY AUTOINCREMENT, tag TEXT NOT NULL, hashtag_id TEXT, media_count INTEGER NOT NULL, cycle_bucket TEXT NOT NULL, recorded_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, CONSTRAINT uq_tag_bucket UNIQUE (tag, cycle_bucket) ) """) cursor.execute("CREATE INDEX IF NOT EXISTS idx_tag_bucket ON hashtag_snapshots(tag, cycle_bucket);") cursor.execute("CREATE INDEX IF NOT EXISTS idx_recorded_at ON hashtag_snapshots(recorded_at);") cursor.execute(""" CREATE TABLE IF NOT EXISTS discovered_tags ( name TEXT PRIMARY KEY, source TEXT, media_count INTEGER, mentions INTEGER DEFAULT 1, status TEXT DEFAULT 'pending', first_seen TIMESTAMP DEFAULT CURRENT_TIMESTAMP, last_seen TIMESTAMP ) """) cursor.execute(""" CREATE TABLE IF NOT EXISTS leaderboard_history ( id INTEGER PRIMARY KEY AUTOINCREMENT, bucket TEXT, tag TEXT, rank INTEGER, weekly_velocity REAL, growth_rate_pct REAL, acceleration REAL, r_squared REAL, snapshots INTEGER, recorded_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) """) cursor.execute(""" CREATE TABLE IF NOT EXISTS trend_alerts ( tag TEXT PRIMARY KEY, reason TEXT, detail TEXT, alerted_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) """) cursor.execute(""" CREATE TABLE IF NOT EXISTS meta ( key TEXT PRIMARY KEY, value TEXT ) """) conn.commit() def calculate_cycle_bucket(hours_per_cycle: int) -> str: now = datetime.now(timezone.utc) bucket_hour = (now.hour // hours_per_cycle) * hours_per_cycle bucket_dt = now.replace(hour=bucket_hour, minute=0, second=0, microsecond=0) return bucket_dt.strftime("%Y-%m-%d %H:%M:%S") def log_snapshot(tag_data: Dict[str, Any], hours_per_cycle: int, db_path: Optional[str] = None): tag = tag_data.get("name", "").lower().strip("#") hashtag_id = str(tag_data.get("id") or "") media_count = int(tag_data.get("media_count") or 0) cycle_bucket = calculate_cycle_bucket(hours_per_cycle) recorded_at = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S") with get_db_connection(db_path) as conn: cursor = conn.cursor() cursor.execute(""" INSERT INTO hashtag_snapshots (tag, hashtag_id, media_count, cycle_bucket, recorded_at) VALUES (?, ?, ?, ?, ?) ON CONFLICT(tag, cycle_bucket) DO UPDATE SET hashtag_id = excluded.hashtag_id, media_count = excluded.media_count, recorded_at = excluded.recorded_at """, (tag, hashtag_id, media_count, cycle_bucket, recorded_at)) conn.commit() def fetch_recent_snapshots(limit: int = 10, db_path: Optional[str] = None) -> pd.DataFrame: query = f""" SELECT tag, hashtag_id, media_count, cycle_bucket, recorded_at FROM hashtag_snapshots ORDER BY id DESC LIMIT {limit} """ with get_db_connection(db_path) as conn: return pd.read_sql_query(query, conn) # ---------- Tag discovery & lifecycle ---------- def upsert_discovered_tag(name: str, source: str, media_count: int, db_path: Optional[str] = None): """Record a hashtag seen during discovery; repeat sightings bump `mentions`.""" name = name.lower().strip("#") if not name: return with get_db_connection(db_path) as conn: conn.execute(""" INSERT INTO discovered_tags (name, source, media_count, mentions, last_seen) VALUES (?, ?, ?, 1, CURRENT_TIMESTAMP) ON CONFLICT(name) DO UPDATE SET mentions = mentions + 1, media_count = excluded.media_count, last_seen = CURRENT_TIMESTAMP """, (name, source, media_count)) conn.commit() def promote_pending_tags(min_mentions: int, db_path: Optional[str] = None) -> list: """Promote discovered tags seen often enough; return the newly promoted names.""" with get_db_connection(db_path) as conn: rows = conn.execute( "SELECT name FROM discovered_tags WHERE status = 'pending' AND mentions >= ?", (min_mentions,) ).fetchall() if rows: conn.execute(""" UPDATE discovered_tags SET status = 'promoted' WHERE status = 'pending' AND mentions >= ? """, (min_mentions,)) conn.commit() return [r[0] for r in rows] def get_active_discovered_tags(db_path: Optional[str] = None) -> list: with get_db_connection(db_path) as conn: rows = conn.execute( "SELECT name FROM discovered_tags WHERE status = 'promoted'" ).fetchall() return [r[0] for r in rows] def get_pending_discovered_tags(limit: int = 50, db_path: Optional[str] = None) -> pd.DataFrame: with get_db_connection(db_path) as conn: return pd.read_sql_query( "SELECT name, source, media_count, mentions, status FROM discovered_tags " "ORDER BY mentions DESC, media_count DESC LIMIT ?", conn, params=(limit,)) # ---------- Leaderboard history & alerts ---------- def save_leaderboard(df: pd.DataFrame, bucket: str, db_path: Optional[str] = None): """Persist the current top-k snapshot so rank changes can be tracked over time.""" if df.empty: return rows = [] for rank, (_, r) in enumerate(df.iterrows(), start=1): rows.append(( bucket, str(r["hashtag"]).lstrip("#"), rank, float(r.get("weekly_velocity", 0) or 0), float(r.get("growth_rate_pct", 0) or 0), float(r.get("acceleration", 0) or 0), float(r.get("r_squared", 0) or 0), int(r.get("snapshots", 0) or 0), )) with get_db_connection(db_path) as conn: conn.executemany(""" INSERT INTO leaderboard_history (bucket, tag, rank, weekly_velocity, growth_rate_pct, acceleration, r_squared, snapshots) VALUES (?, ?, ?, ?, ?, ?, ?, ?) """, rows) conn.commit() def get_rank_changes(current_tags: list, db_path: Optional[str] = None) -> pd.DataFrame: """Previous rank for each currently-ranked tag (from the most recent earlier bucket).""" with get_db_connection(db_path) as conn: return pd.read_sql_query(""" WITH LastBuckets AS ( SELECT DISTINCT bucket FROM leaderboard_history ORDER BY bucket DESC LIMIT 2 ), Prev AS ( SELECT tag, rank FROM leaderboard_history WHERE bucket = (SELECT MIN(bucket) FROM LastBuckets) ) SELECT tag, rank AS prev_rank FROM Prev WHERE tag IN """ + "(" + ",".join("?" * len(current_tags)) + ")", conn, params=current_tags) def get_alerted_tags(db_path: Optional[str] = None) -> set: with get_db_connection(db_path) as conn: rows = conn.execute("SELECT tag FROM trend_alerts").fetchall() return {r[0] for r in rows} def mark_trend_alerted(tag: str, reason: str, detail: str, db_path: Optional[str] = None): with get_db_connection(db_path) as conn: conn.execute(""" INSERT OR REPLACE INTO trend_alerts (tag, reason, detail) VALUES (?, ?, ?) """, (tag, reason, detail)) conn.commit() def get_meta(key: str, db_path: Optional[str] = None) -> Optional[str]: with get_db_connection(db_path) as conn: row = conn.execute("SELECT value FROM meta WHERE key = ?", (key,)).fetchone() return row[0] if row else None def set_meta(key: str, value: str, db_path: Optional[str] = None): with get_db_connection(db_path) as conn: conn.execute(""" INSERT INTO meta (key, value) VALUES (?, ?) ON CONFLICT(key) DO UPDATE SET value = excluded.value """, (key, value)) conn.commit()