"""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, 'image_url': _image(entry, content_html, base)[:2000], } def is_teaser(content_html): return len(plain(content_html)) < TEASER_CHARS def _drop_title(html_text): """trafilatura repeats the headline as the first

; the reader shows it already.""" return re.sub(r'^\s*(]*>\s*)?

.*?

', lambda m: m.group(1) or '', html_text, count=1, flags=re.S) def fetch_article(url, http=None): """The article on `url` as (clean HTML, image URL, paywalled), or None. trafilatura finds the main text and leaves out navigation, ads, comments. `http`: a logged-in session for a paywalled site (logins.py); paywalled says the page showed only the free part anyway.""" import trafilatura # heavy import, only where articles are fetched from .logins import adapter_for try: r = (http or requests).get(url, timeout=PAGE_TIMEOUT, headers={'User-Agent': USER_AGENT}) except requests.RequestException as e: log.info('article %s: %s', url, type(e).__name__) return None if r.status_code >= 400 or 'html' not in r.headers.get('Content-Type', 'text/html'): return None page = r.text adapter = adapter_for(r.url) paywalled = bool(adapter and adapter.paywalled(page)) try: body = trafilatura.extract(page, url=r.url, output_format='html', include_images=True, include_links=True, include_tables=True, favor_precision=True) except Exception: log.exception('article %s: extraction failed', url) return None if not body: return None body = re.sub(r'^\s*\s*|\s*\s*$', '', body.strip()) content = clean_html(_drop_title(body), r.url) if len(plain(content)) < 200 and not paywalled: return None image = '' try: meta = trafilatura.extract_metadata(page, default_url=r.url) image = (meta.image if meta else '') or '' except Exception: pass return content, image, paywalled # ---- finding a site's feed FEED_TYPES = ('application/rss+xml', 'application/atom+xml', 'application/feed+json', 'application/xml', 'text/xml') class _FeedLinks(HTMLParser): def __init__(self): super().__init__() self.feeds = [] self.icon = '' def handle_starttag(self, tag, attrs): if tag != 'link': return a = {k.lower(): (v or '') for k, v in attrs} rel = a.get('rel', '').lower().split() if 'alternate' in rel and a.get('type', '').lower() in FEED_TYPES and a.get('href'): self.feeds.append((a['href'], a.get('title', ''))) if ('icon' in rel or 'apple-touch-icon' in rel) and a.get('href') and not self.icon: self.icon = a['href'] def discover(page_html, base): """[(feed url, title)] a page links to, and its icon URL.""" p = _FeedLinks() try: p.feed(page_html) except Exception: pass seen, out = set(), [] for href, title in p.feeds: u = urljoin(base, href) if u not in seen: seen.add(u) out.append((u, title)) return out, (urljoin(base, p.icon) if p.icon else '') def short_title(title): """A feed's name without the slogan: "tagesschau.de - die erste Adresse für …" -> "tagesschau.de", "Aktuell - FAZ.NET" -> "FAZ.NET", "DIE ZEIT | Nachrichten, …" -> "DIE ZEIT".""" parts = [p.strip() for p in re.split(r'\s+[-–—|:]\s+', title or '') if p.strip()] if len(parts) < 2: return (title or '').strip() branded = [p for p in parts if re.search(r'\w\.[a-z]{2,}$', p, re.I)] return (branded or parts)[0] def site_icon(site_url): if not site_url: return '' u = urlparse(site_url) return f'{u.scheme}://{u.netloc}/favicon.ico' if u.scheme and u.netloc else ''