Spaces:
Sleeping
Sleeping
| """ | |
| 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 :  | |
| 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) | |
| 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  | |
| 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() | |
| 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', | |
| ] | |
| 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 | |
| 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 | |
| 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) | |
| 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 | |
| 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 | |
| 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 | |
| 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'} | |
| 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 | |
| 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 |