seo_analyzer.py

import os
import re
import logging
import requests
import PyPDF2
import numpy as np
import pandas as pd
from io import BytesIO
from typing import List, Dict, 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, util
import torch
import spacy
import matplotlib.pyplot as plt
from utils import sanitize_filename

# Palabras no permitidas en SEO financiero/bancario
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
        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 = {}

    def _configure_session(self):
        session = requests.Session()
        retry = Retry(total=3, backoff_factor=1,
                      status_forcelist=[500, 502, 503, 504],
                      allowed_methods=["GET"])
        session.mount("http://", HTTPAdapter(max_retries=retry))
        session.mount("https://", HTTPAdapter(max_retries=retry))
        session.headers.update({
            "User-Agent": "SEOAnalyzer/1.0",
            "Accept-Language": "es-ES,es;q=0.9"
        })
        return session

    def _load_models(self):
        device = 0 if torch.cuda.is_available() else -1
        return {
            "spacy": spacy.load("es_core_news_lg"),
            "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"),
            "zeroshot": pipeline("zero-shot-classification", model="facebook/bart-large-mnli")
        }

    def analyze_sitemap(self, sitemap_url: str) -> Tuple:
        urls = self._parse_sitemap(sitemap_url)
        if not 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)})

        summaries, entities = self._apply_nlp(results)
        similarities = self._compute_similarity(results)
        flags = self._flag_prohibited_terms(results)
        topics = self._classify_topics(results)
        seo_tags = self._generate_seo_tags(results, summaries, topics, flags)

        self.current_analysis = {
            "stats": self._calculate_stats(results),
            "content_analysis": self._analyze_content(results),
            "links": self._analyze_links(results),
            "recommendations": self._generate_recommendations(results),
            "details": results,
            "summaries": summaries,
            "entities": entities,
            "similarities": similarities,
            "flags": flags,
            "topics": topics,
            "seo_tags": seo_tags,
            "timestamp": datetime.now().isoformat()
        }

        a = self.current_analysis
        return (
            a["stats"], a["recommendations"], a["content_analysis"],
            a["links"], a["details"], a["summaries"],
            a["similarities"], a["seo_tags"]
        )

    def _process_url(self, url: str) -> Dict:
        try:
            response = self.session.get(url, timeout=10)
            content_type = response.headers.get("Content-Type", "")
            if "application/pdf" in content_type:
                return self._process_pdf(url, response.content)
            return self._process_html(url, response.text)
        except Exception as e:
            return {"url": url, "status": "error", "error": str(e)}

    def _process_html(self, url: str, html: str) -> Dict:
        soup = BeautifulSoup(html, "html.parser")
        text = re.sub(r"\\s+", " ", soup.get_text())
        return {
            "url": url,
            "type": "html",
            "status": "success",
            "content": text,
            "word_count": len(text.split()),
            "metadata": self._extract_metadata(soup),
            "links": self._extract_links(soup, url)
        }

    def _process_pdf(self, url: str, content: bytes) -> Dict:
        try:
            reader = PyPDF2.PdfReader(BytesIO(content))
            text = "".join(p.extract_text() or "" for p in reader.pages)
            return {
                "url": url,
                "type": "pdf",
                "status": "success",
                "content": text,
                "word_count": len(text.split()),
                "page_count": len(reader.pages)
            }
        except Exception as e:
            return {"url": url, "status": "error", "error": str(e)}

    def _extract_metadata(self, soup) -> Dict:
        meta = {"title": "", "description": ""}
        if soup.title:
            meta["title"] = soup.title.string.strip()
        for tag in soup.find_all("meta"):
            if tag.get("name") == "description":
                meta["description"] = tag.get("content", "")
        return meta

    def _extract_links(self, soup, base_url) -> List[Dict]:
        links = []
        base_domain = urlparse(base_url).netloc
        for tag in soup.find_all("a", href=True):
            href = tag["href"]
            full_url = urljoin(base_url, href)
            netloc = urlparse(full_url).netloc
            links.append({
                "url": full_url,
                "type": "internal" if netloc == base_domain else "external",
                "anchor": tag.get_text(strip=True)
            })
        return links

    def _parse_sitemap(self, sitemap_url: str) -> List[str]:
        try:
            r = self.session.get(sitemap_url)
            soup = BeautifulSoup(r.text, "lxml-xml")
            return [loc.text for loc in soup.find_all("loc")]
        except:
            return []

    def _apply_nlp(self, results) -> Tuple[Dict, Dict]:
        summaries, entities = {}, {}
        for r in results:
            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:
                continue
        return summaries, entities

    def _compute_similarity(self, results) -> 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 {}
        urls, texts = zip(*docs)
        emb = self.models["semantic"].encode(texts, convert_to_tensor=True)
        sim = util.pytorch_cos_sim(emb, emb)
        return {
            urls[i]: [{"url": urls[j], "score": float(sim[i][j])}
                      for j in np.argsort(-sim[i]) if i != j][:3]
            for i in range(len(urls))
        }

    def _flag_prohibited_terms(self, results) -> Dict[str, List[str]]:
        flags = {}
        for r in results:
            found = [term for term in PROHIBITED_TERMS if term in r.get("content", "").lower()]
            if found:
                flags[r["url"]] = found
        return flags

    def _classify_topics(self, results) -> 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
            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:
                continue
        return topics

    def _generate_seo_tags(self, results, summaries, topics, flags) -> 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]
            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:
                title, desc = "", ""
            seo_tags[url] = {
                "title": title,
                "meta_description": desc,
                "flags": flags.get(url, [])
            }
        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):
        recs = []
        if any(r.get("word_count", 0) < 300 for r in results):
            recs.append("✍️ Algunos contenidos son demasiado breves (<300 palabras)")
        if any("gratis" in r.get("content", "").lower() for r in results):
            recs.append("⚠️ Detectado uso de lenguaje no permitido")
        return recs or ["✅ Todo parece correcto"]

    def plot_internal_links(self, links: Dict):
        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")
            return fig
        top = links["internal_links"]
        fig, ax = plt.subplots()
        ax.barh(list(top.keys()), list(top.values()))
        ax.set_title("Top Enlaces Internos")
        plt.tight_layout()
        return fig