import ipaddress import re import socket from urllib.parse import urlparse import requests from bs4 import BeautifulSoup from django.conf import settings BLOCKED_HOSTNAMES = {'localhost', '127.0.0.1', '0.0.0.0', '::1'} MAX_DOWNLOAD_BYTES = getattr(settings, 'ARTICLE_FETCH_MAX_BYTES', 2_000_000) TARGET_TEXT_CHARS = getattr(settings, 'ARTICLE_FETCH_TARGET_CHARS', 8_000) MAX_OUTPUT_CHARS = getattr(settings, 'ARTICLE_FETCH_OUTPUT_CHARS', 12_000) PARTIAL_PARSE_BYTES = getattr(settings, 'ARTICLE_FETCH_PARTIAL_BYTES', 96_000) FETCH_TIMEOUT = getattr(settings, 'ARTICLE_FETCH_TIMEOUT', 15) USER_AGENT = 'FaktenkompassBot/1.0 (+https://faktenkompass.local)' REMOVE_TAGS = { 'script', 'style', 'nav', 'footer', 'header', 'aside', 'noscript', 'iframe', 'svg', 'form', 'button', 'input', 'select', 'textarea', 'menu', 'dialog', 'figure', 'picture', 'video', 'audio', 'canvas', 'embed', 'object', } NOISE_PATTERN = re.compile( r'comment|sidebar|related|newsletter|cookie|banner|social|share|widget|' r'menu|breadcrumb|promo|advert|tracking|popup|modal|consent|paywall|' r'recommend|trending|tag-cloud|author-box|meta-|sharing|outbrain|taboola', re.IGNORECASE, ) MIN_PARAGRAPH_LENGTH = 40 class ArticleFetchError(Exception): pass def _hostname_resolves_to_private_ip(hostname): try: infos = socket.getaddrinfo(hostname, None) except socket.gaierror as exc: raise ArticleFetchError(f'Domain konnte nicht aufgelöst werden: {hostname}') from exc for info in infos: ip = ipaddress.ip_address(info[4][0]) if ( ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_reserved or ip.is_multicast ): return True return False def validate_article_url(url): parsed = urlparse(url.strip()) if parsed.scheme not in {'http', 'https'}: raise ArticleFetchError('Nur http- und https-URLs sind erlaubt.') if not parsed.netloc: raise ArticleFetchError('Ungültige URL.') hostname = parsed.hostname or '' if hostname.lower() in BLOCKED_HOSTNAMES: raise ArticleFetchError('Diese URL ist nicht erlaubt.') if _hostname_resolves_to_private_ip(hostname): raise ArticleFetchError('Interne oder private URLs sind aus Sicherheitsgründen blockiert.') def _remove_noise_elements(soup): for tag_name in REMOVE_TAGS: for tag in soup.find_all(tag_name): tag.decompose() to_remove = [] for element in soup.find_all(True): element_attrs = getattr(element, 'attrs', None) or {} role = element_attrs.get('role') if role in {'navigation', 'banner', 'complementary', 'contentinfo'}: to_remove.append(element) continue class_val = element_attrs.get('class', []) if isinstance(class_val, list): class_str = ' '.join(class_val) else: class_str = str(class_val or '') id_str = str(element_attrs.get('id', '') or '') combined = f'{class_str} {id_str}'.strip() if combined and NOISE_PATTERN.search(combined): to_remove.append(element) for element in to_remove: element.decompose() def _find_content_root(soup): for selector in ( ('article', {}), ('main', {}), ('div', {'role': 'main'}), ('div', {'class': re.compile(r'article|content|post|entry|story', re.I)}), ): tag, attrs = selector match = soup.find(tag, attrs=attrs) if attrs else soup.find(tag) if match: return match return soup.body or soup def _collect_paragraphs(root): paragraphs = [] seen = set() for element in root.find_all(['h1', 'h2', 'h3', 'p', 'li', 'blockquote']): text = element.get_text(' ', strip=True) text = re.sub(r'\s+', ' ', text) if len(text) < MIN_PARAGRAPH_LENGTH: continue key = text.lower() if key in seen: continue seen.add(key) paragraphs.append(text) return paragraphs def _extract_readable_text(html): soup = BeautifulSoup(html, 'html.parser') _remove_noise_elements(soup) root = _find_content_root(soup) paragraphs = _collect_paragraphs(root) if not paragraphs: fallback = root.get_text('\n', strip=True) paragraphs = [] for line in fallback.splitlines(): line = re.sub(r'\s+', ' ', line.strip()) if len(line) >= MIN_PARAGRAPH_LENGTH: paragraphs.append(line) content = '\n\n'.join(paragraphs) content = re.sub(r'\n{3,}', '\n\n', content) return content.strip() def _finalize_text(text, *, partial=False): if len(text) < 100: raise ArticleFetchError('Aus dem Artikel konnte kein ausreichender Text extrahiert werden.') trimmed = text[:MAX_OUTPUT_CHARS] if partial and len(text) > MAX_OUTPUT_CHARS: trimmed += '\n\n[… Text gekürzt …]' return trimmed def _decode_html(chunks, encoding): return b''.join(chunks).decode(encoding or 'utf-8', errors='replace') def fetch_article_text(url): validate_article_url(url) try: response = requests.get( url, timeout=FETCH_TIMEOUT, headers={'User-Agent': USER_AGENT}, allow_redirects=True, stream=True, ) except requests.RequestException as exc: raise ArticleFetchError(f'Artikel konnte nicht geladen werden: {exc}') from exc if response.status_code >= 400: raise ArticleFetchError(f'Artikel nicht erreichbar (HTTP {response.status_code}).') final_host = urlparse(response.url).hostname or '' if _hostname_resolves_to_private_ip(final_host): raise ArticleFetchError('Weiterleitung auf eine blockierte URL.') chunks = [] size = 0 encoding = response.encoding truncated_download = False for chunk in response.iter_content(chunk_size=8192): if not chunk: continue size += len(chunk) if size > MAX_DOWNLOAD_BYTES: truncated_download = True break chunks.append(chunk) if size >= PARTIAL_PARSE_BYTES and size % PARTIAL_PARSE_BYTES < 8192: text = _extract_readable_text(_decode_html(chunks, encoding)) if len(text) >= TARGET_TEXT_CHARS: return _finalize_text(text, partial=True) if not chunks: raise ArticleFetchError('Artikel konnte nicht geladen werden.') text = _extract_readable_text(_decode_html(chunks, encoding)) return _finalize_text(text, partial=truncated_download)