Update routers/searchterm.py

routers/searchterm.py

@@ -10,17 +10,13 @@ import uuid
 import time
 from pathlib import Path
 from urllib.parse import urlparse
-from typing import List, Dict, Any, Optional, Set, Tuple
+from typing import List, Dict, Any, Optional
 from fastapi import APIRouter, HTTPException, Body
 from fastapi.responses import FileResponse
 from newspaper import Article
 from threading import Timer
 from google import genai
 from google.genai import types
-from asyncio import Queue, create_task, gather
-from concurrent.futures import ThreadPoolExecutor
-import aiofiles
-import ujson  # JSON mais rápido
 
 router = APIRouter()
 
@@ -45,16 +41,10 @@ USER_AGENTS = [
     "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36",
 ]
 
-BLOCKED_DOMAINS = frozenset({  # frozenset é mais rápido para lookup
-    "reddit.com", "www.reddit.com", "old.reddit.com",
-    "quora.com", "www.quora.com"
-})
+BLOCKED_DOMAINS = {"reddit.com", "www.reddit.com", "old.reddit.com",
+                   "quora.com", "www.quora.com"}
 
 MAX_TEXT_LENGTH = 4000
-MAX_CONCURRENT_SEARCHES = 30  # Aumentado
-MAX_CONCURRENT_EXTRACTIONS = 80  # Aumentado significativamente
-EXTRACTION_TIMEOUT = 8  # Reduzido
-HTTP_TIMEOUT = 10  # Reduzido
 
 # Diretório para arquivos temporários
 TEMP_DIR = Path("/tmp")
@@ -63,30 +53,22 @@ TEMP_DIR.mkdir(exist_ok=True)
 # Dicionário para controlar arquivos temporários
 temp_files = {}
 
-# Pool de threads para operações CPU-intensive
-thread_pool = ThreadPoolExecutor(max_workers=20)
-
-# Cache de domínios bloqueados para evitar verificações repetidas
-domain_cache = {}
 
 def is_blocked_domain(url: str) -> bool:
     try:
         host = urlparse(url).netloc.lower()
-        
-        # Cache lookup
-        if host in domain_cache:
-            return domain_cache[host]
-        
-        is_blocked = any(host == b or host.endswith("." + b) for b in BLOCKED_DOMAINS)
-        domain_cache[host] = is_blocked
-        return is_blocked
+        return any(host == b or host.endswith("." + b) for b in BLOCKED_DOMAINS)
     except Exception:
         return False
 
+
 def clamp_text(text: str) -> str:
-    if not text or len(text) <= MAX_TEXT_LENGTH:
-        return text
-    return text[:MAX_TEXT_LENGTH]
+    if not text:
+        return ""
+    if len(text) > MAX_TEXT_LENGTH:
+        return text[:MAX_TEXT_LENGTH]
+    return text
+
 
 def get_realistic_headers() -> Dict[str, str]:
     return {
@@ -94,9 +76,9 @@ def get_realistic_headers() -> Dict[str, str]:
         "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8",
         "Accept-Language": "en-US,en;q=0.7,pt-BR;q=0.6",
         "Connection": "keep-alive",
-        "Accept-Encoding": "gzip, deflate, br",
     }
 
+
 def delete_temp_file(file_id: str, file_path: Path):
     """Remove arquivo temporário após expiração"""
     try:
@@ -107,16 +89,17 @@ def delete_temp_file(file_id: str, file_path: Path):
     except Exception as e:
         print(f"Erro ao remover arquivo temporário: {e}")
 
-async def create_temp_file(data: Dict[str, Any]) -> Dict[str, str]:
-    """Cria arquivo temporário assíncrono e agenda sua remoção"""
+
+def create_temp_file(data: Dict[str, Any]) -> Dict[str, str]:
+    """Cria arquivo temporário e agenda sua remoção"""
     file_id = str(uuid.uuid4())
     file_path = TEMP_DIR / f"fontes_{file_id}.txt"
     
-    # Salva o JSON no arquivo de forma assíncrona
-    async with aiofiles.open(file_path, 'w', encoding='utf-8') as f:
-        await f.write(ujson.dumps(data, ensure_ascii=False, indent=2))
+    # Salva o JSON no arquivo
+    with open(file_path, 'w', encoding='utf-8') as f:
+        json.dump(data, f, ensure_ascii=False, indent=2)
     
-    # Agenda remoção em 24 horas
+    # Agenda remoção em 24 horas (86400 segundos)
     timer = Timer(86400, delete_temp_file, args=[file_id, file_path])
     timer.start()
     
@@ -133,6 +116,7 @@ async def create_temp_file(data: Dict[str, Any]) -> Dict[str, str]:
         "expires_in_hours": 24
     }
 
+
 async def generate_search_terms(context: str) -> List[str]:
     """Gera termos de pesquisa usando o modelo Gemini"""
     try:
@@ -179,7 +163,9 @@ Retorne apenas o JSON, sem mais nenhum texto."""
         ]
         
         generate_content_config = types.GenerateContentConfig(
-            thinking_config=types.ThinkingConfig(thinking_budget=0),
+            thinking_config=types.ThinkingConfig(
+                thinking_budget=0,
+            ),
         )
         
         # Coletamos toda a resposta em stream
@@ -194,6 +180,7 @@ Retorne apenas o JSON, sem mais nenhum texto."""
         
         # Tenta extrair o JSON da resposta
         try:
+            # Remove possíveis ```json e ``` da resposta
             clean_response = full_response.strip()
             if clean_response.startswith("```json"):
                 clean_response = clean_response[7:]
@@ -201,170 +188,92 @@ Retorne apenas o JSON, sem mais nenhum texto."""
                 clean_response = clean_response[:-3]
             clean_response = clean_response.strip()
             
-            response_data = ujson.loads(clean_response)
+            # Parse do JSON
+            response_data = json.loads(clean_response)
             terms = response_data.get("terms", [])
             
+            # Validação básica
             if not isinstance(terms, list):
                 raise ValueError("Terms deve ser uma lista")
             
-            return terms[:20]
+            return terms[:20]  # Garante máximo de 20 termos
             
-        except (ujson.JSONDecodeError, ValueError) as e:
+        except (json.JSONDecodeError, ValueError) as e:
             print(f"Erro ao parsear resposta do Gemini: {e}")
+            print(f"Resposta recebida: {full_response}")
+            # Retorna uma lista vazia em caso de erro
             return []
         
     except Exception as e:
         print(f"Erro ao gerar termos de pesquisa: {str(e)}")
         return []
 
-async def search_brave_batch(client: httpx.AsyncClient, terms: List[str]) -> List[Tuple[str, List[Dict[str, str]]]]:
-    """Busca múltiplos termos em paralelo com otimizações"""
-    semaphore = asyncio.Semaphore(MAX_CONCURRENT_SEARCHES)
+
+async def search_brave_term(client: httpx.AsyncClient, term: str) -> List[Dict[str, str]]:
+    params = {"q": term, "count": 10, "safesearch": "off", "summary": "false"}
     
-    async def search_single_term(term: str) -> Tuple[str, List[Dict[str, str]]]:
-        async with semaphore:
-            params = {"q": term, "count": 10, "safesearch": "off", "summary": "false"}
-            
-            try:
-                resp = await client.get(BRAVE_SEARCH_URL, headers=BRAVE_HEADERS, params=params)
-                if resp.status_code != 200:
-                    return (term, [])
+    try:
+        resp = await client.get(BRAVE_SEARCH_URL, headers=BRAVE_HEADERS, params=params)
+        if resp.status_code != 200:
+            return []
 
-                data = resp.json()
-                results = []
+        data = resp.json()
+        results: List[Dict[str, str]] = []
+        
+        if "web" in data and "results" in data["web"]:
+            for item in data["web"]["results"]:
+                url = item.get("url")
+                age = item.get("age", "Unknown")
                 
-                if "web" in data and "results" in data["web"]:
-                    for item in data["web"]["results"]:
-                        url = item.get("url")
-                        age = item.get("age", "Unknown")
-                        
-                        if url and not is_blocked_domain(url):
-                            results.append({"url": url, "age": age})
-
-                return (term, results)
-            except Exception as e:
-                print(f"Erro na busca do termo '{term}': {e}")
-                return (term, [])
-    
-    # Executa todas as buscas em paralelo
-    tasks = [search_single_term(term) for term in terms]
-    return await gather(*tasks, return_exceptions=False)
+                if url and not is_blocked_domain(url):
+                    results.append({"url": url, "age": age})
 
-def extract_with_trafilatura(html: str) -> str:
-    """Extração CPU-intensive executada em thread pool"""
-    try:
-        extracted = trafilatura.extract(html)
-        return extracted.strip() if extracted else ""
+        return results
     except Exception:
-        return ""
+        return []
+
 
-def extract_with_newspaper(url: str) -> str:
-    """Extração com newspaper executada em thread pool"""
+async def extract_article_text(url: str, session: aiohttp.ClientSession) -> str:
     try:
         art = Article(url)
         art.config.browser_user_agent = random.choice(USER_AGENTS)
-        art.config.request_timeout = 6
+        art.config.request_timeout = 8
         art.config.number_threads = 1
+
         art.download()
         art.parse()
-        return (art.text or "").strip()
+        txt = (art.text or "").strip()
+        if txt and len(txt) > 100:
+            return clamp_text(txt)
     except Exception:
-        return ""
+        pass
 
-async def extract_article_text_optimized(url: str, session: aiohttp.ClientSession) -> str:
-    """Extração de artigo otimizada com fallback robusto"""
-    
-    # Método 1: Tentar com trafilatura primeiro (mais rápido)
     try:
+        await asyncio.sleep(random.uniform(0.1, 0.3))
+        
         headers = get_realistic_headers()
-        async with session.get(url, headers=headers, timeout=EXTRACTION_TIMEOUT) as resp:
-            if resp.status == 200:
-                html = await resp.text()
+        async with session.get(url, headers=headers, timeout=12) as resp:
+            if resp.status != 200:
+                return ""
                 
-                # Verifica paywall rapidamente
-                if not re.search(r"(paywall|subscribe|metered|registration|captcha|access denied)", 
-                            html[:2000], re.I):
-                    
-                    # Extração com trafilatura em thread pool
-                    try:
-                        trafilatura_result = await asyncio.get_event_loop().run_in_executor(
-                            thread_pool, extract_with_trafilatura, html
-                        )
-                        
-                        if trafilatura_result and len(trafilatura_result.strip()) > 100:
-                            return clamp_text(trafilatura_result.strip())
-                    except Exception as e:
-                        print(f"Erro trafilatura para {url}: {e}")
-                        
-    except Exception as e:
-        print(f"Erro HTTP para {url}: {e}")
-    
-    # Método 2: Fallback para newspaper
-    try:
-        newspaper_result = await asyncio.get_event_loop().run_in_executor(
-            thread_pool, extract_with_newspaper, url
-        )
-        
-        if newspaper_result and len(newspaper_result.strip()) > 100:
-            return clamp_text(newspaper_result.strip())
+            html = await resp.text()
             
-    except Exception as e:
-        print(f"Erro newspaper para {url}: {e}")
-    
+            if re.search(r"(paywall|subscribe|metered|registration|captcha|access denied)", html, re.I):
+                return ""
+
+            extracted = trafilatura.extract(html) or ""
+            extracted = extracted.strip()
+            if extracted and len(extracted) > 100:
+                return clamp_text(extracted)
+                
+    except Exception:
+        pass
+
     return ""
 
-async def process_urls_batch(session: aiohttp.ClientSession, urls_data: List[Tuple[str, str, str]]) -> List[Dict[str, Any]]:
-    """Processa URLs em lotes otimizados com logging detalhado"""
-    semaphore = asyncio.Semaphore(MAX_CONCURRENT_EXTRACTIONS)
-    results = []
-    used_urls: Set[str] = set()
-    success_count = 0
-    
-    async def process_single_url(term: str, url: str, age: str) -> Optional[Dict[str, Any]]:
-        nonlocal success_count
-        async with semaphore:
-            if url in used_urls:
-                return None
-            
-            try:
-                text = await extract_article_text_optimized(url, session)
-                if text:
-                    used_urls.add(url)
-                    success_count += 1
-                    print(f"✓ Extraído: {url[:60]}... ({len(text)} chars)")
-                    return {
-                        "term": term,
-                        "age": age,
-                        "url": url,
-                        "text": text
-                    }
-                else:
-                    print(f"✗ Falhou: {url[:60]}... (sem conteúdo)")
-            except Exception as e:
-                print(f"✗ Erro: {url[:60]}... - {str(e)[:50]}")
-            
-            return None
-    
-    # Cria todas as tasks de uma vez
-    tasks = []
-    for term, url, age in urls_data:
-        tasks.append(process_single_url(term, url, age))
-    
-    print(f"Processando {len(tasks)} URLs com semáforo de {MAX_CONCURRENT_EXTRACTIONS}...")
-    
-    # Processa tudo em paralelo
-    processed_results = await gather(*tasks, return_exceptions=True)
-    
-    # Filtra resultados válidos
-    valid_results = [r for r in processed_results if r is not None and not isinstance(r, Exception)]
-    
-    print(f"Sucesso: {success_count}/{len(urls_data)} URLs extraídas")
-    return valid_results
 
 @router.post("/search-terms")
 async def search_terms(payload: Dict[str, str] = Body(...)) -> Dict[str, Any]:
-    start_time = time.time()
-    
     context = payload.get("context")
     if not context or not isinstance(context, str):
         raise HTTPException(status_code=400, detail="Campo 'context' é obrigatório e deve ser uma string.")
@@ -372,73 +281,78 @@ async def search_terms(payload: Dict[str, str] = Body(...)) -> Dict[str, Any]:
     if len(context.strip()) == 0:
         raise HTTPException(status_code=400, detail="Campo 'context' não pode estar vazio.")
     
-    print(f"Iniciando geração de termos...")
     # Gera os termos de pesquisa usando o Gemini
     terms = await generate_search_terms(context)
     
     if not terms:
         raise HTTPException(status_code=500, detail="Não foi possível gerar termos de pesquisa válidos.")
 
-    print(f"Termos gerados em {time.time() - start_time:.2f}s. Iniciando buscas...")
+    used_urls = set()
+    search_semaphore = asyncio.Semaphore(20)
+    extract_semaphore = asyncio.Semaphore(50)
     
-    # Configurações otimizadas para conexões
-    connector = aiohttp.TCPConnector(
-        limit=200,  # Aumentado
-        limit_per_host=30,  # Aumentado
-        ttl_dns_cache=300,
-        use_dns_cache=True,
-        enable_cleanup_closed=True
-    )
-    timeout = aiohttp.ClientTimeout(total=HTTP_TIMEOUT, connect=5)
+    async def search_with_limit(client, term):
+        async with search_semaphore:
+            return await search_brave_term(client, term)
     
-    # Cliente HTTP otimizado
-    http_client = httpx.AsyncClient(
-        timeout=HTTP_TIMEOUT,
-        limits=httpx.Limits(
-            max_connections=200,  # Aumentado
-            max_keepalive_connections=50  # Aumentado
-        ),
-        http2=True  # Ativa HTTP/2 para melhor performance
-    )
+    async def process_term(session, term, search_results):
+        async with extract_semaphore:
+            for result in search_results:
+                url = result["url"]
+                age = result["age"]
+                
+                if url in used_urls:
+                    continue
+                    
+                text = await extract_article_text(url, session)
+                if text:
+                    used_urls.add(url)
+                    return {
+                        "term": term,
+                        "age": age,
+                        "url": url,
+                        "text": text
+                    }
+            return None
+
+    connector = aiohttp.TCPConnector(limit=100, limit_per_host=15)
+    timeout = aiohttp.ClientTimeout(total=15)
     
-    try:
+    async with httpx.AsyncClient(
+        timeout=15.0, 
+        limits=httpx.Limits(max_connections=100, max_keepalive_connections=25)
+    ) as http_client:
         async with aiohttp.ClientSession(connector=connector, timeout=timeout) as session:
-            # Fase 1: Busca todos os termos em paralelo
-            search_results = await search_brave_batch(http_client, terms)
-            print(f"Buscas concluídas em {time.time() - start_time:.2f}s. Iniciando extrações...")
-            
-            # Fase 2: Prepara dados para extração em lote
-            urls_data = []
-            for term, results in search_results:
-                for result in results:
-                    urls_data.append((term, result["url"], result["age"]))
             
-            print(f"Processando {len(urls_data)} URLs...")
+            search_tasks = [search_with_limit(http_client, term) for term in terms]
+            search_results = await asyncio.gather(*search_tasks, return_exceptions=True)
             
-            # Fase 3: Processa todas as URLs em paralelo
-            final_results = await process_urls_batch(session, urls_data)
+            process_tasks = []
+            for term, results in zip(terms, search_results):
+                if isinstance(results, list) and results:
+                    process_tasks.append(process_term(session, term, results))
             
-            print(f"Extração concluída em {time.time() - start_time:.2f}s. Salvando arquivo...")
-    
-    finally:
-        await http_client.aclose()
+            if process_tasks:
+                processed_results = await asyncio.gather(*process_tasks, return_exceptions=True)
+                final_results = [r for r in processed_results if r is not None and not isinstance(r, Exception)]
+            else:
+                final_results = []
 
-    # Fase 4: Cria arquivo temporário assíncrono
+    # Cria o JSON final
     result_data = {"results": final_results}
-    temp_file_info = await create_temp_file(result_data)
     
-    total_time = time.time() - start_time
-    print(f"Processo completo em {total_time:.2f}s")
+    # Cria arquivo temporário
+    temp_file_info = create_temp_file(result_data)
     
     return {
         "message": "Dados salvos em arquivo temporário",
         "total_results": len(final_results),
         "context": context,
         "generated_terms": terms,
-        "file_info": temp_file_info,
-        "processing_time": f"{total_time:.2f}s"
+        "file_info": temp_file_info
     }
 
+
 @router.get("/download-temp/{file_id}")
 async def download_temp_file(file_id: str):
     """Endpoint para download do arquivo temporário"""