Create app.py

app.py

@@ -0,0 +1,251 @@
+import os
+import json
+import logging
+import re
+import requests
+import hashlib
+from urllib.parse import urlparse, urljoin
+from typing import List, Dict, Optional, Tuple
+from concurrent.futures import ThreadPoolExecutor, as_completed
+from bs4 import BeautifulSoup
+import PyPDF2
+from io import BytesIO
+from transformers import pipeline, AutoTokenizer, AutoModelForQuestionAnswering
+import numpy as np
+from sklearn.feature_extraction.text import TfidfVectorizer
+from sentence_transformers import SentenceTransformer
+import spacy
+import gradio as gr
+
+# Configuración avanzada
+logging.basicConfig(level=logging.INFO,
+                    format='%(asctime)s - %(levelname)s - %(message)s')
+logger = logging.getLogger(__name__)
+
+class AdvancedSEOAanalyzer:
+    def __init__(self, sitemap_url: str):
+        self.sitemap_url = sitemap_url
+        self.session = self._configure_session()
+        self.models = self._load_models()
+        self.processed_urls = set()
+        self.link_graph = defaultdict(list)
+        self.content_store = {}
+        self.documents = []
+
+    def _configure_session(self) -> requests.Session:
+        session = requests.Session()
+        retry = Retry(
+            total=5,
+            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)'
+        })
+        return session
+
+    def _load_models(self) -> Dict:
+        return {
+            'summarization': pipeline("summarization", 
+                                    model="facebook/bart-large-cnn",
+                                    device=0 if torch.cuda.is_available() else -1),
+            '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")
+        }
+
+    async def download_content(self, url: str) -> Optional[Dict]:
+        if url in self.processed_urls:
+            return None
+        
+        try:
+            response = self.session.get(url, timeout=15)
+            response.raise_for_status()
+            content_type = response.headers.get('Content-Type', '')
+
+            if 'application/pdf' in content_type:
+                return self._process_pdf(url, 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
+
+        except Exception as e:
+            logger.error(f"Error downloading {url}: {str(e)}")
+            return None
+
+    def _process_pdf(self, url: str, content: bytes) -> Dict:
+        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,
+            'type': 'pdf',
+            'content': text,
+            'hash': doc_hash,
+            'links': []
+        }
+
+    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]:
+        links = []
+        base_domain = urlparse(base_url).netloc
+        
+        for tag in soup.find_all(['a', 'link'], href=True):
+            href = tag['href']
+            full_url = urljoin(base_url, href)
+            parsed = urlparse(full_url)
+            
+            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()
+            })
+            
+        return links
+
+    def _extract_metadata(self, soup: BeautifulSoup) -> Dict:
+        metadata = {
+            'title': soup.title.string if soup.title else '',
+            'description': '',
+            'keywords': [],
+            'open_graph': {}
+        }
+        
+        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)
+        }
+        
+        if content['type'] == 'pdf':
+            analysis['document_analysis'] = self._analyze_document_structure(content)
+            
+        return analysis
+
+    def _perform_seo_analysis(self, content: Dict) -> Dict:
+        text = content['content']
+        doc = self.models['spacy'](text)
+        
+        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)
+        }
+
+    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]
+        
+        return feature_array[tfidf_sorting][:5].tolist()
+
+    def run_analysis(self, max_workers: int = 4) -> Dict:
+        sitemap_urls = self._parse_sitemap()
+        results = []
+        
+        with ThreadPoolExecutor(max_workers=max_workers) as executor:
+            futures = [executor.submit(self.download_content, url) 
+                      for url in sitemap_urls]
+            
+            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)
+        
+        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"
+        
+        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"
+        )
+        interface.launch()
+
+if __name__ == "__main__":
+    analyzer = AdvancedSEOAanalyzer("https://www.example.com/sitemap.xml")
+    analyzer.launch_interface()