diora-web/feeds/refresh.py

"""Refreshing sources: new entries become items, teasers get their article,
news is bundled into stories, old news expires."""

import logging
import time
from collections import Counter
from concurrent.futures import FIRST_COMPLETED, ThreadPoolExecutor, as_completed, wait
from datetime import timedelta

from django.db import IntegrityError, transaction
from django.utils import timezone

from podcasts.views import FeedFetchError, _clear_error, _error_text, _fetch_feed

from . import fetch
from .cluster import Clusterer
from .models import FeedSettings, Item, Source, Story

log = logging.getLogger(__name__)

FEED_WORKERS = 12
ARTICLE_WORKERS = 8
ARTICLES_PER_RUN = 80        # articles fetched per refresh at most; the rest when opened
BLOG_KEEP = 1000             # items kept per blog
BLOG_FIRST_UNREAD_DAYS = 7   # a new blog's older articles start out read
UNDATED_FIRST_UNREAD = 3     # of a new feed without dates, only the first few
GONE_KEEP_DAYS = 30          # expired news rows kept (without text) against re-adding


def settings_for(user):
    s, _ = FeedSettings.objects.get_or_create(user=user)
    return s


def _apply(source, result, now, keep_days):
    """Write one download; returns the new items."""
    if result.status == 304:
        source.last_refreshed_at = now
        _clear_error(source)
        source.save(update_fields=['last_refreshed_at', 'last_error', 'last_error_at', 'error_count',
                                   'http_etag', 'http_last_modified'])
        return []
    parsed = result.parsed
    feed = parsed.feed
    if feed.get('title') and (not source.title or source.title == source.url):
        source.title = fetch.short_title(fetch.plain(feed.title))[:300]
    if feed.get('link'):
        source.site_url = feed.link[:1000]
    if not source.icon_url:
        icon = (feed.get('image') or {}).get('href') or feed.get('icon') or fetch.site_icon(source.site_url)
        source.icon_url = (icon or '')[:1000]
    if result.permanent_url and result.permanent_url != source.url and \
            not Source.objects.filter(user=source.user, url=result.permanent_url).exists():
        source.url = result.permanent_url[:1000]
    source.http_etag = result.etag
    source.http_last_modified = result.last_modified
    source.http_body_hash = result.body_hash
    source.last_refreshed_at = now
    _clear_error(source)
    first = not source.items.exists()
    source.save()

    known = set(source.items.values_list('guid_hash', flat=True))
    oldest_kept = None
    if source.kind == Source.BLOG and len(known) >= BLOG_KEEP:
        oldest_kept = source.items.order_by('published').values_list('published', flat=True).first()
    news_cutoff = now - timedelta(days=keep_days)
    new = []
    entries = [fetch.parse_entry(e, source.url) for e in parsed.entries]
    # Some feeds (Deutschlandfunk) give every entry the time the feed was made:
    # then the dates say nothing, and the time an entry first shows up counts.
    stamps = Counter(f['published'] for f in entries if f['dated'])
    bogus_dates = len(entries) >= 5 and bool(stamps) and stamps.most_common(1)[0][1] >= 0.8 * len(entries)
    for n, fields in enumerate(entries):
        dated = fields.pop('dated')
        undated = bogus_dates or not dated
        if undated:
            # First time: the feed's order stands for the time (newest first), so the
            # list doesn't fill up with "just now"; afterwards, when it showed up.
            fields['published'] = now - timedelta(minutes=n) if first else now
            fields['undated'] = first and n >= UNDATED_FIRST_UNREAD
        if fields['guid_hash'] in known:
            continue
        known.add(fields['guid_hash'])
        if source.kind == Source.NEWS and fields['published'] < news_cutoff:
            continue
        if oldest_kept and fields['published'] < oldest_kept:
            continue
        old_undated = fields.pop('undated', False)
        item = Item(source=source, user=source.user, kind=source.kind, **fields)
        if source.kind == Source.BLOG and first and (
                old_undated or fields['published'] < now - timedelta(days=BLOG_FIRST_UNREAD_DAYS)):
            item.read_at = now
        new.append(item)
    if new:
        try:
            with transaction.atomic():
                Item.objects.bulk_create(new)
        except IntegrityError:
            # Two entries of one feed with the same id: one by one, skipping the doubles.
            kept = []
            for item in new:
                try:
                    with transaction.atomic():
                        item.save()
                    kept.append(item)
                except IntegrityError:
                    pass
            new = kept
    return new


def needs_article(item, source):
    if not item.url or item.content_full or item.fulltext_failed:
        return False
    if source.fulltext == Source.FULLTEXT_NEVER:
        return False
    return source.fulltext == Source.FULLTEXT_ALWAYS or fetch.is_teaser(item.content)


def load_article(item):
    """Fetch the article for one item now (when opened); saves it."""
    got = fetch.fetch_article(item.url)
    _store_article(item, got)
    return item


def _store_article(item, got):
    if got is None:
        item.fulltext_failed = True
        item.save(update_fields=['fulltext_failed'])
        return
    content, image = got
    if len(fetch.plain(content)) <= len(fetch.plain(item.content)):
        item.fulltext_failed = True  # the feed had more
        item.save(update_fields=['fulltext_failed'])
        return
    item.content = content
    item.content_full = True
    if not item.image_url and image:
        item.image_url = image[:2000]
    item.save(update_fields=['content', 'content_full', 'image_url'])


def _articles(items, deadline):
    todo = [i for i in items if needs_article(i, i.source)][:ARTICLES_PER_RUN]
    if not todo:
        return
    pool = ThreadPoolExecutor(ARTICLE_WORKERS)
    futures = {pool.submit(fetch.fetch_article, i.url): i for i in todo}
    pending = set(futures)
    try:
        while pending:
            left = None if deadline is None else deadline - time.monotonic()
            if left is not None and left <= 0:
                break
            done, pending = wait(pending, timeout=left, return_when=FIRST_COMPLETED)
            for f in done:
                try:
                    _store_article(futures[f], f.result())
                except Exception:
                    log.exception('article for item %s', futures[f].pk)
    finally:
        pool.shutdown(wait=False, cancel_futures=True)


def expire(user, now=None):
    """News past the user's keep time loses its text (rows stay GONE_KEEP_DAYS
    as markers), stories without visible items go."""
    now = now or timezone.now()
    keep = settings_for(user).news_keep_days
    Item.objects.filter(user=user, kind=Source.NEWS, gone=False, starred=False,
                        published__lt=now - timedelta(days=keep)).update(
        gone=True, content='', summary='', story=None)
    Item.objects.filter(user=user, gone=True, published__lt=now - timedelta(days=GONE_KEEP_DAYS)).delete()
    Story.objects.filter(user=user).exclude(items__gone=False).delete()
    # Blogs: the oldest read articles beyond BLOG_KEEP per blog.
    for source in Source.objects.filter(user=user, kind=Source.BLOG):
        ids = list(source.items.order_by('-published').values_list('id', flat=True)[BLOG_KEEP:])
        if ids:
            Item.objects.filter(id__in=ids, starred=False).exclude(read_at=None).delete()


def refresh(sources, deadline=None):
    """Refresh `sources` (downloads in parallel, writes one by one).
    Returns (refreshed, failed, new items)."""
    sources = list(sources)
    now = timezone.now()
    keep = {}
    refreshed = failed = 0
    new_items = []
    pool = ThreadPoolExecutor(FEED_WORKERS)
    try:
        futures = {
            pool.submit(_fetch_feed, s.url, s.http_etag, s.http_last_modified, s.http_body_hash,
                        user_agent=fetch.USER_AGENT): s
            for s in sources
        }
        try:
            timeout = None if deadline is None else max(0.0, deadline - time.monotonic())
            for f in as_completed(futures, timeout=timeout):
                s = futures[f]
                try:
                    result = f.result()
                    if s.user_id not in keep:
                        keep[s.user_id] = settings_for(s.user).news_keep_days
                    new_items += _apply(s, result, now, keep[s.user_id])
                    refreshed += 1
                except Exception as e:
                    if not isinstance(e, FeedFetchError):
                        log.exception('feed source %s', s.pk)
                    s.last_error = _error_text(e)
                    s.last_error_at = now
                    s.error_count += 1
                    s.save(update_fields=['last_error', 'last_error_at', 'error_count'])
                    failed += 1
        except TimeoutError:
            pass
    finally:
        # Past the deadline: don't wait for the slow ones (they stay as they are).
        pool.shutdown(wait=False, cancel_futures=True)

    for item in new_items:
        item.source = next(s for s in sources if s.pk == item.source_id)
    _articles(new_items, deadline)

    by_user = {}
    for item in new_items:
        if item.kind == Source.NEWS:
            by_user.setdefault(item.user_id, []).append(item)
    for user_id, items in by_user.items():
        c = Clusterer(items[0].user, now)
        for item in sorted(items, key=lambda i: i.published):
            c.add(item)
    for user_id in {s.user_id for s in sources}:
        expire(next(s.user for s in sources if s.user_id == user_id), now)
    return refreshed, failed, len(new_items)