Something went wrong. Try again.
Something went wrong. Try again.
4.3 kB · 123 lines
Python
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124"""SQLite persistence for episodes + transcripts."""import jsonimport osimport sqlite3import time
DB_PATH = os.environ.get( "DB_PATH", os.path.join(os.path.dirname(os.path.abspath(__file__)), "data", "simcluster.db"))
SCHEMA = """CREATE TABLE IF NOT EXISTS episodes ( guid TEXT PRIMARY KEY, episode_num INTEGER, title TEXT NOT NULL, guest TEXT, guest_handle TEXT, description TEXT, audio_url TEXT NOT NULL, pub_date TEXT, duration_sec INTEGER, status TEXT NOT NULL DEFAULT 'pending', -- pending|processing|done|error error TEXT, cost_usd REAL, word_count INTEGER, names_resolved INTEGER, transcript TEXT, -- JSON: {turns:[{t,end,speaker,text}], mapping, confidence} created_at INTEGER NOT NULL, updated_at INTEGER NOT NULL);"""
def connect(): os.makedirs(os.path.dirname(os.path.abspath(DB_PATH)), exist_ok=True) con = sqlite3.connect(DB_PATH, timeout=30) con.row_factory = sqlite3.Row con.execute("PRAGMA journal_mode=WAL") return con
def init(): with connect() as con: con.executescript(SCHEMA)
def upsert_episode(ep, status="pending"): """Insert a feed item if its guid is new. Returns True if inserted.""" now = int(time.time()) with connect() as con: cur = con.execute( """INSERT OR IGNORE INTO episodes (guid, episode_num, title, guest, guest_handle, description, audio_url, pub_date, duration_sec, status, created_at, updated_at) VALUES (?,?,?,?,?,?,?,?,?,?,?,?)""", (ep["guid"], ep["episode_num"], ep["title"], ep["guest"], ep["guest_handle"], ep["description"], ep["audio_url"], ep["pub_date"], ep["duration_sec"], status, now, now)) return cur.rowcount > 0
def next_episode_num(): """Sequential fallback for feeds that don't number their episodes.""" with connect() as con: r = con.execute("SELECT COALESCE(MAX(episode_num), 0) + 1 FROM episodes" ).fetchone() return r[0]
def set_guest(guid, guest): with connect() as con: con.execute("UPDATE episodes SET guest=?, updated_at=? WHERE guid=?", (guest, int(time.time()), guid))
def set_status(guid, status, error=None): with connect() as con: con.execute("UPDATE episodes SET status=?, error=?, updated_at=? WHERE guid=?", (status, error, int(time.time()), guid))
def save_transcript(guid, transcript, cost_usd, word_count, names_resolved): with connect() as con: con.execute( """UPDATE episodes SET status='done', error=NULL, transcript=?, cost_usd=?, word_count=?, names_resolved=?, updated_at=? WHERE guid=?""", (json.dumps(transcript), cost_usd, word_count, int(names_resolved), int(time.time()), guid))
def list_episodes(): with connect() as con: rows = con.execute( """SELECT guid, episode_num, title, guest, guest_handle, description, audio_url, pub_date, duration_sec, status, error, cost_usd, word_count, names_resolved, updated_at FROM episodes ORDER BY episode_num DESC""").fetchall() return [dict(r) for r in rows]
def get_episode(num): with connect() as con: r = con.execute("SELECT * FROM episodes WHERE episode_num=?", (num,)).fetchone() if not r: return None d = dict(r) if d.get("transcript"): d["transcript"] = json.loads(d["transcript"]) return d
def claim_pending(): """Atomically grab one pending episode and mark it processing.""" with connect() as con: r = con.execute( "SELECT guid FROM episodes WHERE status='pending' ORDER BY episode_num LIMIT 1" ).fetchone() if not r: return None con.execute("UPDATE episodes SET status='processing', updated_at=? WHERE guid=?", (int(time.time()), r["guid"])) row = con.execute("SELECT * FROM episodes WHERE guid=?", (r["guid"],)).fetchone() return dict(row)