mrwnslz
69442f4a4d
TranscriptionSettings (Token Fernet-verschluesselt wie SiteLogin), TranscriptionJob und EpisodeArticle in podcasts; Source.synthetic in feeds, das Refresh, Aufraeumen, OPML-Export, Kommando und Umstellen auf Nachrichten ueberspringt bzw. verweigert. GET/POST /podcasts/transcription/settings/ und POST /podcasts/transcription/test/ (SSRF-Pruefung ueber chapters._open, jetzt auch fuer POST; eingebaute Instanz aus DIORA_WHISPER_URL ausgenommen). Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
278 lines
12 KiB
Python
278 lines
12 KiB
Python
"""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, logins
|
|
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
|
|
NOTIFY_MAX_AGE = timedelta(days=2) # older posts are never announced (feeds dig up old ones)
|
|
|
|
|
|
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
|
|
# The bell: never for what a first fetch brings, never for old posts.
|
|
item.notify = source.notify and not first and not undated and fields['published'] >= now - NOTIFY_MAX_AGE
|
|
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
|
|
if item.paywalled:
|
|
# Only the free part came: again once there's a working subscriber login.
|
|
return logins.has_login(item.user_id, item.url)
|
|
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."""
|
|
http = logins.session_for(item.user_id, item.url)
|
|
got = fetch.fetch_article(item.url, http)
|
|
if got and got[2] and http:
|
|
# Logged in, still the paywall: the session ran out, log in anew once.
|
|
http = logins.session_for(item.user_id, item.url, fresh=True)
|
|
if http:
|
|
got = fetch.fetch_article(item.url, http)
|
|
_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, paywalled = got
|
|
longer = len(fetch.plain(content)) > len(fetch.plain(item.content))
|
|
if paywalled:
|
|
# The free part, if it's more than the feed's teaser; the rest waits for a login.
|
|
item.paywalled = True
|
|
if longer:
|
|
item.content = content
|
|
item.words = fetch.word_count(content)
|
|
if not item.image_url and image:
|
|
item.image_url = image[:2000]
|
|
item.save(update_fields=['paywalled', 'content', 'words', 'image_url'])
|
|
return
|
|
if not longer:
|
|
item.fulltext_failed = True # the feed had more
|
|
item.save(update_fields=['fulltext_failed'])
|
|
return
|
|
item.content = content
|
|
item.words = fetch.word_count(content)
|
|
item.content_full = True
|
|
item.paywalled = False
|
|
if not item.image_url and image:
|
|
item.image_url = image[:2000]
|
|
item.save(update_fields=['content', 'words', 'content_full', 'paywalled', '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
|
|
# Logged-in sessions for paywalled sites, one per account and site (made here, not in the threads).
|
|
sessions = {}
|
|
for i in todo:
|
|
adapter = logins.adapter_for(i.url)
|
|
if adapter and (i.user_id, adapter.domain) not in sessions:
|
|
sessions[(i.user_id, adapter.domain)] = logins.session_for(i.user_id, i.url)
|
|
|
|
def http_for(i):
|
|
adapter = logins.adapter_for(i.url)
|
|
return sessions.get((i.user_id, adapter.domain)) if adapter else None
|
|
|
|
pool = ThreadPoolExecutor(ARTICLE_WORKERS)
|
|
futures = {pool.submit(fetch.fetch_article, i.url, http_for(i)): 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:
|
|
i = futures[f]
|
|
try:
|
|
got = f.result()
|
|
if got and got[2] and http_for(i) is not None:
|
|
# Logged in and still the paywall: next time with a fresh login.
|
|
logins.forget(i.user_id, logins.adapter_for(i.url).domain)
|
|
_store_article(i, got)
|
|
except Exception:
|
|
log.exception('article for item %s', i.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, synthetic=False):
|
|
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 = [s for s in sources if not s.synthetic] # synthetic sources have no feed to fetch
|
|
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)
|