/ middleware_yacy_gpt_openai.py
middleware_yacy_gpt_openai.py
  1  import os
  2  import re
  3  import time
  4  import json
  5  import logging
  6  import requests
  7  from bs4 import BeautifulSoup, Comment
  8  from typing import Optional, Dict, List, Any
  9  from urllib.parse import urljoin
 10  from requests.adapters import HTTPAdapter
 11  from requests.packages.urllib3.util.retry import Retry
 12  
 13  # Configuración de logging
 14  logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
 15  
 16  # Configuración de headers y timeout
 17  HEADERS = {
 18      "User-Agent": ("Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
 19                     "AppleWebKit/537.36 (KHTML, like Gecko) "
 20                     "Chrome/90.0.4430.85 Safari/537.36")
 21  }
 22  TIMEOUT = 10
 23  MAX_RETRIES = 3
 24  
 25  # Configuración de la sesión con reintentos
 26  session = requests.Session()
 27  retries = Retry(total=MAX_RETRIES, backoff_factor=1, status_forcelist=[429, 500, 502, 503, 504])
 28  adapter = HTTPAdapter(max_retries=retries)
 29  session.mount("http://", adapter)
 30  session.mount("https://", adapter)
 31  session.headers.update(HEADERS)
 32  
 33  # Intentar usar Readability para extraer contenido relevante
 34  try:
 35      from readability import Document
 36      USE_READABILITY = True
 37      logging.info("Readability detectado. Se usará para extraer el contenido principal.")
 38  except ImportError:
 39      USE_READABILITY = False
 40      logging.info("Readability no instalado. Se utilizarán métodos alternativos.")
 41  
 42  # Configuración del LLM (OpenAI) usando la nueva interfaz de chat.
 43  USE_LLM = False
 44  try:
 45      from openai import OpenAI
 46      # Se recomienda usar variables de entorno para la API key.
 47      llm_client = OpenAI(api_key="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")
 48      USE_LLM = True
 49      logging.info("LLM activado. Se usará OpenAI para describir el contenido del enlace.")
 50  except Exception as e:
 51      logging.warning("No se pudo configurar OpenAI: %s", e)
 52  
 53  def is_valid_description(desc: str) -> bool:
 54      """
 55      Considera válida una descripción si tiene al menos 50 caracteres y no contiene avisos (ej. 'cookie').
 56      """
 57      return desc and len(desc) >= 50 and "cookie" not in desc.lower()
 58  
 59  def generar_descripcion_llm(texto: str) -> str:
 60      """
 61      Solicita al LLM que, basándose en el contenido de la página, genere una descripción breve (140-160 caracteres)
 62      que informe de forma clara y cercana qué encontrará el usuario al visitar el enlace.
 63      
 64      Args:
 65          texto (str): Fragmento representativo del contenido de la página.
 66          
 67      Returns:
 68          str: Descripción generada por el LLM.
 69      """
 70      prompt = (
 71          "Lee el siguiente contenido extraído de una página web y, basándote en él, "
 72          "genera una descripción de entre 140 y 160 caracteres en la que se indique de forma "
 73          "formal pero cercana qué encontrará el usuario al visitar esta página (por ejemplo, "
 74          "si es un portal de noticias, un banco de imágenes, una plataforma de streaming, repositorios, etc.).\n\n"
 75          f"{texto}\n\nDescripción:"
 76      )
 77      try:
 78          response = llm_client.chat.completions.create(
 79              model="gpt-3.5-turbo",
 80              messages=[
 81                  {"role": "system", "content": "Eres un experto en analizar y describir contenido web."},
 82                  {"role": "user", "content": prompt}
 83              ],
 84              temperature=0.7,
 85              max_tokens=120
 86          )
 87          descripcion_llm = response.choices[0].message.content.strip()
 88          logging.info("LLM generó una descripción de longitud %d", len(descripcion_llm))
 89          return descripcion_llm
 90      except Exception as e:
 91          logging.error("Error al generar descripción con LLM: %s", e)
 92          return ""
 93  
 94  def obtener_contenido(url: str) -> Optional[str]:
 95      """
 96      Descarga el contenido HTML de una URL.
 97      """
 98      try:
 99          logging.info("Obteniendo contenido de %s", url)
100          respuesta = session.get(url, timeout=TIMEOUT)
101          respuesta.raise_for_status()
102          return respuesta.text
103      except requests.RequestException as e:
104          logging.error("Error al obtener contenido de %s: %s", url, e)
105          return None
106  
107  def limpiar_html(html: str) -> str:
108      """
109      Elimina etiquetas, scripts, estilos y comentarios para obtener el texto limpio.
110      """
111      soup = BeautifulSoup(html, "html.parser")
112      for tag in soup(["script", "style", "iframe", "form", "header", "footer", "nav", "aside"]):
113          tag.decompose()
114      for comentario in soup.find_all(string=lambda text: isinstance(text, Comment)):
115          comentario.extract()
116      for tag in soup.find_all(['br', 'hr', 'input']):
117          tag.extract()
118      texto = soup.get_text(separator=' ', strip=True)
119      return re.sub(r'\s+', ' ', texto).strip()
120  
121  def extraer_metadatos(soup: BeautifulSoup, html_raw: Optional[str] = None) -> Dict[str, str]:
122      """
123      Extrae el título y utiliza al LLM para generar una descripción basada en lo que encontrará el usuario
124      al visitar la página.
125      """
126      # Título: se obtiene del <title> o de meta og:title
127      titulo = ""
128      if soup.title and soup.title.string:
129          titulo = soup.title.string.strip()
130      else:
131          og_title = soup.find("meta", property="og:title")
132          if og_title and og_title.get("content"):
133              titulo = og_title["content"].strip()
134      
135      descripcion = ""
136      # Usamos el LLM para generar la descripción a partir del contenido limpio
137      if USE_LLM and html_raw:
138          contenido_para_llm = limpiar_html(html_raw)[:2000]
139          descripcion_llm = generar_descripcion_llm(contenido_para_llm)
140          if is_valid_description(descripcion_llm):
141              descripcion = descripcion_llm
142      if not descripcion:
143          descripcion = "No se pudo generar una descripción."
144      
145      # Limitar la descripción a 160 caracteres
146      if descripcion and len(descripcion) > 160:
147          descripcion = descripcion[:160]
148      
149      # Extraer palabras clave de forma sencilla
150      palabras_clave = ""
151      meta_keywords = soup.find("meta", attrs={"name": "keywords"})
152      if meta_keywords and meta_keywords.get("content"):
153          palabras_clave = meta_keywords["content"].strip()
154      
155      return {
156          "titulo": titulo,
157          "descripcion": descripcion,
158          "palabras_clave": palabras_clave
159      }
160  
161  def filtrar_enlaces_interesantes(soup: BeautifulSoup, base_url: str) -> List[str]:
162      """
163      Devuelve una lista de enlaces interesantes basados en palabras clave en el texto del enlace.
164      """
165      palabras_clave = ['blog', 'feed', 'publicación', 'producto', 'tienda', 'noticia']
166      enlaces_interesantes = set()
167      for link in soup.find_all('a', href=True):
168          url_completa = urljoin(base_url, link['href'])
169          texto_link = link.get_text().strip().lower()
170          if any(palabra in texto_link for palabra in palabras_clave):
171              enlaces_interesantes.add(url_completa)
172      return list(enlaces_interesantes)
173  
174  def resumir_contenido(contenido: str, max_palabras: int = 200) -> str:
175      """
176      Resume el contenido limitando el número de palabras.
177      """
178      palabras = contenido.split()
179      if len(palabras) > max_palabras:
180          return ' '.join(palabras[:max_palabras]) + "..."
181      return contenido
182  
183  def eliminar_contenido_repetido(contenido: str) -> str:
184      """
185      Elimina oraciones repetidas, manteniendo el orden.
186      """
187      oraciones = re.split(r'(?<=[.!?])\s+', contenido)
188      oraciones_unicas = []
189      vistas = set()
190      for oracion in oraciones:
191          oracion_limpia = oracion.strip()
192          if oracion_limpia and oracion_limpia not in vistas:
193              oraciones_unicas.append(oracion_limpia)
194              vistas.add(oracion_limpia)
195      return ' '.join(oraciones_unicas)
196  
197  def procesar_pagina(url: str) -> Dict[str, Any]:
198      """
199      Descarga el contenido de una página, extrae el título, genera la descripción mediante el LLM
200      y filtra enlaces interesantes.
201      """
202      html_raw = obtener_contenido(url)
203      if html_raw is None:
204          return {"error": "No se pudo obtener el contenido de la URL"}
205      soup = BeautifulSoup(html_raw, "html.parser")
206      metadatos = extraer_metadatos(soup, html_raw)
207      contenido_limpio = limpiar_html(html_raw)
208      contenido_sin_repetir = eliminar_contenido_repetido(contenido_limpio)
209      # Se mantiene el resumen de contenido para fines internos (no se mostrará)
210      contenido_resumido = resumir_contenido(contenido_sin_repetir, max_palabras=200)
211      enlaces_interesantes = filtrar_enlaces_interesantes(soup, url)
212      
213      return {
214          "url": url,
215          "titulo": metadatos.get('titulo', ''),
216          "descripcion": metadatos.get('descripcion', ''),
217          "contenido": contenido_resumido,
218          "enlaces_interesantes": enlaces_interesantes,
219          "palabras_clave": metadatos.get("palabras_clave", "")
220      }
221  
222  def formatear_resultado_yacy(resultado: Dict[str, Any]) -> Dict[str, Any]:
223      """
224      Formatea el resultado para que tenga las claves estándar utilizadas por YaCy.
225      """
226      return {
227          "url": resultado.get("url", ""),
228          "title": resultado.get("titulo", ""),
229          "description": resultado.get("descripcion", ""),
230          "links": resultado.get("enlaces_interesantes", []),
231          "keywords": resultado.get("palabras_clave", "")
232      }
233  
234  if __name__ == "__main__":
235      import argparse
236      parser = argparse.ArgumentParser(description="Script para indexar en YaCy usando LLM para describir el enlace.")
237      parser.add_argument("url", help="URL de la página a procesar")
238      parser.add_argument("--json", action="store_true", help="Salida en formato JSON para YaCy")
239      args = parser.parse_args()
240      
241      resultado = procesar_pagina(args.url)
242      if "error" in resultado:
243          logging.error(resultado["error"])
244      else:
245          if args.json:
246              resultado_yacy = formatear_resultado_yacy(resultado)
247              print(json.dumps(resultado_yacy, indent=2, ensure_ascii=False))
248          else:
249              print("----- Resultado del Análisis -----")
250              print(f"Título: {resultado['titulo']}")
251              print(f"Descripción: {resultado['descripcion']}")
252              print(f"Enlaces Interesantes: {resultado['enlaces_interesantes']}")
253