Update routers/searchterm.py

routers/searchterm.py

@@ -10,13 +10,17 @@ import uuid
 import time
 from pathlib import Path
 from urllib.parse import urlparse
-from typing import List, Dict, Any, Optional
+from typing import List, Dict, Any, Optional, Set, Tuple
 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()
 
@@ -41,10 +45,16 @@ 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 = {"reddit.com", "www.reddit.com", "old.reddit.com",
-                   "quora.com", "www.quora.com"}
+BLOCKED_DOMAINS = frozenset({  # frozenset é mais rápido para lookup
+    "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")
@@ -53,22 +63,30 @@ 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()
-        return any(host == b or host.endswith("." + b) for b in BLOCKED_DOMAINS)
+        
+        # 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
     except Exception:
         return False
 
-
 def clamp_text(text: str) -> str:
-    if not text:
-        return ""
-    if len(text) > MAX_TEXT_LENGTH:
-        return text[:MAX_TEXT_LENGTH]
-    return text
-
+    if not text or len(text) <= MAX_TEXT_LENGTH:
+        return text
+    return text[:MAX_TEXT_LENGTH]
 
 def get_realistic_headers() -> Dict[str, str]:
     return {
@@ -76,9 +94,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:
@@ -89,17 +107,16 @@ def delete_temp_file(file_id: str, file_path: Path):
     except Exception as e:
         print(f"Erro ao remover arquivo temporário: {e}")
 
-
-def create_temp_file(data: Dict[str, Any]) -> Dict[str, str]:
-    """Cria arquivo temporário e agenda sua remoção"""
+async def create_temp_file(data: Dict[str, Any]) -> Dict[str, str]:
+    """Cria arquivo temporário assíncrono e agenda sua remoção"""
     file_id = str(uuid.uuid4())
     file_path = TEMP_DIR / f"fontes_{file_id}.txt"
     
-    # Salva o JSON no arquivo
-    with open(file_path, 'w', encoding='utf-8') as f:
-        json.dump(data, f, ensure_ascii=False, indent=2)
+    # 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))
     
-    # Agenda remoção em 24 horas (86400 segundos)
+    # Agenda remoção em 24 horas
     timer = Timer(86400, delete_temp_file, args=[file_id, file_path])
     timer.start()
     
@@ -116,7 +133,6 @@ 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:
@@ -163,9 +179,7 @@ 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
@@ -180,7 +194,6 @@ 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:]
@@ -188,92 +201,163 @@ Retorne apenas o JSON, sem mais nenhum texto."""
                 clean_response = clean_response[:-3]
             clean_response = clean_response.strip()
             
-            # Parse do JSON
-            response_data = json.loads(clean_response)
+            response_data = ujson.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]  # Garante máximo de 20 termos
+            return terms[:20]
             
-        except (json.JSONDecodeError, ValueError) as e:
+        except (ujson.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_term(client: httpx.AsyncClient, term: str) -> List[Dict[str, str]]:
-    params = {"q": term, "count": 10, "safesearch": "off", "summary": "false"}
+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)
     
-    try:
-        resp = await client.get(BRAVE_SEARCH_URL, headers=BRAVE_HEADERS, params=params)
-        if resp.status_code != 200:
-            return []
+    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, [])
 
-        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")
+                data = resp.json()
+                results = []
                 
-                if url and not is_blocked_domain(url):
-                    results.append({"url": url, "age": age})
+                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)
 
-        return results
+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 ""
     except Exception:
-        return []
-
+        return ""
 
-async def extract_article_text(url: str, session: aiohttp.ClientSession) -> str:
+def extract_with_newspaper(url: str) -> str:
+    """Extração com newspaper executada em thread pool"""
     try:
         art = Article(url)
         art.config.browser_user_agent = random.choice(USER_AGENTS)
-        art.config.request_timeout = 8
+        art.config.request_timeout = 6
         art.config.number_threads = 1
-
         art.download()
         art.parse()
-        txt = (art.text or "").strip()
-        if txt and len(txt) > 100:
-            return clamp_text(txt)
+        return (art.text or "").strip()
     except Exception:
-        pass
+        return ""
 
+async def extract_article_text_optimized(url: str, session: aiohttp.ClientSession) -> str:
+    """Extração de artigo otimizada com paralelização de métodos"""
+    
+    # Tentativa 1: Newspaper em thread pool (paralelo com download HTTP)
+    newspaper_task = asyncio.create_task(
+        asyncio.get_event_loop().run_in_executor(thread_pool, extract_with_newspaper, url)
+    )
+    
+    # Tentativa 2: Download HTTP e trafilatura
     try:
-        await asyncio.sleep(random.uniform(0.1, 0.3))
-        
         headers = get_realistic_headers()
-        async with session.get(url, headers=headers, timeout=12) as resp:
+        async with session.get(url, headers=headers, timeout=EXTRACTION_TIMEOUT) as resp:
             if resp.status != 200:
-                return ""
+                # Aguarda newspaper se HTTP falhou
+                newspaper_result = await newspaper_task
+                return clamp_text(newspaper_result) if newspaper_result and len(newspaper_result) > 100 else ""
                 
             html = await resp.text()
             
-            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)
+            # Verifica paywall rapidamente
+            if re.search(r"(paywall|subscribe|metered|registration|captcha|access denied)", 
+                        html[:2000], re.I):  # Verifica apenas o início
+                newspaper_result = await newspaper_task
+                return clamp_text(newspaper_result) if newspaper_result and len(newspaper_result) > 100 else ""
+
+            # Extração com trafilatura em thread pool
+            trafilatura_task = asyncio.create_task(
+                asyncio.get_event_loop().run_in_executor(thread_pool, extract_with_trafilatura, html)
+            )
+            
+            # Aguarda ambos os métodos e pega o melhor resultado
+            newspaper_result, trafilatura_result = await gather(newspaper_task, trafilatura_task)
+            
+            # Escolhe o melhor resultado
+            best_result = ""
+            if trafilatura_result and len(trafilatura_result) > 100:
+                best_result = trafilatura_result
+            elif newspaper_result and len(newspaper_result) > 100:
+                best_result = newspaper_result
+                
+            return clamp_text(best_result) if best_result else ""
                 
     except Exception:
-        pass
-
-    return ""
-
+        # Se tudo falhar, tenta pelo menos o newspaper
+        try:
+            newspaper_result = await newspaper_task
+            return clamp_text(newspaper_result) if newspaper_result and len(newspaper_result) > 100 else ""
+        except Exception:
+            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"""
+    semaphore = asyncio.Semaphore(MAX_CONCURRENT_EXTRACTIONS)
+    results = []
+    used_urls: Set[str] = set()
+    
+    async def process_single_url(term: str, url: str, age: str) -> Optional[Dict[str, Any]]:
+        async with semaphore:
+            if url in used_urls:
+                return None
+                
+            text = await extract_article_text_optimized(url, session)
+            if text:
+                used_urls.add(url)
+                return {
+                    "term": term,
+                    "age": age,
+                    "url": url,
+                    "text": text
+                }
+            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))
+    
+    # Processa tudo em paralelo
+    processed_results = await gather(*tasks, return_exceptions=True)
+    
+    # Filtra resultados válidos
+    return [r for r in processed_results if r is not None and not isinstance(r, Exception)]
 
 @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.")
@@ -281,78 +365,73 @@ 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.")
 
-    used_urls = set()
-    search_semaphore = asyncio.Semaphore(20)
-    extract_semaphore = asyncio.Semaphore(50)
+    print(f"Termos gerados em {time.time() - start_time:.2f}s. Iniciando buscas...")
     
-    async def search_with_limit(client, term):
-        async with search_semaphore:
-            return await search_brave_term(client, term)
+    # 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 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)
+    # 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
+    )
     
-    async with httpx.AsyncClient(
-        timeout=15.0, 
-        limits=httpx.Limits(max_connections=100, max_keepalive_connections=25)
-    ) as http_client:
+    try:
         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"]))
             
-            search_tasks = [search_with_limit(http_client, term) for term in terms]
-            search_results = await asyncio.gather(*search_tasks, return_exceptions=True)
+            print(f"Processando {len(urls_data)} URLs...")
             
-            process_tasks = []
-            for term, results in zip(terms, search_results):
-                if isinstance(results, list) and results:
-                    process_tasks.append(process_term(session, term, results))
+            # Fase 3: Processa todas as URLs em paralelo
+            final_results = await process_urls_batch(session, urls_data)
             
-            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 = []
+            print(f"Extração concluída em {time.time() - start_time:.2f}s. Salvando arquivo...")
+    
+    finally:
+        await http_client.aclose()
 
-    # Cria o JSON final
+    # Fase 4: Cria arquivo temporário assíncrono
     result_data = {"results": final_results}
+    temp_file_info = await create_temp_file(result_data)
     
-    # Cria arquivo temporário
-    temp_file_info = create_temp_file(result_data)
+    total_time = time.time() - start_time
+    print(f"Processo completo em {total_time:.2f}s")
     
     return {
         "message": "Dados salvos em arquivo temporário",
         "total_results": len(final_results),
         "context": context,
         "generated_terms": terms,
-        "file_info": temp_file_info
+        "file_info": temp_file_info,
+        "processing_time": f"{total_time:.2f}s"
     }
 
-
 @router.get("/download-temp/{file_id}")
 async def download_temp_file(file_id: str):
     """Endpoint para download do arquivo temporário"""