/ middleware_yacy_tensaflow.py
middleware_yacy_tensaflow.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 el contenido principal
 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 para extraer el contenido principal.")
 41  
 42  def obtener_contenido(url: str) -> Optional[str]:
 43      """
 44      Descarga el contenido HTML de una URL utilizando la sesión configurada.
 45      """
 46      try:
 47          logging.info("Obteniendo contenido de %s", url)
 48          respuesta = session.get(url, timeout=TIMEOUT)
 49          respuesta.raise_for_status()
 50          return respuesta.text
 51      except requests.RequestException as e:
 52          logging.error("Error al obtener contenido de %s: %s", url, e)
 53          return None
 54  
 55  def limpiar_html(html: str) -> str:
 56      """
 57      Limpia el HTML eliminando etiquetas, scripts, estilos y comentarios, y normaliza los espacios.
 58      """
 59      soup = BeautifulSoup(html, "html.parser")
 60      for tag in soup(["script", "style", "iframe", "form", "header", "footer", "nav", "aside"]):
 61          tag.decompose()
 62      for comentario in soup.find_all(string=lambda text: isinstance(text, Comment)):
 63          comentario.extract()
 64      for tag in soup.find_all(['br', 'hr', 'input']):
 65          tag.extract()
 66      texto = soup.get_text(separator=' ', strip=True)
 67      return re.sub(r'\s+', ' ', texto).strip()
 68  
 69  def extraer_metadatos(soup: BeautifulSoup, html_raw: Optional[str] = None) -> Dict[str, str]:
 70      """
 71      Extrae el título y la descripción de la página.
 72      Se intenta obtener el título desde la etiqueta <title> o meta og:title.
 73      Para la descripción se usan, en orden de prioridad, la meta description, og:description,
 74      el primer párrafo de <main> o <article>, Readability (si está disponible) y, como último recurso,
 75      se utiliza el contenido limpio.
 76      La descripción se trunca a 160 caracteres y se añade "..." si se excede.
 77      """
 78      # Extracción del título
 79      titulo = ""
 80      if soup.title and soup.title.string:
 81          titulo = soup.title.string.strip()
 82      else:
 83          og_title = soup.find("meta", property="og:title")
 84          if og_title and og_title.get("content"):
 85              titulo = og_title["content"].strip()
 86      
 87      descripcion = ""
 88      # Fuente 1: meta description
 89      meta_desc = soup.find("meta", attrs={"name": "description"})
 90      if meta_desc and meta_desc.get("content"):
 91          temp_desc = meta_desc["content"].strip()
 92          if "cookie" not in temp_desc.lower():
 93              descripcion = temp_desc
 94      # Fuente 2: og:description
 95      if not descripcion:
 96          og_desc = soup.find("meta", property="og:description")
 97          if og_desc and og_desc.get("content"):
 98              temp_desc = og_desc["content"].strip()
 99              if "cookie" not in temp_desc.lower():
100                  descripcion = temp_desc
101      # Fuente 3: Primer párrafo en <main>
102      if not descripcion:
103          main = soup.find("main")
104          if main:
105              p = main.find("p")
106              if p and p.get_text(strip=True):
107                  descripcion = p.get_text(strip=True)
108      # Fuente 4: Primer párrafo en <article>
109      if not descripcion:
110          article = soup.find("article")
111          if article:
112              p = article.find("p")
113              if p and p.get_text(strip=True):
114                  descripcion = p.get_text(strip=True)
115      # Fuente 5: Readability
116      if not descripcion and USE_READABILITY and html_raw:
117          try:
118              doc = Document(html_raw)
119              contenido_principal = BeautifulSoup(doc.summary(), "html.parser").get_text(separator=' ', strip=True)
120              if contenido_principal and len(contenido_principal) > 50:
121                  descripcion = contenido_principal[:160]
122          except Exception as e:
123              logging.warning("Error al usar Readability: %s", e)
124      # Fuente 6: Contenido limpio
125      if not descripcion and html_raw:
126          cleaned_text = limpiar_html(html_raw)
127          descripcion = cleaned_text[:160] + "..."
128      if not descripcion:
129          descripcion = " ".join(soup.stripped_strings)[:160] + "..."
130      
131      # Truncar la descripción a 160 caracteres, agregando "..." al final si se trunca.
132      max_desc_len = 160
133      if descripcion and len(descripcion) > max_desc_len:
134          descripcion = descripcion[:max_desc_len-3].rstrip() + "..."
135      
136      # Extracción de palabras clave (simple)
137      palabras_clave = ""
138      meta_keywords = soup.find("meta", attrs={"name": "keywords"})
139      if meta_keywords and meta_keywords.get("content"):
140          palabras_clave = meta_keywords["content"].strip()
141      
142      return {
143          "titulo": titulo,
144          "descripcion": descripcion,
145          "palabras_clave": palabras_clave
146      }
147  
148  def filtrar_enlaces_interesantes(soup: BeautifulSoup, base_url: str) -> List[str]:
149      """
150      Devuelve una lista de enlaces interesantes basados en palabras clave en el texto del enlace.
151      """
152      palabras_clave = ['blog', 'feed', 'publicación', 'producto', 'tienda', 'noticia']
153      enlaces_interesantes = set()
154      for link in soup.find_all('a', href=True):
155          url_completa = urljoin(base_url, link['href'])
156          texto_link = link.get_text().strip().lower()
157          if any(palabra in texto_link for palabra in palabras_clave):
158              enlaces_interesantes.add(url_completa)
159      return list(enlaces_interesantes)
160  
161  def resumir_contenido(contenido: str, max_palabras: int = 200) -> str:
162      """
163      Resume el contenido limitando el número de palabras.
164      """
165      palabras = contenido.split()
166      if len(palabras) > max_palabras:
167          return ' '.join(palabras[:max_palabras]) + "..."
168      return contenido
169  
170  def eliminar_contenido_repetido(contenido: str) -> str:
171      """
172      Elimina oraciones repetidas, manteniendo el orden original.
173      """
174      oraciones = re.split(r'(?<=[.!?])\s+', contenido)
175      oraciones_unicas = []
176      vistas = set()
177      for oracion in oraciones:
178          oracion_limpia = oracion.strip()
179          if oracion_limpia and oracion_limpia not in vistas:
180              oraciones_unicas.append(oracion_limpia)
181              vistas.add(oracion_limpia)
182      return ' '.join(oraciones_unicas)
183  
184  def procesar_pagina(url: str) -> Dict[str, Any]:
185      """
186      Descarga el contenido de una página, extrae el título, la descripción (usando solo heurísticas)
187      y filtra enlaces interesantes.
188      """
189      html_raw = obtener_contenido(url)
190      if html_raw is None:
191          return {"error": "No se pudo obtener el contenido de la URL"}
192      
193      soup = BeautifulSoup(html_raw, "html.parser")
194      metadatos = extraer_metadatos(soup, html_raw)
195      contenido_limpio = limpiar_html(html_raw)
196      contenido_sin_repetir = eliminar_contenido_repetido(contenido_limpio)
197      # Se genera un resumen interno (no se muestra)
198      contenido_resumido = resumir_contenido(contenido_sin_repetir, max_palabras=200)
199      enlaces_interesantes = filtrar_enlaces_interesantes(soup, url)
200      
201      return {
202          "url": url,
203          "titulo": metadatos.get('titulo', ''),
204          "descripcion": metadatos.get('descripcion', ''),
205          "contenido": contenido_resumido,
206          "enlaces_interesantes": enlaces_interesantes,
207          "palabras_clave": metadatos.get("palabras_clave", "")
208      }
209  
210  def formatear_resultado_yacy(resultado: Dict[str, Any]) -> Dict[str, Any]:
211      """
212      Formatea el resultado para que tenga las claves estándar utilizadas por YaCy.
213      """
214      return {
215          "url": resultado.get("url", ""),
216          "title": resultado.get("titulo", ""),
217          "description": resultado.get("descripcion", ""),
218          "links": resultado.get("enlaces_interesantes", []),
219          "keywords": resultado.get("palabras_clave", "")
220      }
221  
222  if __name__ == "__main__":
223      import argparse
224      parser = argparse.ArgumentParser(description="Script para indexar en YaCy sin LLM, usando heurísticas para describir el enlace.")
225      parser.add_argument("url", help="URL de la página a procesar")
226      parser.add_argument("--json", action="store_true", help="Salida en formato JSON para YaCy")
227      args = parser.parse_args()
228      
229      resultado = procesar_pagina(args.url)
230      
231      if "error" in resultado:
232          logging.error(resultado["error"])
233      else:
234          if args.json:
235              resultado_yacy = formatear_resultado_yacy(resultado)
236              print(json.dumps(resultado_yacy, indent=2, ensure_ascii=False))
237          else:
238              print("----- Resultado del Análisis -----")
239              print(f"Título: {resultado['titulo']}")
240              print(f"Descripción: {resultado['descripcion']}")
241              print(f"Enlaces Interesantes: {resultado['enlaces_interesantes']}")
242