chore(deps): update dependency fastapi to v0.137.1

.gitignore

@@ -1,35 +0,0 @@
-# Python
-__pycache__/
-*.py[cod]
-*$py.class
-*.egg-info/
-.eggs/
-build/
-dist/
-
-# Virtualenvs
-.venv/
-venv/
-env/
-
-# Secrets / config local
-.env
-.env.*
-!.env.example
-
-# Datos / runtime
-*.db
-*.db-wal
-*.db-shm
-/data/
-
-# Tests / coverage
-.pytest_cache/
-.coverage
-htmlcov/
-
-# Editor / OS
-.vscode/
-.idea/
-*.swp
-.DS_Store

requirements.txt

@@ -1,5 +1,5 @@
 # Core
-fastapi==0.137.2
+fastapi==0.137.1
 uvicorn==0.30.0
 python-telegram-bot==21.5
 httpx==0.27.0

src/bot/bot.py

@@ -18,7 +18,7 @@ from telegram.ext import (
 from telegram.constants import ParseMode
 
 from src.config import settings
-from src.db.database import get_db, close_db, ResearchDB, ResearchStatus, OutputType
+from src.db.database import get_db, ResearchDB, ResearchStatus, OutputType
 from src.scraper.exhaustive import ExhaustiveScraper
 from src.processor.processor import OllamaClient, ContentProcessor
 from src.generator.generator import OutputGenerator
@@ -849,7 +849,7 @@ async def _scheduler_loop(app: Application):
                 _active_sessions[chat_id] = session_id
                 await db.update_watch_run(watch["id"])
 
-                async def _task(c=chat_id, t=topic, s=session_id):
+                async def _task(c=chat_id, t=topic, s=session_id, w_id=watch["id"]):
                     inner_db_conn = await get_db()
                     inner_db = ResearchDB(inner_db_conn)
                     try:
@@ -913,10 +913,6 @@ async def _on_startup(app: Application) -> None:
     await _start_scheduler(app)
 
 
-async def _on_shutdown(app: Application) -> None:
-    await close_db()
-
-
 async def cmd_export(update: Update, ctx: ContextTypes.DEFAULT_TYPE):
     if not is_authorized(update.effective_user.id):
         return
@@ -1263,7 +1259,6 @@ def create_bot() -> Application:
         Application.builder()
         .token(settings.telegram_bot_token)
         .post_init(_on_startup)
-        .post_shutdown(_on_shutdown)
         .build()
     )
 

src/config.py

@@ -24,19 +24,10 @@ class Settings(BaseSettings):
     max_depth: int = Field(3, env="MAX_DEPTH")                      # recursion depth
     max_sources: int = Field(150, env="MAX_SOURCES")                # hard cap
     max_pages_per_search: int = Field(5, env="MAX_PAGES_PER_SEARCH")
-    searxng_url: str = Field(
-        "http://searxng-svc.researchowl.svc.cluster.local:8080/search",
-        env="SEARXNG_URL"
-    )
     request_timeout: int = Field(30, env="REQUEST_TIMEOUT")
     request_delay: float = Field(1.0, env="REQUEST_DELAY")          # seconds between requests
     min_content_length: int = Field(200, env="MIN_CONTENT_LENGTH")  # chars
 
-    # Fuentes opcionales — desactivadas por defecto: la IP del homelab está
-    # bloqueada por Reddit (403) y YouTube (transcripts vacíos), eran peso muerto.
-    enable_youtube: bool = Field(False, env="ENABLE_YOUTUBE")
-    enable_reddit: bool = Field(False, env="ENABLE_REDDIT")
-
     # Processing
     chunk_size: int = Field(800, env="CHUNK_SIZE")                  # tokens per chunk
     chunk_overlap: int = Field(100, env="CHUNK_OVERLAP")

src/db/database.py

@@ -1,4 +1,3 @@
-import asyncio
 import aiosqlite
 import json
 import time
@@ -118,65 +117,15 @@ CREATE TABLE IF NOT EXISTS watched_topics (
 """
 
 
-# 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 conn.commit()
-                _shared_conn = conn
-                logger.info("Shared DB connection initialized", path=settings.db_path)
-    return _shared_conn
-
-
 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
+    Path(settings.db_path).parent.mkdir(parents=True, exist_ok=True)
+    db = await aiosqlite.connect(settings.db_path)
+    db.row_factory = aiosqlite.Row
+    await db.execute("PRAGMA journal_mode=WAL")
+    await db.execute("PRAGMA synchronous=NORMAL")
+    await db.executescript(SCHEMA)
+    await db.commit()
+    return db
 
 
 class ResearchDB:
@@ -392,26 +341,11 @@ class ResearchDB:
 
     # --- 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
+        # Precios Claude Haiku (claude-haiku-4-5):
+        # input: $0.80 / 1M tokens  output: $4.00 / 1M tokens
+        cost = (input_tokens * 0.80 + output_tokens * 4.00) / 1_000_000
         await self.db.execute(
             """INSERT INTO api_usage
                (session_id, call_type, model, input_tokens, output_tokens, cost_usd, created_at)

src/generator/generator.py

@@ -12,7 +12,6 @@ import time
 import structlog
 
 from src.config import settings
-from src.llm import get_anthropic_client
 from src.processor.processor import OllamaClient, ContentProcessor
 from src.db.database import ResearchDB, OutputType
 
@@ -443,9 +442,10 @@ class OutputGenerator:
 
     async def _generate_with_claude(self, prompt: str, system: str, output_type: OutputType,
                                      session_id: int | None = None) -> str:
+        import anthropic
         max_tokens = 4096 if output_type == OutputType.THREAD else 16000
         try:
-            client = get_anthropic_client()
+            client = anthropic.AsyncAnthropic(api_key=settings.anthropic_api_key)
             msg = await client.messages.create(
                 model=settings.claude_model,
                 max_tokens=max_tokens,
@@ -613,8 +613,9 @@ class OutputGenerator:
     async def _generate_raw(self, prompt: str,
                              session_id: int | None = None) -> str:
         if settings.anthropic_api_key:
+            import anthropic
             try:
-                client = get_anthropic_client()
+                client = anthropic.AsyncAnthropic(api_key=settings.anthropic_api_key)
                 msg = await client.messages.create(
                     model=settings.claude_model,
                     max_tokens=2048,
@@ -818,7 +819,8 @@ async def generate_diff_summary(
         )
 
     try:
-        client = get_anthropic_client()
+        import anthropic
+        client = anthropic.AsyncAnthropic(api_key=settings.anthropic_api_key)
         prompt = (
             f'Analiza el siguiente material de investigación sobre "{topic}" '
             f'y genera un resumen BREVE (máximo 300 palabras) de las novedades '
@@ -904,7 +906,8 @@ async def generate_comparison(
     )
 
     try:
-        client = get_anthropic_client()
+        import anthropic
+        client = anthropic.AsyncAnthropic(api_key=settings.anthropic_api_key)
         msg = await client.messages.create(
             model=settings.claude_model,
             max_tokens=8192,

src/llm.py

@@ -1,15 +0,0 @@
-"""Cliente Anthropic compartido y cacheado.
-
-Antes cada llamada (scoring de cada chunk, generación, outline, diff…)
-instanciaba un AsyncAnthropic nuevo — cientos de veces por sesión, cada uno
-con su propio pool de conexiones. Se reutiliza una única instancia por proceso.
-"""
-from functools import lru_cache
-
-from src.config import settings
-
-
-@lru_cache(maxsize=1)
-def get_anthropic_client():
-    import anthropic
-    return anthropic.AsyncAnthropic(api_key=settings.anthropic_api_key)

src/processor/processor.py

@@ -23,7 +23,6 @@ class OllamaClient:
     def __init__(self):
         self.base_url = settings.ollama_url.rstrip("/")
         self.model = settings.ollama_model
-        self.embed_model = settings.ollama_embed_model
 
     async def generate(self, prompt: str, system: str = None,
                        timeout: int = 120, temperature: float = 0.7) -> str:
@@ -48,7 +47,7 @@ class OllamaClient:
 
     async def embed(self, text: str) -> Optional[list[float]]:
         """Get embedding vector for a text"""
-        payload = {"model": self.embed_model, "prompt": text}
+        payload = {"model": self.model, "prompt": text}
         try:
             async with httpx.AsyncClient(timeout=60) as client:
                 resp = await client.post(f"{self.base_url}/api/embeddings", json=payload)
@@ -70,28 +69,9 @@ class OllamaClient:
 def simple_chunk(text: str, chunk_size: int = 800, overlap: int = 100) -> list[str]:
     """
     Split text into overlapping chunks by approximate word count.
-    Respects paragraph/line boundaries when possible.
-
-    Acepta párrafos separados por uno o más saltos de línea (Wikipedia y
-    trafilatura usan '\n' simple, lo que antes dejaba el documento entero como
-    un único 'párrafo' → un solo chunk gigante). Además subdivide por palabras
-    cualquier párrafo que por sí solo supere chunk_size.
+    Respects paragraph boundaries when possible.
     """
-    raw_paragraphs = [p.strip() for p in re.split(r"\n+", text) if p.strip()]
-
-    # Subdivide párrafos sobredimensionados en piezas de (chunk_size - overlap)
-    # palabras; así, al reinyectar 'overlap' palabras de solapamiento, ningún
-    # chunk resultante supera chunk_size.
-    piece_size = max(1, chunk_size - max(0, overlap))
-    paragraphs: list[str] = []
-    for para in raw_paragraphs:
-        words = para.split()
-        if len(words) <= chunk_size:
-            paragraphs.append(para)
-        else:
-            for i in range(0, len(words), piece_size):
-                paragraphs.append(" ".join(words[i:i + piece_size]))
-
+    paragraphs = [p.strip() for p in text.split("\n\n") if p.strip()]
     chunks = []
     current = []
     current_words = 0
@@ -100,12 +80,10 @@ def simple_chunk(text: str, chunk_size: int = 800, overlap: int = 100) -> list[s
         para_words = len(para.split())
         if current_words + para_words > chunk_size and current:
             chunks.append("\n\n".join(current))
-            # overlap: arrastra un tail de 'overlap' palabras (no el párrafo
-            # completo — eso duplicaba el tamaño cuando los párrafos eran grandes)
-            if overlap > 0:
-                tail = "\n\n".join(current).split()[-overlap:]
-                current = [" ".join(tail)]
-                current_words = len(tail)
+            # overlap: keep last paragraph
+            if overlap > 0 and current:
+                current = [current[-1]]
+                current_words = len(current[0].split())
             else:
                 current = []
                 current_words = 0
@@ -146,6 +124,7 @@ class ContentProcessor:
     async def process_session(self, session_id: int, topic: str,
                                progress_callback=None) -> dict:
         """Process all scraped sources for a session"""
+        from src.db.database import ResearchDB
         sources = await self.db.get_all_sources(session_id)
         scraped = [s for s in sources if s["status"] == "scraped"]
 
@@ -153,6 +132,7 @@ class ContentProcessor:
         scraped = await self._dedup_sources(session_id, scraped)
         logger.info("After dedup", unique=len(scraped))
         total_chunks = 0
+        total_words = 0
 
         semaphore = asyncio.Semaphore(3)  # process 3 sources at once
 
@@ -243,45 +223,33 @@ class ContentProcessor:
             return 0
 
         chunks = simple_chunk(content, settings.chunk_size, settings.chunk_overlap)
-        # Solo se puntúan/almacenan chunks con suficiente contenido
-        candidates = [(i, ch) for i, ch in enumerate(chunks)
-                      if len(ch.split()) >= 30]
         logger.info("Processing source", source_id=source_id,
                     content_len=len(content), num_chunks=len(chunks),
-                    candidates=len(candidates),
                     quality_threshold=settings.quality_threshold)
-        if not candidates:
-            return 0
-
-        # Scoring en lote: 1 (o pocas) llamadas a Claude por fuente en vez de
-        # una por chunk — antes una fuente de 19 chunks = 19 llamadas.
-        qualities = await self._score_quality_batch(
-            [ch for _, ch in candidates], topic, session_id
-        )
-
         stored = 0
         filtered_quality = 0
 
-        for (i, chunk), quality in zip(candidates, qualities):
+        for i, chunk in enumerate(chunks):
+            words = len(chunk.split())
+            if words < 30:
+                continue
+
+            quality = await self._score_quality(chunk, topic, session_id)
             if quality < settings.quality_threshold:
                 filtered_quality += 1
                 logger.debug("Chunk filtered by quality", source_id=source_id,
                              chunk_index=i, quality=round(quality, 2),
-                             threshold=settings.quality_threshold,
-                             words=len(chunk.split()))
+                             threshold=settings.quality_threshold, words=words)
                 continue
 
-            # Embeber el chunk completo (ya acotado a ~chunk_size palabras).
-            # Antes truncaba a 1000 chars → el vector solo representaba el
-            # principio de cada chunk, degradando el ranking del RAG.
-            embedding = await self.ollama.embed(chunk)
+            embedding = await self.ollama.embed(chunk[:1000])
 
             await self.db.add_chunk(
                 session_id=session_id,
                 source_id=source_id,
                 content=chunk,
                 chunk_index=i,
-                token_count=len(chunk.split()),
+                token_count=words,
                 quality_score=quality,
                 embedding=embedding
             )
@@ -306,84 +274,9 @@ class ContentProcessor:
             return await self._score_with_claude(chunk, topic, session_id)
         return await self._score_with_ollama(chunk, topic)
 
-    # ─── Batch scoring ──────────────────────────────────────────────────────
-
-    _BATCH_SCORE_SIZE = 25
-
-    async def _score_quality_batch(self, chunk_texts: list[str], topic: str,
-                                    session_id: int | None = None) -> list[float]:
-        """Puntúa varios chunks a la vez. Devuelve scores 0-1 en el mismo orden."""
-        if not chunk_texts:
-            return []
-        if not settings.anthropic_api_key:
-            # Ollama es local; no merece la pena batchear, se mantiene por-chunk
-            return [await self._score_with_ollama(c, topic) for c in chunk_texts]
-
-        results: list[float] = []
-        for i in range(0, len(chunk_texts), self._BATCH_SCORE_SIZE):
-            sub = chunk_texts[i:i + self._BATCH_SCORE_SIZE]
-            scores = await self._score_with_claude_batch(sub, topic, session_id)
-            if scores is None:  # fallo del batch → fallback por-chunk con Ollama
-                scores = [await self._score_with_ollama(c, topic) for c in sub]
-            results.extend(scores)
-        return results
-
-    @staticmethod
-    def _parse_batch_scores(text: str, n: int) -> list[float]:
-        """Extrae n scores normalizados (0-1) de la respuesta del modelo.
-
-        Toma el último número de cada línea (robusto ante numeración tipo
-        '1. 8'); rellena con 0.6 neutro si faltan líneas."""
-        scores: list[float] = []
-        for line in text.splitlines():
-            nums = re.findall(r'\d+(?:\.\d+)?', line)
-            if nums:
-                scores.append(min(1.0, float(nums[-1]) / 10.0))
-        if len(scores) >= n:
-            return scores[:n]
-        return scores + [0.6] * (n - len(scores))
-
-    async def _score_with_claude_batch(self, chunks: list[str], topic: str,
-                                        session_id: int | None = None):
-        """Puntúa hasta _BATCH_SCORE_SIZE chunks en una sola llamada.
-        Devuelve lista de scores 0-1, o None si la llamada falla."""
-        from src.llm import get_anthropic_client
-        listing = "\n\n".join(f"[{i + 1}]\n{c[:400]}" for i, c in enumerate(chunks))
-        prompt = (
-            f'Rate each of the following {len(chunks)} texts 0-10 for relevance '
-            f'to the topic "{topic}". Be generous — tangentially related = 4+, '
-            f'only below 3 if completely unrelated.\n'
-            f'Reply with EXACTLY {len(chunks)} lines, one integer 0-10 per line, '
-            f'in the same order as the texts. Just the number (e.g. 7), '
-            f'no labels, no other text.\n\n{listing}'
-        )
-        try:
-            client = get_anthropic_client()
-            msg = await client.messages.create(
-                model=settings.claude_model,
-                max_tokens=8 * len(chunks) + 50,
-                messages=[{"role": "user", "content": prompt}]
-            )
-            if session_id is not None:
-                try:
-                    await self.db.log_api_call(
-                        session_id, "scoring", settings.claude_model,
-                        msg.usage.input_tokens, msg.usage.output_tokens
-                    )
-                except Exception as log_err:
-                    logger.warning("Failed to log API usage", error=str(log_err))
-            scores = self._parse_batch_scores(msg.content[0].text.strip(), len(chunks))
-            logger.debug("Claude batch scored", n=len(chunks),
-                         avg=round(sum(scores) / len(scores), 2))
-            return scores
-        except Exception as e:
-            logger.warning("Claude batch scoring failed, will fallback",
-                           error=str(e), n=len(chunks))
-            return None
-
     async def _score_with_claude(self, chunk: str, topic: str,
                                   session_id: int | None = None) -> float:
-        from src.llm import get_anthropic_client
+        import anthropic
         prompt = (
             f'Rate 0-10 how relevant this text is to the topic "{topic}". '
             f'Be generous — if the text is tangentially related, score 4+. '
@@ -391,7 +284,7 @@ class ContentProcessor:
             f'Reply with only a number.\n\nText:\n{chunk[:500]}'
         )
         try:
-            client = get_anthropic_client()
+            client = anthropic.AsyncAnthropic(api_key=settings.anthropic_api_key)
             msg = await client.messages.create(
                 model=settings.claude_model,
                 max_tokens=10,

src/scraper/exhaustive.py

@@ -89,22 +89,15 @@ def detect_source_type(url: str) -> str:
 
 def is_blacklisted(url: str) -> bool:
     try:
-        domain = urlparse(url).netloc.lower().split(":")[0]
-        if domain.startswith("www."):
-            domain = domain[4:]
-        # Exact domain or subdomain match — NOT substring (evitaba bloquear
-        # netflix.com / phoenix.com por contener "x.com", etc.)
-        return any(domain == bl or domain.endswith("." + bl)
-                   for bl in BLACKLIST_DOMAINS)
+        domain = urlparse(url).netloc.lower().replace("www.", "")
+        return any(bl in domain for bl in BLACKLIST_DOMAINS)
     except Exception:
         return True
 
 
 def normalize_url(url: str) -> str:
-    # Strip only the fragment. NO borrar el query string: rompía URLs de
-    # YouTube (watch?v=...) y artículos que enrutan por query (?id=, ?p=).
     parsed = urlparse(url)
-    clean = parsed._replace(fragment="")
+    clean = parsed._replace(fragment="", query="")
     return clean.geturl().rstrip("/")
 
 
@@ -162,16 +155,13 @@ class ExhaustiveScraper:
 
     async def seed(self):
         """Initial broad search across multiple sources"""
-        logger.info("Seeding research", topic=self.topic,
-                    youtube=settings.enable_youtube, reddit=settings.enable_reddit)
+        logger.info("Seeding research", topic=self.topic)
         tasks = [
             self._seed_search(),
             self._seed_wikipedia(),
+            self._seed_reddit(),
+            self._seed_youtube(),
         ]
-        if settings.enable_reddit:
-            tasks.append(self._seed_reddit())
-        if settings.enable_youtube:
-            tasks.append(self._seed_youtube())
         await asyncio.gather(*tasks, return_exceptions=True)
 
     async def _generate_ddg_queries(self) -> list[str]:
@@ -190,9 +180,9 @@ class ExhaustiveScraper:
             return fallback
 
         try:
-            from src.llm import get_anthropic_client
+            import anthropic
             logger.info("Generating DDG queries with Claude", topic=self.topic)
-            client = get_anthropic_client()
+            client = anthropic.AsyncAnthropic(api_key=settings.anthropic_api_key)
             prompt = (
                 f'Generate exactly 8 DuckDuckGo search queries to research: "{self.topic}"\n\n'
                 f'Rules:\n'
@@ -225,7 +215,7 @@ class ExhaustiveScraper:
     async def _search_searxng(self, query: str) -> list[dict]:
         """Busca en SearXNG y retorna lista de {href, title}. Retorna [] si no disponible."""
         import aiohttp
-        searxng_url = settings.searxng_url
+        searxng_url = "http://searxng-svc.researchowl.svc.cluster.local:8080/search"
         params = {
             "q": query,
             "format": "json",
@@ -423,14 +413,6 @@ class ExhaustiveScraper:
             url = source["url"]
             source_id = source["id"]
 
-            # Saltar fuentes desactivadas (también las descubiertas dentro de
-            # páginas web, no solo las del seed) sin gastar una petición de red.
-            if ((source_type == "youtube" and not settings.enable_youtube) or
-                    (source_type == "reddit" and not settings.enable_reddit)):
-                await self.db.update_source(source_id, status="skipped",
-                                            error=f"{source_type} disabled")
-                return 0
-
             try:
                 try:
                     cached = await self.db.get_cached_content(url)
@@ -551,8 +533,7 @@ class ExhaustiveScraper:
 
         # Extract title and new URLs with BS4
         soup = BeautifulSoup(html, "lxml")
-        # .string es None si el <title> tiene tags anidados; get_text es robusto
-        title = soup.title.get_text(strip=True) if soup.title else url
+        title = soup.title.string.strip() if soup.title else url
 
         new_urls = []
         if depth < settings.max_depth:
@@ -612,7 +593,7 @@ class ExhaustiveScraper:
             return None, None
 
         video_id = match.group(1)
-        loop = asyncio.get_running_loop()
+        loop = asyncio.get_event_loop()
 
         def _fetch():
             return YouTubeTranscriptApi.get_transcript(