diora-web/feeds/fetch.py

"""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 <h1>; the reader shows it already."""
    return re.sub(r'^\s*(<img[^>]*>\s*)?<h1>.*?</h1>', 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*<html>\s*<body>|</body>\s*</html>\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 ''