chore(deps): update dependency pydantic-settings to v2.14.2

get_db() devuelve un proxy sobre una única conexión real reutilizada durante
toda la vida del proceso, en vez de abrir una conexión nueva y re-ejecutar
todo el SCHEMA en cada comando/scoring.

- _SharedConnection: proxy con close() no-op → los handlers conservan el
  patrón get_db()/finally close() sin cambios
- aiosqlite serializa las operaciones en el hilo de la conexión: compartirla
  entre coroutines es seguro y elimina la contención del lock de escritura a
  nivel de fichero (scheduler solapado, /compare con 2 sesiones)
- close_db() vía post_shutdown para checkpoint WAL limpio al apagar

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

.gitignore

@@ -0,0 +1,35 @@
+# 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

@@ -32,7 +32,7 @@ reportlab==4.2.5
 
 # Utilities
 pydantic==2.8.0
-pydantic-settings==2.14.1
+pydantic-settings==2.14.2
 tenacity==9.0.0
 structlog==24.4.0
 python-dotenv==1.0.1

src/bot/bot.py

@@ -5,8 +5,9 @@ Main user interface — all commands handled here
 import asyncio
 import os
 import time
-from datetime import datetime
+from datetime import datetime, timedelta
 from typing import Optional
+from zoneinfo import ZoneInfo
 
 import structlog
 from telegram import Update, Message
@@ -17,7 +18,7 @@ from telegram.ext import (
 from telegram.constants import ParseMode
 
 from src.config import settings
-from src.db.database import get_db, ResearchDB, ResearchStatus, OutputType
+from src.db.database import get_db, close_db, ResearchDB, ResearchStatus, OutputType
 from src.scraper.exhaustive import ExhaustiveScraper
 from src.processor.processor import OllamaClient, ContentProcessor
 from src.generator.generator import OutputGenerator
@@ -536,22 +537,62 @@ async def cmd_costs(update: Update, ctx: ContextTypes.DEFAULT_TYPE):
         await db_conn.close()
 
 
+def _parse_at_time(time_str: str) -> tuple[float, int, int, bool]:
+    """Parse 'HH:MM' in the configured timezone.
+
+    Returns (next_run_at_unix, hour, minute, is_today). If the time has
+    already passed today, schedules for tomorrow (is_today=False).
+    Raises ValueError if the string is not a valid HH:MM time.
+    """
+    parts = time_str.split(":")
+    if len(parts) != 2 or not (parts[0].isdigit() and parts[1].isdigit()):
+        raise ValueError(f"Hora inválida: {time_str!r}")
+    hour, minute = int(parts[0]), int(parts[1])
+    if not (0 <= hour <= 23 and 0 <= minute <= 59):
+        raise ValueError(f"Hora fuera de rango: {time_str!r}")
+
+    tz = ZoneInfo(settings.timezone)
+    now_local = datetime.now(tz)
+    target = now_local.replace(hour=hour, minute=minute, second=0, microsecond=0)
+    is_today = True
+    if target <= now_local:
+        target += timedelta(days=1)
+        is_today = False
+    return target.timestamp(), hour, minute, is_today
+
+
 async def cmd_watch(update: Update, ctx: ContextTypes.DEFAULT_TYPE):
     if not is_authorized(update.effective_user.id):
         return
 
     chat_id = update.effective_chat.id
-    args = ctx.args or []
+    args = list(ctx.args or [])
 
     if not args:
         await update.message.reply_text(
-            "❌ Uso: `/watch <tema> [horas]`\nEjemplo: `/watch Incidente Roswell 24`",
+            "❌ Uso: `/watch <tema> [horas]` o `/watch <tema> --at HH:MM [horas]`\n"
+            "Ejemplo: `/watch Incidente Roswell 24`\n"
+            "Ejemplo: `/watch Incidente Roswell --at 19:30 6`",
             parse_mode=ParseMode.MARKDOWN
         )
         return
 
+    # Extract optional --at HH:MM from anywhere in the args
+    at_time_str: Optional[str] = None
+    if "--at" in args:
+        idx = args.index("--at")
+        if idx + 1 >= len(args):
+            await update.message.reply_text(
+                "❌ `--at` requiere una hora en formato HH:MM. Ejemplo: `--at 19:30`",
+                parse_mode=ParseMode.MARKDOWN
+            )
+            return
+        at_time_str = args[idx + 1]
+        # Remove '--at' and its value from args
+        del args[idx:idx + 2]
+
     interval_hours = 24
-    if args[-1].isdigit():
+    if args and args[-1].isdigit():
         interval_hours = int(args[-1])
         topic = " ".join(args[:-1]).strip()
     else:
@@ -567,14 +608,31 @@ async def cmd_watch(update: Update, ctx: ContextTypes.DEFAULT_TYPE):
         )
         return
 
+    next_run_at: Optional[float] = None
+    when_msg = f"Primera ejecución en ~{interval_hours}h."
+    if at_time_str is not None:
+        try:
+            next_run_at, hour, minute, is_today = _parse_at_time(at_time_str)
+        except ValueError:
+            await update.message.reply_text(
+                "❌ Hora inválida. Usa el formato HH:MM (24h). Ejemplo: `--at 19:30`",
+                parse_mode=ParseMode.MARKDOWN
+            )
+            return
+        day_word = "hoy" if is_today else "mañana"
+        when_msg = (
+            f"Primera ejecución: {day_word} a las {hour:02d}:{minute:02d} "
+            f"({settings.timezone})"
+        )
+
     db_conn = await get_db()
     db = ResearchDB(db_conn)
     try:
         try:
-            await db.add_watch(topic, chat_id, interval_hours)
+            await db.add_watch(topic, chat_id, interval_hours, next_run_at=next_run_at)
             await update.message.reply_text(
                 f"👁 Watching: `{topic}` — cada {interval_hours}h\n"
-                f"Primera ejecución en ~{interval_hours}h.\n"
+                f"{when_msg}\n"
                 f"Usa /watches para ver todos tus temas.",
                 parse_mode=ParseMode.MARKDOWN
             )
@@ -633,14 +691,27 @@ async def cmd_watches(update: Update, ctx: ContextTypes.DEFAULT_TYPE):
             return
 
         now = time.time()
+        tz = ZoneInfo(settings.timezone)
+        today_local = datetime.now(tz).date()
         lines = ["👁 *Tus temas vigilados:*\n"]
         for i, w in enumerate(watches, 1):
             secs_remaining = max(0.0, w["next_run_at"] - now)
             hours_remaining = secs_remaining / 3600
             eta = f"{int(secs_remaining / 60)}min" if hours_remaining < 1 else f"{hours_remaining:.1f}h"
             status = "✅" if w["enabled"] else "⏸"
+
+            nxt = datetime.fromtimestamp(w["next_run_at"], tz)
+            if nxt.date() == today_local:
+                day_word = "hoy"
+            elif nxt.date() == today_local + timedelta(days=1):
+                day_word = "mañana"
+            else:
+                day_word = nxt.strftime("%d/%m")
+            local_time = nxt.strftime("%H:%M")
+
             lines.append(
-                f"{i}. {status} `{w['topic']}` — cada {w['interval_hours']}h · próxima en {eta}"
+                f"{i}. {status} `{w['topic']}` — cada {w['interval_hours']}h · "
+                f"próxima a las {local_time} ({day_word}) · en {eta}"
             )
 
         await update.message.reply_text("\n".join(lines), parse_mode=ParseMode.MARKDOWN)
@@ -745,6 +816,23 @@ async def _purge_on_startup(app: Application) -> None:
         await db_conn.close()
 
 
+async def _safe_send(bot, chat_id, text: str):
+    """Envía un mensaje, con fallback a texto plano si falla el parseo Markdown.
+
+    Los resúmenes generados por Claude suelen contener entidades Markdown
+    desbalanceadas (*, _, [, `) que hacen que Telegram rechace el mensaje con
+    BadRequest. Sin este fallback, el envío fallaba en silencio.
+    """
+    try:
+        await bot.send_message(chat_id, text, parse_mode=ParseMode.MARKDOWN)
+    except Exception as e:
+        logger.warning("Envío Markdown falló, reintentando en texto plano", error=str(e))
+        try:
+            await bot.send_message(chat_id, text)
+        except Exception as e2:
+            logger.error("Envío en texto plano también falló", chat_id=chat_id, error=str(e2))
+
+
 async def _scheduler_loop(app: Application):
     while True:
         db_conn = None
@@ -761,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, w_id=watch["id"]):
+                async def _task(c=chat_id, t=topic, s=session_id):
                     inner_db_conn = await get_db()
                     inner_db = ResearchDB(inner_db_conn)
                     try:
@@ -787,24 +875,23 @@ async def _scheduler_loop(app: Application):
                             )
 
                         if summary:
-                            await app.bot.send_message(
-                                c, summary, parse_mode=ParseMode.MARKDOWN
-                            )
+                            await _safe_send(app.bot, c, summary)
                         else:
-                            await app.bot.send_message(
-                                c,
-                                f"🔄 *{t}* — sin novedades significativas esta vez.",
-                                parse_mode=ParseMode.MARKDOWN
+                            await _safe_send(
+                                app.bot, c,
+                                f"🔄 *{t}* — sin novedades significativas esta vez."
                             )
+                    except Exception as e:
+                        logger.error("Tarea programada falló",
+                                     topic=t, session_id=s, error=str(e))
                     finally:
                         await inner_db_conn.close()
 
                 task = asyncio.create_task(_task())
                 _active_tasks[chat_id] = task
-                await app.bot.send_message(
-                    chat_id,
-                    f"🔄 Investigación automática iniciada: `{topic}`",
-                    parse_mode=ParseMode.MARKDOWN
+                await _safe_send(
+                    app.bot, chat_id,
+                    f"🔄 Investigación automática iniciada: `{topic}`"
                 )
         except Exception as e:
             logger.warning("Scheduler loop error", error=str(e))
@@ -826,6 +913,10 @@ 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
@@ -1172,6 +1263,7 @@ def create_bot() -> Application:
         Application.builder()
         .token(settings.telegram_bot_token)
         .post_init(_on_startup)
+        .post_shutdown(_on_shutdown)
         .build()
     )
 

src/config.py

@@ -24,10 +24,19 @@ 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")
@@ -44,6 +53,7 @@ class Settings(BaseSettings):
 
     # App
     log_level: str = Field("INFO", env="LOG_LEVEL")
+    timezone: str = Field("Europe/Madrid", env="TIMEZONE")
 
     @property
     def allowed_user_ids(self) -> list[int]:

src/db/database.py

@@ -1,3 +1,4 @@
+import asyncio
 import aiosqlite
 import json
 import time
@@ -117,15 +118,65 @@ 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:
-    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
+    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:
@@ -341,11 +392,26 @@ 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):
-        # 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
+        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)
@@ -378,13 +444,16 @@ class ResearchDB:
 
     # --- Watched Topics ---
 
-    async def add_watch(self, topic: str, chat_id: int, interval_hours: int) -> int:
+    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, now + interval_hours * 3600, now)
+            (topic, chat_id, interval_hours, next_run_at, now)
         )
         await self.db.commit()
         return cursor.lastrowid
@@ -439,7 +508,7 @@ class ResearchDB:
         )
         session_ids = [row[0] for row in await cursor.fetchall()]
 
-        counts = {"sessions": 0, "sources": 0, "chunks": 0, "outputs": 0}
+        counts = {"sessions": 0, "sources": 0, "chunks": 0, "outputs": 0, "api_usage": 0}
 
         for sid in session_ids:
             await self.db.execute(
@@ -450,6 +519,8 @@ class ResearchDB:
             counts["chunks"] += cur.rowcount
             cur = await self.db.execute("DELETE FROM outputs WHERE session_id = ?", (sid,))
             counts["outputs"] += 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,))

src/generator/generator.py

@@ -12,6 +12,7 @@ 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
 
@@ -442,10 +443,9 @@ 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 = anthropic.AsyncAnthropic(api_key=settings.anthropic_api_key)
+            client = get_anthropic_client()
             msg = await client.messages.create(
                 model=settings.claude_model,
                 max_tokens=max_tokens,
@@ -613,9 +613,8 @@ class OutputGenerator:
     async def _generate_raw(self, prompt: str,
                              session_id: int | None = None) -> str:
         if settings.anthropic_api_key:
-            import anthropic
             try:
-                client = anthropic.AsyncAnthropic(api_key=settings.anthropic_api_key)
+                client = get_anthropic_client()
                 msg = await client.messages.create(
                     model=settings.claude_model,
                     max_tokens=2048,
@@ -819,8 +818,7 @@ async def generate_diff_summary(
         )
 
     try:
-        import anthropic
-        client = anthropic.AsyncAnthropic(api_key=settings.anthropic_api_key)
+        client = get_anthropic_client()
         prompt = (
             f'Analiza el siguiente material de investigación sobre "{topic}" '
             f'y genera un resumen BREVE (máximo 300 palabras) de las novedades '
@@ -906,8 +904,7 @@ async def generate_comparison(
     )
 
     try:
-        import anthropic
-        client = anthropic.AsyncAnthropic(api_key=settings.anthropic_api_key)
+        client = get_anthropic_client()
         msg = await client.messages.create(
             model=settings.claude_model,
             max_tokens=8192,

src/llm.py

@@ -0,0 +1,15 @@
+"""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,6 +23,7 @@ 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:
@@ -47,7 +48,7 @@ class OllamaClient:
 
     async def embed(self, text: str) -> Optional[list[float]]:
         """Get embedding vector for a text"""
-        payload = {"model": self.model, "prompt": text}
+        payload = {"model": self.embed_model, "prompt": text}
         try:
             async with httpx.AsyncClient(timeout=60) as client:
                 resp = await client.post(f"{self.base_url}/api/embeddings", json=payload)
@@ -69,9 +70,28 @@ 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 boundaries when possible.
+    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.
     """
-    paragraphs = [p.strip() for p in text.split("\n\n") if p.strip()]
+    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]))
+
     chunks = []
     current = []
     current_words = 0
@@ -80,10 +100,12 @@ 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: keep last paragraph
-            if overlap > 0 and current:
-                current = [current[-1]]
-                current_words = len(current[0].split())
+            # 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)
             else:
                 current = []
                 current_words = 0
@@ -124,7 +146,6 @@ 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"]
 
@@ -132,7 +153,6 @@ 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
 
@@ -223,33 +243,45 @@ 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 in enumerate(chunks):
-            words = len(chunk.split())
-            if words < 30:
-                continue
-
-            quality = await self._score_quality(chunk, topic, session_id)
+        for (i, chunk), quality in zip(candidates, qualities):
             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=words)
+                             threshold=settings.quality_threshold,
+                             words=len(chunk.split()))
                 continue
 
-            embedding = await self.ollama.embed(chunk[:1000])
+            # 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)
 
             await self.db.add_chunk(
                 session_id=session_id,
                 source_id=source_id,
                 content=chunk,
                 chunk_index=i,
-                token_count=words,
+                token_count=len(chunk.split()),
                 quality_score=quality,
                 embedding=embedding
             )
@@ -274,9 +306,84 @@ 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:
-        import anthropic
+        from src.llm import get_anthropic_client
         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+. '
@@ -284,7 +391,7 @@ class ContentProcessor:
             f'Reply with only a number.\n\nText:\n{chunk[:500]}'
         )
         try:
-            client = anthropic.AsyncAnthropic(api_key=settings.anthropic_api_key)
+            client = get_anthropic_client()
             msg = await client.messages.create(
                 model=settings.claude_model,
                 max_tokens=10,

src/scraper/exhaustive.py

@@ -89,15 +89,22 @@ def detect_source_type(url: str) -> str:
 
 def is_blacklisted(url: str) -> bool:
     try:
-        domain = urlparse(url).netloc.lower().replace("www.", "")
-        return any(bl in domain for bl in BLACKLIST_DOMAINS)
+        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)
     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="", query="")
+    clean = parsed._replace(fragment="")
     return clean.geturl().rstrip("/")
 
 
@@ -155,13 +162,16 @@ class ExhaustiveScraper:
 
     async def seed(self):
         """Initial broad search across multiple sources"""
-        logger.info("Seeding research", topic=self.topic)
+        logger.info("Seeding research", topic=self.topic,
+                    youtube=settings.enable_youtube, reddit=settings.enable_reddit)
         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]:
@@ -180,9 +190,9 @@ class ExhaustiveScraper:
             return fallback
 
         try:
-            import anthropic
+            from src.llm import get_anthropic_client
             logger.info("Generating DDG queries with Claude", topic=self.topic)
-            client = anthropic.AsyncAnthropic(api_key=settings.anthropic_api_key)
+            client = get_anthropic_client()
             prompt = (
                 f'Generate exactly 8 DuckDuckGo search queries to research: "{self.topic}"\n\n'
                 f'Rules:\n'
@@ -215,7 +225,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 = "http://searxng-svc.researchowl.svc.cluster.local:8080/search"
+        searxng_url = settings.searxng_url
         params = {
             "q": query,
             "format": "json",
@@ -413,6 +423,14 @@ 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)
@@ -533,7 +551,8 @@ class ExhaustiveScraper:
 
         # Extract title and new URLs with BS4
         soup = BeautifulSoup(html, "lxml")
-        title = soup.title.string.strip() if soup.title else url
+        # .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
 
         new_urls = []
         if depth < settings.max_depth:
@@ -593,7 +612,7 @@ class ExhaustiveScraper:
             return None, None
 
         video_id = match.group(1)
-        loop = asyncio.get_event_loop()
+        loop = asyncio.get_running_loop()
 
         def _fetch():
             return YouTubeTranscriptApi.get_transcript(