""" WebSearch — Módulo para busca de notícias (WebScraping) e pesquisa geral (API Placeholder). - Angola News: Fontes fixas (Angop, Novo Jornal, Jornal de Angola, etc.) - Busca Geral: Placeholder para integração de API externa (ex: Google Search API, Serper API) - Cache: 15 minutos (900 segundos) """ import time import re import requests from typing import List, Dict, Any from loguru import logger from bs4 import BeautifulSoup import os # Importa o config para possível uso futuro de chaves de API try: # Assumindo que o config está em modules/config.py import modules.config as config except ImportError: # Fallback se config.py não estiver disponível class ConfigMock: pass config = ConfigMock() # Configuração do logger para este módulo logger.add("web_search.log", rotation="10 MB", level="INFO") class SimpleCache: """Cache simples em memória com Time-To-Live (TTL).""" def __init__(self, ttl: int = 900): # 15 min self.ttl = ttl self._data: Dict[str, Any] = {} def get(self, key: str): if key in self._data: value, timestamp = self._data[key] if time.time() - timestamp < self.ttl: return value del self._data[key] return None def set(self, key: str, value: Any): self._data[key] = (value, time.time()) class WebSearch: """Gerenciador de buscas para notícias de Angola e pesquisa geral.""" def __init__(self): self.cache = SimpleCache(ttl=900) self.session = requests.Session() # Header para simular um navegador real e evitar bloqueios de scraping self.session.headers.update({ "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36", "Accept-Language": "pt-BR,pt;q=0.9,en-US;q=0.8,en;q=0.7" }) # Fontes de notícias de Angola (Web Scraping) self.fontes_angola = [ "https://www.angop.ao/ultimas", "https://www.novojornal.co.ao/", "https://www.jornaldeangola.ao/", "https://www.verangola.net/va/noticias" ] def _limpar_texto(self, texto: str) -> str: """Limpa e formata o texto para o LLM.""" if not texto: return "" # Remove espaços múltiplos, quebras de linha e caracteres de formatação texto = re.sub(r'[\s\n\t]+', ' ', texto) # Limita o tamanho para o contexto do LLM return texto.strip()[:200] # --- FUNÇÃO PRINCIPAL DE BUSCA GERAL (PLACEHOLDER) --- def buscar_geral(self, query: str) -> str: """ Retorna resultados de pesquisa na web para cultura geral. ATENÇÃO: Esta função é um PLACEHOLDER. Para funcionar, você DEVE integrar uma API de busca externa paga (ex: Serper, Google Search API, ou outra) para substituir o bloco de fallback. """ cache_key = f"busca_geral_{query.lower()}" cached = self.cache.get(cache_key) if cached: return cached logger.warning(f"PLACEHOLDER: Executando busca geral para '{query}'. É necessária integração de API externa.") # O BLOCO ABAIXO DEVE SER SUBSTITUÍDO PELA CHAMADA REAL DA API DE BUSCA # --- COMEÇO DO PLACEHOLDER --- fallback_response = "Sem informações de cultura geral disponíveis. Para ativar a pesquisa em tempo real, configure e integre uma API de busca (como Serper ou Google Search API) na função 'buscar_geral' do web_search.py." # --- FIM DO PLACEHOLDER --- self.cache.set(cache_key, fallback_response) return fallback_response # --- IMPLEMENTAÇÃO DE BUSCA DE NOTÍCIAS DE ANGOLA (WEB SCRAPING) --- def _buscar_angop(self) -> List[Dict]: """Extrai notícias da Angop.""" try: r = self.session.get(self.fontes_angola[0], timeout=8) if r.status_code != 200: return [] soup = BeautifulSoup(r.text, 'html.parser') itens = soup.select('.ultimas-noticias .item')[:3] noticias = [] for item in itens: titulo = item.select_one('h3 a') link = item.select_one('a') if titulo and link: noticias.append({ "titulo": self._limpar_texto(titulo.get_text()), "link": "https://www.angop.ao" + link.get('href', '') if link.get('href', '').startswith('/') else link.get('href', '') }) return noticias except Exception as e: logger.warning(f"Angop falhou: {e}") return [] def _buscar_novojornal(self) -> List[Dict]: """Extrai notícias do Novo Jornal.""" try: r = self.session.get(self.fontes_angola[1], timeout=8) if r.status_code != 200: return [] soup = BeautifulSoup(r.text, 'html.parser') itens = soup.select('.noticia-lista .titulo')[:3] noticias = [] for item in itens: a = item.find('a') if a: noticias.append({ "titulo": self._limpar_texto(a.get_text()), "link": a.get('href', '') }) return noticias except Exception as e: logger.warning(f"Novo Jornal falhou: {e}") return [] def _buscar_jornaldeangola(self) -> List[Dict]: """Extrai notícias do Jornal de Angola.""" try: r = self.session.get(self.fontes_angola[2], timeout=8) if r.status_code != 200: return [] soup = BeautifulSoup(r.text, 'html.parser') itens = soup.select('.ultimas .titulo a')[:3] noticias = [] for a in itens: noticias.append({ "titulo": self._limpar_texto(a.get_text()), "link": a.get('href', '') }) return noticias except Exception as e: logger.warning(f"Jornal de Angola falhou: {e}") return [] def _buscar_verangola(self) -> List[Dict]: """Extrai notícias do VerAngola.""" try: r = self.session.get(self.fontes_angola[3], timeout=8) if r.status_code != 200: return [] soup = BeautifulSoup(r.text, 'html.parser') # Seletores podem mudar, mas .noticia-item geralmente é um bom ponto de partida itens = soup.select('.noticia-item')[:3] noticias = [] for item in itens: titulo = item.select_one('h3 a') if titulo: link = titulo.get('href', '') noticias.append({ "titulo": self._limpar_texto(titulo.get_text()), "link": link if link.startswith('http') else "https://www.verangola.net" + link }) return noticias except Exception as e: logger.warning(f"VerAngola falhou: {e}") return [] def pesquisar_noticias_angola(self) -> str: """ Retorna as notícias mais recentes de Angola através de Web Scraping. Esta é a função usada no api.py quando detecta intenção de notícias. """ cache_key = "noticias_angola" cached = self.cache.get(cache_key) if cached: return cached todas = [] try: todas.extend(self._buscar_angop()) todas.extend(self._buscar_novojornal()) todas.extend(self._buscar_jornaldeangola()) todas.extend(self._buscar_verangola()) except Exception as e: logger.error(f"Erro no pipeline de scraping: {e}") # Filtra e remove duplicatas vistos = set() unicas = [] for n in todas: t = n["titulo"].lower() if t not in vistos and len(t) > 20: vistos.add(t) unicas.append(n) if len(unicas) >= 5: break if not unicas: fallback = "Sem notícias recentes de Angola disponíveis no momento." self.cache.set(cache_key, fallback) return fallback # Formata a resposta para injeção no prompt do LLM texto = "NOTÍCIAS RECENTES DE ANGOLA (CONTEXTO):\n" for i, n in enumerate(unicas, 1): # Apenas o título é relevante para o contexto do LLM texto += f"[{i}] {n['titulo']}\n" self.cache.set(cache_key, texto.strip()) return texto.strip()