seo_analyzer.py

import os
import logging
import re
import requests
import hashlib
import PyPDF2
import numpy as np
import pandas as pd
from io import BytesIO
from typing import List, Dict, Any, Tuple
from urllib.parse import urlparse, urljoin
from concurrent.futures import ThreadPoolExecutor, as_completed
from bs4 import BeautifulSoup
from pathlib import Path
from datetime import datetime
from sklearn.feature_extraction.text import TfidfVectorizer
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
from transformers import pipeline
from sentence_transformers import SentenceTransformer
import torch
import subprocess
import sys
import spacy
import matplotlib.pyplot as plt

from utils import sanitize_filename

# Configuración de logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

class SEOSpaceAnalyzer:
    def __init__(self, max_urls: int = 20, max_workers: int = 4) -> None:
        """
        Inicializa la sesión HTTP, carga modelos NLP y prepara el directorio de almacenamiento.
        
        Args:
            max_urls: Número máximo de URLs a procesar por análisis.
            max_workers: Número de hilos para la ejecución concurrente.
        """
        self.max_urls = max_urls
        self.max_workers = max_workers
        self.session = self._configure_session()
        self.models = self._load_models()
        self.base_dir = Path("content_storage")
        self.base_dir.mkdir(parents=True, exist_ok=True)
        self.current_analysis: Dict[str, Any] = {}

    def _load_models(self) -> Dict[str, Any]:
        """Carga los modelos NLP de Hugging Face y spaCy."""
        try:
            device = 0 if torch.cuda.is_available() else -1
            logger.info("Cargando modelos NLP...")
            models = {
                'summarizer': pipeline("summarization", model="facebook/bart-large-cnn", device=device),
                'ner': pipeline("ner", model="dslim/bert-base-NER", aggregation_strategy="simple", device=device),
                'semantic': SentenceTransformer('paraphrase-multilingual-MiniLM-L12-v2'),
                'spacy': spacy.load("es_core_news_lg")
            }
            logger.info("Modelos cargados correctamente.")
            return models
        except Exception as e:
            logger.error(f"Error cargando modelos: {e}")
            raise

    def _configure_session(self) -> requests.Session:
        """Configura una sesión HTTP con reintentos y headers personalizados."""
        session = requests.Session()
        retry = Retry(
            total=3,
            backoff_factor=1,
            status_forcelist=[500, 502, 503, 504],
            allowed_methods=['GET', 'HEAD']
        )
        adapter = HTTPAdapter(max_retries=retry)
        session.mount('http://', adapter)
        session.mount('https://', adapter)
        session.headers.update({
            'User-Agent': 'Mozilla/5.0 (compatible; SEOBot/1.0)',
            'Accept-Language': 'es-ES,es;q=0.9'
        })
        return session

    def analyze_sitemap(self, sitemap_url: str) -> Tuple[Dict, List[str], Dict, Dict, List[Dict]]:
        """
        Procesa el sitemap: extrae URLs, analiza cada página individualmente y devuelve datos agregados.
        
        Args:
            sitemap_url: URL del sitemap XML.
        
        Returns:
            Una tupla con 5 elementos:
              - Estadísticas generales (dict)
              - Recomendaciones SEO (lista de strings)
              - Análisis de contenido agregado (dict)
              - Análisis de enlaces (dict)
              - Detalle individual de cada URL procesada (lista de dicts)
        """
        try:
            urls = self._parse_sitemap(sitemap_url)
            if not urls:
                return {"error": "No se pudieron extraer URLs del sitemap"}, [], {}, {}, []
            results: List[Dict] = []
            with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
                futures = {executor.submit(self._process_url, url): url for url in urls[:self.max_urls]}
                for future in as_completed(futures):
                    url = futures[future]
                    try:
                        res = future.result()
                        results.append(res)
                        logger.info(f"Procesado: {url}")
                    except Exception as e:
                        logger.error(f"Error procesando {url}: {e}")
                        results.append({'url': url, 'status': 'error', 'error': str(e)})

            self.current_analysis = {
                'stats': self._calculate_stats(results),
                'content_analysis': self._analyze_content(results),
                'links': self._analyze_links(results),
                'recommendations': self._generate_seo_recommendations(results),
                'details': results,
                'timestamp': datetime.now().isoformat()
            }
            analysis = self.current_analysis
            return analysis['stats'], analysis['recommendations'], analysis['content_analysis'], analysis['links'], analysis['details']
        except Exception as e:
            logger.error(f"Error en análisis: {e}")
            return {"error": str(e)}, [], {}, {}, []

    def _process_url(self, url: str) -> Dict:
        """Procesa una URL individual extrayendo contenido, metadatos y enlaces."""
        try:
            response = self.session.get(url, timeout=15)
            response.raise_for_status()
            content_type = response.headers.get('Content-Type', '')
            result: Dict[str, Any] = {'url': url, 'status': 'success'}
            if 'application/pdf' in content_type:
                result.update(self._process_pdf(response.content))
            elif 'text/html' in content_type:
                result.update(self._process_html(response.text, url))
            else:
                result.update({'type': 'unknown', 'content': '', 'word_count': 0})
            self._save_content(url, response.content)
            return result
        except requests.exceptions.RequestException as e:
            logger.warning(f"Error procesando {url}: {str(e)}")
            return {'url': url, 'status': 'error', 'error': str(e)}
        except Exception as e:
            logger.error(f"Error inesperado en {url}: {str(e)}")
            return {'url': url, 'status': 'error', 'error': str(e)}

    def _process_html(self, html: str, base_url: str) -> Dict:
        """Extrae y limpia el contenido HTML, metadatos y enlaces de la página."""
        soup = BeautifulSoup(html, 'html.parser')
        clean_text = self._clean_text(soup.get_text())
        return {
            'type': 'html',
            'content': clean_text,
            'word_count': len(clean_text.split()),
            'metadata': self._extract_metadata(soup),
            'links': self._extract_links(soup, base_url)
        }

    def _process_pdf(self, content: bytes) -> Dict:
        """Extrae texto de un documento PDF y calcula estadísticas básicas."""
        try:
            text = ""
            with BytesIO(content) as pdf_file:
                reader = PyPDF2.PdfReader(pdf_file)
                for page in reader.pages:
                    extracted = page.extract_text()
                    text += extracted if extracted else ""
            clean_text = self._clean_text(text)
            return {
                'type': 'pdf',
                'content': clean_text,
                'word_count': len(clean_text.split()),
                'page_count': len(reader.pages)
            }
        except PyPDF2.PdfReadError as e:
            logger.error(f"Error leyendo PDF: {e}")
            return {'type': 'pdf', 'error': str(e)}

    def _clean_text(self, text: str) -> str:
        """Limpia y normaliza el texto removiendo espacios y caracteres especiales."""
        if not text:
            return ""
        text = re.sub(r'\s+', ' ', text)
        return re.sub(r'[^\w\sáéíóúñÁÉÍÓÚÑ]', ' ', text).strip()

    def _extract_metadata(self, soup: BeautifulSoup) -> Dict:
        """Extrae metadatos relevantes (título, descripción, keywords, Open Graph) de la página."""
        metadata = {'title': '', 'description': '', 'keywords': [], 'og': {}}
        if soup.title and soup.title.string:
            metadata['title'] = soup.title.string.strip()[:200]
        for meta in soup.find_all('meta'):
            name = meta.get('name', '').lower()
            prop = meta.get('property', '').lower()
            content = meta.get('content', '')
            if name == 'description':
                metadata['description'] = content[:300]
            elif name == 'keywords':
                metadata['keywords'] = [kw.strip() for kw in content.split(',') if kw.strip()]
            elif prop.startswith('og:'):
                metadata['og'][prop[3:]] = content
        return metadata

    def _extract_links(self, soup: BeautifulSoup, base_url: str) -> List[Dict]:
        """Extrae enlaces de la página, distinguiendo entre internos y externos."""
        links: List[Dict] = []
        base_netloc = urlparse(base_url).netloc
        for tag in soup.find_all('a', href=True):
            try:
                href = tag['href'].strip()
                if not href or href.startswith('javascript:'):
                    continue
                full_url = urljoin(base_url, href)
                parsed = urlparse(full_url)
                links.append({
                    'url': full_url,
                    'type': 'internal' if parsed.netloc == base_netloc else 'external',
                    'anchor': self._clean_text(tag.get_text())[:100],
                    'file_type': self._get_file_type(parsed.path)
                })
            except Exception as e:
                logger.warning(f"Error procesando enlace {tag.get('href')}: {e}")
                continue
        return links

    def _get_file_type(self, path: str) -> str:
        """Determina el tipo de archivo según la extensión."""
        ext = Path(path).suffix.lower()
        return ext[1:] if ext else 'html'

    def _parse_sitemap(self, sitemap_url: str) -> List[str]:
        """Parsea un sitemap XML (y posibles índices de sitemaps) para extraer URLs."""
        try:
            response = self.session.get(sitemap_url, timeout=10)
            response.raise_for_status()
            if 'xml' not in response.headers.get('Content-Type', ''):
                logger.warning(f"El sitemap no parece ser XML: {sitemap_url}")
                return []
            soup = BeautifulSoup(response.text, 'lxml-xml')
            urls: List[str] = []
            if soup.find('sitemapindex'):
                for sitemap in soup.find_all('loc'):
                    url = sitemap.text.strip()
                    if url.endswith('.xml'):
                        urls.extend(self._parse_sitemap(url))
            else:
                urls = [loc.text.strip() for loc in soup.find_all('loc')]
            filtered_urls = list({url for url in urls if url.startswith('http')})
            return filtered_urls
        except Exception as e:
            logger.error(f"Error al parsear sitemap {sitemap_url}: {e}")
            return []

    def _save_content(self, url: str, content: bytes) -> None:
        """
        Guarda el contenido descargado en una estructura de directorios organizada por dominio,
        sanitizando el nombre del archivo y evitando sobrescribir archivos idénticos mediante hash.
        """
        try:
            parsed = urlparse(url)
            domain_dir = self.base_dir / parsed.netloc
            path = parsed.path.lstrip('/')
            if not path or path.endswith('/'):
                path = os.path.join(path, 'index.html')
            safe_path = sanitize_filename(path)
            save_path = domain_dir / safe_path
            save_path.parent.mkdir(parents=True, exist_ok=True)
            new_hash = hashlib.md5(content).hexdigest()
            if save_path.exists():
                with open(save_path, 'rb') as f:
                    existing_content = f.read()
                existing_hash = hashlib.md5(existing_content).hexdigest()
                if new_hash == existing_hash:
                    logger.debug(f"El contenido de {url} ya está guardado.")
                    return
            with open(save_path, 'wb') as f:
                f.write(content)
            logger.info(f"Guardado contenido en: {save_path}")
        except Exception as e:
            logger.error(f"Error guardando contenido para {url}: {e}")

    def _calculate_stats(self, results: List[Dict]) -> Dict:
        """Calcula estadísticas generales del análisis."""
        successful = [r for r in results if r.get('status') == 'success']
        content_types = [r.get('type', 'unknown') for r in successful]
        avg_word_count = round(np.mean([r.get('word_count', 0) for r in successful]) if successful else 0, 1)
        return {
            'total_urls': len(results),
            'successful': len(successful),
            'failed': len(results) - len(successful),
            'content_types': pd.Series(content_types).value_counts().to_dict(),
            'avg_word_count': avg_word_count,
            'failed_urls': [r['url'] for r in results if r.get('status') != 'success']
        }

    def _analyze_content(self, results: List[Dict]) -> Dict:
        """Genera un análisis de contenido agregado usando TF-IDF para extraer las palabras clave principales y muestras."""
        successful = [r for r in results if r.get('status') == 'success' and r.get('content')]
        texts = [r['content'] for r in successful if len(r['content'].split()) > 10]
        if not texts:
            return {'top_keywords': [], 'content_samples': []}
        try:
            stop_words = list(self.models['spacy'].Defaults.stop_words)
            vectorizer = TfidfVectorizer(stop_words=stop_words, max_features=50, ngram_range=(1, 2))
            tfidf = vectorizer.fit_transform(texts)
            feature_names = vectorizer.get_feature_names_out()
            sorted_indices = np.argsort(np.asarray(tfidf.sum(axis=0)).ravel())[-10:]
            top_keywords = feature_names[sorted_indices][::-1].tolist()
        except Exception as e:
            logger.error(f"Error en análisis TF-IDF: {e}")
            top_keywords = []
        samples = [{'url': r['url'], 'sample': (r['content'][:500] + '...') if len(r['content']) > 500 else r['content']} for r in successful[:3]]
        return {'top_keywords': top_keywords, 'content_samples': samples}

    def _analyze_links(self, results: List[Dict]) -> Dict:
        """Genera un análisis de enlaces internos, dominios externos, anclas y tipos de archivos."""
        all_links = []
        for result in results:
            if result.get('links'):
                all_links.extend(result['links'])
        if not all_links:
            return {'internal_links': {}, 'external_domains': {}, 'common_anchors': {}, 'file_types': {}}
        df = pd.DataFrame(all_links)
        return {
            'internal_links': df[df['type'] == 'internal']['url'].value_counts().head(20).to_dict(),
            'external_domains': df[df['type'] == 'external']['url'].apply(lambda x: urlparse(x).netloc).value_counts().head(10).to_dict(),
            'common_anchors': df['anchor'].value_counts().head(10).to_dict(),
            'file_types': df['file_type'].value_counts().to_dict()
        }

    def _generate_seo_recommendations(self, results: List[Dict]) -> List[str]:
        """Genera recomendaciones SEO en base a las deficiencias encontradas en el análisis."""
        successful = [r for r in results if r.get('status') == 'success']
        if not successful:
            return ["No se pudo analizar ningún contenido exitosamente"]
        recs = []
        missing_titles = sum(1 for r in successful if not r.get('metadata', {}).get('title'))
        if missing_titles:
            recs.append(f"📌 Añadir títulos a {missing_titles} páginas")
        short_descriptions = sum(1 for r in successful if not r.get('metadata', {}).get('description'))
        if short_descriptions:
            recs.append(f"📌 Añadir meta descripciones a {short_descriptions} páginas")
        short_content = sum(1 for r in successful if r.get('word_count', 0) < 300)
        if short_content:
            recs.append(f"📝 Ampliar contenido en {short_content} páginas (menos de 300 palabras)")
        all_links = [link for r in results for link in r.get('links', [])]
        if all_links:
            df_links = pd.DataFrame(all_links)
            internal_links = df_links[df_links['type'] == 'internal']
            if len(internal_links) > 100:
                recs.append(f"🔗 Optimizar estructura de enlaces internos ({len(internal_links)} enlaces)")
        return recs if recs else ["✅ No se detectaron problemas críticos de SEO"]

    def plot_internal_links(self, links_data: Dict) -> Any:
        """Genera un gráfico de barras horizontales mostrando los 20 principales enlaces internos."""
        internal_links = links_data.get('internal_links', {})
        if not internal_links:
            return {}
        fig, ax = plt.subplots()
        names = list(internal_links.keys())
        counts = list(internal_links.values())
        ax.barh(names, counts)
        ax.set_xlabel("Cantidad de enlaces")
        ax.set_title("Top 20 Enlaces Internos")
        plt.tight_layout()
        return fig