chemavx/researchowl
1
0
Fork 0
Code Issues Pull Requests 10 Actions Packages Projects Releases Wiki Activity
Files
a14e5b99f9d4e191aebbbf75302f1369421bdd77
researchowl/src/db/database.py
T
ChemaVXandClaude Opus 5 a14e5b99f9 fix(db): un spec no envejece con la investigación que lo originó 
La retención iba por la edad de la SESIÓN y cascadeaba
`DELETE FROM outputs WHERE session_id = ?`, así que un output generado ayer
sobre una sesión de julio moría en el siguiente arranque del bot.

Pasó hoy, 2026-09-01, al desplegar 198b0e62:

    Startup purge done  sessions=12 outputs=27 chunks=2030 sources=3466
                        api_usage=1118 shorts=6

El pod llevaba 18 días sin reiniciarse, así que tres semanas de material
cumplieron los 30 días de golpe. Y lo que se llevó por delante no fue lo viejo:
los outputs 132-139 tenían 18,8 días —los tres Shorts re-renderizados el día
antes entre ellos— pero colgaban de las sesiones 161-165, de hace 40. Murieron
por la edad de su madre mientras 128-131, más antiguos, sobrevivían.

Ahora la purga va en dos fases:

1. Outputs por SU propia fecha, vivan en la sesión que vivan.
2. Sesiones viejas que ya no sostienen ningún output.

El orden importa: la sesión cuyos outputs eran todos viejos se queda sin
ninguno en la fase 1 y resulta purgable en la fase 2, así que el caso normal
—sesión vieja, material viejo— sigue limpiándose entero en UNA pasada. Hay un
test que lo fija, y es el único de los cuatro nuevos que pasa también con la
lógica anterior: está para probar que no se rompió lo que funcionaba.

Una sesión con un output vivo sobrevive entera, con sus sources y sus chunks.
No es generosidad: los chunks son contra lo que se comprueba el fundamento de
ese output, y conservar el spec tirando aquello con lo que se verifica deja
algo que ya no se puede auditar. El precio es que la retención afloja, y se
paga a sabiendas.

Dos cosas más que salieron al mirarlo:

- `_purge_on_startup` sólo escribía en el log `if result["sessions"] > 0`. Con
  la retención por output, una pasada puede borrar 27 outputs y CERO sesiones,
  y eso no habría dejado ni una línea. Una purga silenciosa es como se
  descubre tres semanas tarde. Ahora informa si borró cualquier cosa.
- El MP4 se llama por sesión, así que cuando la fase 1 se lleva el último
  short_en de una sesión que sigue viva, el fichero queda sin nada que lo
  nombre. Se borra ahí también, o el PVC acumula vídeos que no aparecen en
  ninguna fila.

`purge_old_sessions` pasa a llamarse `purge_old_data`: ya no purga sólo por
sesiones y el nombre viejo describía justo el defecto.

La BD anterior a la purga está a salvo y verificada en
~/rescates/researchowl-purga-2026-09-01 (integrity_check ok, 31 outputs, los
17 short_en). Con esta lógica, un restore conserva 16 de los 17.

Suite: 278 pasan.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-09-01 17:27:21 +00:00

747 lines
29 KiB
Python
Raw Blame History

import asyncio
import aiosqlite
import json
import time
from pathlib import Path
from typing import Optional
from enum import Enum
import structlog
from src.config import settings
logger = structlog.get_logger()
class ResearchStatus(str, Enum):
RUNNING = "running"
SATURATED = "saturated"
FINISHED = "finished"
ERROR = "error"
INTERRUPTED = "interrupted" # el pod se reinició con el research en marcha
class OutputType(str, Enum):
PODCAST = "podcast"
BLOG = "blog"
REPORT = "report"
THREAD = "thread"
REPORT_EXTENDED = "report_extended"
BLOG_EXTENDED = "blog_extended"
PODCAST_EXTENDED = "podcast_extended"
# El contenido de un short_en NO es prosa: es el shot spec (JSON) que
# shortsmith convierte en vídeo. El MP4 vive en disco, nunca en SQLite.
SHORT_EN = "short_en"
SCHEMA = """
CREATE TABLE IF NOT EXISTS research_sessions (
id INTEGER PRIMARY KEY AUTOINCREMENT,
topic TEXT NOT NULL,
status TEXT NOT NULL DEFAULT 'running',
telegram_chat_id INTEGER NOT NULL,
telegram_message_id INTEGER,
created_at REAL NOT NULL,
updated_at REAL NOT NULL,
iterations INTEGER DEFAULT 0,
total_sources INTEGER DEFAULT 0,
total_chunks INTEGER DEFAULT 0,
total_words INTEGER DEFAULT 0,
meta JSON DEFAULT '{}'
);
CREATE TABLE IF NOT EXISTS sources (
id INTEGER PRIMARY KEY AUTOINCREMENT,
session_id INTEGER NOT NULL REFERENCES research_sessions(id),
url TEXT NOT NULL,
title TEXT,
source_type TEXT, -- wikipedia, reddit, youtube, pdf, web, rss
depth INTEGER DEFAULT 0,
quality_score REAL DEFAULT 0,
word_count INTEGER DEFAULT 0,
scraped_at REAL,
status TEXT DEFAULT 'pending', -- pending, scraped, failed, skipped
error TEXT,
UNIQUE(session_id, url)
);
CREATE TABLE IF NOT EXISTS chunks (
id INTEGER PRIMARY KEY AUTOINCREMENT,
session_id INTEGER NOT NULL REFERENCES research_sessions(id),
source_id INTEGER NOT NULL REFERENCES sources(id),
content TEXT NOT NULL,
chunk_index INTEGER NOT NULL,
token_count INTEGER,
quality_score REAL DEFAULT 0,
embedding JSON, -- stored as JSON array for sqlite-vec compat
created_at REAL NOT NULL
);
CREATE TABLE IF NOT EXISTS outputs (
id INTEGER PRIMARY KEY AUTOINCREMENT,
session_id INTEGER NOT NULL REFERENCES research_sessions(id),
output_type TEXT NOT NULL,
content TEXT NOT NULL,
created_at REAL NOT NULL,
published_url TEXT -- URL del artículo publicado (blog -> Ghost)
);
CREATE TABLE IF NOT EXISTS source_contents (
id INTEGER PRIMARY KEY AUTOINCREMENT,
source_id INTEGER NOT NULL UNIQUE REFERENCES sources(id),
content TEXT NOT NULL,
created_at REAL NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_sources_session ON sources(session_id);
CREATE INDEX IF NOT EXISTS idx_chunks_session ON chunks(session_id);
CREATE INDEX IF NOT EXISTS idx_chunks_quality ON chunks(session_id, quality_score DESC);
CREATE INDEX IF NOT EXISTS idx_source_contents ON source_contents(source_id);
CREATE TABLE IF NOT EXISTS api_usage (
id INTEGER PRIMARY KEY AUTOINCREMENT,
session_id INTEGER REFERENCES research_sessions(id),
call_type TEXT NOT NULL,
model TEXT NOT NULL,
input_tokens INTEGER NOT NULL,
output_tokens INTEGER NOT NULL,
cost_usd REAL NOT NULL,
created_at REAL NOT NULL
);
CREATE TABLE IF NOT EXISTS watched_topics (
id INTEGER PRIMARY KEY AUTOINCREMENT,
topic TEXT NOT NULL,
chat_id INTEGER NOT NULL,
interval_hours INTEGER NOT NULL DEFAULT 24,
next_run_at REAL NOT NULL,
last_run_at REAL,
enabled INTEGER NOT NULL DEFAULT 1,
created_at REAL NOT NULL,
UNIQUE(topic, chat_id)
);
CREATE TABLE IF NOT EXISTS news_seen (
id INTEGER PRIMARY KEY,
source TEXT NOT NULL,
guid TEXT NOT NULL,
title TEXT,
link TEXT,
published_at TEXT,
matched_keywords TEXT,
seen_at TEXT DEFAULT (datetime('now')),
notified INTEGER DEFAULT 0,
UNIQUE(source, guid)
);
CREATE INDEX IF NOT EXISTS idx_news_seen_source ON news_seen(source);
CREATE INDEX IF NOT EXISTS idx_news_seen_seen ON news_seen(seen_at);
"""
# Conexión única compartida por proceso. aiosqlite serializa todas las
# operaciones en el hilo de la conexión, así que compartirla entre coroutines
# es seguro y además evita la contención del lock de escritura a nivel de
# fichero que sufrían las conexiones múltiples (scheduler, /compare).
_shared_conn: Optional[aiosqlite.Connection] = None
_init_lock = asyncio.Lock()
class _SharedConnection:
"""Proxy sobre la conexión compartida cuyo close() es no-op.
Permite que los handlers sigan usando el patrón `db = await get_db()` …
`finally: await db.close()` sin cambios, mientras por debajo se reutiliza
una sola conexión real durante toda la vida del proceso.
"""
def __init__(self, conn: aiosqlite.Connection):
self._conn = conn
def __getattr__(self, name):
return getattr(self._conn, name)
async def close(self):
# No cerrar: la conexión es compartida por todo el proceso.
pass
async def _init_shared() -> aiosqlite.Connection:
global _shared_conn
if _shared_conn is None:
async with _init_lock:
if _shared_conn is None: # doble check tras adquirir el lock
Path(settings.db_path).parent.mkdir(parents=True, exist_ok=True)
conn = await aiosqlite.connect(settings.db_path)
conn.row_factory = aiosqlite.Row
await conn.execute("PRAGMA journal_mode=WAL")
await conn.execute("PRAGMA synchronous=NORMAL")
# Espera hasta 5s si OTRO proceso tiene el lock de escritura
# antes de fallar con "database is locked".
await conn.execute("PRAGMA busy_timeout=5000")
await conn.executescript(SCHEMA)
await _ensure_columns(conn)
await conn.commit()
_shared_conn = conn
logger.info("Shared DB connection initialized", path=settings.db_path)
return _shared_conn
#: Columnas añadidas después de que la tabla existiera en producción. El
#: `CREATE TABLE IF NOT EXISTS` no toca una tabla ya creada, así que una
#: columna nueva necesita su ALTER — guardado por PRAGMA table_info para que
#: sea idempotente. No es una migración: no hay versiones ni orden, sólo
#: "¿existe la columna? si no, créala".
_ADDED_COLUMNS: dict[str, dict[str, str]] = {
"outputs": {"published_url": "TEXT"},
}
async def _ensure_columns(conn: aiosqlite.Connection) -> None:
for table, columns in _ADDED_COLUMNS.items():
async with conn.execute(f"PRAGMA table_info({table})") as cur:
existing = {row[1] for row in await cur.fetchall()}
for name, decl in columns.items():
if name not in existing:
await conn.execute(f"ALTER TABLE {table} ADD COLUMN {name} {decl}")
logger.info("Columna añadida", table=table, column=name)
async def get_db() -> aiosqlite.Connection:
conn = await _init_shared()
return _SharedConnection(conn)
async def close_db() -> None:
"""Cierra la conexión compartida (checkpoint WAL). Llamar al apagar el bot."""
global _shared_conn
if _shared_conn is not None:
try:
await _shared_conn.close()
finally:
_shared_conn = None
class ResearchDB:
def __init__(self, db: aiosqlite.Connection):
self.db = db
# --- Sessions ---
async def create_session(self, topic: str, chat_id: int) -> int:
now = time.time()
cursor = await self.db.execute(
"""INSERT INTO research_sessions (topic, status, telegram_chat_id, created_at, updated_at)
VALUES (?, ?, ?, ?, ?)""",
(topic, ResearchStatus.RUNNING, chat_id, now, now)
)
await self.db.commit()
return cursor.lastrowid
async def get_session(self, session_id: int) -> Optional[dict]:
cursor = await self.db.execute(
"SELECT * FROM research_sessions WHERE id = ?", (session_id,)
)
row = await cursor.fetchone()
return dict(row) if row else None
async def get_latest_session(self, chat_id: int) -> Optional[dict]:
cursor = await self.db.execute(
"SELECT * FROM research_sessions WHERE telegram_chat_id = ? ORDER BY created_at DESC LIMIT 1",
(chat_id,)
)
row = await cursor.fetchone()
return dict(row) if row else None
async def get_session_urls(self, session_id: int) -> set:
async with self.db.execute(
"SELECT url FROM sources WHERE session_id = ?", (session_id,)
) as cur:
rows = await cur.fetchall()
return {r[0] for r in rows}
async def get_previous_session(self, chat_id: int, topic: str,
exclude_session_id: int) -> Optional[dict]:
async with self.db.execute(
"""SELECT id, topic, status, created_at FROM research_sessions
WHERE telegram_chat_id = ? AND topic = ? AND id != ?
ORDER BY created_at DESC LIMIT 1""",
(chat_id, topic, exclude_session_id)
) as cur:
row = await cur.fetchone()
if not row:
return None
return {"id": row[0], "topic": row[1],
"status": row[2], "created_at": row[3]}
async def get_active_session(self, chat_id: int) -> Optional[dict]:
cursor = await self.db.execute(
"""SELECT * FROM research_sessions
WHERE telegram_chat_id = ? AND status = 'running'
ORDER BY created_at DESC LIMIT 1""",
(chat_id,)
)
row = await cursor.fetchone()
return dict(row) if row else None
async def update_session(self, session_id: int, **kwargs):
kwargs["updated_at"] = time.time()
sets = ", ".join(f"{k} = ?" for k in kwargs)
values = list(kwargs.values()) + [session_id]
await self.db.execute(
f"UPDATE research_sessions SET {sets} WHERE id = ?", values
)
await self.db.commit()
async def get_session_stats(self, session_id: int) -> dict:
cursor = await self.db.execute(
"""SELECT
COUNT(*) as total,
SUM(CASE WHEN status='scraped' THEN 1 ELSE 0 END) as scraped,
SUM(CASE WHEN status='failed' THEN 1 ELSE 0 END) as failed,
SUM(CASE WHEN status='pending' THEN 1 ELSE 0 END) as pending,
SUM(CASE WHEN status='skipped' THEN 1 ELSE 0 END) as skipped
FROM sources WHERE session_id = ?""",
(session_id,)
)
row = await cursor.fetchone()
return dict(row) if row else {}
# --- Sources ---
async def add_source(self, session_id: int, url: str, source_type: str,
depth: int = 0, title: str = None) -> Optional[int]:
try:
cursor = await self.db.execute(
"""INSERT OR IGNORE INTO sources (session_id, url, title, source_type, depth)
VALUES (?, ?, ?, ?, ?)""",
(session_id, url, title, source_type, depth)
)
await self.db.commit()
return cursor.lastrowid if cursor.rowcount > 0 else None
except Exception:
return None
async def update_source(self, source_id: int, **kwargs):
sets = ", ".join(f"{k} = ?" for k in kwargs)
values = list(kwargs.values()) + [source_id]
await self.db.execute(f"UPDATE sources SET {sets} WHERE id = ?", values)
await self.db.commit()
async def get_pending_sources(self, session_id: int, limit: int = 10) -> list[dict]:
cursor = await self.db.execute(
"""SELECT * FROM sources WHERE session_id = ? AND status = 'pending'
ORDER BY depth ASC, id ASC LIMIT ?""",
(session_id, limit)
)
rows = await cursor.fetchall()
return [dict(r) for r in rows]
async def get_all_sources(self, session_id: int) -> list[dict]:
cursor = await self.db.execute(
"SELECT * FROM sources WHERE session_id = ? ORDER BY quality_score DESC",
(session_id,)
)
rows = await cursor.fetchall()
return [dict(r) for r in rows]
async def source_exists(self, session_id: int, url: str) -> bool:
cursor = await self.db.execute(
"SELECT 1 FROM sources WHERE session_id = ? AND url = ?",
(session_id, url)
)
return await cursor.fetchone() is not None
# --- Chunks ---
async def add_chunk(self, session_id: int, source_id: int, content: str,
chunk_index: int, token_count: int, quality_score: float,
embedding: Optional[list] = None) -> int:
cursor = await self.db.execute(
"""INSERT INTO chunks (session_id, source_id, content, chunk_index,
token_count, quality_score, embedding, created_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)""",
(session_id, source_id, content, chunk_index,
token_count, quality_score,
json.dumps(embedding) if embedding else None,
time.time())
)
await self.db.commit()
return cursor.lastrowid
async def get_top_chunks(self, session_id: int, limit: int = 50) -> list[dict]:
cursor = await self.db.execute(
"""SELECT c.*, s.url, s.title, s.source_type FROM chunks c
JOIN sources s ON c.source_id = s.id
WHERE c.session_id = ? AND c.quality_score >= ?
ORDER BY c.quality_score DESC LIMIT ?""",
(session_id, settings.quality_threshold, limit)
)
rows = await cursor.fetchall()
return [dict(r) for r in rows]
async def get_chunks_count(self, session_id: int) -> int:
cursor = await self.db.execute(
"SELECT COUNT(*) FROM chunks WHERE session_id = ?", (session_id,)
)
row = await cursor.fetchone()
return row[0]
# --- Outputs ---
async def save_output(self, session_id: int, output_type: str, content: str) -> int:
cursor = await self.db.execute(
"INSERT INTO outputs (session_id, output_type, content, created_at) VALUES (?, ?, ?, ?)",
(session_id, output_type, content, time.time())
)
await self.db.commit()
return cursor.lastrowid
async def save_source_content(self, source_id: int, content: str):
await self.db.execute(
"""INSERT OR REPLACE INTO source_contents (source_id, content, created_at)
VALUES (?, ?, ?)""",
(source_id, content, time.time())
)
await self.db.commit()
async def get_source_content(self, source_id: int) -> Optional[str]:
cursor = await self.db.execute(
"SELECT content FROM source_contents WHERE source_id = ?", (source_id,)
)
row = await cursor.fetchone()
return row[0] if row else None
async def get_cached_content(self, url: str,
max_age_days: int = 7) -> Optional[str]:
threshold = time.time() - (max_age_days * 86400)
async with self.db.execute(
"""SELECT sc.content FROM source_contents sc
JOIN sources s ON s.id = sc.source_id
WHERE s.url = ? AND sc.created_at > ?
ORDER BY sc.created_at DESC LIMIT 1""",
(url, threshold)
) as cur:
row = await cur.fetchone()
return row[0] if row else None
async def set_output_url(self, output_id: int, url: str) -> None:
"""Guarda la URL pública del artículo en su fila de `outputs`.
Hace falta porque el Short enlaza al artículo en su descripción, y hasta
ahora la URL sólo viajaba en el aviso de Telegram: se perdía en cuanto
se cerraba la conversación.
"""
await self.db.execute(
"UPDATE outputs SET published_url = ? WHERE id = ?", (url, output_id))
await self.db.commit()
async def get_latest_output(self, session_id: int,
output_type: Optional[str] = None) -> Optional[dict]:
query = "SELECT * FROM outputs WHERE session_id = ?"
params: list = [session_id]
if output_type:
query += " AND output_type = ?"
params.append(output_type)
query += " ORDER BY created_at DESC LIMIT 1"
cursor = await self.db.execute(query, params)
row = await cursor.fetchone()
return dict(row) if row else None
async def get_article_url(self, session_id: int) -> Optional[str]:
"""La URL del artículo publicado más reciente de la sesión, si la hay.
Excluye las filas `short_en`: su `published_url` es la del vídeo en
YouTube, no la de un artículo. Sin este filtro, subir un Short y volver
a generar otro haría que el segundo enlazara al primero — un bucle
silencioso, porque la URL es válida y nadie la miraría dos veces.
"""
cursor = await self.db.execute(
"""SELECT published_url FROM outputs
WHERE session_id = ? AND published_url IS NOT NULL AND published_url != ''
AND output_type NOT IN (?)
ORDER BY created_at DESC LIMIT 1""",
(session_id, OutputType.SHORT_EN.value)
)
row = await cursor.fetchone()
return row[0] if row else None
async def get_outputs(self, session_id: int) -> list[dict]:
cursor = await self.db.execute(
"SELECT * FROM outputs WHERE session_id = ? ORDER BY created_at DESC",
(session_id,)
)
rows = await cursor.fetchall()
return [dict(r) for r in rows]
# --- API Usage ---
# Precios Claude en USD por 1M de tokens (input, output).
# Se busca por substring del id del modelo; fallback a Haiku.
_MODEL_PRICING = {
"opus": (15.00, 75.00),
"sonnet": (3.00, 15.00),
"haiku": (0.80, 4.00),
}
@classmethod
def _price_for_model(cls, model: str) -> tuple[float, float]:
m = (model or "").lower()
for key, price in cls._MODEL_PRICING.items():
if key in m:
return price
return cls._MODEL_PRICING["haiku"]
async def log_api_call(self, session_id, call_type: str, model: str,
input_tokens: int, output_tokens: int):
in_price, out_price = self._price_for_model(model)
cost = (input_tokens * in_price + output_tokens * out_price) / 1_000_000
await self.db.execute(
"""INSERT INTO api_usage
(session_id, call_type, model, input_tokens, output_tokens, cost_usd, created_at)
VALUES (?,?,?,?,?,?,?)""",
(session_id, call_type, model, input_tokens, output_tokens, cost, time.time())
)
await self.db.commit()
async def get_usage_stats(self, session_id: int) -> list[dict]:
cursor = await self.db.execute(
"""SELECT call_type,
COUNT(*) as calls,
SUM(input_tokens + output_tokens) as total_tokens,
SUM(cost_usd) as total_cost
FROM api_usage WHERE session_id = ?
GROUP BY call_type""",
(session_id,)
)
rows = await cursor.fetchall()
return [dict(r) for r in rows]
async def get_total_usage_stats(self) -> dict:
cursor = await self.db.execute(
"""SELECT COUNT(DISTINCT session_id) as sessions,
SUM(cost_usd) as total_cost
FROM api_usage"""
)
row = await cursor.fetchone()
return dict(row) if row else {"sessions": 0, "total_cost": 0}
# --- Watched Topics ---
async def add_watch(self, topic: str, chat_id: int, interval_hours: int,
next_run_at: Optional[float] = None) -> int:
now = time.time()
if next_run_at is None:
next_run_at = now + interval_hours * 3600
cursor = await self.db.execute(
"""INSERT OR REPLACE INTO watched_topics
(topic, chat_id, interval_hours, next_run_at, created_at)
VALUES (?, ?, ?, ?, ?)""",
(topic, chat_id, interval_hours, next_run_at, now)
)
await self.db.commit()
return cursor.lastrowid
async def remove_watch(self, topic: str, chat_id: int) -> bool:
cursor = await self.db.execute(
"DELETE FROM watched_topics WHERE topic = ? AND chat_id = ?",
(topic, chat_id)
)
await self.db.commit()
return cursor.rowcount > 0
async def list_watches(self, chat_id: int) -> list[dict]:
cursor = await self.db.execute(
"SELECT * FROM watched_topics WHERE chat_id = ? ORDER BY created_at ASC",
(chat_id,)
)
rows = await cursor.fetchall()
return [dict(r) for r in rows]
async def get_due_watches(self) -> list[dict]:
cursor = await self.db.execute(
"SELECT * FROM watched_topics WHERE enabled = 1 AND next_run_at <= ?",
(time.time(),)
)
rows = await cursor.fetchall()
return [dict(r) for r in rows]
async def update_watch_run(self, watch_id: int):
cursor = await self.db.execute(
"SELECT interval_hours FROM watched_topics WHERE id = ?", (watch_id,)
)
row = await cursor.fetchone()
if not row:
return
now = time.time()
await self.db.execute(
"UPDATE watched_topics SET last_run_at = ?, next_run_at = ? WHERE id = ?",
(now, now + row[0] * 3600, watch_id)
)
await self.db.commit()
# --- News monitor ---
async def source_seeded(self, source: str) -> bool:
"""True si ya hemos visto algún item de esta fuente (cold-start por fuente:
el primer poll siembra sin notificar)."""
cursor = await self.db.execute(
"SELECT 1 FROM news_seen WHERE source = ? LIMIT 1", (source,)
)
return await cursor.fetchone() is not None
async def record_news_item(self, source: str, guid: str, title: str,
link: str, published_at: str, matched: str,
notified: int) -> bool:
"""INSERT OR IGNORE de un item. Devuelve True solo si es una inserción
real (item nuevo); False si ya existía (UNIQUE(source, guid))."""
cursor = await self.db.execute(
"""INSERT OR IGNORE INTO news_seen
(source, guid, title, link, published_at, matched_keywords, notified)
VALUES (?, ?, ?, ?, ?, ?, ?)""",
(source, guid, title, link, published_at, matched, notified)
)
await self.db.commit()
return cursor.rowcount > 0
async def mark_news_notified(self, ids: list[int]):
if not ids:
return
placeholders = ",".join("?" for _ in ids)
await self.db.execute(
f"UPDATE news_seen SET notified = 1 WHERE id IN ({placeholders})",
tuple(ids)
)
await self.db.commit()
async def get_recent_news(self, hours: int = 24) -> list[dict]:
"""Items matcheados cuya seen_at (o published_at parseable) cae en la
ventana. Orden por published_at desc, con seen_at como desempate."""
window = f"-{int(hours)} hours"
cursor = await self.db.execute(
"""SELECT * FROM news_seen
WHERE seen_at >= datetime('now', ?)
OR (published_at IS NOT NULL AND published_at != ''
AND datetime(published_at) >= datetime('now', ?))
ORDER BY published_at DESC, seen_at DESC
LIMIT 50""",
(window, window)
)
rows = await cursor.fetchall()
return [dict(r) for r in rows]
async def get_unnotified(self, limit: Optional[int] = None) -> list[dict]:
"""Items aún sin notificar (notified=0), más recientes primero. El
scheduler (F2) los empuja a Telegram y luego los marca con
mark_news_notified. Aplica LIMIT solo si `limit` no es None."""
sql = (
"SELECT id, source, title, link, published_at, matched_keywords "
"FROM news_seen WHERE notified = 0 "
"ORDER BY published_at DESC NULLS LAST, seen_at DESC"
)
params: tuple = ()
if limit is not None:
sql += " LIMIT ?"
params = (int(limit),)
cursor = await self.db.execute(sql, params)
rows = await cursor.fetchall()
return [dict(r) for r in rows]
# --- Maintenance ---
async def purge_old_data(self, max_age_days: int = 30) -> dict:
"""Retención en dos fases: primero los outputs por SU fecha, luego las
sesiones que ya no sostienen nada.
Antes iba todo por la edad de la sesión, y la cascada
`DELETE FROM outputs WHERE session_id = ?` se llevaba por delante lo
generado ayer si colgaba de una sesión de hace dos meses. El
2026-09-01 eso borró 27 outputs, entre ellos los tres Shorts
re-renderizados el día antes: sobrevivieron los outputs 128-131 y
murieron los 132-139, que eran **más nuevos**. Un spec no envejece con
la investigación que lo originó.
Las dos fases van en este orden por una razón: la sesión cuyos outputs
eran todos viejos se queda sin ninguno en la fase 1 y resulta purgable
en la fase 2, así que el caso normal —sesión vieja, material viejo—
sigue limpiándose igual que antes en una sola pasada.
**Y una sesión con un output vivo sobrevive entera**, con sus sources y
sus chunks. No es generosidad: los `chunks` son contra lo que se
comprueba el fundamento de ese output, así que conservar el spec y tirar
aquello con lo que se verifica deja algo que ya no se puede auditar. El
precio es que la retención afloja — una sesión de julio con un Short de
ayer mantiene vivos sus cientos de sources — y ese precio se paga a
sabiendas.
"""
await self.db.execute("PRAGMA foreign_keys = ON")
threshold = time.time() - max_age_days * 86400
counts = {"sessions": 0, "sources": 0, "chunks": 0, "outputs": 0,
"api_usage": 0, "shorts": 0}
# --- fase 1: outputs por su propia fecha, vivan donde vivan ---------
# Se apuntan las sesiones tocadas antes de borrar: si una se queda sin
# ningún short_en, su MP4 no lo referencia ya nadie.
cursor = await self.db.execute(
"SELECT DISTINCT session_id FROM outputs WHERE created_at < ?",
(threshold,)
)
touched = [row[0] for row in await cursor.fetchall()]
cur = await self.db.execute("DELETE FROM outputs WHERE created_at < ?",
(threshold,))
counts["outputs"] += cur.rowcount
for sid in touched:
cursor = await self.db.execute(
"SELECT 1 FROM outputs WHERE session_id = ? AND output_type = ?"
" LIMIT 1", (sid, "short_en")
)
if await cursor.fetchone() is None:
counts["shorts"] += self._drop_short(sid)
# --- fase 2: sesiones viejas que ya no sostienen ningún output ------
cursor = await self.db.execute(
"SELECT id FROM research_sessions WHERE created_at < ?"
" AND status != 'running'"
" AND NOT EXISTS (SELECT 1 FROM outputs WHERE session_id ="
" research_sessions.id)",
(threshold,)
)
session_ids = [row[0] for row in await cursor.fetchall()]
for sid in session_ids:
counts["shorts"] += self._drop_short(sid)
await self.db.execute(
"DELETE FROM source_contents WHERE source_id IN (SELECT id FROM sources WHERE session_id = ?)",
(sid,)
)
cur = await self.db.execute("DELETE FROM chunks WHERE session_id = ?", (sid,))
counts["chunks"] += cur.rowcount
cur = await self.db.execute("DELETE FROM api_usage WHERE session_id = ?", (sid,))
counts["api_usage"] += cur.rowcount
cur = await self.db.execute("DELETE FROM sources WHERE session_id = ?", (sid,))
counts["sources"] += cur.rowcount
cur = await self.db.execute("DELETE FROM research_sessions WHERE id = ?", (sid,))
counts["sessions"] += cur.rowcount
await self.db.commit()
logger.info("Purga por antigüedad", days=max_age_days, **counts)
return counts
def _drop_short(self, sid: int) -> int:
"""Borra el MP4 de una sesión, si queda. Devuelve 1 si borró algo.
El vídeo vive en disco (los blobs en SQLite hacen patológico el WAL),
así que su borrado no lo arrastra ninguna FK y hay que hacerlo aquí.
Best-effort: un fichero que no se puede borrar no va a impedir la purga.
"""
try:
video = Path(settings.shorts_dir) / f"{sid}.mp4"
if video.is_file():
video.unlink()
return 1
except OSError as e:
logger.warning("No se pudo borrar el Short de una sesión purgada",
session_id=sid, error=str(e))
return 0
Reference in New Issue View Git Blame Copy Permalink