""" Utilitaires de scraping haute performance - Version finale patchée. Nettoyage de texte renforcé pour éliminer tous les artefacts résiduels. """ import re import hashlib from typing import Optional, List, Dict, Any from urllib.parse import urljoin, urlparse import logging from selectolax.parser import HTMLParser from trafilatura import extract, bare_extraction from readability import Document import justext import ftfy from langdetect import detect_langs from w3lib.html import remove_tags, replace_entities logger = logging.getLogger(__name__) class TextCleaner: """Pipeline de nettoyage de texte ultra-agressif.""" # Images Markdown : ![alt](url) MD_IMAGE = re.compile(r'!\[.*?\]\(.*?\)', re.DOTALL) # Liens Markdown : [texte](url) MD_LINK = re.compile(r'\[([^\]]*)\]\([^)]*\)', re.DOTALL) # Références Wikipedia : [[1]](url) ou [1] ou [[note]] WIKI_REF = re.compile(r'\[\[?\d+\]?\]\([^)]*\)|\[\[?\d+\]?\]') # URLs brutes RAW_URL = re.compile( r'https?://[^\s\)\]\,\"\'<>]+|' r'//[^\s\)\]\,\"\'<>]+' ) # Balises HTML résiduelles HTML_TAGS = re.compile(r'<[^>]+>') # Espaces et lignes multiples MULTI_SPACE = re.compile(r' {2,}') MULTI_NEWLINE = re.compile(r'\n{3,}') # Caractères de contrôle (sauf \n et \t) CONTROL_CHARS = re.compile(r'[\x00-\x08\x0b-\x0c\x0e-\x1f\x7f]') # Lignes ne contenant que de la ponctuation ou des séparateurs JUNK_LINE = re.compile(r'^\s*[\|\-\=\*\#\~\^]{2,}\s*$', re.MULTILINE) # Pipe tables (tableaux Markdown) TABLE_ROW = re.compile(r'^\|.*\|$', re.MULTILINE) @classmethod def deep_clean(cls, text: str) -> str: """ Nettoyage profond du texte extrait. Pipeline complet dans l'ordre optimal. Args: text: Texte brut à nettoyer Returns: Texte propre et normalisé """ if not text: return "" # 1. Fix encodage cassé (mojibake, caractères mal encodés) text = ftfy.fix_text(text) # 2. Décoder les entités HTML (& < etc.) text = replace_entities(text) # 3. Supprimer les images Markdown ![](url) text = cls.MD_IMAGE.sub('', text) # 4. Supprimer les références Wikipedia [[1]](url) text = cls.WIKI_REF.sub('', text) # 5. Transformer les liens Markdown en texte seul text = cls.MD_LINK.sub(r'\1', text) # 6. Supprimer les URLs brutes résiduelles text = cls.RAW_URL.sub('', text) # 7. Supprimer les balises HTML résiduelles text = cls.HTML_TAGS.sub('', text) # 8. Supprimer les caractères de contrôle text = cls.CONTROL_CHARS.sub('', text) # 9. Supprimer les lignes de tableaux Markdown text = cls.TABLE_ROW.sub('', text) # 10. Supprimer les lignes parasites (séparateurs, pipes, etc.) text = cls.JUNK_LINE.sub('', text) # 11. Nettoyer ligne par ligne lines = [] for line in text.splitlines(): line = line.strip() # Ignorer les lignes trop courtes ou parasites if len(line) < 2: continue # Ignorer les lignes qui ne sont que des chiffres/ponctuation if re.match(r'^[\d\s\.\,\;\:\!\?\-\|\=\[\]\(\)]+$', line): continue lines.append(line) text = '\n'.join(lines) # 12. Normaliser les espaces multiples text = cls.MULTI_SPACE.sub(' ', text) # 13. Supprimer les lignes vides multiples text = cls.MULTI_NEWLINE.sub('\n\n', text) return text.strip() @classmethod def extract_clean_sentences(cls, text: str, min_length: int = 30) -> str: """ Extrait uniquement les phrases valides du texte. Args: text: Texte à filtrer min_length: Longueur minimale d'une phrase valide Returns: Texte composé uniquement de phrases valides """ # D'abord nettoyer text = cls.deep_clean(text) # Filtrer par longueur de paragraphe paragraphs = text.split('\n\n') valid = [ p.strip() for p in paragraphs if len(p.strip()) >= min_length ] return '\n\n'.join(valid) class ContentCleaner: """Nettoyeur de contenu HTML ultra-performant.""" WHITESPACE_PATTERN = re.compile(r'\s+') UNWANTED_TAGS = { 'script', 'style', 'noscript', 'iframe', 'embed', 'object', 'svg', 'canvas', 'applet', 'link', 'meta', 'head', 'nav', 'footer', 'aside', 'advertisement', '.mw-editsection', # Wikipedia edit links '.reference', # Wikipedia references '.reflist', # Wikipedia reference lists '.navbox', # Wikipedia navigation boxes '.toc', # Wikipedia table of contents '.hatnote', # Wikipedia hatnotes '.mbox-small', # Wikipedia message boxes '.sidebar', '.noprint', '#coordinates', } UNWANTED_CSS_SELECTORS = [ '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', ] @staticmethod def clean_html_fast(html: str) -> str: """ Nettoyage ultra-rapide avec selectolax. Supprime les tags indésirables avant extraction. Args: html: HTML brut Returns: HTML nettoyé """ try: tree = HTMLParser(html) # Supprimer les sélecteurs CSS spécifiques (Wikipedia, etc.) for selector in ContentCleaner.UNWANTED_CSS_SELECTORS: try: for node in tree.css(selector): node.decompose() except Exception: pass # Supprimer les tags génériques indésirables simple_tags = { 'script', 'style', 'noscript', 'iframe', 'embed', 'object', 'svg', 'canvas', 'applet', 'head' } for tag in simple_tags: for node in tree.css(tag): node.decompose() return tree.html or html except Exception as e: logger.warning(f"Erreur nettoyage selectolax: {e}") return html @staticmethod def extract_main_content(html: str, url: str) -> Dict[str, Any]: """ Extraction intelligente du contenu principal. Pipeline multi-méthodes avec nettoyage profond. Args: html: HTML source url: URL de la page Returns: Dict avec le contenu extrait et métadonnées """ result = { 'text': '', 'title': '', 'author': '', 'date': None, 'description': '', 'language': 'unknown', 'method': 'unknown' } # Pré-nettoyage HTML (supprime les sections de référence Wikipedia, etc.) clean_html = ContentCleaner.clean_html_fast(html) try: # Méthode 1: Trafilatura (SOTA en précision) extracted = bare_extraction( clean_html, url=url, include_comments=False, include_tables=False, # Désactivé : génère du bruit include_images=False, # Désactivé : génère du bruit include_links=False, # Désactivé : génère du bruit deduplicate=True, favor_precision=True, no_fallback=False ) if extracted and extracted.get('text') and len(extracted['text']) > 100: raw_text = extracted['text'] # Nettoyage profond du texte extrait clean_text = TextCleaner.deep_clean(raw_text) result.update({ 'text': clean_text, 'title': extracted.get('title', ''), 'author': extracted.get('author', ''), 'date': str(extracted.get('date', '')) if extracted.get('date') else None, 'description': extracted.get('description', ''), 'method': 'trafilatura' }) # Détection de langue sur le texte propre try: if result['text']: langs = detect_langs(result['text'][:500]) result['language'] = langs[0].lang if langs else 'unknown' except Exception: pass return result except Exception as e: logger.debug(f"Trafilatura failed: {e}") try: # Méthode 2: Readability (fallback) doc = Document(clean_html) raw_text = remove_tags(doc.summary()) clean_text = TextCleaner.deep_clean(raw_text) if clean_text and len(clean_text) > 50: result.update({ 'text': clean_text, 'title': doc.title(), 'method': 'readability' }) return result except Exception as e: logger.debug(f"Readability failed: {e}") try: # Méthode 3: Justext (dernier recours) 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 ) paragraphs_text = [p.text for p in paragraphs if not p.is_boilerplate] raw_text = '\n\n'.join(paragraphs_text) clean_text = TextCleaner.deep_clean(raw_text) if clean_text: result.update({ 'text': clean_text, 'method': 'justext' }) return result except Exception as e: logger.debug(f"Justext failed: {e}") # Fallback final raw_text = remove_tags(clean_html) result['text'] = TextCleaner.deep_clean(raw_text) result['method'] = 'basic' return result @staticmethod def normalize_text(text: str) -> str: """ Normalisation finale du texte. Appelé après extract_main_content pour une passe supplémentaire. Args: text: Texte extrait Returns: Texte normalisé """ return TextCleaner.deep_clean(text) @staticmethod def extract_metadata(html: str) -> Dict[str, Any]: """ Extraction des métadonnées (Open Graph, Twitter Cards). Args: html: HTML source Returns: Dict de métadonnées """ metadata: Dict[str, Any] = {} try: tree = HTMLParser(html) # Open Graph 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 # Twitter Cards 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 # Meta standards 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 URL canonical = tree.css_first('link[rel="canonical"]') if canonical: href = canonical.attributes.get('href', '').strip() if href: metadata['canonical'] = href # Title tag 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 e: logger.warning(f"Erreur extraction métadonnées: {e}") return metadata @staticmethod def extract_links(html: str, base_url: str) -> List[Dict[str, str]]: """ Extraction optimisée des liens avec résolution des URLs relatives. Args: html: HTML source base_url: URL de base Returns: Liste de liens avec métadonnées """ links = [] seen_urls: set = set() try: tree = HTMLParser(html) for link in tree.css('a[href]'): href = link.attributes.get('href', '').strip() text = link.text(strip=True) if not href or href.startswith(('#', 'javascript:', 'mailto:', 'tel:')): continue try: absolute_url = urljoin(base_url, href) except Exception: continue # Déduplication if absolute_url in seen_urls: continue seen_urls.add(absolute_url) # Valider que c'est une URL HTTP parsed = urlparse(absolute_url) if parsed.scheme not in ('http', 'https'): continue links.append({ 'url': absolute_url, 'text': text[:200] if text else '', 'rel': link.attributes.get('rel', ''), 'title': link.attributes.get('title', '') }) except Exception as e: logger.warning(f"Erreur extraction liens: {e}") return links @staticmethod def extract_images(html: str, base_url: str) -> List[Dict[str, str]]: """ Extraction des images avec support lazy loading et srcset. Args: html: HTML source base_url: URL de base Returns: Liste d'images avec métadonnées """ images = [] seen_urls: set = set() try: tree = HTMLParser(html) for img in tree.css('img'): src = ( img.attributes.get('src') or img.attributes.get('data-src') or img.attributes.get('data-lazy-src') or img.attributes.get('data-original') or img.attributes.get('data-lazy') or '' ) src = src.strip() if not src or src.startswith('data:'): continue try: absolute_url = urljoin(base_url, src) except Exception: continue parsed = urlparse(absolute_url) if parsed.scheme not in ('http', 'https'): continue if absolute_url in seen_urls: continue seen_urls.add(absolute_url) images.append({ 'url': absolute_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 e: logger.warning(f"Erreur extraction images: {e}") return images @staticmethod def compute_content_hash(text: str) -> str: """ Calcule un hash SHA256 du contenu pour la déduplication. Args: text: Texte du contenu Returns: Hash SHA256 hexadécimal """ return hashlib.sha256(text.encode('utf-8')).hexdigest() class URLValidator: """Validateur d'URL avec filtres avancés.""" BLOCKED_EXTENSIONS = { '.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 = {'http', 'https'} @staticmethod def is_valid_url(url: str) -> bool: """Vérifie si l'URL est valide et scrapable.""" try: parsed = urlparse(url) if parsed.scheme not in URLValidator.ALLOWED_SCHEMES: return False path_lower = parsed.path.lower() if any(path_lower.endswith(ext) for ext in URLValidator.BLOCKED_EXTENSIONS): return False if not parsed.netloc: return False return True except Exception: return False @staticmethod def normalize_url(url: str) -> str: """Normalise l'URL (supprimer fragments, trier params).""" try: from urllib.parse import urlunparse, parse_qs, urlencode parsed = urlparse(url) parsed = parsed._replace(fragment='') if parsed.query: params = parse_qs(parsed.query) sorted_params = sorted(params.items()) new_query = urlencode(sorted_params, doseq=True) parsed = parsed._replace(query=new_query) return urlunparse(parsed) except Exception: return url