import sqlite3 from contextlib import contextmanager from datetime import date from .config import settings SCHEMA = """ CREATE TABLE IF NOT EXISTS jokes ( day TEXT PRIMARY KEY, article_title TEXT NOT NULL, article_url TEXT NOT NULL, jokes TEXT NOT NULL, generated_at TEXT NOT NULL ); """ def init_db() -> None: with _conn() as conn: conn.executescript(SCHEMA) @contextmanager def _conn(): conn = sqlite3.connect(settings.db_path) try: yield conn conn.commit() finally: conn.close() def save_batch(day: date, article_title: str, article_url: str, jokes: list[str]) -> None: import json from datetime import datetime, timezone with _conn() as conn: conn.execute( "INSERT OR REPLACE INTO jokes (day, article_title, article_url, jokes, generated_at) " "VALUES (?, ?, ?, ?, ?)", ( day.isoformat(), article_title, article_url, json.dumps(jokes, ensure_ascii=False), datetime.now(timezone.utc).isoformat(timespec="seconds"), ), ) def get_day(day: date) -> dict | None: import json with _conn() as conn: row = conn.execute( "SELECT day, article_title, article_url, jokes, generated_at FROM jokes WHERE day = ?", (day.isoformat(),), ).fetchone() if row is None: return None return { "day": row[0], "article_title": row[1], "article_url": row[2], "jokes": json.loads(row[3]), "generated_at": row[4], } def get_latest() -> dict | None: import json with _conn() as conn: row = conn.execute( "SELECT day, article_title, article_url, jokes, generated_at " "FROM jokes ORDER BY day DESC LIMIT 1" ).fetchone() if row is None: return None return { "day": row[0], "article_title": row[1], "article_url": row[2], "jokes": json.loads(row[3]), "generated_at": row[4], }