Update seo_analyzer.py

seo_analyzer.py

@@ -6,7 +6,7 @@ import PyPDF2
 import numpy as np
 import pandas as pd
 from io import BytesIO
-from typing import List, Dict, Tuple
+from typing import List, Dict, Tuple, Optional
 from urllib.parse import urlparse, urljoin
 from concurrent.futures import ThreadPoolExecutor, as_completed
 from bs4 import BeautifulSoup
@@ -22,15 +22,15 @@ import spacy
 import matplotlib.pyplot as plt
 from utils import sanitize_filename
 
-# Palabras no permitidas en SEO financiero/bancario
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+# Términos prohibidos (ejemplo)
 PROHIBITED_TERMS = [
     "gratis", "garantizado", "rentabilidad asegurada", "sin compromiso",
     "resultados inmediatos", "cero riesgo", "sin letra pequeña"
 ]
 
-logging.basicConfig(level=logging.INFO)
-logger = logging.getLogger(__name__)
-
 class SEOSpaceAnalyzer:
     def __init__(self, max_urls: int = 20, max_workers: int = 4):
         self.max_urls = max_urls
@@ -64,20 +64,36 @@ class SEOSpaceAnalyzer:
             "zeroshot": pipeline("zero-shot-classification", model="facebook/bart-large-mnli")
         }
 
-    def analyze_sitemap(self, sitemap_url: str) -> Tuple:
+    def analyze_sitemap(
+        self,
+        sitemap_url: str,
+        progress_callback: Optional[callable] = None,
+        status_callback: Optional[callable] = None
+    ) -> Tuple:
         urls = self._parse_sitemap(sitemap_url)
         if not urls:
-            return {"error": "No se pudieron extraer URLs"}, [], {}, {}, [], {}, {}, {}
-
+            return {"error": "No se pudieron extraer URLs"}, [], {}, {}, {}, {}, {}
         results = []
-        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):
-                try:
-                    results.append(future.result())
-                except Exception as e:
-                    results.append({"url": futures[future], "status": "error", "error": str(e)})
+        batch_size = 5
+        num_urls = min(len(urls), self.max_urls)
+        total_batches = (num_urls + batch_size - 1) // batch_size
 
+        for batch_index in range(total_batches):
+            start = batch_index * batch_size
+            batch_urls = urls[start:start+batch_size]
+            if status_callback:
+                status_callback(f"Procesando batch {batch_index+1}/{total_batches}: {batch_urls}")
+            with ThreadPoolExecutor(max_workers=len(batch_urls)) as executor:
+                futures = {executor.submit(self._process_url, url): url for url in batch_urls}
+                for future in as_completed(futures):
+                    try:
+                        results.append(future.result())
+                    except Exception as e:
+                        results.append({"url": futures[future], "status": "error", "error": str(e)})
+            if progress_callback:
+                progress_callback(batch_index+1, total_batches)
+
+        # Aplicar procesos de NLP a los resultados
         summaries, entities = self._apply_nlp(results)
         similarities = self._compute_similarity(results)
         flags = self._flag_prohibited_terms(results)
@@ -100,10 +116,11 @@ class SEOSpaceAnalyzer:
         }
 
         a = self.current_analysis
+        # Retornamos 7 outputs (sin summaries, que no se muestran en la UI)
         return (
             a["stats"], a["recommendations"], a["content_analysis"],
-            a["links"], a["details"], a["summaries"],
-            a["similarities"], a["seo_tags"]
+            a["links"], a["details"], a["similarities"],
+            a["seo_tags"]
         )
 
     def _process_url(self, url: str) -> Dict:
@@ -118,7 +135,7 @@ class SEOSpaceAnalyzer:
 
     def _process_html(self, url: str, html: str) -> Dict:
         soup = BeautifulSoup(html, "html.parser")
-        text = re.sub(r"\\s+", " ", soup.get_text())
+        text = re.sub(r"\s+", " ", soup.get_text())
         return {
             "url": url,
             "type": "html",
@@ -144,7 +161,7 @@ class SEOSpaceAnalyzer:
         except Exception as e:
             return {"url": url, "status": "error", "error": str(e)}
 
-    def _extract_metadata(self, soup) -> Dict:
+    def _extract_metadata(self, soup: BeautifulSoup) -> Dict:
         meta = {"title": "", "description": ""}
         if soup.title:
             meta["title"] = soup.title.string.strip()
@@ -153,7 +170,7 @@ class SEOSpaceAnalyzer:
                 meta["description"] = tag.get("content", "")
         return meta
 
-    def _extract_links(self, soup, base_url) -> List[Dict]:
+    def _extract_links(self, soup: BeautifulSoup, base_url: str) -> List[Dict]:
         links = []
         base_domain = urlparse(base_url).netloc
         for tag in soup.find_all("a", href=True):
@@ -172,25 +189,80 @@ class SEOSpaceAnalyzer:
             r = self.session.get(sitemap_url)
             soup = BeautifulSoup(r.text, "lxml-xml")
             return [loc.text for loc in soup.find_all("loc")]
-        except:
+        except Exception as e:
+            logger.error(f"Error al parsear sitemap {sitemap_url}: {e}")
             return []
 
-    def _apply_nlp(self, results) -> Tuple[Dict, Dict]:
+    def _save_content(self, url: str, content: bytes) -> None:
+        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 = hash(content)
+            if save_path.exists():
+                with open(save_path, "rb") as f:
+                    if hash(f.read()) == new_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:
+        success = [r for r in results if r.get("status") == "success"]
+        return {
+            "total": len(results),
+            "success": len(success),
+            "failed": len(results) - len(success),
+            "avg_words": round(np.mean([r.get("word_count", 0) for r in success]) if success else 0, 1)
+        }
+
+    def _analyze_content(self, results: List[Dict]) -> Dict:
+        texts = [r["content"] for r in results if r.get("status") == "success" and r.get("content")]
+        if not texts:
+            return {}
+        tfidf = TfidfVectorizer(max_features=20, stop_words=list(self.models["spacy"].Defaults.stop_words))
+        tfidf.fit(texts)
+        top = tfidf.get_feature_names_out().tolist()
+        return {"top_keywords": top, "samples": texts[:3]}
+
+    def _analyze_links(self, results: List[Dict]) -> Dict:
+        all_links = []
+        for r in results:
+            all_links.extend(r.get("links", []))
+        if not all_links:
+            return {}
+        df = pd.DataFrame(all_links)
+        return {
+            "internal_links": df[df["type"] == "internal"]["url"].value_counts().head(10).to_dict(),
+            "external_links": df[df["type"] == "external"]["url"].value_counts().head(10).to_dict()
+        }
+
+    def _apply_nlp(self, results: List[Dict]) -> Tuple[Dict, Dict]:
         summaries, entities = {}, {}
         for r in results:
-            if r.get("status") != "success" or not r.get("content"): continue
+            if r.get("status") != "success" or not r.get("content"):
+                continue
             text = r["content"][:1024]
             try:
                 summaries[r["url"]] = self.models["summarizer"](text, max_length=100, min_length=30)[0]["summary_text"]
                 ents = self.models["ner"](text)
                 entities[r["url"]] = list({e["word"] for e in ents if e["score"] > 0.8})
-            except:
+            except Exception as e:
                 continue
         return summaries, entities
 
-    def _compute_similarity(self, results) -> Dict[str, List[Dict]]:
+    def _compute_similarity(self, results: List[Dict]) -> Dict[str, List[Dict]]:
         docs = [(r["url"], r["content"]) for r in results if r.get("status") == "success" and r.get("content")]
-        if len(docs) < 2: return {}
+        if len(docs) < 2:
+            return {}
         urls, texts = zip(*docs)
         emb = self.models["semantic"].encode(texts, convert_to_tensor=True)
         sim = util.pytorch_cos_sim(emb, emb)
@@ -200,7 +272,7 @@ class SEOSpaceAnalyzer:
             for i in range(len(urls))
         }
 
-    def _flag_prohibited_terms(self, results) -> Dict[str, List[str]]:
+    def _flag_prohibited_terms(self, results: List[Dict]) -> Dict[str, List[str]]:
         flags = {}
         for r in results:
             found = [term for term in PROHIBITED_TERMS if term in r.get("content", "").lower()]
@@ -208,32 +280,33 @@ class SEOSpaceAnalyzer:
                 flags[r["url"]] = found
         return flags
 
-    def _classify_topics(self, results) -> Dict[str, List[str]]:
+    def _classify_topics(self, results: List[Dict]) -> Dict[str, List[str]]:
         labels = [
             "hipotecas", "préstamos", "cuentas", "tarjetas",
             "seguros", "inversión", "educación financiera"
         ]
         topics = {}
         for r in results:
-            if r.get("status") != "success": continue
+            if r.get("status") != "success":
+                continue
             try:
                 res = self.models["zeroshot"](r["content"][:1000], candidate_labels=labels, multi_label=True)
                 topics[r["url"]] = [l for l, s in zip(res["labels"], res["scores"]) if s > 0.5]
-            except:
+            except Exception as e:
                 continue
         return topics
 
-    def _generate_seo_tags(self, results, summaries, topics, flags) -> Dict[str, Dict]:
+    def _generate_seo_tags(self, results: List[Dict], summaries: Dict, topics: Dict, flags: Dict) -> Dict[str, Dict]:
         seo_tags = {}
         for r in results:
             url = r["url"]
             base = summaries.get(url, r.get("content", "")[:300])
-            topic = topics.get(url, ["contenido"])[0]
+            topic = topics.get(url, ["contenido"])[0] if topics.get(url) else "contenido"
             try:
                 prompt = f"Genera un título SEO formal y una meta descripción para contenido sobre {topic}: {base}"
                 output = self.models["summarizer"](prompt, max_length=60, min_length=20)[0]["summary_text"]
                 title, desc = output.split(".")[0], output
-            except:
+            except Exception as e:
                 title, desc = "", ""
             seo_tags[url] = {
                 "title": title,
@@ -242,37 +315,7 @@ class SEOSpaceAnalyzer:
             }
         return seo_tags
 
-    def _calculate_stats(self, results):
-        success = [r for r in results if r.get("status") == "success"]
-        return {
-            "total": len(results),
-            "success": len(success),
-            "failed": len(results) - len(success),
-            "avg_words": round(np.mean([r.get("word_count", 0) for r in success]), 1)
-        }
-
-    def _analyze_content(self, results):
-        texts = [r["content"] for r in results if r.get("status") == "success" and r.get("content")]
-        if not texts:
-            return {}
-        tfidf = TfidfVectorizer(max_features=20, stop_words=list(self.models["spacy"].Defaults.stop_words))
-        tfidf.fit(texts)
-        top = tfidf.get_feature_names_out().tolist()
-        return {"top_keywords": top, "samples": texts[:3]}
-
-    def _analyze_links(self, results):
-        all_links = []
-        for r in results:
-            all_links.extend(r.get("links", []))
-        if not all_links:
-            return {}
-        df = pd.DataFrame(all_links)
-        return {
-            "internal_links": df[df["type"] == "internal"]["url"].value_counts().head(10).to_dict(),
-            "external_links": df[df["type"] == "external"]["url"].value_counts().head(10).to_dict()
-        }
-
-    def _generate_recommendations(self, results):
+    def _generate_recommendations(self, results: List[Dict]) -> List[str]:
         recs = []
         if any(r.get("word_count", 0) < 300 for r in results):
             recs.append("✍️ Algunos contenidos son demasiado breves (<300 palabras)")
@@ -280,7 +323,7 @@ class SEOSpaceAnalyzer:
             recs.append("⚠️ Detectado uso de lenguaje no permitido")
         return recs or ["✅ Todo parece correcto"]
 
-    def plot_internal_links(self, links: Dict):
+    def plot_internal_links(self, links: Dict) -> any:
         if not links or not links.get("internal_links"):
             fig, ax = plt.subplots()
             ax.text(0.5, 0.5, "No hay enlaces internos", ha="center")