utils.py

"""
Pipeline d'extraction de contenu — version refactorisée.

- TextCleaner     : nettoyage profond du texte (inchangé, déjà très bon)
- ContentCleaner  : extraction multi-méthodes (trafilatura → readability → justext)
- URLValidator    : validation et normalisation d'URL
"""

from __future__ import annotations

import hashlib
import logging
import re
from typing import Any, Optional
from urllib.parse import parse_qs, urlencode, urljoin, urlparse, urlunparse

import ftfy
import justext
from langdetect import detect_langs
from readability import Document
from selectolax.parser import HTMLParser
from trafilatura import bare_extraction
from w3lib.html import remove_tags, replace_entities

logger = logging.getLogger(__name__)


# ---------------------------------------------------------------------------
# TextCleaner
# ---------------------------------------------------------------------------


class TextCleaner:
    """Pipeline de nettoyage de texte ultra-agressif."""

    MD_IMAGE = re.compile(r"!\[.*?\]\(.*?\)", re.DOTALL)
    MD_LINK = re.compile(r"\[([^\]]*)\]\([^)]*\)", re.DOTALL)
    WIKI_REF = re.compile(r"\[\[?\d+\]?\]\([^)]*\)|\[\[?\d+\]?\]")
    RAW_URL = re.compile(r"https?://[^\s\)\]\,\"\'<>]+|//[^\s\)\]\,\"\'<>]+")
    HTML_TAGS = re.compile(r"<[^>]+>")
    MULTI_SPACE = re.compile(r" {2,}")
    MULTI_NEWLINE = re.compile(r"\n{3,}")
    CONTROL_CHARS = re.compile(r"[\x00-\x08\x0b-\x0c\x0e-\x1f\x7f]")
    JUNK_LINE = re.compile(r"^\s*[\|\-\=\*\#\~\^]{2,}\s*$", re.MULTILINE)
    TABLE_ROW = re.compile(r"^\|.*\|$", re.MULTILINE)
    ONLY_PUNCTUATION = re.compile(r"^[\d\s\.\,\;\:\!\?\-\|\=\[\]\(\)]+$")

    @classmethod
    def deep_clean(cls, text: str) -> str:
        if not text:
            return ""

        text = ftfy.fix_text(text)
        text = replace_entities(text)
        text = cls.MD_IMAGE.sub("", text)
        text = cls.WIKI_REF.sub("", text)
        text = cls.MD_LINK.sub(r"\1", text)
        text = cls.RAW_URL.sub("", text)
        text = cls.HTML_TAGS.sub("", text)
        text = cls.CONTROL_CHARS.sub("", text)
        text = cls.TABLE_ROW.sub("", text)
        text = cls.JUNK_LINE.sub("", text)

        lines = []
        for line in text.splitlines():
            line = line.strip()
            if len(line) < 2:
                continue
            if cls.ONLY_PUNCTUATION.match(line):
                continue
            lines.append(line)

        text = "\n".join(lines)
        text = cls.MULTI_SPACE.sub(" ", text)
        text = cls.MULTI_NEWLINE.sub("\n\n", text)
        return text.strip()

    @classmethod
    def extract_clean_sentences(cls, text: str, min_length: int = 30) -> str:
        text = cls.deep_clean(text)
        paragraphs = text.split("\n\n")
        valid = [p.strip() for p in paragraphs if len(p.strip()) >= min_length]
        return "\n\n".join(valid)


# ---------------------------------------------------------------------------
# ContentCleaner
# ---------------------------------------------------------------------------


class ContentCleaner:
    """Extraction et nettoyage de contenu HTML."""

    UNWANTED_CSS_SELECTORS: list[str] = [
        "sup.reference",
        "div.reflist",
        "div.navbox",
        "div.toc",
        "div.hatnote",
        "table.navbox",
        "table.wikitable",
        "div.mw-references-wrap",
        "ol.references",
        "span.mw-editsection",
        "div.sidebar",
        "div.noprint",
        ".navigation-not-searchable",
        "script",
        "style",
        "noscript",
        "iframe",
        "embed",
        "object",
        "svg",
        "canvas",
        "head",
    ]

    @classmethod
    def clean_html_fast(cls, html: str) -> str:
        """Supprime les éléments parasites avant extraction."""
        try:
            tree = HTMLParser(html)
            for selector in cls.UNWANTED_CSS_SELECTORS:
                try:
                    for node in tree.css(selector):
                        node.decompose()
                except Exception:
                    pass
            return tree.html or html
        except Exception as exc:
            logger.warning("clean_html_fast error=%s", exc)
            return html

    @classmethod
    def extract_main_content(cls, html: str, url: str) -> dict[str, Any]:
        """
        Extraction multi-méthodes : trafilatura → readability → justext → basic.
        Retourne toujours un dict même si toutes les méthodes échouent.
        """
        result: dict[str, Any] = {
            "text": "",
            "title": "",
            "author": "",
            "date": None,
            "description": "",
            "language": "unknown",
            "method": "unknown",
        }

        clean_html = cls.clean_html_fast(html)

        # --- Méthode 1 : trafilatura (SOTA précision) ---
        try:
            extracted = bare_extraction(
                clean_html,
                url=url,
                include_comments=False,
                include_tables=False,
                include_images=False,
                include_links=False,
                deduplicate=True,
                favor_precision=True,
                no_fallback=False,
            )
            if extracted and len(extracted.get("text") or "") > 100:
                clean_text = TextCleaner.deep_clean(extracted["text"])
                result.update(
                    {
                        "text": clean_text,
                        "title": extracted.get("title") or "",
                        "author": extracted.get("author") or "",
                        "date": str(extracted["date"]) if extracted.get("date") else None,
                        "description": extracted.get("description") or "",
                        "method": "trafilatura",
                    }
                )
                result["language"] = cls._detect_language(clean_text)
                return result
        except Exception as exc:
            logger.debug("trafilatura_failed error=%s", exc)

        # --- Méthode 2 : readability ---
        try:
            doc = Document(clean_html)
            raw_text = remove_tags(doc.summary())
            clean_text = TextCleaner.deep_clean(raw_text)
            if len(clean_text) > 50:
                result.update(
                    {
                        "text": clean_text,
                        "title": doc.title(),
                        "method": "readability",
                    }
                )
                result["language"] = cls._detect_language(clean_text)
                return result
        except Exception as exc:
            logger.debug("readability_failed error=%s", exc)

        # --- Méthode 3 : justext ---
        try:
            paragraphs = justext.justext(
                clean_html.encode("utf-8", errors="replace"),
                justext.get_stoplist("English"),
                length_low=50,
                length_high=200,
                stopwords_low=0.20,
                stopwords_high=0.30,
                max_link_density=0.3,
                no_headings=False,
            )
            texts = [p.text for p in paragraphs if not p.is_boilerplate]
            clean_text = TextCleaner.deep_clean("\n\n".join(texts))
            if clean_text:
                result.update({"text": clean_text, "method": "justext"})
                result["language"] = cls._detect_language(clean_text)
                return result
        except Exception as exc:
            logger.debug("justext_failed error=%s", exc)

        # --- Fallback ultime ---
        result["text"] = TextCleaner.deep_clean(remove_tags(clean_html))
        result["method"] = "basic"
        return result

    @staticmethod
    def _detect_language(text: str) -> str:
        try:
            if text:
                langs = detect_langs(text[:500])
                return langs[0].lang if langs else "unknown"
        except Exception:
            pass
        return "unknown"

    @staticmethod
    def normalize_text(text: str) -> str:
        return TextCleaner.deep_clean(text)

    @staticmethod
    def extract_metadata(html: str) -> dict[str, Any]:
        """Open Graph + Twitter Cards + meta standards."""
        metadata: dict[str, Any] = {}
        try:
            tree = HTMLParser(html)

            for meta in tree.css('meta[property^="og:"]'):
                prop = meta.attributes.get("property", "").replace("og:", "").strip()
                content = meta.attributes.get("content", "").strip()
                if prop and content:
                    metadata[f"og_{prop}"] = content

            for meta in tree.css('meta[name^="twitter:"]'):
                name = meta.attributes.get("name", "").replace("twitter:", "").strip()
                content = meta.attributes.get("content", "").strip()
                if name and content:
                    metadata[f"twitter_{name}"] = content

            for meta in tree.css("meta[name]"):
                name = meta.attributes.get("name", "").strip()
                content = meta.attributes.get("content", "").strip()
                if name in {"description", "keywords", "author"} and content:
                    metadata[name] = content

            canonical = tree.css_first('link[rel="canonical"]')
            if canonical:
                href = canonical.attributes.get("href", "").strip()
                if href:
                    metadata["canonical"] = href

            title_node = tree.css_first("title")
            if title_node and not metadata.get("og_title"):
                metadata["page_title"] = title_node.text(strip=True)

        except Exception as exc:
            logger.warning("extract_metadata error=%s", exc)

        return metadata

    @staticmethod
    def extract_links(html: str, base_url: str) -> list[dict[str, str]]:
        links: list[dict[str, str]] = []
        seen: set[str] = set()
        try:
            tree = HTMLParser(html)
            for link in tree.css("a[href]"):
                href = link.attributes.get("href", "").strip()
                if not href or href.startswith(("#", "javascript:", "mailto:", "tel:")):
                    continue
                try:
                    abs_url = urljoin(base_url, href)
                except Exception:
                    continue
                if abs_url in seen:
                    continue
                seen.add(abs_url)
                parsed = urlparse(abs_url)
                if parsed.scheme not in {"http", "https"}:
                    continue
                links.append(
                    {
                        "url": abs_url,
                        "text": (link.text(strip=True) or "")[:200],
                        "rel": link.attributes.get("rel", ""),
                        "title": link.attributes.get("title", ""),
                    }
                )
        except Exception as exc:
            logger.warning("extract_links error=%s", exc)
        return links

    @staticmethod
    def extract_images(html: str, base_url: str) -> list[dict[str, str]]:
        images: list[dict[str, str]] = []
        seen: set[str] = set()
        src_attrs = ("src", "data-src", "data-lazy-src", "data-original", "data-lazy")
        try:
            tree = HTMLParser(html)
            for img in tree.css("img"):
                src = next(
                    (img.attributes.get(a, "") for a in src_attrs if img.attributes.get(a)),
                    "",
                ).strip()
                if not src or src.startswith("data:"):
                    continue
                try:
                    abs_url = urljoin(base_url, src)
                except Exception:
                    continue
                if abs_url in seen:
                    continue
                seen.add(abs_url)
                if urlparse(abs_url).scheme not in {"http", "https"}:
                    continue
                images.append(
                    {
                        "url": abs_url,
                        "alt": img.attributes.get("alt", "").strip(),
                        "title": img.attributes.get("title", "").strip(),
                        "width": img.attributes.get("width", ""),
                        "height": img.attributes.get("height", ""),
                    }
                )
        except Exception as exc:
            logger.warning("extract_images error=%s", exc)
        return images

    @staticmethod
    def compute_content_hash(text: str) -> str:
        return hashlib.sha256(text.encode("utf-8")).hexdigest()


# ---------------------------------------------------------------------------
# URLValidator
# ---------------------------------------------------------------------------


class URLValidator:
    BLOCKED_EXTENSIONS: frozenset[str] = frozenset(
        {
            ".pdf", ".zip", ".exe", ".dmg", ".pkg", ".deb", ".rpm",
            ".jpg", ".jpeg", ".png", ".gif", ".svg", ".webp", ".ico",
            ".mp4", ".avi", ".mov", ".mp3", ".wav", ".flac",
            ".css", ".js", ".woff", ".woff2", ".ttf", ".eot",
        }
    )
    ALLOWED_SCHEMES: frozenset[str] = frozenset({"http", "https"})

    @classmethod
    def is_valid_url(cls, url: str) -> bool:
        try:
            parsed = urlparse(url)
            if parsed.scheme not in cls.ALLOWED_SCHEMES:
                return False
            if not parsed.netloc:
                return False
            path_lower = parsed.path.lower()
            if any(path_lower.endswith(ext) for ext in cls.BLOCKED_EXTENSIONS):
                return False
            return True
        except Exception:
            return False

    @staticmethod
    def normalize_url(url: str) -> str:
        try:
            parsed = urlparse(url)._replace(fragment="")
            if parsed.query:
                params = parse_qs(parsed.query)
                new_query = urlencode(sorted(params.items()), doseq=True)
                parsed = parsed._replace(query=new_query)
            return urlunparse(parsed)
        except Exception:
            return url