"""Reading feeds: entries into items, the whole article from its page, finding a site's feed. Downloading the feed itself is podcasts.views._fetch_feed (ETag, body hash, redirects), shared with the podcast refresh.""" import calendar import hashlib import html import logging import re from datetime import datetime, timezone as dt_timezone from html.parser import HTMLParser from urllib.parse import urljoin, urlparse import requests from django.utils import timezone from podcasts.shownotes import _Sanitizer, html_to_text log = logging.getLogger(__name__) USER_AGENT = 'Mozilla/5.0 (compatible; diora feed reader; +https://diora.creamfresh.xyz)' PAGE_TIMEOUT = 15 # Shorter text than this in the feed counts as a teaser: the article is fetched ("auto"). TEASER_CHARS = 700 SUMMARY_CHARS = 400 class _AbsoluteSanitizer(_Sanitizer): """The podcast sanitizer, with links and images made absolute first (article HTML is full of relative ones, which the allowlist would drop).""" def __init__(self, base): super().__init__() self.base = base def _start(self, tag, attrs, closed): if tag == 'img': # Lazy-loaded images keep the real address in data-src, src is a placeholder. a = dict(attrs) real = a.get('data-src') or a.get('data-original') if real: attrs = [(n, v) for n, v in attrs if n not in ('src', 'data-src', 'data-original')] + [('src', real)] fixed = [] for name, value in attrs: if name in ('href', 'src') and value and not value.strip().startswith(('data:', 'javascript:', '#')): value = urljoin(self.base, value.strip()) fixed.append((name, value)) super()._start(tag, fixed, closed) def clean_html(text, base=''): if not text: return '' s = _AbsoluteSanitizer(base) s.feed(text) return s.result().strip() def plain(text): return re.sub(r'\s+', ' ', html_to_text(text or '')).strip() def _short(text, n=SUMMARY_CHARS): if len(text) <= n: return text cut = text[:n].rsplit(' ', 1)[0] return cut + ' …' def _when(entry): """The entry's time (published, else updated), never in the future, or None.""" now = timezone.now() for key in ('published_parsed', 'updated_parsed', 'created_parsed'): t = entry.get(key) if t: try: when = datetime.fromtimestamp(calendar.timegm(t), dt_timezone.utc) except (OverflowError, ValueError, TypeError): continue return min(when, now) return None class _FirstImage(HTMLParser): def __init__(self): super().__init__() self.src = '' def handle_starttag(self, tag, attrs): if tag == 'img' and not self.src: a = dict(attrs) w, h = a.get('width', ''), a.get('height', '') if w in ('1', '0') or h in ('1', '0'): return # tracking pixel self.src = a.get('src') or '' def _image(entry, content_html, base): for m in entry.get('media_thumbnail') or []: if m.get('url'): return m['url'] for m in entry.get('media_content') or []: if m.get('url') and (m.get('medium') == 'image' or (m.get('type') or '').startswith('image/')): return m['url'] for enc in entry.get('enclosures') or []: if (enc.get('type') or '').startswith('image/') and enc.get('href'): return enc['href'] p = _FirstImage() try: p.feed(content_html or '') except Exception: pass return urljoin(base, p.src) if p.src else '' def guid_hash(entry): key = entry.get('id') or entry.get('link') or f"{entry.get('title', '')}|{entry.get('published', '')}" return hashlib.sha256(key.encode('utf-8', 'replace')).hexdigest() def parse_entry(entry, feed_url): """A feedparser entry -> the fields of an Item (without source/user).""" link = entry.get('link') or '' base = link or feed_url contents = [c.get('value') or '' for c in entry.get('content') or []] content_html = max(contents, key=len) if contents else '' summary_html = entry.get('summary') or '' if len(plain(summary_html)) > len(plain(content_html)): content_html = summary_html content = clean_html(content_html, base) text = plain(content) summary = plain(summary_html) or text title = plain(html.unescape(entry.get('title') or '')) or _short(text, 80) or 'Ohne Titel' when = _when(entry) return { 'guid_hash': guid_hash(entry), 'url': link[:2000], 'title': title[:500], 'author': plain(entry.get('author') or '')[:300], 'published': when or timezone.now(), 'dated': when is not None, # not an Item field: refresh._apply uses and drops it 'summary': _short(summary), 'content': content, 'words': word_count(content), 'image_url': _image(entry, content_html, base)[:2000], } def word_count(content_html): return len(plain(content_html).split()) def is_teaser(content_html): return len(plain(content_html)) < TEASER_CHARS def _drop_title(html_text): """trafilatura repeats the headline as the first