From bf275b7f82bcb07974fec9a0311a855bfb132463 Mon Sep 17 00:00:00 2001 From: ChemaVX Date: Mon, 15 Jun 2026 14:38:03 +0000 Subject: [PATCH] fix: correcciones de scraping/DB y mejoras de robustez MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Sección crítica: - is_blacklisted: match por dominio/subdominio exacto (antes "x.com" como substring bloqueaba netflix.com, phoenix.com, etc.) - normalize_url: conserva el query string (rompía YouTube watch?v= y URLs con ?id=); solo borra el fragment - get_db: PRAGMA busy_timeout=5000 para evitar "database is locked" en /compare y watches solapados - OllamaClient.embed: usa OLLAMA_EMBED_MODEL en vez del modelo de chat - log_api_call: coste por modelo (opus/sonnet/haiku) en vez de Haiku fijo Mejoras: - src/llm.py: cliente Anthropic compartido y cacheado (antes se instanciaba uno por cada llamada/chunk) - SEARXNG_URL configurable via env - get_running_loop() en vez de get_event_loop() (deprecado) - soup.title.get_text() robusto ante con tags anidados - limpieza: import muerto, total_words duplicado, w_id no usado Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> --- src/bot/bot.py | 2 +- src/config.py | 4 ++++ src/db/database.py | 25 ++++++++++++++++++++++--- src/generator/generator.py | 13 +++++-------- src/llm.py | 15 +++++++++++++++ src/processor/processor.py | 9 ++++----- src/scraper/exhaustive.py | 24 ++++++++++++++++-------- 7 files changed, 67 insertions(+), 25 deletions(-) create mode 100644 src/llm.py diff --git a/src/bot/bot.py b/src/bot/bot.py index 0be9c17..6c63c8b 100644 --- a/src/bot/bot.py +++ b/src/bot/bot.py @@ -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, 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: diff --git a/src/config.py b/src/config.py index 676f9e5..821bed3 100644 --- a/src/config.py +++ b/src/config.py @@ -24,6 +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 diff --git a/src/db/database.py b/src/db/database.py index 64ca0ab..a59b472 100644 --- a/src/db/database.py +++ b/src/db/database.py @@ -123,6 +123,10 @@ async def get_db() -> aiosqlite.Connection: db.row_factory = aiosqlite.Row await db.execute("PRAGMA journal_mode=WAL") await db.execute("PRAGMA synchronous=NORMAL") + # Espera hasta 5s si otra conexión tiene el lock de escritura (scheduler + # solapado con research activa, o las 2 sesiones de /compare) en vez de + # fallar al instante con "database is locked". + await db.execute("PRAGMA busy_timeout=5000") await db.executescript(SCHEMA) await db.commit() return db @@ -341,11 +345,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) diff --git a/src/generator/generator.py b/src/generator/generator.py index 5729a43..e969381 100644 --- a/src/generator/generator.py +++ b/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, diff --git a/src/llm.py b/src/llm.py new file mode 100644 index 0000000..8abc0db --- /dev/null +++ b/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) diff --git a/src/processor/processor.py b/src/processor/processor.py index 9b3db7f..98ab1e6 100644 --- a/src/processor/processor.py +++ b/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) @@ -124,7 +125,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 +132,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 @@ -276,7 +275,7 @@ class ContentProcessor: 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 +283,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, diff --git a/src/scraper/exhaustive.py b/src/scraper/exhaustive.py index 710a4b5..690d557 100644 --- a/src/scraper/exhaustive.py +++ b/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("/") @@ -180,9 +187,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 +222,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", @@ -533,7 +540,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 +601,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(