Faktenkompass: UI-Rebrand, Admin-Assistent und Produktions-Setup.
Modernisiert die öffentliche Oberfläche mit Inter, Dark Mode und Live-Suche, ergänzt KI-gestützte Entwürfe im Admin mit Nachweis-Uploads und dokumentiert Daphne/Nginx für den Serverbetrieb. Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
Pirmin Hinderling (fedora)
@@ -0,0 +1,203 @@
|
||||
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)
|
||||
Reference in New Issue
Block a user