digest_store

digest_store

DigestStore — SQLite-backed storage for pre-computed digest artifacts.

Classes

DigestArtifact dataclass

DigestArtifact(
    text: str,
    audio_path: Path,
    sections: Dict[str, str],
    sources_used: List[str],
    generated_at: datetime,
    model_used: str,
    voice_used: str,
    quality_score: float = 0.0,
    evaluator_feedback: str = "",
)

A pre-computed morning digest ready for delivery.

DigestStore

DigestStore(db_path: str = '')

SQLite store for digest artifacts.

Source code in src/diapason/agents/digest_store.py

def __init__(self, db_path: str = "") -> None:
    if not db_path:
        db_path = str(get_config_dir() / "digest.db")
    self._db_path = db_path
    if db_path != ":memory:":
        secure_create(Path(db_path).expanduser())
    self._conn = sqlite3.connect(db_path, check_same_thread=False)
    self._conn.execute("PRAGMA journal_mode=WAL")
    self._conn.execute(
        """
        CREATE TABLE IF NOT EXISTS digests (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            text TEXT NOT NULL,
            audio_path TEXT NOT NULL,
            sections TEXT NOT NULL,
            sources_used TEXT NOT NULL,
            generated_at TEXT NOT NULL,
            model_used TEXT NOT NULL,
            voice_used TEXT NOT NULL,
            quality_score REAL NOT NULL DEFAULT 0.0,
            evaluator_feedback TEXT NOT NULL DEFAULT ''
        )
        """
    )
    self._migrate()
    self._conn.commit()

Methods:

save

save(artifact: DigestArtifact) -> None

Save a digest artifact.

Source code in src/diapason/agents/digest_store.py

def save(self, artifact: DigestArtifact) -> None:
    """Save a digest artifact."""
    self._conn.execute(
        """
        INSERT INTO digests
            (text, audio_path, sections, sources_used,
             generated_at, model_used, voice_used,
             quality_score, evaluator_feedback)
        VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
        """,
        (
            artifact.text,
            str(artifact.audio_path),
            json.dumps(artifact.sections),
            json.dumps(artifact.sources_used),
            artifact.generated_at.isoformat(),
            artifact.model_used,
            artifact.voice_used,
            artifact.quality_score,
            artifact.evaluator_feedback,
        ),
    )
    self._conn.commit()

get_latest

get_latest() -> Optional[DigestArtifact]

Return the most recent digest, or None.

Source code in src/diapason/agents/digest_store.py

def get_latest(self) -> Optional[DigestArtifact]:
    """Return the most recent digest, or None."""
    row = self._conn.execute(
        "SELECT text, audio_path, sections, sources_used,"
        " generated_at, model_used, voice_used,"
        " quality_score, evaluator_feedback"
        " FROM digests ORDER BY id DESC LIMIT 1"
    ).fetchone()
    if row is None:
        return None
    return self._row_to_artifact(row)

get_today

get_today(
    timezone_name: str = "",
) -> Optional[DigestArtifact]

Return today's digest if it exists, or None.

timezone_name must be the timezone the digest was generated in — digest.timezone in the config. The stored generated_at carries that zone, so its date prefix is the reader's calendar day, and comparing it against today in the same zone is like for like.

This defaulted to "UTC" while generated_at was written with a naive datetime.now(), i.e. machine-local. West of Greenwich the two dates diverge every evening: at 21:00 in Toronto it is already tomorrow in UTC, so the morning's digest stopped matching and the API answered "No digest for today" about a digest sitting in the table. The window was four hours a day, every day. East of Greenwich it fails the other way round, in the early morning.

The fallback is the machine's own zone rather than UTC: rows written before this fix are naive local, and local is what they mean.

Source code in src/diapason/agents/digest_store.py

def get_today(self, timezone_name: str = "") -> Optional[DigestArtifact]:
    """Return today's digest if it exists, or None.

    ``timezone_name`` must be the timezone the digest was *generated* in
    — ``digest.timezone`` in the config. The stored ``generated_at``
    carries that zone, so its date prefix is the reader's calendar day,
    and comparing it against today in the same zone is like for like.

    This defaulted to ``"UTC"`` while ``generated_at`` was written with a
    naive ``datetime.now()``, i.e. machine-local. West of Greenwich the
    two dates diverge every evening: at 21:00 in Toronto it is already
    tomorrow in UTC, so the morning's digest stopped matching and the API
    answered "No digest for today" about a digest sitting in the table.
    The window was four hours a day, every day. East of Greenwich it
    fails the other way round, in the early morning.

    The fallback is the machine's own zone rather than UTC: rows written
    before this fix are naive local, and local is what they mean.
    """
    try:
        from zoneinfo import ZoneInfo

        now = (
            datetime.now(ZoneInfo(timezone_name))
            if timezone_name
            else datetime.now().astimezone()
        )
    except Exception:  # noqa: BLE001 - un fuseau inconnu n'est pas fatal
        now = datetime.now().astimezone()
    today = now.strftime("%Y-%m-%d")

    row = self._conn.execute(
        "SELECT text, audio_path, sections, sources_used,"
        " generated_at, model_used, voice_used,"
        " quality_score, evaluator_feedback"
        " FROM digests WHERE generated_at LIKE ? ORDER BY id DESC LIMIT 1",
        (f"{today}%",),
    ).fetchone()
    if row is None:
        return None
    return self._row_to_artifact(row)

history

history(limit: int = 10) -> List[DigestArtifact]

Return the N most recent digests.

Source code in src/diapason/agents/digest_store.py

def history(self, limit: int = 10) -> List[DigestArtifact]:
    """Return the N most recent digests."""
    rows = self._conn.execute(
        "SELECT text, audio_path, sections, sources_used,"
        " generated_at, model_used, voice_used,"
        " quality_score, evaluator_feedback"
        " FROM digests ORDER BY id DESC LIMIT ?",
        (limit,),
    ).fetchall()
    return [self._row_to_artifact(r) for r in rows]

Functions: