103 lines
3.6 KiB
Python
103 lines
3.6 KiB
Python
"""SQLite persistence for weekly episode tracking."""
|
|
import os
|
|
import sqlite3
|
|
from datetime import datetime, timezone, timedelta
|
|
from contextlib import contextmanager
|
|
|
|
DB_PATH = os.environ.get("DB_PATH", "/app/data/episodes.db")
|
|
SEED_EPISODE = 928
|
|
|
|
|
|
@contextmanager
|
|
def _conn():
|
|
os.makedirs(os.path.dirname(DB_PATH), exist_ok=True)
|
|
con = sqlite3.connect(DB_PATH)
|
|
con.row_factory = sqlite3.Row
|
|
try:
|
|
yield con
|
|
con.commit()
|
|
finally:
|
|
con.close()
|
|
|
|
|
|
def init_db():
|
|
"""Create table and seed with the initial previous-week episode if empty."""
|
|
with _conn() as con:
|
|
con.execute("""
|
|
CREATE TABLE IF NOT EXISTS weekly_episodes (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
episode_number INTEGER NOT NULL,
|
|
week_start TEXT NOT NULL,
|
|
week_end TEXT NOT NULL,
|
|
item_id TEXT,
|
|
media_name TEXT,
|
|
watched_at TEXT,
|
|
recorded_at TEXT DEFAULT (datetime('now'))
|
|
)
|
|
""")
|
|
count = con.execute("SELECT COUNT(*) FROM weekly_episodes").fetchone()[0]
|
|
if count == 0:
|
|
_seed(con)
|
|
|
|
|
|
def _seed(con):
|
|
"""Seed with episode 928 for the week before the current week."""
|
|
from datetime import date
|
|
PHT = timezone(timedelta(hours=8))
|
|
now_pht = datetime.now(PHT)
|
|
wd = now_pht.weekday()
|
|
days_since_sunday = (wd + 1) % 7
|
|
if wd == 6:
|
|
days_since_sunday = 7
|
|
current_sunday = (now_pht - timedelta(days=days_since_sunday)).date()
|
|
prev_sunday = current_sunday - timedelta(weeks=1)
|
|
prev_saturday = prev_sunday + timedelta(days=6)
|
|
con.execute(
|
|
"INSERT INTO weekly_episodes (episode_number, week_start, week_end, media_name) VALUES (?,?,?,?)",
|
|
(SEED_EPISODE, prev_sunday.isoformat(), prev_saturday.isoformat(), "seed"),
|
|
)
|
|
|
|
|
|
def save_episode(episode_number: int, week_start: str, week_end: str,
|
|
item_id: str = None, media_name: str = None, watched_at: str = None):
|
|
"""Insert or replace the episode record for this week."""
|
|
with _conn() as con:
|
|
existing = con.execute(
|
|
"SELECT id FROM weekly_episodes WHERE week_start = ?", (week_start,)
|
|
).fetchone()
|
|
if existing:
|
|
con.execute(
|
|
"""UPDATE weekly_episodes
|
|
SET episode_number=?, item_id=?, media_name=?, watched_at=?, recorded_at=datetime('now')
|
|
WHERE week_start=?""",
|
|
(episode_number, item_id, media_name, watched_at, week_start),
|
|
)
|
|
else:
|
|
con.execute(
|
|
"""INSERT INTO weekly_episodes
|
|
(episode_number, week_start, week_end, item_id, media_name, watched_at)
|
|
VALUES (?,?,?,?,?,?)""",
|
|
(episode_number, week_start, week_end, item_id, media_name, watched_at),
|
|
)
|
|
|
|
|
|
def get_previous_episode(week_start: str) -> dict | None:
|
|
"""Return the most recent episode record strictly before week_start."""
|
|
with _conn() as con:
|
|
row = con.execute(
|
|
"""SELECT * FROM weekly_episodes
|
|
WHERE week_start < ?
|
|
ORDER BY week_start DESC LIMIT 1""",
|
|
(week_start,),
|
|
).fetchone()
|
|
return dict(row) if row else None
|
|
|
|
|
|
def get_all_history() -> list:
|
|
"""Return all weekly episode records, newest first."""
|
|
with _conn() as con:
|
|
rows = con.execute(
|
|
"SELECT * FROM weekly_episodes ORDER BY week_start DESC"
|
|
).fetchall()
|
|
return [dict(r) for r in rows]
|