Podcasts: Feeds bedingt abrufen, Fehler speichern, Umzüge folgen, Backoff
- Bedingter Abruf: ETag/Last-Modified werden nach einer 200 gespeichert und als If-None-Match/If-Modified-Since mitgeschickt. Bei 304 wird nichts geparst und keine Folge geschrieben, nur last_refreshed_at aktualisiert und der Fehler gelöscht. Teilen sich mehrere Abos eine URL, gehen die Validatoren nur mit, wenn alle schon einmal abgerufen wurden und dieselben haben. - Fehler (Netz, HTTP >= 400, kein Feed) landen in last_error, last_error_at und error_count; ein erfolgreicher Abruf setzt sie zurück. Feed-Liste und /podcasts/state/ liefern die drei Felder mit. - Dauerhafte Weiterleitungen (301/308 am Anfang der Kette) ändern rss_url, außer der Nutzer hat die neue URL schon abonniert; 302/307 ändern nichts. - Backoff im stündlichen Job: ab 3 Fehlern in Folge nur alle 2, 4, 8, 16, höchstens 24 Stunden. „Alle aktualisieren“ und --feed-id versuchen es trotzdem (max_age gilt weiter). - _fetch_feed liefert jetzt ein FeedFetch-Objekt bzw. wirft FeedFetchError. - Migration 0007_feed_http_cache. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
parent
f9b3a0ce04
commit
bbe5f6c5c2
5 changed files with 596 additions and 29 deletions
|
|
@ -1,16 +1,15 @@
|
|||
import sys
|
||||
|
||||
from django.core.management.base import BaseCommand
|
||||
from django.utils import timezone
|
||||
|
||||
from podcasts.models import PodcastFeed
|
||||
from podcasts.views import _refresh_feeds
|
||||
from podcasts.views import _in_backoff_q, _refresh_feeds
|
||||
|
||||
|
||||
class Command(BaseCommand):
|
||||
help = 'Refresh podcast feeds (fetch new episodes from RSS)'
|
||||
|
||||
def add_arguments(self, parser):
|
||||
parser.add_argument('--feed-id', type=int, help='Refresh only this feed ID')
|
||||
parser.add_argument('--feed-id', type=int, help='Refresh only this feed ID (even when backing off)')
|
||||
parser.add_argument('--limit', type=int, default=0, help='Max number of feeds to refresh')
|
||||
|
||||
def handle(self, *args, **options):
|
||||
|
|
@ -18,6 +17,10 @@ class Command(BaseCommand):
|
|||
|
||||
if options['feed_id']:
|
||||
qs = qs.filter(pk=options['feed_id'])
|
||||
else:
|
||||
# A feed failing 3+ times in a row is tried every 2, 4, 8, 16,
|
||||
# at most 24 hours instead of every hour.
|
||||
qs = qs.exclude(_in_backoff_q(timezone.now()))
|
||||
|
||||
if options['limit']:
|
||||
qs = qs[: options['limit']]
|
||||
|
|
|
|||
38
podcasts/migrations/0007_feed_http_cache.py
Normal file
38
podcasts/migrations/0007_feed_http_cache.py
Normal file
|
|
@ -0,0 +1,38 @@
|
|||
# Generated by Django 6.1 on 2026-10-02 12:00
|
||||
|
||||
from django.db import migrations, models
|
||||
|
||||
|
||||
class Migration(migrations.Migration):
|
||||
|
||||
dependencies = [
|
||||
('podcasts', '0006_episodeprogress_favorite'),
|
||||
]
|
||||
|
||||
operations = [
|
||||
migrations.AddField(
|
||||
model_name='podcastfeed',
|
||||
name='http_etag',
|
||||
field=models.CharField(blank=True, max_length=500),
|
||||
),
|
||||
migrations.AddField(
|
||||
model_name='podcastfeed',
|
||||
name='http_last_modified',
|
||||
field=models.CharField(blank=True, max_length=100),
|
||||
),
|
||||
migrations.AddField(
|
||||
model_name='podcastfeed',
|
||||
name='last_error',
|
||||
field=models.CharField(blank=True, max_length=500),
|
||||
),
|
||||
migrations.AddField(
|
||||
model_name='podcastfeed',
|
||||
name='last_error_at',
|
||||
field=models.DateTimeField(blank=True, null=True),
|
||||
),
|
||||
migrations.AddField(
|
||||
model_name='podcastfeed',
|
||||
name='error_count',
|
||||
field=models.PositiveIntegerField(default=0),
|
||||
),
|
||||
]
|
||||
|
|
@ -13,6 +13,15 @@ class PodcastFeed(models.Model):
|
|||
last_refreshed_at = models.DateTimeField(null=True, blank=True)
|
||||
added_at = models.DateTimeField(auto_now_add=True)
|
||||
auto_queue = models.BooleanField(default=False)
|
||||
# HTTP validators of the last full (200) response, sent back as
|
||||
# If-None-Match / If-Modified-Since so an unchanged feed costs a 304.
|
||||
http_etag = models.CharField(max_length=500, blank=True)
|
||||
http_last_modified = models.CharField(max_length=100, blank=True)
|
||||
# The last failed refresh (network, HTTP status, not a feed); cleared on
|
||||
# success. error_count counts failures in a row and drives the backoff.
|
||||
last_error = models.CharField(max_length=500, blank=True)
|
||||
last_error_at = models.DateTimeField(null=True, blank=True)
|
||||
error_count = models.PositiveIntegerField(default=0)
|
||||
|
||||
class Meta:
|
||||
unique_together = ('user', 'rss_url')
|
||||
|
|
|
|||
|
|
@ -7,11 +7,16 @@ from django.contrib.auth.models import User
|
|||
from django.test import TestCase
|
||||
|
||||
from .models import EpisodeProgress, PodcastEpisode, PodcastFeed, PodcastQueue, PodcastSettings
|
||||
from .views import _refresh_feed
|
||||
from .views import FeedFetch, _refresh_feed
|
||||
|
||||
_parse = feedparser.parse
|
||||
|
||||
|
||||
def _ok(rss, **kw):
|
||||
"""What _fetch_feed returns for a 200 with this RSS."""
|
||||
return FeedFetch(200, _parse(rss), **kw)
|
||||
|
||||
|
||||
def _rss(items):
|
||||
entries = ''.join(
|
||||
f'<item><guid>{guid}</guid><title>{guid}</title><pubDate>{date}</pubDate>'
|
||||
|
|
@ -157,7 +162,7 @@ class AutoQueueTests(TestCase):
|
|||
])
|
||||
|
||||
def refresh(self):
|
||||
with patch('podcasts.views._fetch_feed', lambda url: _parse(self.rss)):
|
||||
with patch('podcasts.views._fetch_feed', lambda url, **kw: _ok(self.rss)):
|
||||
_refresh_feed(self.feed)
|
||||
order = PodcastQueue.objects.filter(user=self.user).order_by('position')
|
||||
return [q.episode.guid for q in order]
|
||||
|
|
@ -189,7 +194,7 @@ class FirstFetchTests(TestCase):
|
|||
self.rss = _rss([(f'e{i}', f'Mon, {i + 1:02d} Jan 2026 10:00:00 GMT') for i in range(5)])
|
||||
|
||||
def subscribe(self):
|
||||
with patch('podcasts.views._fetch_feed', lambda url: _parse(self.rss)):
|
||||
with patch('podcasts.views._fetch_feed', lambda url, **kw: _ok(self.rss)):
|
||||
return self.client.post('/podcasts/feeds/add/', json.dumps({'rss_url': 'https://example.org/f'}),
|
||||
content_type='application/json')
|
||||
|
||||
|
|
@ -207,7 +212,7 @@ class FirstFetchTests(TestCase):
|
|||
self.subscribe()
|
||||
feed = PodcastFeed.objects.get(user=self.user)
|
||||
self.rss = _rss([(f'e{i}', f'Mon, {i + 1:02d} Jan 2026 10:00:00 GMT') for i in range(8)])
|
||||
with patch('podcasts.views._fetch_feed', lambda url: _parse(self.rss)):
|
||||
with patch('podcasts.views._fetch_feed', lambda url, **kw: _ok(self.rss)):
|
||||
_refresh_feed(feed)
|
||||
inbox = self.client.get('/podcasts/inbox/').json()['episodes']
|
||||
self.assertEqual([e['title'] for e in inbox], ['e7', 'e6', 'e5', 'e4'])
|
||||
|
|
@ -225,11 +230,11 @@ class RefreshAllTests(TestCase):
|
|||
self.bobs = PodcastFeed.objects.create(user=bob, rss_url='https://example.org/a', title='A')
|
||||
self.fetched = []
|
||||
|
||||
def fake_fetch(self, url):
|
||||
def fake_fetch(self, url, **kw):
|
||||
self.fetched.append(url)
|
||||
if url.endswith('/b'):
|
||||
raise ValueError('broken feed')
|
||||
return _parse(_rss([(url[-1] + '1', 'Mon, 05 Jan 2026 10:00:00 GMT')]))
|
||||
return _ok(_rss([(url[-1] + '1', 'Mon, 05 Jan 2026 10:00:00 GMT')]))
|
||||
|
||||
def test_refresh_all_reports_counts_and_skips_recent(self):
|
||||
with patch('podcasts.views._fetch_feed', self.fake_fetch):
|
||||
|
|
@ -318,7 +323,7 @@ class FeedExtrasParsingTests(TestCase):
|
|||
self.feed = PodcastFeed.objects.create(user=self.user, rss_url='https://example.org/feed', title='Feed')
|
||||
|
||||
def refresh(self, rss):
|
||||
with patch('podcasts.views._fetch_feed', lambda url: _parse(rss)):
|
||||
with patch('podcasts.views._fetch_feed', lambda url, **kw: _ok(rss)):
|
||||
_refresh_feed(self.feed)
|
||||
|
||||
def test_feedparser_keys(self):
|
||||
|
|
@ -894,3 +899,329 @@ class OpmlExportTests(TestCase):
|
|||
def test_login_required(self):
|
||||
self.client.logout()
|
||||
self.assertEqual(self.client.get('/podcasts/feeds/export.opml').status_code, 401)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Feed refresh: conditional GET, errors, permanent redirects, backoff
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class _Hop:
|
||||
def __init__(self, url, status):
|
||||
self.url, self.status_code = url, status
|
||||
|
||||
|
||||
class _HttpResponse:
|
||||
"""Just enough of a requests.Response for _fetch_feed."""
|
||||
|
||||
def __init__(self, url, status=200, content=b'', headers=None, history=()):
|
||||
self.url, self.status_code, self.content = url, status, content
|
||||
self.headers = headers or {}
|
||||
self.history = list(history)
|
||||
|
||||
|
||||
class FetchFeedTests(TestCase):
|
||||
"""_fetch_feed itself, with requests.get faked."""
|
||||
|
||||
RSS = _rss([('a', 'Mon, 05 Jan 2026 10:00:00 GMT')]).encode()
|
||||
|
||||
def fetch(self, response, url='https://example.org/feed', **kw):
|
||||
calls = []
|
||||
|
||||
def fake_get(u, **kwargs):
|
||||
calls.append(kwargs['headers'])
|
||||
if isinstance(response, Exception):
|
||||
raise response
|
||||
return response
|
||||
|
||||
from .views import _fetch_feed
|
||||
with patch('podcasts.views.requests.get', fake_get):
|
||||
result = _fetch_feed(url, **kw)
|
||||
return result, calls[0]
|
||||
|
||||
def error(self, response):
|
||||
from .views import FeedFetchError
|
||||
with self.assertRaises(FeedFetchError) as cm:
|
||||
self.fetch(response)
|
||||
return str(cm.exception)
|
||||
|
||||
def test_200_returns_parsed_feed_and_validators(self):
|
||||
r, headers = self.fetch(_HttpResponse('https://example.org/feed', content=self.RSS, headers={
|
||||
'ETag': '"v1"', 'Last-Modified': 'Mon, 05 Jan 2026 10:00:00 GMT'}))
|
||||
self.assertEqual(r.status, 200)
|
||||
self.assertEqual([e.title for e in r.parsed.entries], ['a'])
|
||||
self.assertEqual((r.etag, r.last_modified), ('"v1"', 'Mon, 05 Jan 2026 10:00:00 GMT'))
|
||||
self.assertEqual(r.permanent_url, 'https://example.org/feed')
|
||||
self.assertNotIn('If-None-Match', headers)
|
||||
self.assertNotIn('If-Modified-Since', headers)
|
||||
|
||||
def test_conditional_request_and_304(self):
|
||||
r, headers = self.fetch(_HttpResponse('https://example.org/feed', status=304),
|
||||
etag='"v1"', last_modified='Mon, 05 Jan 2026 10:00:00 GMT')
|
||||
self.assertEqual(headers['If-None-Match'], '"v1"')
|
||||
self.assertEqual(headers['If-Modified-Since'], 'Mon, 05 Jan 2026 10:00:00 GMT')
|
||||
self.assertEqual(r.status, 304)
|
||||
self.assertIsNone(r.parsed)
|
||||
|
||||
def test_304_without_validators_is_an_error(self):
|
||||
self.assertEqual(self.error(_HttpResponse('https://example.org/feed', status=304)), 'HTTP 304')
|
||||
|
||||
def test_errors(self):
|
||||
import requests
|
||||
self.assertEqual(self.error(_HttpResponse('https://example.org/feed', status=404)), 'HTTP 404')
|
||||
self.assertEqual(self.error(_HttpResponse('https://example.org/feed', status=503)), 'HTTP 503')
|
||||
self.assertEqual(self.error(requests.Timeout()), 'timeout')
|
||||
self.assertEqual(self.error(requests.ConnectionError()), 'connection failed')
|
||||
self.assertEqual(self.error(requests.TooManyRedirects()), 'too many redirects')
|
||||
self.assertEqual(self.error(_HttpResponse('https://example.org/feed', content=b'<html><body><p>Hi')),
|
||||
'not a feed')
|
||||
self.assertEqual(self.error(_HttpResponse('https://example.org/feed', content=b'\x00\xff garbage')),
|
||||
'not a feed')
|
||||
|
||||
def test_permanent_redirect_chain(self):
|
||||
r, _ = self.fetch(_HttpResponse('https://c.org/feed', content=self.RSS, history=[
|
||||
_Hop('https://example.org/feed', 301), _Hop('https://b.org/feed', 308)]))
|
||||
self.assertEqual(r.permanent_url, 'https://c.org/feed')
|
||||
self.assertEqual(r.final_url, 'https://c.org/feed')
|
||||
|
||||
def test_temporary_hop_ends_the_permanent_part(self):
|
||||
r, _ = self.fetch(_HttpResponse('https://c.org/feed', content=self.RSS, history=[
|
||||
_Hop('https://example.org/feed', 301), _Hop('https://b.org/feed', 302)]))
|
||||
self.assertEqual(r.permanent_url, 'https://b.org/feed')
|
||||
for status in (302, 307):
|
||||
r, _ = self.fetch(_HttpResponse('https://c.org/feed', content=self.RSS, history=[
|
||||
_Hop('https://example.org/feed', status), _Hop('https://b.org/feed', 301)]))
|
||||
self.assertEqual(r.permanent_url, 'https://example.org/feed')
|
||||
|
||||
|
||||
class FeedRefreshTests(TestCase):
|
||||
"""What a refresh stores: validators, 304s, errors, moved feeds."""
|
||||
|
||||
V = {'etag': '"v1"', 'last_modified': 'Mon, 05 Jan 2026 10:00:00 GMT'}
|
||||
|
||||
def setUp(self):
|
||||
self.user = User.objects.create_user(username='alice', password='pw12345678')
|
||||
self.client.force_login(self.user)
|
||||
self.feed = PodcastFeed.objects.create(
|
||||
user=self.user, rss_url='https://example.org/feed', title='Feed',
|
||||
last_refreshed_at=datetime(2026, 1, 1, tzinfo=timezone.utc),
|
||||
http_etag=self.V['etag'], http_last_modified=self.V['last_modified'],
|
||||
)
|
||||
PodcastEpisode.objects.create(feed=self.feed, guid='old', title='Old', audio_url='https://example.org/old.mp3')
|
||||
self.calls = []
|
||||
self.answer = None
|
||||
|
||||
def fake_fetch(self, url, **kw):
|
||||
self.calls.append((url, kw))
|
||||
if isinstance(self.answer, Exception):
|
||||
raise self.answer
|
||||
return self.answer
|
||||
|
||||
def refresh(self, feeds=None):
|
||||
from .views import _refresh_feeds
|
||||
with patch('podcasts.views._fetch_feed', self.fake_fetch):
|
||||
return _refresh_feeds(feeds or [self.feed])
|
||||
|
||||
def test_304_touches_no_episodes(self):
|
||||
from .views import FeedFetch
|
||||
self.feed.error_count, self.feed.last_error = 2, 'timeout'
|
||||
self.feed.last_error_at = datetime(2026, 1, 2, tzinfo=timezone.utc)
|
||||
self.feed.save()
|
||||
self.answer = FeedFetch(304, None, **self.V)
|
||||
with patch('podcasts.views.PodcastEpisode.objects.get_or_create') as create, \
|
||||
patch('podcasts.views.PodcastEpisode.objects.bulk_update') as update:
|
||||
self.assertEqual(self.refresh(), (1, 0, 0))
|
||||
create.assert_not_called()
|
||||
update.assert_not_called()
|
||||
self.assertEqual(self.calls, [('https://example.org/feed', self.V)])
|
||||
self.feed.refresh_from_db()
|
||||
self.assertGreater(self.feed.last_refreshed_at, datetime(2026, 1, 1, tzinfo=timezone.utc))
|
||||
self.assertEqual((self.feed.error_count, self.feed.last_error, self.feed.last_error_at), (0, '', None))
|
||||
self.assertEqual((self.feed.http_etag, self.feed.http_last_modified), (self.V['etag'], self.V['last_modified']))
|
||||
self.assertEqual(self.feed.episodes.count(), 1)
|
||||
|
||||
def test_200_stores_new_validators(self):
|
||||
self.answer = _ok(_rss([('new', 'Mon, 05 Jan 2026 10:00:00 GMT')]), etag='"v2"', last_modified='')
|
||||
self.assertEqual(self.refresh(), (1, 0, 1))
|
||||
self.feed.refresh_from_db()
|
||||
self.assertEqual((self.feed.http_etag, self.feed.http_last_modified), ('"v2"', ''))
|
||||
|
||||
def test_first_fetch_is_never_conditional(self):
|
||||
self.feed.last_refreshed_at = None
|
||||
self.feed.save()
|
||||
self.answer = _ok(_rss([]))
|
||||
self.refresh()
|
||||
self.assertEqual(self.calls, [('https://example.org/feed', {})])
|
||||
# Nothing stored yet either: unconditional.
|
||||
self.calls.clear()
|
||||
new = PodcastFeed.objects.create(user=self.user, rss_url='https://example.org/n', title='N')
|
||||
self.refresh([new])
|
||||
self.assertEqual(self.calls, [('https://example.org/n', {})])
|
||||
|
||||
def test_shared_url_validators_only_when_all_agree(self):
|
||||
from .views import FeedFetch
|
||||
bob = User.objects.create_user(username='bob', password='pw12345678')
|
||||
bobs = PodcastFeed.objects.create(
|
||||
user=bob, rss_url='https://example.org/feed', title='Feed',
|
||||
last_refreshed_at=datetime(2026, 1, 1, tzinfo=timezone.utc),
|
||||
http_etag=self.V['etag'], http_last_modified=self.V['last_modified'],
|
||||
)
|
||||
self.answer = FeedFetch(304, None, **self.V)
|
||||
self.assertEqual(self.refresh([self.feed, bobs]), (2, 0, 0))
|
||||
self.assertEqual(self.calls, [('https://example.org/feed', self.V)])
|
||||
|
||||
# Bob's copy has other validators: one full download for both.
|
||||
bobs.http_etag = '"v0"'
|
||||
bobs.save()
|
||||
self.calls.clear()
|
||||
self.answer = _ok(_rss([('new', 'Mon, 05 Jan 2026 10:00:00 GMT')]), etag='"v2"')
|
||||
self.assertEqual(self.refresh([self.feed, bobs]), (2, 0, 2))
|
||||
self.assertEqual(self.calls, [('https://example.org/feed', {})])
|
||||
self.assertEqual(set(PodcastFeed.objects.values_list('http_etag', flat=True)), {'"v2"'})
|
||||
|
||||
# Carol just subscribed (never fetched): a full download as well.
|
||||
carol = User.objects.create_user(username='carol', password='pw12345678')
|
||||
carols = PodcastFeed.objects.create(user=carol, rss_url='https://example.org/feed', title='Feed',
|
||||
http_etag='"v2"')
|
||||
self.calls.clear()
|
||||
self.refresh([self.feed, PodcastFeed.objects.get(pk=bobs.pk), carols])
|
||||
self.assertEqual(self.calls, [('https://example.org/feed', {})])
|
||||
self.assertEqual(PodcastEpisode.objects.filter(feed=carols).count(), 1)
|
||||
|
||||
def test_errors_are_counted_and_cleared(self):
|
||||
from .views import FeedFetchError
|
||||
self.answer = FeedFetchError('HTTP 404')
|
||||
self.assertEqual(self.refresh(), (0, 1, 0))
|
||||
self.feed.refresh_from_db()
|
||||
self.assertEqual(self.refresh([self.feed]), (0, 1, 0))
|
||||
self.feed.refresh_from_db()
|
||||
self.assertEqual((self.feed.last_error, self.feed.error_count), ('HTTP 404', 2))
|
||||
self.assertIsNotNone(self.feed.last_error_at)
|
||||
# Unchanged otherwise: still the last good refresh.
|
||||
self.assertEqual(self.feed.last_refreshed_at, datetime(2026, 1, 1, tzinfo=timezone.utc))
|
||||
|
||||
feeds = self.client.get('/podcasts/feeds/').json()['feeds']
|
||||
self.assertEqual(feeds[0]['last_error'], 'HTTP 404')
|
||||
self.assertEqual(feeds[0]['error_count'], 2)
|
||||
self.assertEqual(feeds[0]['last_error_at'], self.feed.last_error_at.isoformat())
|
||||
state = self.client.get('/podcasts/state/').json()['feeds']
|
||||
self.assertEqual(state[0]['error_count'], 2)
|
||||
|
||||
self.answer = _ok(_rss([('new', 'Mon, 05 Jan 2026 10:00:00 GMT')]))
|
||||
self.assertEqual(self.refresh([self.feed]), (1, 0, 1))
|
||||
feeds = self.client.get('/podcasts/feeds/').json()['feeds']
|
||||
self.assertEqual((feeds[0]['last_error'], feeds[0]['last_error_at'], feeds[0]['error_count']), ('', None, 0))
|
||||
|
||||
def test_refresh_now_reports_and_stores_the_error(self):
|
||||
from .views import FeedFetchError
|
||||
self.answer = FeedFetchError('not a feed')
|
||||
with patch('podcasts.views._fetch_feed', self.fake_fetch):
|
||||
r = self.client.post('/podcasts/feeds/refresh/', json.dumps({'feed_id': self.feed.id}),
|
||||
content_type='application/json')
|
||||
self.assertEqual(r.status_code, 502)
|
||||
self.assertEqual(r.json()['error'], 'not a feed')
|
||||
self.assertEqual(self.calls, [('https://example.org/feed', self.V)])
|
||||
self.feed.refresh_from_db()
|
||||
self.assertEqual((self.feed.last_error, self.feed.error_count), ('not a feed', 1))
|
||||
|
||||
def test_failed_subscription_is_not_kept(self):
|
||||
from .views import FeedFetchError
|
||||
self.answer = FeedFetchError('HTTP 404')
|
||||
with patch('podcasts.views._fetch_feed', self.fake_fetch):
|
||||
r = self.client.post('/podcasts/feeds/add/', json.dumps({'rss_url': 'https://example.org/gone'}),
|
||||
content_type='application/json')
|
||||
self.assertEqual(r.status_code, 400)
|
||||
self.assertIn('HTTP 404', r.json()['error'])
|
||||
self.assertEqual(self.calls, [('https://example.org/gone', {})])
|
||||
self.assertFalse(PodcastFeed.objects.filter(rss_url='https://example.org/gone').exists())
|
||||
|
||||
def test_permanent_redirect_moves_the_subscription(self):
|
||||
self.answer = _ok(_rss([('new', 'Mon, 05 Jan 2026 10:00:00 GMT')]),
|
||||
final_url='https://new.org/feed', permanent_url='https://new.org/feed')
|
||||
self.assertEqual(self.refresh(), (1, 0, 1))
|
||||
self.feed.refresh_from_db()
|
||||
self.assertEqual(self.feed.rss_url, 'https://new.org/feed')
|
||||
self.assertEqual(self.feed.episodes.count(), 2)
|
||||
|
||||
def test_permanent_redirect_to_a_url_the_user_has_already(self):
|
||||
other = PodcastFeed.objects.create(user=self.user, rss_url='https://new.org/feed', title='Other')
|
||||
self.answer = _ok(_rss([('new', 'Mon, 05 Jan 2026 10:00:00 GMT')]), etag='"v2"',
|
||||
final_url='https://new.org/feed', permanent_url='https://new.org/feed')
|
||||
self.assertEqual(self.refresh(), (1, 0, 1))
|
||||
self.feed.refresh_from_db()
|
||||
self.assertEqual(self.feed.rss_url, 'https://example.org/feed')
|
||||
self.assertEqual((self.feed.error_count, self.feed.http_etag), (0, '"v2"'))
|
||||
self.assertEqual(PodcastFeed.objects.get(pk=other.pk).rss_url, 'https://new.org/feed')
|
||||
|
||||
def test_temporary_redirect_keeps_the_url(self):
|
||||
self.answer = _ok(_rss([]), final_url='https://tmp.org/feed', permanent_url='https://example.org/feed')
|
||||
self.refresh()
|
||||
self.feed.refresh_from_db()
|
||||
self.assertEqual(self.feed.rss_url, 'https://example.org/feed')
|
||||
|
||||
|
||||
class FeedBackoffTests(TestCase):
|
||||
"""The hourly job tries a feed failing 3+ times in a row less often."""
|
||||
|
||||
def setUp(self):
|
||||
from django.utils import timezone as dj_tz
|
||||
self.user = User.objects.create_user(username='alice', password='pw12345678')
|
||||
self.client.force_login(self.user)
|
||||
now = dj_tz.now()
|
||||
long_ago = datetime(2026, 1, 1, tzinfo=timezone.utc)
|
||||
# (name, error_count, hours since the last error) -> due in the hourly job?
|
||||
self.cases = {
|
||||
'ok': (0, None, True),
|
||||
'two': (2, 0.2, True),
|
||||
'three-early': (3, 1, False), # waits 2 h
|
||||
'three-due': (3, 2.5, True),
|
||||
'four-early': (4, 3, False), # waits 4 h
|
||||
'four-due': (4, 4.5, True),
|
||||
'six-early': (6, 15, False), # waits 16 h
|
||||
'six-due': (6, 17, True),
|
||||
'ten-early': (10, 23, False), # capped at 24 h
|
||||
'ten-due': (10, 25, True),
|
||||
}
|
||||
self.feeds = {}
|
||||
for name, (count, hours, _) in self.cases.items():
|
||||
self.feeds[name] = PodcastFeed.objects.create(
|
||||
user=self.user, rss_url=f'https://example.org/{name}', title=name,
|
||||
last_refreshed_at=long_ago, error_count=count, last_error='timeout' if count else '',
|
||||
last_error_at=now - timedelta(hours=hours) if hours is not None else None,
|
||||
)
|
||||
self.fetched = []
|
||||
|
||||
def fake_fetch(self, url, **kw):
|
||||
self.fetched.append(url.rsplit('/', 1)[1])
|
||||
return _ok(_rss([]))
|
||||
|
||||
def run_command(self, *args):
|
||||
from io import StringIO
|
||||
from django.core.management import call_command
|
||||
out = StringIO()
|
||||
with patch('podcasts.views._fetch_feed', self.fake_fetch):
|
||||
call_command('refresh_feeds', *args, stdout=out)
|
||||
return out.getvalue()
|
||||
|
||||
def test_hourly_job_skips_feeds_in_backoff(self):
|
||||
out = self.run_command()
|
||||
due = {name for name, (_, _, is_due) in self.cases.items() if is_due}
|
||||
self.assertEqual(set(self.fetched), due)
|
||||
self.assertIn(f'{len(due)} feeds refreshed, 0 failed', out)
|
||||
|
||||
def test_explicit_feed_id_ignores_backoff(self):
|
||||
self.run_command('--feed-id', str(self.feeds['ten-early'].id))
|
||||
self.assertEqual(self.fetched, ['ten-early'])
|
||||
|
||||
def test_user_refresh_all_tries_them_but_respects_max_age(self):
|
||||
from django.utils import timezone as dj_tz
|
||||
recent = self.feeds['three-early']
|
||||
recent.last_refreshed_at = dj_tz.now() - timedelta(seconds=60)
|
||||
recent.save()
|
||||
with patch('podcasts.views._fetch_feed', self.fake_fetch):
|
||||
r = self.client.post('/podcasts/feeds/refresh-all/', '{}', content_type='application/json')
|
||||
self.assertEqual(set(self.fetched), set(self.cases) - {'three-early'})
|
||||
self.assertEqual(r.json()['refreshed'], len(self.cases) - 1)
|
||||
|
||||
def test_backoff_hours(self):
|
||||
from .views import _backoff_hours
|
||||
self.assertEqual([_backoff_hours(n) for n in range(0, 9)], [0, 0, 0, 2, 4, 8, 16, 24, 24])
|
||||
|
|
|
|||
|
|
@ -2,12 +2,14 @@ import json
|
|||
import logging
|
||||
import urllib.parse
|
||||
from concurrent.futures import ThreadPoolExecutor
|
||||
from dataclasses import dataclass
|
||||
from datetime import timedelta
|
||||
import xml.etree.ElementTree as ET
|
||||
|
||||
import feedparser
|
||||
import requests
|
||||
from django.conf import settings
|
||||
from django.db import IntegrityError, transaction
|
||||
from django.db.models import (
|
||||
Case, Count, Exists, F, FilteredRelation, IntegerField, Max, OuterRef, Q, Sum, Value, When,
|
||||
)
|
||||
|
|
@ -214,11 +216,141 @@ EXTRA_FIELDS = ['description', 'transcripts', 'chapters_url', 'chapters', 'chapt
|
|||
# Shared feed refresh helper
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _fetch_feed(url):
|
||||
"""Download and parse a feed (with a timeout, unlike feedparser's own fetch)."""
|
||||
r = requests.get(url, timeout=FEED_TIMEOUT, headers={'User-Agent': 'diora podcast refresh'})
|
||||
r.raise_for_status()
|
||||
return feedparser.parse(r.content, response_headers={'content-location': r.url, **r.headers})
|
||||
class FeedFetchError(Exception):
|
||||
"""A feed that could not be fetched or is not a feed; str() is the short
|
||||
message stored in PodcastFeed.last_error ("HTTP 404", "timeout", …)."""
|
||||
|
||||
|
||||
@dataclass
|
||||
class FeedFetch:
|
||||
"""What _fetch_feed got. status 200: `parsed` is the feedparser result and
|
||||
etag/last_modified are the new validators ('' when the server sent none).
|
||||
status 304: unchanged since the validators that were sent, parsed is None.
|
||||
final_url is where the content came from after all redirects;
|
||||
permanent_url is the requested URL after following only the 301/308 hops
|
||||
at the start of the redirect chain (== the requested URL without one)."""
|
||||
status: int
|
||||
parsed: object = None
|
||||
etag: str = ''
|
||||
last_modified: str = ''
|
||||
final_url: str = ''
|
||||
permanent_url: str = ''
|
||||
|
||||
|
||||
PERMANENT_REDIRECTS = (301, 308)
|
||||
|
||||
|
||||
def _fetch_feed(url, etag='', last_modified=''):
|
||||
"""Download and parse a feed (with a timeout, unlike feedparser's own
|
||||
fetch). With etag/last_modified it is a conditional GET. Returns a
|
||||
FeedFetch; raises FeedFetchError when there is nothing usable."""
|
||||
headers = {'User-Agent': 'diora podcast refresh'}
|
||||
if etag:
|
||||
headers['If-None-Match'] = etag
|
||||
if last_modified:
|
||||
headers['If-Modified-Since'] = last_modified
|
||||
try:
|
||||
r = requests.get(url, timeout=FEED_TIMEOUT, headers=headers)
|
||||
except requests.Timeout:
|
||||
raise FeedFetchError('timeout')
|
||||
except requests.TooManyRedirects:
|
||||
raise FeedFetchError('too many redirects')
|
||||
except requests.ConnectionError:
|
||||
raise FeedFetchError('connection failed')
|
||||
except requests.RequestException as e:
|
||||
raise FeedFetchError(f'request failed ({type(e).__name__})')
|
||||
|
||||
# Each hop's target is the next hop's URL (the last one's is r.url).
|
||||
permanent_url = url
|
||||
targets = [h.url for h in r.history[1:]] + [r.url]
|
||||
for hop, target in zip(r.history, targets):
|
||||
if hop.status_code not in PERMANENT_REDIRECTS:
|
||||
break
|
||||
permanent_url = target
|
||||
|
||||
if r.status_code == 304:
|
||||
if not (etag or last_modified):
|
||||
raise FeedFetchError('HTTP 304') # nothing to be unchanged from
|
||||
return FeedFetch(304, None, etag, last_modified, r.url, permanent_url)
|
||||
if r.status_code >= 400:
|
||||
raise FeedFetchError(f'HTTP {r.status_code}')
|
||||
|
||||
parsed = feedparser.parse(r.content, response_headers={'content-location': r.url, **r.headers})
|
||||
if parsed.get('bozo') and not parsed.entries and not parsed.feed.get('title'):
|
||||
raise FeedFetchError('not a feed')
|
||||
new_etag = r.headers.get('ETag', '')
|
||||
new_modified = r.headers.get('Last-Modified', '')
|
||||
return FeedFetch(
|
||||
r.status_code, parsed,
|
||||
# A validator too long to store is dropped (a truncated one never matches).
|
||||
new_etag if len(new_etag) <= 500 else '',
|
||||
new_modified if len(new_modified) <= 100 else '',
|
||||
r.url, permanent_url,
|
||||
)
|
||||
|
||||
|
||||
def _validators(feeds):
|
||||
"""If-None-Match / If-Modified-Since for one download that serves all
|
||||
`feeds` (several users' subscriptions of the same URL): only when every
|
||||
one of them was fetched before and stored the same validators, because a
|
||||
304 then skips all of them. Otherwise {} (a full response)."""
|
||||
pairs = {(f.http_etag, f.http_last_modified) for f in feeds}
|
||||
if len(pairs) != 1 or any(f.last_refreshed_at is None for f in feeds):
|
||||
return {}
|
||||
etag, last_modified = pairs.pop()
|
||||
if not (etag or last_modified):
|
||||
return {}
|
||||
return {'etag': etag, 'last_modified': last_modified}
|
||||
|
||||
|
||||
def _error_text(e):
|
||||
if isinstance(e, FeedFetchError):
|
||||
return str(e)[:500]
|
||||
return (str(e) or type(e).__name__)[:500]
|
||||
|
||||
|
||||
def _record_error(feed_obj, e):
|
||||
feed_obj.last_error = _error_text(e)
|
||||
feed_obj.last_error_at = timezone.now()
|
||||
feed_obj.error_count += 1
|
||||
try:
|
||||
feed_obj.save(update_fields=['last_error', 'last_error_at', 'error_count'])
|
||||
except Exception:
|
||||
log.exception('could not store the refresh error of feed %s', feed_obj.pk)
|
||||
|
||||
|
||||
ERROR_FIELDS = ['last_error', 'last_error_at', 'error_count']
|
||||
|
||||
|
||||
def _clear_error(feed_obj):
|
||||
feed_obj.last_error = ''
|
||||
feed_obj.last_error_at = None
|
||||
feed_obj.error_count = 0
|
||||
|
||||
|
||||
BACKOFF_AFTER = 3 # failures in a row before the hourly job backs off
|
||||
BACKOFF_MAX_HOURS = 24
|
||||
|
||||
|
||||
def _backoff_hours(error_count):
|
||||
"""Hours between hourly-job attempts for a feed failing error_count
|
||||
times in a row: 2, 4, 8, 16, 24 … from the third failure on."""
|
||||
if error_count < BACKOFF_AFTER:
|
||||
return 0
|
||||
return min(2 ** (error_count - 2), BACKOFF_MAX_HOURS)
|
||||
|
||||
|
||||
def _in_backoff_q(now=None):
|
||||
"""Q for feeds that failed too often to be tried again yet."""
|
||||
now = now or timezone.now()
|
||||
n = BACKOFF_AFTER
|
||||
while _backoff_hours(n) < BACKOFF_MAX_HOURS:
|
||||
n += 1
|
||||
# From n failures on the wait is capped; below, one term per count.
|
||||
q = Q(error_count__gte=n, last_error_at__gt=now - timedelta(hours=BACKOFF_MAX_HOURS))
|
||||
for count in range(BACKOFF_AFTER, n):
|
||||
q |= Q(error_count=count, last_error_at__gt=now - timedelta(hours=_backoff_hours(count)))
|
||||
return q
|
||||
|
||||
|
||||
def _refresh_feeds(feeds, chunk=32):
|
||||
|
|
@ -229,27 +361,31 @@ def _refresh_feeds(feeds, chunk=32):
|
|||
refreshed = failed = new = 0
|
||||
for start in range(0, len(feeds), chunk):
|
||||
part = feeds[start:start + chunk]
|
||||
urls = sorted({f.rss_url for f in part})
|
||||
by_url = {}
|
||||
for f in part:
|
||||
by_url.setdefault(f.rss_url, []).append(f)
|
||||
results = {}
|
||||
|
||||
def fetch(url):
|
||||
try:
|
||||
return url, _fetch_feed(url), None
|
||||
return url, _fetch_feed(url, **_validators(by_url[url])), None
|
||||
except Exception as e: # network, HTTP or parser trouble: report and go on
|
||||
return url, None, e
|
||||
|
||||
urls = sorted(by_url)
|
||||
with ThreadPoolExecutor(max_workers=min(FEED_WORKERS, len(urls) or 1)) as pool:
|
||||
for url, parsed, err in pool.map(fetch, urls):
|
||||
results[url] = (parsed, err)
|
||||
for url, fetched, err in pool.map(fetch, urls):
|
||||
results[url] = (fetched, err)
|
||||
|
||||
for feed in part:
|
||||
parsed, err = results[feed.rss_url]
|
||||
fetched, err = results[feed.rss_url]
|
||||
if err is not None:
|
||||
failed += 1
|
||||
log.warning('refreshing %s failed: %s', feed.rss_url, err)
|
||||
_record_error(feed, err)
|
||||
continue
|
||||
try:
|
||||
new += _refresh_feed(feed, parsed)
|
||||
new += _refresh_feed(feed, fetched)
|
||||
refreshed += 1
|
||||
except Exception as e:
|
||||
failed += 1
|
||||
|
|
@ -257,10 +393,47 @@ def _refresh_feeds(feeds, chunk=32):
|
|||
return refreshed, failed, new
|
||||
|
||||
|
||||
def _refresh_feed(feed_obj, parsed=None):
|
||||
"""Parse rss_url and upsert episodes. Returns count of new episodes."""
|
||||
if parsed is None:
|
||||
parsed = _fetch_feed(feed_obj.rss_url)
|
||||
def _refresh_feed(feed_obj, fetched=None):
|
||||
"""Fetch rss_url (unless `fetched`, a FeedFetch, is given) and upsert
|
||||
episodes. Returns count of new episodes. A failure is stored on the feed
|
||||
(last_error, error_count) and raised again."""
|
||||
if fetched is None:
|
||||
try:
|
||||
fetched = _fetch_feed(feed_obj.rss_url, **_validators([feed_obj]))
|
||||
except Exception as e:
|
||||
_record_error(feed_obj, e)
|
||||
raise
|
||||
try:
|
||||
return _apply_feed(feed_obj, fetched)
|
||||
except Exception as e:
|
||||
_record_error(feed_obj, e)
|
||||
raise
|
||||
|
||||
|
||||
def _move_to_permanent_url(feed_obj, new_url):
|
||||
"""The feed answered with a permanent redirect: subscribe to the new URL,
|
||||
unless the user already has that one too (then nothing changes)."""
|
||||
if not new_url or new_url == feed_obj.rss_url or len(new_url) > 1000:
|
||||
return
|
||||
try:
|
||||
with transaction.atomic():
|
||||
PodcastFeed.objects.filter(pk=feed_obj.pk).update(rss_url=new_url)
|
||||
except IntegrityError: # the user's other subscription has it
|
||||
return
|
||||
log.info('feed %s moved permanently: %s -> %s', feed_obj.pk, feed_obj.rss_url, new_url)
|
||||
feed_obj.rss_url = new_url
|
||||
|
||||
|
||||
def _apply_feed(feed_obj, fetched):
|
||||
now = timezone.now()
|
||||
if fetched.status == 304:
|
||||
# Unchanged since the last full download: nothing to parse or write.
|
||||
feed_obj.last_refreshed_at = now
|
||||
_clear_error(feed_obj)
|
||||
feed_obj.save(update_fields=['last_refreshed_at', *ERROR_FIELDS])
|
||||
return 0
|
||||
|
||||
parsed = fetched.parsed
|
||||
channel = parsed.feed
|
||||
# The first fetch of a new subscription brings the whole back catalogue.
|
||||
first_fetch = feed_obj.last_refreshed_at is None
|
||||
|
|
@ -278,7 +451,7 @@ def _refresh_feed(feed_obj, parsed=None):
|
|||
or image.get('url', '')
|
||||
)[:1000]
|
||||
|
||||
feed_obj.last_refreshed_at = timezone.now()
|
||||
feed_obj.last_refreshed_at = now
|
||||
feed_obj.save()
|
||||
|
||||
max_ep = getattr(settings, 'PODCAST_MAX_EPISODES_PER_FEED', 200)
|
||||
|
|
@ -393,6 +566,13 @@ def _refresh_feed(feed_obj, parsed=None):
|
|||
ordered = newest_first if where != PodcastSettings.BACK else newest_first[::-1]
|
||||
_enqueue(feed_obj.user, [e.id for e in ordered], where)
|
||||
|
||||
# Only now, with every episode written: the validators (a 304 next time
|
||||
# skips all of the above) and a clean error state.
|
||||
feed_obj.http_etag = fetched.etag
|
||||
feed_obj.http_last_modified = fetched.last_modified
|
||||
_clear_error(feed_obj)
|
||||
feed_obj.save(update_fields=['http_etag', 'http_last_modified', *ERROR_FIELDS])
|
||||
_move_to_permanent_url(feed_obj, fetched.permanent_url)
|
||||
return new_count
|
||||
|
||||
|
||||
|
|
@ -443,12 +623,17 @@ def _feeds_data(user):
|
|||
feeds = list(
|
||||
user.podcast_feeds
|
||||
.annotate(latest_episode_at=Max('episodes__pub_date'))
|
||||
.values('id', 'title', 'artwork_url', 'rss_url', 'last_refreshed_at', 'author', 'added_at', 'auto_queue', 'latest_episode_at')
|
||||
.values(
|
||||
'id', 'title', 'artwork_url', 'rss_url', 'last_refreshed_at', 'author', 'added_at', 'auto_queue',
|
||||
'latest_episode_at', 'last_error', 'last_error_at', 'error_count',
|
||||
)
|
||||
)
|
||||
|
||||
for f in feeds:
|
||||
if f['last_refreshed_at']:
|
||||
f['last_refreshed_at'] = f['last_refreshed_at'].isoformat()
|
||||
if f['last_error_at']:
|
||||
f['last_error_at'] = f['last_error_at'].isoformat()
|
||||
if f['added_at']:
|
||||
f['added_at'] = f['added_at'].isoformat()
|
||||
if f['latest_episode_at']:
|
||||
|
|
@ -957,7 +1142,8 @@ def refresh_feed_now(request):
|
|||
def refresh_all_feeds(request):
|
||||
"""Refresh all of the user's feeds on the server in one go. Feeds the
|
||||
hourly job (or an earlier call) refreshed within `max_age` seconds are
|
||||
skipped. Clients then fetch the result once (state/, inbox/ …)."""
|
||||
skipped. Clients then fetch the result once (state/, inbox/ …).
|
||||
The user asked, so feeds the hourly job backs off from are tried too."""
|
||||
if not request.user.is_authenticated:
|
||||
return JsonResponse({'error': 'authentication required'}, status=401)
|
||||
try:
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue