import os
import sqlite3
import tempfile
from datetime import datetime, timedelta, timezone
from io import StringIO
from unittest.mock import patch

from django.contrib.auth.models import User
from django.core.management import call_command
from django.core.management.base import CommandError
from django.test import TestCase

from .antennapod import filter_words, normalize_url, parse_filter_terms
from .models import EpisodeProgress, PodcastEpisode, PodcastFeed, PodcastQueue, PodcastSettings

BACKUP_DATE = datetime(2026, 9, 1, 12, 0, tzinfo=timezone.utc)
T0 = datetime(2026, 8, 1, tzinfo=timezone.utc)


def ms(dt):
    return int(dt.timestamp() * 1000)


class Backup:
    """A small AntennaPod database: the tables and columns the import reads."""

    SCHEMA = """
        CREATE TABLE Feeds (id INTEGER PRIMARY KEY AUTOINCREMENT, title TEXT, custom_title TEXT,
            download_url TEXT, link TEXT, author TEXT, image_url TEXT, keep_updated INTEGER DEFAULT 1,
            include_filter TEXT DEFAULT '', exclude_filter TEXT DEFAULT '',
            minimal_duration_filter INTEGER DEFAULT -1, new_episodes_action INTEGER DEFAULT 0,
            feed_playback_speed REAL DEFAULT -1, feed_skip_intro INTEGER DEFAULT 0,
            feed_skip_ending INTEGER DEFAULT 0, feed_volume_adaption INTEGER DEFAULT 0,
            auto_download INTEGER DEFAULT 0, state INTEGER DEFAULT 0, tags TEXT);
        CREATE TABLE FeedItems (id INTEGER PRIMARY KEY AUTOINCREMENT, title TEXT, pubDate INTEGER,
            read INTEGER, link TEXT, description TEXT, media INTEGER, feed INTEGER,
            item_identifier TEXT, image_url TEXT);
        CREATE TABLE FeedMedia (id INTEGER PRIMARY KEY AUTOINCREMENT, duration INTEGER, download_url TEXT,
            position INTEGER, playback_completion_date INTEGER, feeditem INTEGER, played_duration INTEGER,
            last_played_time INTEGER, downloaded INTEGER DEFAULT 0, filesize INTEGER DEFAULT 0);
        CREATE TABLE Queue (id INTEGER PRIMARY KEY AUTOINCREMENT, feeditem INTEGER, feed INTEGER);
        CREATE TABLE Favorites (id INTEGER PRIMARY KEY AUTOINCREMENT, feeditem INTEGER, feed INTEGER);
    """

    def __init__(self, path):
        self.path = path
        self.db = sqlite3.connect(path)
        self.db.executescript(self.SCHEMA)

    def feed(self, url, title='Feed', **cols):
        cols = {'download_url': url, 'title': title, **cols}
        cur = self.db.execute(f'INSERT INTO Feeds ({",".join(cols)}) VALUES ({",".join("?" * len(cols))})',
                              list(cols.values()))
        self.db.commit()
        return cur.lastrowid

    def item(self, feed, guid, read=0, audio=None, pub=T0, duration=1800, position=0, completed=None,
             last_played=None, title=None, description='', image_url=None, media=True, downloaded=0, size=0):
        cur = self.db.execute(
            'INSERT INTO FeedItems (title, pubDate, read, description, feed, item_identifier, image_url) '
            'VALUES (?,?,?,?,?,?,?)',
            (title or f'Episode {guid}', ms(pub) if pub else 0, read, description, feed, guid, image_url))
        item_id = cur.lastrowid
        if media:
            self.db.execute(
                'INSERT INTO FeedMedia (duration, download_url, position, playback_completion_date, feeditem, '
                'played_duration, last_played_time, downloaded, filesize) VALUES (?,?,?,?,?,?,?,?,?)',
                (duration * 1000, audio or f'https://cdn.example.org/{guid}.mp3', position * 1000,
                 ms(completed) if completed else 0, item_id, 0, ms(last_played) if last_played else 0,
                 downloaded, size))
        self.db.commit()
        return item_id

    def queue(self, *items):
        for i in items:
            self.db.execute('INSERT INTO Queue (feeditem, feed) SELECT id, feed FROM FeedItems WHERE id=?', (i,))
        self.db.commit()

    def favorite(self, *items):
        for i in items:
            self.db.execute('INSERT INTO Favorites (feeditem, feed) SELECT id, feed FROM FeedItems WHERE id=?', (i,))
        self.db.commit()

    def close(self):
        self.db.close()


class ImportTestCase(TestCase):
    def setUp(self):
        self.user = User.objects.create_user(username='alice', password='pw12345678')
        self.tmp = tempfile.TemporaryDirectory()
        self.ap = Backup(os.path.join(self.tmp.name, 'AntennaPodBackup.db'))

    def tearDown(self):
        self.ap.close()
        self.tmp.cleanup()

    def run_import(self, *args, user='alice'):
        out = StringIO()
        call_command('import_antennapod', self.ap.path, '--user', user,
                     '--backup-date', BACKUP_DATE.isoformat(), *args, stdout=out)
        return out.getvalue()

    def diora_feed(self, url='https://example.org/feed', **kw):
        return PodcastFeed.objects.create(user=self.user, rss_url=url, title=kw.pop('title', 'Diora'), **kw)

    def episode(self, feed, guid, audio=None, **kw):
        return PodcastEpisode.objects.create(feed=feed, guid=guid, title=guid,
                                             audio_url=audio or f'https://cdn.example.org/{guid}.mp3', **kw)

    def progress(self, episode):
        return EpisodeProgress.objects.filter(user=self.user, episode=episode).first()

    def snapshot(self):
        return (
            list(PodcastFeed.objects.order_by('id').values()),
            list(PodcastEpisode.objects.order_by('id').values()),
            list(EpisodeProgress.objects.order_by('id').values()),
            list(PodcastQueue.objects.order_by('id').values('id', 'user_id', 'episode_id', 'position')),
        )

    def queue(self):
        return list(PodcastQueue.objects.filter(user=self.user).order_by('position')
                    .values_list('episode_id', flat=True))


class HelperTests(TestCase):
    def test_normalize_url(self):
        same = ['https://www.Example.org/feed/', 'http://example.org/feed', 'HTTPS://EXAMPLE.ORG/feed',
                'https://example.org:443/feed#x', 'http://example.org:80/feed//']
        self.assertEqual({normalize_url(u) for u in same}, {'example.org/feed'})
        self.assertNotEqual(normalize_url('https://example.org/Feed'), normalize_url('https://example.org/feed'))
        self.assertEqual(normalize_url('https://example.org:8080/f?a=1'), 'example.org:8080/f?a=1')
        for bad in ('antennapod_local:content://x', '', None, 'ftp://example.org/f', 'https://'):
            self.assertIsNone(normalize_url(bad))

    def test_filter_terms(self):
        self.assertEqual(parse_filter_terms('news "long talk" Teaser'), ['news', 'long talk', 'Teaser'])
        self.assertEqual(parse_filter_terms('"news headlines" '), ['news headlines'])
        self.assertEqual(parse_filter_terms('  a   b '), ['a', 'b'])
        self.assertEqual(parse_filter_terms('"open quote'), ['open', 'quote'])
        self.assertEqual(parse_filter_terms('a,b A'), ['a b', 'A'])
        self.assertEqual(parse_filter_terms('x X'), ['x'])
        self.assertEqual(parse_filter_terms(''), [])
        self.assertEqual(filter_words('news "long talk"'), 'news, long talk')
        self.assertEqual(len(filter_words(' '.join(['word%03d' % i for i in range(100)]))), 493)


class FeedImportTests(ImportTestCase):
    def test_feeds_match_by_normalised_url_and_missing_ones_are_created(self):
        mine = self.diora_feed('https://www.Example.org/feed/')
        only_diora = self.diora_feed('https://example.org/only-in-diora')
        self.ap.feed('http://example.org/feed', title='Same')
        self.ap.feed('https://new.example.org/rss', title='Original', custom_title='Renamed', author='Ann',
                     link='https://new.example.org/', image_url='https://new.example.org/art.jpg')
        self.ap.feed('antennapod_local:content://folder', title='Local')
        self.ap.feed('https://gone.example.org/rss', title='Not subscribed', state=1)

        out = self.run_import()

        self.assertEqual(PodcastFeed.objects.filter(user=self.user).count(), 3)
        mine.refresh_from_db()
        self.assertEqual(mine.title, 'Diora')  # matched, not renamed
        self.assertTrue(PodcastFeed.objects.filter(pk=only_diora.pk).exists())
        new = PodcastFeed.objects.get(user=self.user, rss_url='https://new.example.org/rss')
        self.assertEqual((new.title, new.author, new.link, new.artwork_url, new.last_refreshed_at),
                         ('Renamed', 'Ann', 'https://new.example.org/', 'https://new.example.org/art.jpg', None))
        self.assertIn('Feeds:      4 in backup, 3 subscribed (1 not subscribed ignored); 1 matched, 1 created, '
                      '1 skipped (not http/https), 0 duplicates', out)
        self.assertIn('Created feeds (1):\n  - Renamed\n', out)
        self.assertIn('Skipped feeds (1):\n  - Local (antennapod_local:content://folder)\n', out)

    def test_two_backup_feeds_for_one_url(self):
        self.ap.feed('https://example.org/feed')
        self.ap.feed('http://www.example.org/feed/')
        out = self.run_import()
        self.assertEqual(PodcastFeed.objects.filter(user=self.user).count(), 1)
        self.assertIn('0 matched, 1 created, 0 skipped (not http/https), 1 duplicates', out)

    def test_settings(self):
        keep = self.diora_feed('https://example.org/keep', playback_speed=1.5, skip_intro=10, volume=2.0,
                               new_episodes='queue', include_words='x', min_duration=5, auto_download=True)
        changed = self.diora_feed('https://example.org/changed', playback_speed=1.5)
        self.ap.feed('https://example.org/keep', auto_download=1)  # all AntennaPod defaults
        self.ap.feed('https://example.org/changed', feed_playback_speed=1.2000000476837158, feed_skip_intro=30,
                     feed_skip_ending=700, feed_volume_adaption=2, include_filter='news "long talk" ',
                     exclude_filter='"Trailer"', minimal_duration_filter=1500, new_episodes_action=2,
                     auto_download=2)
        self.ap.feed('https://example.org/fast', feed_playback_speed=3.5, feed_volume_adaption=5,
                     minimal_duration_filter=59, new_episodes_action=3)
        self.ap.feed('https://example.org/slow', feed_playback_speed=0.25, feed_volume_adaption=3,
                     new_episodes_action=1)
        self.ap.feed('https://example.org/loud', feed_volume_adaption=4)
        self.ap.feed('https://example.org/quiet', feed_volume_adaption=1)

        out = self.run_import()

        keep.refresh_from_db()
        self.assertEqual(
            (keep.playback_speed, keep.skip_intro, keep.volume, keep.new_episodes, keep.auto_queue,
             keep.include_words, keep.min_duration, keep.auto_download),
            (1.5, 10, 2.0, 'queue', True, 'x', 5, False),  # auto_download "global", --global-auto-download off
        )
        changed.refresh_from_db()
        self.assertEqual(
            (changed.playback_speed, changed.skip_intro, changed.skip_outro, changed.volume, changed.include_words,
             changed.exclude_words, changed.min_duration, changed.new_episodes, changed.auto_download),
            (1.2, 30, 600, 0.2, 'news, long talk', 'Trailer', 25, 'ignore', True),
        )
        feeds = {f.rss_url.rsplit('/', 1)[1]: f for f in PodcastFeed.objects.filter(user=self.user)}
        fast, slow = feeds['fast'], feeds['slow']
        self.assertEqual((fast.playback_speed, fast.volume, fast.min_duration, fast.new_episodes, fast.auto_queue),
                         (3.0, 2.5, 0, 'queue', True))
        self.assertEqual((slow.playback_speed, slow.volume, slow.new_episodes, slow.auto_queue),
                         (0.5, 1.5, 'inbox', False))
        self.assertEqual((feeds['loud'].volume, feeds['quiet'].volume), (2.0, 0.5))
        self.assertIn('Settings:   changed: speed 3, skip intro 1, skip outro 1, volume 5, include words 1, '
                      'exclude words 1, min duration 1, new episodes 2, auto-download 2', out)

    def test_not_kept_updated_feeds_bring_nothing_new(self):
        self.ap.feed('https://example.org/a', keep_updated=0)
        self.ap.feed('https://example.org/b', keep_updated=0, new_episodes_action=1)
        self.ap.feed('https://example.org/c', keep_updated=0, new_episodes_action=3)
        self.ap.feed('https://example.org/d', keep_updated=1)
        out = self.run_import()
        new = dict(PodcastFeed.objects.filter(user=self.user).values_list('rss_url', 'new_episodes'))
        self.assertEqual(new, {'https://example.org/a': 'ignore', 'https://example.org/b': 'ignore',
                               'https://example.org/c': 'queue', 'https://example.org/d': 'inbox'})
        self.assertIn('2 feeds not kept updated in AntennaPod -> new episodes "ignore"', out)

    def test_auto_download(self):
        self.diora_feed('https://example.org/off', auto_download=True)
        for name, code in (('off', 0), ('global', 1), ('on', 2), ('global2', 1)):
            self.ap.feed(f'https://example.org/{name}', auto_download=code)

        def values():
            return dict(PodcastFeed.objects.filter(user=self.user).values_list('rss_url', 'auto_download'))

        out = self.run_import()
        self.assertEqual(values(), {'https://example.org/off': False, 'https://example.org/global': False,
                                    'https://example.org/on': True, 'https://example.org/global2': False})
        self.assertIn("Auto-download: on for 1 feeds (1 on in AntennaPod, 2 use AntennaPod's global setting -> off "
                      'by --global-auto-download); default for new subscriptions: off\n', out)
        self.assertFalse(PodcastSettings.for_user(self.user).auto_download_default)

        out = self.run_import('--global-auto-download', 'on')
        self.assertEqual(values(), {'https://example.org/off': False, 'https://example.org/global': True,
                                    'https://example.org/on': True, 'https://example.org/global2': True})
        self.assertIn("Auto-download: on for 3 feeds (1 on in AntennaPod, 2 use AntennaPod's global setting -> on "
                      'by --global-auto-download); default for new subscriptions: on (set by this import)\n', out)
        self.assertTrue(PodcastSettings.for_user(self.user).auto_download_default)
        # Once more: nothing to set any more; "off" leaves the default alone.
        self.assertIn('default for new subscriptions: on\n', self.run_import('--global-auto-download', 'on'))
        self.assertIn('default for new subscriptions: on\n', self.run_import())

    def test_created_feeds_start_with_the_users_default(self):
        PodcastSettings.objects.create(user=self.user, auto_download_default=True)
        self.ap.feed('https://example.org/explicit-off', auto_download=0)
        self.ap.feed('https://example.org/on', auto_download=2)
        self.run_import()
        values = dict(PodcastFeed.objects.filter(user=self.user).values_list('rss_url', 'auto_download'))
        self.assertEqual(values, {'https://example.org/explicit-off': False, 'https://example.org/on': True})


class EpisodeImportTests(ImportTestCase):
    def setUp(self):
        super().setUp()
        self.feed = self.diora_feed()
        self.ap_feed = self.ap.feed('https://example.org/feed')

    def test_matching_and_creating_episodes(self):
        by_guid = self.episode(self.feed, 'g1')
        by_url = self.episode(self.feed, 'other-guid', audio='http://www.cdn.example.org/g2.mp3')
        stateless = self.episode(self.feed, 'g4')
        untouched = self.episode(self.feed, 'g6')
        self.ap.item(self.ap_feed, 'g1', read=1)
        self.ap.item(self.ap_feed, 'g2', read=1, audio='https://cdn.example.org/g2.mp3')
        self.ap.item(self.ap_feed, 'g3', read=1, title='Created', description='<p>Notes</p>', duration=3600,
                     pub=T0 + timedelta(days=3), image_url='https://example.org/g3.jpg')
        self.ap.item(self.ap_feed, 'g4', read=0)
        self.ap.item(self.ap_feed, 'g5', read=0)  # no state, not in diora: not imported
        self.ap.item(self.ap_feed, None, read=-1, audio='https://cdn.example.org/noguid.mp3')
        self.ap.item(self.ap_feed, 'g7', read=1, media=False)

        out = self.run_import()

        self.assertTrue(self.progress(by_guid).played)
        self.assertTrue(self.progress(by_url).played)
        created = PodcastEpisode.objects.get(feed=self.feed, guid='g3')
        self.assertEqual(
            (created.title, created.description, created.audio_url, created.duration_seconds, created.pub_date,
             created.artwork_url),
            ('Created', '<p>Notes</p>', 'https://cdn.example.org/g3.mp3', 3600, T0 + timedelta(days=3),
             'https://example.org/g3.jpg'),
        )
        self.assertTrue(self.progress(created).played)
        # Without a guid, the audio URL is the guid (as when diora reads the feed).
        noguid = PodcastEpisode.objects.get(feed=self.feed, guid='https://cdn.example.org/noguid.mp3')
        self.assertFalse(self.progress(noguid).dismissed)
        self.assertFalse(PodcastEpisode.objects.filter(guid__in=['g5', 'g7']).exists())
        # Known to AntennaPod and not new there: out of the inbox, not played.
        p = self.progress(stateless)
        self.assertEqual((p.dismissed, p.played), (True, False))
        self.assertIsNone(self.progress(untouched))
        self.assertIn('Episodes:   5 with a state: 1 matched by guid, 1 by URL, 2 created, '
                      '1 skipped (no audio URL), 0 in skipped feeds', out)
        self.assertIn('1 without a state matched to diora episodes', out)
        self.assertIn('Inbox:      4 before, 2 after', out)  # g6 and noguid

    def test_states(self):
        played_at = T0 + timedelta(days=10)
        self.ap.item(self.ap_feed, 'played', read=1, position=0, completed=played_at, last_played=played_at)
        self.ap.item(self.ap_feed, 'marked', read=1, pub=T0 + timedelta(days=2))
        self.ap.item(self.ap_feed, 'started', read=0, position=600, last_played=played_at - timedelta(hours=1))
        self.ap.item(self.ap_feed, 'new', read=-1, pub=T0 + timedelta(days=5))
        self.ap.item(self.ap_feed, 'nodate', read=1, pub=None)
        self.ap.item(self.ap_feed, 'future', read=1, pub=BACKUP_DATE + timedelta(days=30))
        self.run_import()

        def state(guid):
            p = self.progress(PodcastEpisode.objects.get(guid=guid))
            return p.played, p.dismissed, p.position_seconds, p.updated_at

        self.assertEqual(state('played'), (True, True, 0, played_at))
        self.assertEqual(state('marked'), (True, True, 0, T0 + timedelta(days=2)))  # no activity: pub date
        self.assertEqual(state('started'), (False, True, 600, played_at - timedelta(hours=1)))
        self.assertEqual(state('new'), (False, False, 0, T0 + timedelta(days=5)))
        self.assertEqual(state('nodate'), (True, True, 0, BACKUP_DATE))
        self.assertEqual(state('future'), (True, True, 0, BACKUP_DATE))

    def test_favourites_are_a_union(self):
        diora_star = self.episode(self.feed, 'a')
        p = EpisodeProgress.objects.create(user=self.user, episode=diora_star, favorite=True)
        EpisodeProgress.objects.filter(pk=p.pk).update(updated_at=BACKUP_DATE - timedelta(days=5))
        both = self.episode(self.feed, 'b')
        self.ap.item(self.ap_feed, 'a', read=1)  # AntennaPod wins, but diora's star stays
        b = self.ap.item(self.ap_feed, 'b', read=0)
        c = self.ap.item(self.ap_feed, 'c', read=0)
        self.ap.favorite(b, c)
        out = self.run_import()
        p = self.progress(diora_star)
        self.assertEqual((p.favorite, p.played), (True, True))
        self.assertTrue(self.progress(both).favorite)
        self.assertTrue(self.progress(PodcastEpisode.objects.get(guid='c')).favorite)
        self.assertIn('Favourites: 2 in backup; 1 before, 3 after', out)

    def test_merge_rule(self):
        newer = self.episode(self.feed, 'newer')
        older = self.episode(self.feed, 'older')
        dismissed = self.episode(self.feed, 'dismissed')
        after_backup = BACKUP_DATE + timedelta(days=1)
        for ep, values in ((newer, dict(position_seconds=100)), (older, dict(position_seconds=100)),
                           (dismissed, dict(dismissed=True))):
            p = EpisodeProgress.objects.create(user=self.user, episode=ep, **values)
            EpisodeProgress.objects.filter(pk=p.pk).update(
                updated_at=after_backup if ep is not older else BACKUP_DATE - timedelta(days=1))
        self.ap.item(self.ap_feed, 'newer', read=1, completed=BACKUP_DATE - timedelta(days=2))
        self.ap.item(self.ap_feed, 'older', read=1, completed=BACKUP_DATE - timedelta(days=2))
        n = self.ap.item(self.ap_feed, 'dismissed', read=-1)
        self.ap.favorite(n)
        out = self.run_import()
        p = self.progress(newer)
        self.assertEqual((p.played, p.position_seconds, p.updated_at), (False, 100, after_backup))
        p = self.progress(older)
        self.assertEqual((p.played, p.position_seconds, p.updated_at), (True, 0, BACKUP_DATE - timedelta(days=2)))
        # diora's newer "not new" stays; the star comes over anyway, without
        # touching updated_at.
        p = self.progress(dismissed)
        self.assertEqual((p.dismissed, p.favorite, p.updated_at), (True, True, after_backup))
        self.assertIn('Progress:   0 new, 1 updated, 0 unchanged, 2 kept (diora newer)', out)

    def test_activity_after_diora_change_wins(self):
        ep = self.episode(self.feed, 'e')
        p = EpisodeProgress.objects.create(user=self.user, episode=ep, position_seconds=50)
        EpisodeProgress.objects.filter(pk=p.pk).update(updated_at=BACKUP_DATE + timedelta(hours=1))
        self.ap.item(self.ap_feed, 'e', read=0, position=300, last_played=BACKUP_DATE + timedelta(hours=2))
        self.run_import()
        self.assertEqual(self.progress(ep).position_seconds, 300)

    def test_queue(self):
        x = self.episode(self.feed, 'x')
        y = self.episode(self.feed, 'y')
        PodcastQueue.objects.create(user=self.user, episode=x, position=0)
        PodcastQueue.objects.create(user=self.user, episode=y, position=1)
        a = self.ap.item(self.ap_feed, 'a', read=-1)
        b = self.ap.item(self.ap_feed, 'b', read=0)
        y_ap = self.ap.item(self.ap_feed, 'y', read=0)
        local = self.ap.feed('antennapod_local:x')
        lost = self.ap.item(local, 'l', read=0)
        self.ap.queue(b, y_ap, lost, a)
        out = self.run_import()
        ids = {e.guid: e.id for e in PodcastEpisode.objects.filter(feed=self.feed)}
        self.assertEqual(self.queue(), [ids['b'], y.id, ids['a'], x.id])
        self.assertEqual(list(PodcastQueue.objects.filter(user=self.user).order_by('position')
                              .values_list('position', flat=True)), [0, 1, 2, 3])
        # Queued is not new, even if AntennaPod still had it as new.
        self.assertTrue(self.progress(PodcastEpisode.objects.get(guid='a')).dismissed)
        self.assertIn('Queue:      4 in backup (1 not found); 2 before, 4 after (1 diora entries after', out)

    def test_queue_kept_sorted(self):
        PodcastSettings.objects.create(user=self.user, keep_sorted='old')
        a = self.ap.item(self.ap_feed, 'a', read=0, pub=T0 + timedelta(days=2))
        b = self.ap.item(self.ap_feed, 'b', read=0, pub=T0 + timedelta(days=1))
        self.ap.queue(a, b)
        out = self.run_import()
        self.assertEqual([PodcastEpisode.objects.get(pk=i).guid for i in self.queue()], ['b', 'a'])
        self.assertIn('kept sorted by "old"', out)

    def test_dry_run_writes_nothing(self):
        self.episode(self.feed, 'g1')
        PodcastQueue.objects.create(user=self.user, episode=self.episode(self.feed, 'q'), position=0)
        g = self.ap.item(self.ap_feed, 'g1', read=1)
        self.ap.item(self.ap_feed, 'g2', read=-1)
        self.ap.feed('https://new.example.org/rss', feed_playback_speed=2.0)
        self.ap.queue(g)
        before = self.snapshot()
        out = self.run_import('--dry-run')
        self.assertEqual(self.snapshot(), before)
        self.assertIn('DRY RUN', out)
        self.assertIn('1 created', out)
        self.assertIn('Queue:      1 in backup (0 not found); 1 before, 2 after', out)
        self.assertNotIn('DRY RUN', self.run_import())
        self.assertNotEqual(self.snapshot(), before)

    def test_running_twice_changes_nothing_more(self):
        self.episode(self.feed, 'g1')
        self.episode(self.feed, 'stateless')
        PodcastQueue.objects.create(user=self.user, episode=self.episode(self.feed, 'q'), position=0)
        g1 = self.ap.item(self.ap_feed, 'g1', read=0, position=60, last_played=T0)
        self.ap.item(self.ap_feed, 'stateless', read=0)
        g2 = self.ap.item(self.ap_feed, 'g2', read=-1)
        self.ap.item(self.ap_feed, 'g3', read=1)
        self.ap.item(self.ap_feed, 'g4', read=1, pub=None)
        self.ap.feed('https://new.example.org/rss', feed_playback_speed=2.0, keep_updated=0)
        self.ap.queue(g2, g1)
        self.ap.favorite(g1)
        self.run_import()
        first = self.snapshot()
        out = self.run_import()
        self.assertEqual(self.snapshot(), first)
        self.assertIn('2 matched, 0 created', out)
        self.assertIn('Settings:   changed: speed 0, skip intro 0, skip outro 0, volume 0, include words 0, '
                      'exclude words 0, min duration 0, new episodes 0, auto-download 0', out)
        self.assertIn('0 created, 0 skipped', out)
        self.assertIn('Progress:   0 new, 0 updated, 4 unchanged, 0 kept (diora newer); 0 taken out', out)


class DownloadImportTests(ImportTestCase):
    """Episodes on the phone in AntennaPod: offered to the app for download."""

    def setUp(self):
        super().setUp()
        self.feed = self.diora_feed()
        self.ap_feed = self.ap.feed('https://example.org/feed')

    def offered(self):
        return PodcastSettings.for_user(self.user).import_downloads

    def test_downloads_are_offered(self):
        played = self.episode(self.feed, 'played')
        known = self.episode(self.feed, 'known')
        self.ap.item(self.ap_feed, 'played', read=1, pub=T0 + timedelta(days=1), downloaded=1, size=30_000_000)
        self.ap.item(self.ap_feed, 'known', read=0, pub=T0 + timedelta(days=3), downloaded=ms(T0), size=20_400_000)
        self.ap.item(self.ap_feed, 'on-phone', read=0, pub=T0 + timedelta(days=2), downloaded=1,
                     size=27_634_898_567_168)  # garbage: unknown
        self.ap.item(self.ap_feed, 'not-downloaded', read=0)
        self.ap.item(self.ap_feed, 'local', read=0, audio='antennapod_local:x', downloaded=1, size=5)
        self.ap.item(self.ap.feed('antennapod_local:folder'), 'file', read=0, downloaded=1, size=7)

        before = self.snapshot()
        out = self.run_import('--dry-run')
        self.assertEqual(self.snapshot(), before)
        self.assertFalse(PodcastSettings.objects.filter(user=self.user).exists())
        self.assertIn('Downloads:  3 on the phone in AntennaPod (50 MB) -> offered to the app for download '
                      '(1 episodes created for them); 2 not found (feeds not imported, no audio URL)\n', out)

        out = self.run_import()
        created = PodcastEpisode.objects.get(feed=self.feed, guid='on-phone')
        self.assertEqual(self.offered(), [  # newest first
            {'episode_id': known.id, 'size': 20_400_000}, {'episode_id': created.id, 'size': 0},
            {'episode_id': played.id, 'size': 30_000_000}])
        # Created only to be downloaded; not new in AntennaPod, so not in the inbox.
        p = self.progress(created)
        self.assertEqual((p.played, p.dismissed), (False, True))
        self.assertFalse(PodcastEpisode.objects.filter(guid__in=['not-downloaded', 'local', 'file']).exists())
        self.assertIn('Episodes:   1 with a state: 1 matched by guid, 0 by URL, 0 created, '
                      '0 skipped (no audio URL), 0 in skipped feeds', out)

        # Another backup later: its downloads replace the list.
        self.ap.db.execute('UPDATE FeedMedia SET downloaded = 0 WHERE filesize != 30000000')
        self.ap.db.commit()
        out = self.run_import()
        self.assertEqual(self.offered(), [{'episode_id': played.id, 'size': 30_000_000}])
        self.assertIn('Downloads:  1 on the phone in AntennaPod (30 MB) -> offered to the app for download\n', out)
        self.ap.db.execute('UPDATE FeedMedia SET downloaded = 0')
        self.ap.db.commit()
        out = self.run_import()
        self.assertEqual(self.offered(), [])
        self.assertIn('Downloads:  0 on the phone in AntennaPod (0 MB)', out)

    def test_no_downloads_offers_nothing(self):
        self.episode(self.feed, 'known')
        self.ap.item(self.ap_feed, 'known', read=0, downloaded=1, size=20_400_000)
        self.run_import()
        self.assertEqual(len(self.offered()), 1)
        out = self.run_import('--no-downloads')
        self.assertEqual(self.offered(), [])
        self.assertIn('Downloads:  1 on the phone in AntennaPod (20 MB) -> not offered (--no-downloads)\n', out)

    def test_one_entry_per_episode(self):
        ep = self.episode(self.feed, 'g')
        self.ap.item(self.ap_feed, 'g', read=1, downloaded=1, size=10)
        self.ap.item(self.ap.feed('http://www.example.org/feed/'), 'g', read=1, downloaded=1, size=10)
        self.run_import()
        self.assertEqual(self.offered(), [{'episode_id': ep.id, 'size': 10}])

    def test_backup_without_download_columns(self):
        self.ap.db.executescript('ALTER TABLE FeedMedia DROP COLUMN downloaded; '
                                 'ALTER TABLE FeedMedia DROP COLUMN filesize;')
        self.ap.db.execute('INSERT INTO FeedItems (title, read, feed, item_identifier) VALUES (?,?,?,?)',
                           ('G', 1, self.ap_feed, 'g'))
        self.ap.db.commit()
        PodcastSettings.objects.create(user=self.user, import_downloads=[{'episode_id': 1, 'size': 1}])
        out = self.run_import()
        self.assertEqual(self.offered(), [])
        self.assertIn('Downloads:  0 on the phone', out)


class ImportDownloadsViewTests(TestCase):
    """GET import-downloads/, its clear/, and the count in state/."""

    def setUp(self):
        self.user = User.objects.create_user(username='alice', password='pw12345678')
        self.client.force_login(self.user)
        feed = PodcastFeed.objects.create(user=self.user, rss_url='https://example.org/a', title='A')
        self.ep = PodcastEpisode.objects.create(feed=feed, guid='e', title='E', audio_url='https://example.org/e.mp3',
                                                duration_seconds=1800, pub_date=T0)
        EpisodeProgress.objects.create(user=self.user, episode=self.ep, position_seconds=120)
        no_audio = PodcastEpisode.objects.create(feed=feed, guid='n', title='N', audio_url='')
        bob = User.objects.create_user(username='bob', password='pw12345678')
        bobs_feed = PodcastFeed.objects.create(user=bob, rss_url='https://example.org/a', title='A')
        bobs = PodcastEpisode.objects.create(feed=bobs_feed, guid='b', title='B', audio_url='https://example.org/b.mp3')
        gone_feed = PodcastFeed.objects.create(user=self.user, rss_url='https://example.org/gone', title='Gone')
        gone = PodcastEpisode.objects.create(feed=gone_feed, guid='g', title='G', audio_url='https://example.org/g.mp3')
        gone_feed.delete()  # unsubscribed since the import
        PodcastSettings.objects.create(user=self.user, import_downloads=[
            {'episode_id': i, 'size': 1000 + n} for n, i in enumerate((bobs.id, self.ep.id, no_audio.id, gone.id))])

    def test_offered_episodes(self):
        data = self.client.get('/podcasts/import-downloads/').json()
        self.assertEqual(data['bytes'], 1001)
        [e] = data['episodes']
        self.assertEqual((e['id'], e['title'], e['feed__title'], e['duration_seconds'], e['pub_date'],
                          e['position_seconds'], e['in_queue'], e['size']),
                         (self.ep.id, 'E', 'A', 1800, T0.isoformat(), 120, False, 1001))
        # The same keys as an inbox entry of state/, plus the size.
        state = self.client.get('/podcasts/state/').json()
        self.assertEqual(set(e), set(state['inbox'][0]) | {'size'})
        self.assertEqual(state['import_downloads'], {'count': 1, 'bytes': 1001})

    def test_clear(self):
        self.assertEqual(self.client.get('/podcasts/import-downloads/clear/').status_code, 405)
        r = self.client.post('/podcasts/import-downloads/clear/', '{}', content_type='application/json')
        self.assertEqual(r.json(), {'ok': True})
        self.assertEqual(PodcastSettings.objects.get(user=self.user).import_downloads, [])
        self.assertEqual(self.client.get('/podcasts/import-downloads/').json(), {'episodes': [], 'bytes': 0})
        self.assertEqual(self.client.get('/podcasts/state/').json()['import_downloads'], {'count': 0, 'bytes': 0})

    def test_other_user_and_login(self):
        self.client.force_login(User.objects.get(username='bob'))
        self.assertEqual(self.client.get('/podcasts/import-downloads/').json(), {'episodes': [], 'bytes': 0})
        self.assertEqual(self.client.get('/podcasts/state/').json()['import_downloads'], {'count': 0, 'bytes': 0})
        self.client.logout()
        self.assertEqual(self.client.get('/podcasts/import-downloads/').status_code, 401)
        self.assertEqual(self.client.post('/podcasts/import-downloads/clear/').status_code, 401)
        self.assertEqual(len(PodcastSettings.objects.get(user=self.user).import_downloads), 4)


class QuietInboxTests(ImportTestCase):
    """What AntennaPod never showed as new leaves diora's inbox too."""

    def inbox(self):
        hidden = EpisodeProgress.objects.filter(user=self.user, dismissed=True).values('episode_id')
        return sorted(PodcastEpisode.objects.filter(feed__user=self.user).exclude(id__in=hidden)
                      .values_list('guid', flat=True))

    def setUp(self):
        super().setUp()
        archived = self.diora_feed('https://example.org/archived')
        self.ap_feed = self.ap.feed('https://example.org/archived', keep_updated=0)
        self.ap.item(self.ap_feed, 'ap-new', read=-1)
        self.episode(archived, 'ap-new', pub_date=T0)
        self.episode(archived, 'unknown-old', pub_date=T0 + timedelta(days=3))   # AntennaPod never fetched it
        self.episode(archived, 'unknown-late', pub_date=BACKUP_DATE + timedelta(days=1))
        self.episode(archived, 'undated')
        self.ap.feed('https://example.org/kept')
        kept = self.diora_feed('https://example.org/kept')
        self.episode(kept, 'kept-unknown', pub_date=T0)      # not archived: stays
        only = self.diora_feed('https://example.org/only-diora')
        self.episode(only, 'only-old', pub_date=T0)
        self.episode(only, 'only-late', pub_date=BACKUP_DATE + timedelta(days=1))

    def test_archived_podcasts(self):
        out = self.run_import()
        self.assertEqual(self.inbox(), ['ap-new', 'kept-unknown', 'only-late', 'only-old', 'unknown-late'])
        self.assertIn('out of it: 2 of podcasts whose new episodes are ignored (AntennaPod never showed them)\n', out)
        p = self.progress(PodcastEpisode.objects.get(guid='unknown-old'))
        self.assertEqual((p.played, p.dismissed, p.updated_at), (False, True, BACKUP_DATE))

    def test_quiet_diora_only(self):
        out = self.run_import('--quiet-diora-only')
        self.assertEqual(self.inbox(), ['ap-new', 'kept-unknown', 'only-late', 'unknown-late'])
        self.assertIn(', 1 of podcasts only diora has', out)

    def test_replace_is_a_clean_slate(self):
        archived = PodcastFeed.objects.get(rss_url='https://example.org/archived')
        ap_new = PodcastEpisode.objects.get(guid='ap-new')
        old = PodcastEpisode.objects.get(guid='unknown-old')
        late = PodcastEpisode.objects.get(guid='unknown-late')
        PodcastQueue.objects.create(user=self.user, episode=late, position=0)
        other = PodcastEpisode.objects.get(guid='kept-unknown')
        PodcastQueue.objects.create(user=self.user, episode=other, position=1)
        EpisodeProgress.objects.create(user=self.user, episode=old, position_seconds=30)  # changed in diora now
        queued = self.ap.item(self.ap_feed, 'unknown-late', read=0, pub=None)
        self.ap.queue(queued)
        out = self.run_import('--replace')
        self.assertFalse(PodcastFeed.objects.filter(rss_url='https://example.org/only-diora').exists())
        self.assertFalse(PodcastEpisode.objects.filter(guid__startswith='only-').exists())
        self.assertEqual(self.queue(), [late.id])
        self.assertTrue(self.progress(old).dismissed)  # AntennaPod wins
        self.assertIn('ap-new', self.inbox())
        self.assertTrue(archived.episodes.filter(id=ap_new.id).exists())
        self.assertIn(', 1 only in diora unsubscribed (--replace)', out)
        self.assertIn(', 1 removed (--replace))', out)
        self.assertIn('Unsubscribed feeds (1):\n  - Diora\n', out)

    def test_diora_newer_stays(self):
        ep = PodcastEpisode.objects.get(guid='unknown-old')
        EpisodeProgress.objects.create(user=self.user, episode=ep, position_seconds=30)  # updated now
        self.run_import()
        self.assertIn('unknown-old', self.inbox())


class AfterFirstRefreshTests(ImportTestCase):
    """The way to move: import, let the new feeds be fetched, import again."""

    def test_second_import_after_the_first_fetch(self):
        import feedparser

        from .views import FeedFetch, _refresh_feed

        ap_feed = self.ap.feed('https://new.example.org/rss')
        days = {'old1': 1, 'old2': 2, 'new': 3, 'newest': 4}
        for guid, read in (('old1', 0), ('old2', 1), ('new', -1), ('newest', 0)):
            self.ap.item(ap_feed, guid, read=read, pub=T0 + timedelta(days=days[guid]))
        self.run_import()
        feed = PodcastFeed.objects.get(user=self.user)
        self.assertEqual(sorted(feed.episodes.values_list('guid', flat=True)), ['new', 'old2'])

        def date(d):
            return (T0 + timedelta(days=d)).strftime('%a, %d %b %Y %H:%M:%S GMT')

        items = ''.join(
            f'<item><guid>{g}</guid><title>{g}</title><pubDate>{date(d)}</pubDate><enclosure url="https://cdn.example.org/{g}.mp3" type="audio/mpeg" length="1"/></item>'
            for g, d in days.items())
        rss = f'<?xml version="1.0"?><rss version="2.0"><channel><title>New</title>{items}</channel></rss>'
        with patch('podcasts.views._fetch_feed', lambda url, **kw: FeedFetch(200, feedparser.parse(rss))):
            _refresh_feed(feed)
        self.assertEqual(feed.episodes.count(), 4)  # nothing twice

        def inbox():
            hidden = EpisodeProgress.objects.filter(user=self.user, dismissed=True).values('episode_id')
            return sorted(feed.episodes.exclude(id__in=hidden).values_list('guid', flat=True))

        self.assertEqual(inbox(), ['new', 'newest'])  # the first-fetch rule leaves the newest
        out = self.run_import()
        self.assertEqual(inbox(), ['new'])
        self.assertIn('2 without a state matched to diora episodes', out)
        self.assertIn('1 taken out of the inbox', out)


class CommandErrorTests(ImportTestCase):
    def test_unknown_user_and_bad_files(self):
        with self.assertRaisesMessage(CommandError, 'No such user'):
            self.run_import(user='nobody')
        with self.assertRaisesMessage(CommandError, 'No such file'):
            call_command('import_antennapod', os.path.join(self.tmp.name, 'missing.db'), '--user', 'alice')
        other = os.path.join(self.tmp.name, 'other.db')
        sqlite3.connect(other).close()
        with self.assertRaisesMessage(CommandError, 'not an AntennaPod backup'):
            call_command('import_antennapod', other, '--user', 'alice')
        text = os.path.join(self.tmp.name, 'text.db')
        with open(text, 'w') as f:
            f.write('not a database at all, just text' * 10)
        with self.assertRaisesMessage(CommandError, 'not'):
            call_command('import_antennapod', text, '--user', 'alice')
        with self.assertRaisesMessage(CommandError, 'Invalid --backup-date'):
            call_command('import_antennapod', self.ap.path, '--user', 'alice', '--backup-date', 'gestern')