Faktenkompass/claims/services/article_fetcher.py

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)