Update app.py

app.py

@@ -4,248 +4,272 @@ 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, Optional
 from urllib.parse import urlparse, urljoin
-from typing import List, Dict, Optional, Tuple
 from concurrent.futures import ThreadPoolExecutor, as_completed
 from bs4 import BeautifulSoup
+from pathlib import Path
+from datetime import datetime
+from collections import defaultdict
+
+import gradio as gr
+import matplotlib.pyplot as plt
+from sklearn.feature_extractioimport json
+import logging
+import re
+import requests
+import hashlib
 import PyPDF2
-from io import BytesIO
-from transformers import pipeline, AutoTokenizer, AutoModelForQuestionAnswering
 import numpy as np
+import pandas as pd
+from io import BytesIO
+from typing import List, Dict, Optional
+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 collections import defaultdict
+
+import gradio as gr
+import matplotlib.pyplot as plt
 from sklearn.feature_extraction.text import TfidfVectorizer
+from requests.adapters import HTTPAdapter
+from requests.packages.urllib3.util.retry import Retry
+from transformers import pipeline, AutoModelForQuestionAnswering, AutoTokenizer
 from sentence_transformers import SentenceTransformer
 import spacy
-import gradio as gr
+import torch
 
-# Configuración avanzada
-logging.basicConfig(level=logging.INFO,
-                    format='%(asctime)s - %(levelname)s - %(message)s')
+# Configuración inicial
+logging.basicConfig(level=logging.INFO)
 logger = logging.getLogger(__name__)
 
-class AdvancedSEOAanalyzer:
-    def __init__(self, sitemap_url: str):
-        self.sitemap_url = sitemap_url
+class SEOSpaceAnalyzer:
+    def __init__(self):
         self.session = self._configure_session()
         self.models = self._load_models()
-        self.processed_urls = set()
-        self.link_graph = defaultdict(list)
-        self.content_store = {}
+        self.base_dir = Path("content_storage")
+        self.link_analysis = defaultdict(list)
         self.documents = []
-
-    def _configure_session(self) -> requests.Session:
+        self.current_analysis = {}
+        
+    def _configure_session(self):
+        """Configuración avanzada de sesión HTTP con reintentos"""
         session = requests.Session()
         retry = Retry(
-            total=5,
+            total=3,
             backoff_factor=1,
             status_forcelist=[500, 502, 503, 504]
         )
         adapter = HTTPAdapter(max_retries=retry)
         session.mount('https://', adapter)
         session.headers.update({
-            'User-Agent': 'Mozilla/5.0 (compatible; SEOBot/1.0; +https://seo.example.com/bot)'
+            'User-Agent': 'Mozilla/5.0 (compatible; SEOBot/1.0)',
+            'Accept-Language': 'es-ES,es;q=0.9'
         })
         return session
-
-    def _load_models(self) -> Dict:
+    
+    def _load_models(self):
+        """Carga modelos de Hugging Face optimizados"""
+        device = 0 if torch.cuda.is_available() else -1
         return {
-            'summarization': pipeline("summarization", 
-                                    model="facebook/bart-large-cnn",
-                                    device=0 if torch.cuda.is_available() else -1),
+            'summarizer': pipeline("summarization", 
+                                 model="facebook/bart-large-cnn",
+                                 device=device),
+            'ner': pipeline("ner", 
+                           model="dslim/bert-base-NER",
+                           aggregation_strategy="simple",
+                           device=device),
             'qa': pipeline("question-answering",
                          model="deepset/roberta-base-squad2",
-                         tokenizer="deepset/roberta-base-squad2"),
-            'ner': pipeline("ner", 
-                          model="dslim/bert-base-NER",
-                          aggregation_strategy="simple"),
-            'semantic': SentenceTransformer('sentence-transformers/all-MiniLM-L6-v2'),
-            'spacy': spacy.load("en_core_web_lg")
+                         device=device),
+            'semantic': SentenceTransformer('paraphrase-multilingual-MiniLM-L12-v2'),
+            'spacy': spacy.load("es_core_news_lg")
         }
-
-    async def download_content(self, url: str) -> Optional[Dict]:
-        if url in self.processed_urls:
-            return None
-        
+    
+    def _process_url(self, url):
+        """Procesa una URL y extrae su contenido"""
         try:
             response = self.session.get(url, timeout=15)
             response.raise_for_status()
+            
             content_type = response.headers.get('Content-Type', '')
-
+            result = {'url': url, 'links': []}
+            
             if 'application/pdf' in content_type:
-                return self._process_pdf(url, response.content)
+                result.update(self._process_pdf(response.content))
             elif 'text/html' in content_type:
-                return await self._process_html(url, response.text)
-            else:
-                logger.warning(f"Unsupported content type: {content_type}")
-                return None
-
+                result.update(self._process_html(response.text, url))
+                
+            self._save_content(url, response.content)
+            return result
+            
         except Exception as e:
-            logger.error(f"Error downloading {url}: {str(e)}")
-            return None
-
-    def _process_pdf(self, url: str, content: bytes) -> Dict:
+            logger.error(f"Error procesando {url}: {str(e)}")
+            return {'url': url, 'error': str(e)}
+    
+    def _process_html(self, html, base_url):
+        """Procesa contenido HTML"""
+        soup = BeautifulSoup(html, 'lxml')
+        return {
+            'content': self._clean_text(soup.get_text()),
+            'type': 'html',
+            'metadata': self._extract_metadata(soup),
+            'links': self._extract_links(soup, base_url)
+        }
+    
+    def _process_pdf(self, content):
+        """Procesa documentos PDF"""
         text = ""
         with BytesIO(content) as pdf_file:
             reader = PyPDF2.PdfReader(pdf_file)
             for page in reader.pages:
                 text += page.extract_text()
-
-        doc_hash = hashlib.sha256(content).hexdigest()
-        self._save_document(url, content, 'pdf')
-
+                
         return {
-            'url': url,
+            'content': self._clean_text(text),
             'type': 'pdf',
-            'content': text,
-            'hash': doc_hash,
-            'links': []
+            'metadata': {'pages': len(reader.pages)}
         }
-
-    async def _process_html(self, url: str, html: str) -> Dict:
-        soup = BeautifulSoup(html, 'lxml')
-        main_content = self._extract_main_content(soup)
-        links = self._extract_links(url, soup)
-        
-        self._save_document(url, html.encode('utf-8'), 'html')
-        
-        return {
-            'url': url,
-            'type': 'html',
-            'content': main_content,
-            'hash': hashlib.sha256(html.encode()).hexdigest(),
-            'links': links,
-            'metadata': self._extract_metadata(soup)
-        }
-
-    def _extract_links(self, base_url: str, soup: BeautifulSoup) -> List[Dict]:
+    
+    def _extract_links(self, soup, base_url):
+        """Extrae y clasifica enlaces"""
         links = []
-        base_domain = urlparse(base_url).netloc
-        
-        for tag in soup.find_all(['a', 'link'], href=True):
+        for tag in soup.find_all('a', href=True):
             href = tag['href']
             full_url = urljoin(base_url, href)
-            parsed = urlparse(full_url)
+            link_type = 'internal' if urlparse(full_url).netloc == urlparse(base_url).netloc else 'external'
             
-            link_type = 'internal' if parsed.netloc == base_domain else 'external'
-            file_type = 'other'
-            
-            if parsed.path.lower().endswith(('.pdf', '.doc', '.docx')):
-                file_type = 'document'
-            elif parsed.path.lower().endswith(('.jpg', '.png', '.gif')):
-                file_type = 'image'
-                
             links.append({
                 'url': full_url,
                 'type': link_type,
-                'file_type': file_type,
-                'anchor': tag.text.strip()
+                'anchor': self._clean_text(tag.text),
+                'file_type': self._get_file_type(href)
             })
-            
         return links
-
-    def _extract_metadata(self, soup: BeautifulSoup) -> Dict:
-        metadata = {
-            'title': soup.title.string if soup.title else '',
-            'description': '',
-            'keywords': [],
-            'open_graph': {}
+    
+    def _get_file_type(self, url):
+        """Determina el tipo de archivo por extensión"""
+        ext = Path(urlparse(url).path).suffix.lower()
+        return ext[1:] if ext else 'html'
+    
+    def _clean_text(self, text):
+        """Limpieza avanzada de texto"""
+        text = re.sub(r'\s+', ' ', text)
+        return re.sub(r'[^\w\sáéíóúñÁÉÍÓÚÑ]', ' ', text).strip()
+    
+    def _save_content(self, url, content):
+        """Almacena el contenido descargado"""
+        path = urlparse(url).path.lstrip('/')
+        save_path = self.base_dir / urlparse(url).netloc / path
+        save_path.parent.mkdir(parents=True, exist_ok=True)
+        
+        with open(save_path.with_suffix(self._get_file_type(url)), 'wb') as f:
+            f.write(content)
+    
+    def analyze_sitemap(self, sitemap_url):
+        """Analiza todo el sitemap y genera reportes"""
+        urls = self._parse_sitemap(sitemap_url)
+        results = []
+        
+        with ThreadPoolExecutor(max_workers=4) as executor:
+            futures = [executor.submit(self._process_url, url) for url in urls]
+            for future in as_completed(futures):
+                results.append(future.result())
+                progress(len(results)/len(urls))
+        
+        self.current_analysis = {
+            'basic_stats': self._calculate_stats(results),
+            'content_analysis': self._analyze_content(results),
+            'link_analysis': self._analyze_links(results),
+            'seo_recommendations': self._generate_recommendations(results)
         }
         
-        meta_tags = soup.find_all('meta')
-        for tag in meta_tags:
-            name = tag.get('name', '').lower()
-            property = tag.get('property', '').lower()
-            content = tag.get('content', '')
-            
-            if name == 'description':
-                metadata['description'] = content
-            elif name == 'keywords':
-                metadata['keywords'] = [kw.strip() for kw in content.split(',')]
-            elif property.startswith('og:'):
-                key = property[3:]
-                metadata['open_graph'][key] = content
-                
-        return metadata
-
-    def analyze_content(self, content: Dict) -> Dict:
-        analysis = {
-            'summary': self.models['summarization'](content['content'], 
-                                                    max_length=150,
-                                                    min_length=30)[0]['summary_text'],
-            'entities': self.models['ner'](content['content']),
-            'semantic_embedding': self.models['semantic'].encode(content['content']),
-            'seo_analysis': self._perform_seo_analysis(content)
+        return self.current_analysis
+    
+    def _parse_sitemap(self, sitemap_url):
+        """Parsea sitemaps XML incluyendo sitemaps indexados"""
+        # Implementación de parsing de sitemap (similar a versiones anteriores)
+        return []
+    
+    def _calculate_stats(self, results):
+        """Calcula estadísticas básicas del análisis"""
+        return {
+            'total_urls': len(results),
+            'content_types': pd.Series([r.get('type', 'unknown') for r in results]).value_counts().to_dict(),
+            'avg_content_length': np.mean([len(r.get('content', '')) for r in results])
+        }
+    
+    def create_report(self):
+        """Crea un reporte descargable en múltiples formatos"""
+        report = {
+            'timestamp': datetime.now().isoformat(),
+            'analysis': self.current_analysis
         }
         
-        if content['type'] == 'pdf':
-            analysis['document_analysis'] = self._analyze_document_structure(content)
+        # Guardar en JSON
+        json_path = self.base_dir / 'seo_report.json'
+        with open(json_path, 'w') as f:
+            json.dump(report, f)
             
-        return analysis
-
-    def _perform_seo_analysis(self, content: Dict) -> Dict:
-        text = content['content']
-        doc = self.models['spacy'](text)
+        # Crear CSV con enlaces
+        df = pd.DataFrame([link for result in self.current_analysis['link_analysis'] for link in result['links']])
+        csv_path = self.base_dir / 'links_analysis.csv'
+        df.to_csv(csv_path, index=False)
         
-        return {
-            'readability_score': self._calculate_readability(text),
-            'keyword_density': self._calculate_keyword_density(text),
-            'heading_structure': self._analyze_headings(doc),
-            'content_length': len(text.split()),
-            'semantic_topics': self._extract_semantic_topics(text)
-        }
+        return [str(json_path), str(csv_path)]
+    
+    def create_visualization(self):
+        """Genera visualizaciones interactivas"""
+        fig, ax = plt.subplots()
+        pd.Series(self.current_analysis['basic_stats']['content_types']).plot.pie(
+            ax=ax, 
+            title='Distribución de Tipos de Contenido',
+            ylabel=''
+        )
+        return fig
 
-    def _extract_semantic_topics(self, text: str) -> List[str]:
-        vectorizer = TfidfVectorizer(stop_words='english', ngram_range=(1,2))
-        tfidf = vectorizer.fit_transform([text])
-        feature_array = np.array(vectorizer.get_feature_names_out())
-        tfidf_sorting = np.argsort(tfidf.toarray()).flatten()[::-1]
+# Interface Gradio
+def create_interface():
+    analyzer = SEOSpaceAnalyzer()
+    
+    with gr.Blocks(title="SEO Analyzer Pro", theme=gr.themes.Soft()) as interface:
+        gr.Markdown("# 🕵️ SEO Analyzer Pro")
         
-        return feature_array[tfidf_sorting][:5].tolist()
-
-    def run_analysis(self, max_workers: int = 4) -> Dict:
-        sitemap_urls = self._parse_sitemap()
-        results = []
+        with gr.Row():
+            sitemap_url = gr.Textbox(label="URL del Sitemap", placeholder="https://www.ing.es/ennaranja/sitemap.xml")
+            analyze_btn = gr.Button("Analizar", variant="primary")
         
-        with ThreadPoolExecutor(max_workers=max_workers) as executor:
-            futures = [executor.submit(self.download_content, url) 
-                      for url in sitemap_urls]
+        with gr.Tab("Resultados"):
+            json_output = gr.JSON(label="Análisis Completo")
+            plot_output = gr.Plot(label="Visualización")
             
-            for future in as_completed(futures):
-                result = future.result()
-                if result:
-                    analyzed = self.analyze_content(result)
-                    results.append({**result, **analyzed})
-                    self._update_link_graph(result)
+        with gr.Tab("Enlaces"):
+            internal_links = gr.Dataframe(label="Enlaces Internos")
+            external_links = gr.Dataframe(label="Enlaces Externos")
+            
+        with gr.Tab("Descargas"):
+            report_download = gr.Files(label="Descargar Reporte")
+            download_btn = gr.Button("Generar Reporte", variant="secondary")
         
-        self._save_full_analysis(results)
-        return {
-            'total_pages': len(results),
-            'document_types': self._count_document_types(results),
-            'link_analysis': self._analyze_link_graph(),
-            'content_analysis': self._aggregate_content_stats(results)
-        }
-
-    def _save_document(self, url: str, content: bytes, file_type: str) -> None:
-        parsed = urlparse(url)
-        path = parsed.path.lstrip('/')
-        filename = f"documents/{parsed.netloc}/{path}" if path else f"documents/{parsed.netloc}/index"
+        analyze_btn.click(
+            fn=analyzer.analyze_sitemap,
+            inputs=sitemap_url,
+            outputs=[json_output, plot_output, internal_links, external_links]
+        )
         
-        os.makedirs(os.path.dirname(filename), exist_ok=True)
-        with open(filename + f'.{file_type}', 'wb') as f:
-            f.write(content)
-
-    def launch_interface(self):
-        interface = gr.Interface(
-            fn=self.run_analysis,
-            inputs=gr.Textbox(label="Sitemap URL"),
-            outputs=[
-                gr.JSON(label="Analysis Results"),
-                gr.File(label="Download Data")
-            ],
-            title="Advanced SEO Analyzer",
-            description="Analyze websites with AI-powered SEO insights"
+        download_btn.click(
+            fn=analyzer.create_report,
+            outputs=report_download
         )
-        interface.launch()
+        
+    return interface
 
 if __name__ == "__main__":
-    analyzer = AdvancedSEOAanalyzer("https://www.ing.es/ennaranja/sitemap.xml")
-    analyzer.launch_interface()
+    interface = create_interface()
+    interface.launch(server_name="0.0.0.0", server_port=7860)