From 8a3e7e2b53d2fe48a063e0ffbe0532bd6ba46503 Mon Sep 17 00:00:00 2001 From: Pluto Date: Thu, 28 May 2026 10:49:24 -0500 Subject: [PATCH] =?UTF-8?q?feat:=20cron=20jobs=20=E2=80=94=20daily=20summa?= =?UTF-8?q?ry,=20speaker=20clustering,=20clip=20purge?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Sonnet 4.6 --- app/cron/cluster_unknowns.py | 64 ++++++++++++++++++++++++++++++++ app/cron/daily_summary.py | 48 ++++++++++++++++++++++++ app/cron/main.py | 35 ++++++++++++++++++ app/cron/purge_clips.py | 31 ++++++++++++++++ tests/test_cron.py | 71 ++++++++++++++++++++++++++++++++++++ 5 files changed, 249 insertions(+) create mode 100644 app/cron/cluster_unknowns.py create mode 100644 app/cron/daily_summary.py create mode 100644 app/cron/main.py create mode 100644 app/cron/purge_clips.py create mode 100644 tests/test_cron.py diff --git a/app/cron/cluster_unknowns.py b/app/cron/cluster_unknowns.py new file mode 100644 index 0000000..c08dd8f --- /dev/null +++ b/app/cron/cluster_unknowns.py @@ -0,0 +1,64 @@ +import sqlite3 +import numpy as np +from app.shared.config import Settings +from app.pipeline.recognition import unpack_embedding, cosine_similarity + + +def cluster_unknown_speakers(settings: Settings, db: sqlite3.Connection) -> None: + """ + Groups unknown/ambiguous utterances from the last 7 days by voice similarity. + Assigns speaker_id from an existing 'Unknown #N' speaker or creates a new one. + Threshold: 0.70 cosine similarity. + """ + CLUSTER_THRESHOLD = 0.70 + + rows = db.execute(""" + SELECT id, embedding FROM utterances + WHERE match_status IN ('unknown', 'ambiguous') + AND embedding IS NOT NULL + AND date(start_time) >= date('now', '-7 days') + ORDER BY start_time ASC + """).fetchall() + + if not rows: + return + + clusters: list = [] + cluster_centroids: list = [] + + for row in rows: + emb = unpack_embedding(row["embedding"]) + best_cluster = None + best_sim = 0.0 + for i, centroid in enumerate(cluster_centroids): + sim = cosine_similarity(emb, centroid) + if sim > best_sim and sim >= CLUSTER_THRESHOLD: + best_sim = sim + best_cluster = i + if best_cluster is not None: + clusters[best_cluster].append(row["id"]) + n = len(clusters[best_cluster]) + cluster_centroids[best_cluster] = ( + (cluster_centroids[best_cluster] * (n - 1) + emb) / n + ) + else: + clusters.append([row["id"]]) + cluster_centroids.append(emb) + + for i, cluster in enumerate(clusters): + if len(cluster) < 2: + continue + name = f"Unknown #{i + 1}" + existing = db.execute("SELECT id FROM speakers WHERE name=?", (name,)).fetchone() + if existing: + speaker_id = existing["id"] + else: + cur = db.execute("INSERT INTO speakers (name) VALUES (?)", (name,)) + db.commit() + speaker_id = cur.lastrowid + for utt_id in cluster: + db.execute( + "UPDATE utterances SET speaker_id=?, match_status='ambiguous' WHERE id=?", + (speaker_id, utt_id), + ) + db.commit() diff --git a/app/cron/daily_summary.py b/app/cron/daily_summary.py new file mode 100644 index 0000000..10c6138 --- /dev/null +++ b/app/cron/daily_summary.py @@ -0,0 +1,48 @@ +import sqlite3 +import httpx +from app.shared.config import Settings + + +def generate_daily_summary(date: str, settings: Settings, db: sqlite3.Connection) -> None: + rows = db.execute(""" + SELECT u.start_time, u.transcript, r.name as room_name, s.name as speaker_name + FROM utterances u + JOIN rooms r ON u.room_id = r.id + LEFT JOIN speakers s ON u.speaker_id = s.id + WHERE date(u.start_time) = ? + ORDER BY u.start_time ASC + """, (date,)).fetchall() + + if not rows: + return + + lines = [ + f"[{r['room_name']} {r['start_time']}] {r['speaker_name'] or 'Unknown'}: {r['transcript']}" + for r in rows + ] + context = "\n".join(lines) + + prompt = ( + f"Summarize the following household conversations from {date} in 2-4 sentences. " + f"Focus on key decisions, activities, and topics discussed. Do not transcribe verbatim.\n\n" + f"{context}" + ) + + resp = httpx.post( + f"{settings.ollama_url}/api/chat", + json={ + "model": settings.ollama_model, + "messages": [{"role": "user", "content": prompt}], + "stream": False, + }, + timeout=120.0, + ) + resp.raise_for_status() + summary = resp.json()["message"]["content"] + + db.execute( + "INSERT INTO daily_summaries (date, summary_text) VALUES (?,?) " + "ON CONFLICT(date) DO UPDATE SET summary_text=?, generated_at=CURRENT_TIMESTAMP", + (date, summary, summary), + ) + db.commit() diff --git a/app/cron/main.py b/app/cron/main.py new file mode 100644 index 0000000..0b237d4 --- /dev/null +++ b/app/cron/main.py @@ -0,0 +1,35 @@ +import time +from datetime import datetime, timezone, timedelta + +from app.shared.config import get_settings +from app.shared.database import get_db, init_schema +from app.cron.daily_summary import generate_daily_summary +from app.cron.cluster_unknowns import cluster_unknown_speakers +from app.cron.purge_clips import purge_old_clips + + +def run_nightly(settings, db): + yesterday = (datetime.now(timezone.utc) - timedelta(days=1)).strftime("%Y-%m-%d") + today = datetime.now(timezone.utc).strftime("%Y-%m-%d") + generate_daily_summary(yesterday, settings, db) + cluster_unknown_speakers(settings, db) + purge_old_clips(today, settings, db) + + +def main(): + settings = get_settings() + init_schema(settings) + db = get_db(settings) + + while True: + now = datetime.now(timezone.utc) + next_midnight = now.replace(hour=0, minute=0, second=0, microsecond=0) + if next_midnight <= now: + next_midnight += timedelta(days=1) + sleep_secs = (next_midnight - now).total_seconds() + time.sleep(sleep_secs) + run_nightly(settings, db) + + +if __name__ == "__main__": + main() diff --git a/app/cron/purge_clips.py b/app/cron/purge_clips.py new file mode 100644 index 0000000..42be309 --- /dev/null +++ b/app/cron/purge_clips.py @@ -0,0 +1,31 @@ +import sqlite3 +from datetime import datetime, timedelta +from pathlib import Path +from app.shared.config import Settings + + +def purge_old_clips(today: str, settings: Settings, db: sqlite3.Connection) -> None: + cutoff = datetime.strptime(today, "%Y-%m-%d") - timedelta(days=settings.clip_retention_days) + clips_root = Path(settings.clips_dir) + if not clips_root.exists(): + return + for date_dir in clips_root.iterdir(): + if not date_dir.is_dir(): + continue + try: + dir_date = datetime.strptime(date_dir.name, "%Y-%m-%d") + except ValueError: + continue + if dir_date < cutoff: + for wav in date_dir.glob("*.wav"): + wav.unlink() + try: + date_dir.rmdir() + except OSError: + pass + cutoff_str = cutoff.strftime("%Y-%m-%d") + db.execute( + "UPDATE utterances SET audio_clip_path=NULL WHERE date(start_time) < ?", + (cutoff_str,), + ) + db.commit() diff --git a/tests/test_cron.py b/tests/test_cron.py new file mode 100644 index 0000000..3f966a5 --- /dev/null +++ b/tests/test_cron.py @@ -0,0 +1,71 @@ +import pytest +import os +from datetime import datetime, timezone +from pathlib import Path +from unittest.mock import MagicMock + + +def _seed_utterances(db, room_id, count=3): + for i in range(count): + db.execute( + "INSERT INTO utterances (room_id, transcript, match_status, start_time, end_time) VALUES (?,?,?,?,?)", + (room_id, f"utterance {i}", "unknown", + "2026-05-27 10:00:00", "2026-05-27 10:00:03"), + ) + db.commit() + + +def test_daily_summary_inserts_row(mocker, settings, db): + db.execute("INSERT INTO rooms (name) VALUES (?)", ("Kitchen",)) + db.commit() + room = db.execute("SELECT id FROM rooms WHERE name='Kitchen'").fetchone() + _seed_utterances(db, room["id"]) + + mock_post = mocker.patch("app.cron.daily_summary.httpx.post") + mock_post.return_value.json.return_value = {"message": {"content": "Busy day."}} + mock_post.return_value.raise_for_status = MagicMock() + + from app.cron.daily_summary import generate_daily_summary + generate_daily_summary("2026-05-27", settings, db) + + row = db.execute("SELECT * FROM daily_summaries WHERE date='2026-05-27'").fetchone() + assert row is not None + assert row["summary_text"] == "Busy day." + + +def test_daily_summary_skips_if_no_utterances(mocker, settings, db): + mock_post = mocker.patch("app.cron.daily_summary.httpx.post") + from app.cron.daily_summary import generate_daily_summary + generate_daily_summary("2026-05-27", settings, db) + mock_post.assert_not_called() + + +def test_purge_clips_deletes_old_files(settings, db, tmp_path): + from app.cron.purge_clips import purge_old_clips + + clips_dir = tmp_path / "clips" + old_dir = clips_dir / "2026-04-01" + old_dir.mkdir(parents=True) + old_file = old_dir / "1.wav" + old_file.write_bytes(b"\x00" * 100) + + new_dir = clips_dir / "2026-05-27" + new_dir.mkdir(parents=True) + new_file = new_dir / "2.wav" + new_file.write_bytes(b"\x00" * 100) + + # Create a settings with clips_dir pointing to tmp + from app.shared.config import Settings + settings_override = Settings( + db_path=settings.db_path, + clips_dir=str(clips_dir), + clip_retention_days=30, + speaches_url=settings.speaches_url, + ollama_url=settings.ollama_url, + livekit_api_key=settings.livekit_api_key, + livekit_api_secret=settings.livekit_api_secret, + ) + + purge_old_clips("2026-05-28", settings_override, db) + assert not old_file.exists() + assert new_file.exists()