feat(news): F2 — scheduler automático del monitor RSS (inerte)
Build & Deploy ResearchOwl / build-and-push (push) Successful in 7s
Build & Deploy ResearchOwl / build-and-push (push) Successful in 7s
Engancha el monitor de noticias al scheduler loop EXISTENTE de watched_topics (sin segunda task asyncio). Cada tick, si NEWS_ENABLED, y si ha pasado NEWS_POLL_INTERVAL_HOURS desde el último poll (estado en memoria), pollea los feeds, empuja los pendientes (notified=0) al chat destino (NEWS_CHAT_ID o 1er TELEGRAM_ALLOWED_USERS) y los marca. Best-effort: la rama de noticias va en su propio try/except — un fallo del news-poll NUNCA tumba el scheduler de watched_topics. Deploy inerte: NEWS_ENABLED sigue False; el flip se hará aparte en k8s-manifests. - db: get_unnotified(limit=None) — pendientes, recientes primero (NULLS LAST) - bot: rama news al final del tick del scheduler, reusa poll_feeds / item_from_row / format_digest de F1; "…y N más" si excede NEWS_MAX_ITEMS Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 4.8
parent
c0b0c1ba0a
commit
b1c5bc737d
+38
-1
@@ -5,7 +5,7 @@ Main user interface — all commands handled here
|
|||||||
import asyncio
|
import asyncio
|
||||||
import os
|
import os
|
||||||
import time
|
import time
|
||||||
from datetime import datetime, timedelta
|
from datetime import datetime, timedelta, timezone
|
||||||
from typing import Optional
|
from typing import Optional
|
||||||
from zoneinfo import ZoneInfo
|
from zoneinfo import ZoneInfo
|
||||||
|
|
||||||
@@ -879,6 +879,9 @@ async def _safe_send(bot, chat_id, text: str):
|
|||||||
|
|
||||||
|
|
||||||
async def _scheduler_loop(app: Application):
|
async def _scheduler_loop(app: Application):
|
||||||
|
# Estado en memoria (no en DB): en un reinicio re-polla una vez, inofensivo
|
||||||
|
# porque solo se empujan items notified=0 y se marcan tras enviarlos.
|
||||||
|
_last_news_poll: Optional[datetime] = None
|
||||||
while True:
|
while True:
|
||||||
db_conn = None
|
db_conn = None
|
||||||
try:
|
try:
|
||||||
@@ -938,6 +941,40 @@ async def _scheduler_loop(app: Application):
|
|||||||
app.bot, chat_id,
|
app.bot, chat_id,
|
||||||
f"🔄 Investigación automática iniciada: `{topic}`"
|
f"🔄 Investigación automática iniciada: `{topic}`"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
# --- Monitor de noticias (F2) — inerte si NEWS_ENABLED=False.
|
||||||
|
# Best-effort: un fallo del news-poll NUNCA tumba el scheduler de
|
||||||
|
# watched_topics (va en su propio try/except).
|
||||||
|
if settings.NEWS_ENABLED:
|
||||||
|
now = datetime.now(timezone.utc)
|
||||||
|
interval = timedelta(hours=settings.NEWS_POLL_INTERVAL_HOURS)
|
||||||
|
due_news = (_last_news_poll is None
|
||||||
|
or (now - _last_news_poll) >= interval)
|
||||||
|
if due_news:
|
||||||
|
_last_news_poll = now # marca antes de pollear: evita reintentos en bucle si falla
|
||||||
|
news_chat_id = settings.news_chat_id
|
||||||
|
if not news_chat_id:
|
||||||
|
logger.warning("NEWS_ENABLED pero sin chat destino; salto poll")
|
||||||
|
else:
|
||||||
|
try:
|
||||||
|
await poll_feeds(db, settings) # inserta novedades notified=0
|
||||||
|
pending = await db.get_unnotified() # TODOS los pendientes
|
||||||
|
if pending:
|
||||||
|
shown = pending[:settings.NEWS_MAX_ITEMS]
|
||||||
|
extra = len(pending) - len(shown)
|
||||||
|
items = [item_from_row(r) for r in shown]
|
||||||
|
chunks = format_digest(items, header="🛸 Novedades UAP/OVNI")
|
||||||
|
if extra > 0 and chunks:
|
||||||
|
chunks[-1] += f"\n…y {extra} más."
|
||||||
|
for c in chunks:
|
||||||
|
await app.bot.send_message(
|
||||||
|
chat_id=news_chat_id, text=c,
|
||||||
|
disable_web_page_preview=False,
|
||||||
|
)
|
||||||
|
# Marca todos los pendientes (incluidos los "…y N más")
|
||||||
|
await db.mark_news_notified([r["id"] for r in pending])
|
||||||
|
except Exception as e:
|
||||||
|
logger.warning("News poll/notify falló", error=str(e))
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
logger.warning("Scheduler loop error", error=str(e))
|
logger.warning("Scheduler loop error", error=str(e))
|
||||||
finally:
|
finally:
|
||||||
|
|||||||
@@ -562,6 +562,23 @@ class ResearchDB:
|
|||||||
rows = await cursor.fetchall()
|
rows = await cursor.fetchall()
|
||||||
return [dict(r) for r in rows]
|
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 ---
|
# --- Maintenance ---
|
||||||
|
|
||||||
async def purge_old_sessions(self, max_age_days: int = 30) -> dict:
|
async def purge_old_sessions(self, max_age_days: int = 30) -> dict:
|
||||||
|
|||||||
Reference in New Issue
Block a user