""" 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