recently_viewed.yaml imports by diffing

recently_viewed.yaml declares one section, its files keyed by
media-relative path, comparing viewed, progress and position.  The
database's rows are the window the config saves (the newest
RECENTLY_VIEWED_LIMIT views, inside the media directory), so an
unchanged import writes nothing.

A file removed from the config has its viewing history cleared
(decision 6), but only when its view is at least as recent as the
config's oldest entry: an older view fell off the end of the window.  A
malformed entry keeps its file's history and is not counted as skipped
(that would stop every recently viewed save).  Without deletes, progress
written after the config is kept, and a newer plain view keeps its
`viewed` but takes the config's position (as before).  prepare finds the
files of new entries on disk before the write; a deleted file's history
goes with it.

Moving a viewed FileGroup counts as a recently_viewed.yaml change (ORM,
the bulk move, and a refresh that changes the primary file).  The media
directory checks are batched (each resolved the media directory again).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

wrolpi/collections/config.py

@@ -11,6 +11,7 @@ from wrolpi.config_import import ConfigSection, SectionDiff, chunks, media_relat
 from wrolpi.config_version import count_config_change
 from wrolpi.db import get_db_session
 from wrolpi.switches import ActivateSwitchMethod, register_switch_handler
+from wrolpi.files.progress import RecentlyViewedConfig, save_recently_viewed_config
 from wrolpi.tags import TagFile, TagsConfig, save_tags_config
 
 logger = logger.getChild(__name__)
@@ -435,9 +436,11 @@ def file_groups_moved(session, file_group_ids):
 
     playlists.yaml references file items by path, so it must be re-saved (and the playlist
     directories re-synced) or the next import will not find the file and will drop the item.
-    tags.yaml saves tagged files by path too.  The move (often raw SQL) is counted as each config's change.
+    tags.yaml and recently_viewed.yaml save files by path too.  The move (often raw SQL) is counted as each config's
+    change.
 
     `file_group_ids` may be an iterable of ids or a select of ids."""
+    from wrolpi.files.models import FileGroup
     from .models import Collection, CollectionItem
 
     if session.query(TagFile.file_group_id).filter(TagFile.file_group_id.in_(file_group_ids)).first():
@@ -448,6 +451,10 @@ def file_groups_moved(session, file_group_ids):
         .join(Collection, Collection.id == CollectionItem.collection_id) \
         .filter(Collection.kind == 'playlist', CollectionItem.file_group_id.in_(file_group_ids)) \
         .first()
+    if session.query(FileGroup.id).filter(FileGroup.id.in_(file_group_ids), FileGroup.viewed.isnot(None)).first():
+        count_config_change(session, RecentlyViewedConfig.file_name)
+        save_recently_viewed_config.activate_switch()
+
     if in_playlist:
         count_config_change(session, PlaylistsConfig.file_name)
         save_playlists_config.activate_switch()

wrolpi/files/models.py

@@ -917,6 +917,10 @@ def _playlist_file_group_moved(file_group: FileGroup) -> bool:
 
 
 register_ownership('playlists.yaml', FileGroup, columns={'primary_path'}, row=_playlist_file_group_moved)
+# recently_viewed.yaml saves viewed files by path.
+register_ownership('recently_viewed.yaml', FileGroup, columns={'primary_path'},
+                   row=lambda file_group: bool(inspect(file_group).attrs.primary_path.history.deleted)
+                   and file_group.viewed is not None)
 
 
 class Directory(ModelHelper, Base):

wrolpi/files/progress.py

@@ -18,13 +18,15 @@ import os
 import pathlib
 from dataclasses import dataclass, field
 from datetime import datetime
-from typing import List, Optional, Tuple
+from typing import Dict, List, Optional, Set, Tuple
 
 import pytz
 
 from wrolpi import dates
-from wrolpi.common import ConfigFile, logger, get_media_directory, get_paths_in_media_directory
-from wrolpi.config_version import SAVE
+from sqlalchemy.orm import Session
+
+from wrolpi.common import ConfigFile, logger, get_media_directory, get_paths_in_media_directory, chunks
+from wrolpi.config_import import ConfigSection
 from wrolpi.db import get_db_session
 from wrolpi.errors import ValidationError, NoPrimaryFile
 from wrolpi.events import Events
@@ -128,8 +130,92 @@ def _parse_entry(entry) -> Optional[Tuple[str, datetime, Optional[float], Option
         return None
 
 
+def _viewed_row(viewed, progress, position) -> dict:
+    return dict(viewed=viewed, progress=progress, position=position or None)
+
+
+def _entry_keys(paths: List[str]) -> Dict[str, str]:
+    """{entry path: its path relative to the media directory}, without those outside it (an absolute `path` replaces
+    the media directory in the join, and `..` can leave it)."""
+    media_directory = get_media_directory()
+    absolute_paths = {i: pathlib.Path(os.path.normpath(media_directory / i)) for i in paths}
+    inside = set(get_paths_in_media_directory(list(absolute_paths.values())))
+    return {i: str(j.relative_to(media_directory)) for i, j in absolute_paths.items() if j in inside}
+
+
+def _malformed_keys(config: dict) -> set:
+    paths = [i['path'] for i in config.get('files') or []
+             if isinstance(i, dict) and isinstance(i.get('path'), str) and not _parse_entry(i)]
+    return set(_entry_keys(paths).values())
+
+
+class ViewedSection(ConfigSection):
+    """recently_viewed.yaml's files, keyed by path (relative to the media directory), comparing when each was viewed,
+    and how far it was read or watched.  Only the newest views are saved (`RECENTLY_VIEWED_LIMIT`); older ones are
+    not in the database's rows."""
+    name = 'viewed'
+
+    def file_rows(self, config: dict) -> Tuple[Dict[str, dict], Set[str]]:
+        # Malformed entries are not counted as skipped: recently viewed is saved after every view, and a skipped
+        # entry would stop that.  (`prepare` keeps their history.)
+        rows = dict()
+        entries = list(filter(None, map(_parse_entry, config.get('files') or [])))
+        keys = _entry_keys([i[0] for i in entries])
+        for path, viewed, progress, position in entries:
+            if not (key := keys.get(path)):
+                logger.warning(f'Not importing recently viewed file outside the media directory: {path}')
+                continue
+            # The first (newest) entry of a file wins.
+            rows.setdefault(key, _viewed_row(viewed, progress, position))
+        return rows, set()
+
+    def db_rows(self, session: Session) -> Dict[str, dict]:
+        # wrolpi.files.models imports this module.
+        from wrolpi.files.models import FileGroup
+        media_directory = get_media_directory()
+        query = session.query(FileGroup.primary_path, FileGroup.viewed, FileGroup.progress, FileGroup.position) \
+            .filter(FileGroup.viewed.isnot(None)) \
+            .order_by(FileGroup.viewed.desc(), FileGroup.id) \
+            .limit(RECENTLY_VIEWED_LIMIT)
+        rows = query.all()
+        # A row outside the media directory is not saved (see `dump_config`).
+        inside = set(get_paths_in_media_directory([i[0] for i in rows]))
+        return {str(path.relative_to(media_directory)): _viewed_row(*values)
+                for path, *values in rows if path in inside}
+
+    def upsert(self, session: Session, rows: Dict[str, dict]):
+        from wrolpi.files.models import FileGroup
+        media_directory = get_media_directory()
+        file_groups = dict()
+        for paths in chunks([str(media_directory / i) for i in rows], 500):
+            file_groups.update((str(i.primary_path.relative_to(media_directory)), i) for i in
+                               session.query(FileGroup).filter(FileGroup.primary_path.in_(paths)))
+        for key, row in rows.items():
+            if not (file_group := file_groups.get(key)):
+                if not row.get('files'):
+                    continue
+                try:
+                    file_group = FileGroup.from_paths(session, *row['files'])
+                except NoPrimaryFile:
+                    logger.warning(f'Cannot import recently viewed file: {key}')
+                    continue
+                session.add(file_group)
+            file_group.viewed, file_group.progress, file_group.position = \
+                row['viewed'], row['progress'], row['position']
+        session.flush()
+
+    def delete(self, session: Session, keys: Set[str]):
+        """A file removed from the config was not viewed."""
+        from wrolpi.files.models import FileGroup
+        media_directory = get_media_directory()
+        for paths in chunks([str(media_directory / i) for i in keys], 500):
+            for file_group in session.query(FileGroup).filter(FileGroup.primary_path.in_(paths)):
+                file_group.viewed = file_group.progress = file_group.position = None
+
+
 class RecentlyViewedConfig(ConfigFile):
     file_name = 'recently_viewed.yaml'
+    sections = (ViewedSection(),)
     default_config = dict(
         version=0,
         files=[],
@@ -180,78 +266,46 @@ class RecentlyViewedConfig(ConfigFile):
             if send_events:
                 Events.send_config_save_failed(message)
 
-    def import_config(self, file=None, send_events=False):
-        """Apply the config to the database.  Progress written after the config (e.g. while WROLPi was starting)
-        is kept; a newer plain view keeps its `viewed`, but takes the config's position."""
-        super().import_config(file, send_events)
-        if self.import_decision() == SAVE:
-            self.save_database_instead()
-            return
-        try:
-            # wrolpi.files.models imports this module.
-            from wrolpi.files.lib import glob_shared_stem
-            from wrolpi.files.models import FileGroup
+    def prepare(self, diffs: dict):
+        diff, db_rows = diffs['viewed'], self.import_db_rows['viewed']
+        file_rows = ViewedSection().file_rows(self._config)[0]
 
-            media_directory = get_media_directory()
-            entries = list()
-            for path, viewed, progress, position in filter(None, map(_parse_entry, self.files)):
-                # An absolute `path` replaces the media directory in the join, and `..` can leave it.
-                absolute_path = pathlib.Path(os.path.normpath(media_directory / path))
-                if not get_paths_in_media_directory([absolute_path]):
-                    logger.warning(f'Not importing recently viewed file outside the media directory: {path}')
-                    continue
-                entries.append((absolute_path, viewed, progress, position))
-            absolute_paths = [i[0] for i in entries]
+        # Only views at least as recent as the oldest entry of the file are removed: an older view fell off the end
+        # of the saved window.  A malformed entry keeps its file's history.
+        if diff.removed:
+            oldest = min(i['viewed'] for i in file_rows.values()) if file_rows else None
+            malformed = _malformed_keys(self._config)
+            diff.removed = {i for i in diff.removed
+                            if oldest and db_rows[i]['viewed'] >= oldest and i not in malformed}
 
-            # Glob the files which are not yet in the DB before the write transaction; stat'ing a thousand files on
-            # a slow disk would block every other writer.
-            with get_db_session() as session:
-                known = {i for (i,) in session.query(FileGroup.primary_path)
-                .filter(FileGroup.primary_path.in_(absolute_paths))}
-            new_files = {i: glob_shared_stem(i) for i in absolute_paths if i not in known and i.is_file()}
+        # Without deletes (no database version), progress written after the config is kept; a newer plain view keeps
+        # its `viewed`, but it is not a newer place to resume.
+        if not self.import_deletes:
+            for key, row in list(diff.changed.items()):
+                if db_rows[key]['viewed'] >= row['viewed']:
+                    if db_rows[key]['progress'] is not None:
+                        del diff.changed[key]
+                    else:
+                        diff.changed[key] = dict(row, viewed=db_rows[key]['viewed'])
 
-            with get_db_session(commit=True) as session:
-                file_groups = session.query(FileGroup).filter(FileGroup.primary_path.in_(absolute_paths))
-                file_groups_by_path = {i.primary_path: i for i in file_groups}
-
-                applied = 0
-                for absolute_path, viewed, progress, position in entries:
-                    file_group = file_groups_by_path.get(absolute_path)
-                    if not file_group:
-                        if not (files := new_files.pop(absolute_path, None)):
-                            # The file was deleted; its history goes with it.
-                            logger.debug(f'Not importing recently viewed file which does not exist: {absolute_path}')
-                            continue
-                        try:
-                            file_group = FileGroup.from_paths(session, *files)
-                        except NoPrimaryFile:
-                            logger.warning(f'Cannot import recently viewed file: {absolute_path}')
-                            continue
-                        session.add(file_group)
-                        session.flush([file_group])
-                        file_groups_by_path[absolute_path] = file_group
-
-                    db_is_newer = file_group.viewed is not None and file_group.viewed >= viewed
-                    if db_is_newer and file_group.progress is not None:
-                        # Progress was written after the config (e.g. while WROLPi was starting).
-                        continue
-                    # A newer plain view is kept, but it is not a newer place to resume.
-                    if not db_is_newer:
-                        file_group.viewed = viewed
-                    file_group.progress = progress
-                    file_group.position = position
-                    applied += 1
-
-            logger.info(f'Imported {applied} of {len(entries)} recently viewed files')
-            self.record_import_version()
-            self.successful_import = True
-        except Exception as e:
-            self.successful_import = False
-            message = f'Failed to import {self.get_relative_file()} config'
-            logger.error(message, exc_info=e)
-            if send_events:
-                Events.send_config_import_failed(message)
-            raise
+        # The files of new entries which are not yet in the database; a deleted file's history goes with it.
+        from wrolpi.files.lib import glob_shared_stem
+        from wrolpi.files.models import FileGroup
+        media_directory = get_media_directory()
+        known = set()
+        with get_db_session() as session:
+            for paths in chunks([str(media_directory / i) for i in diff.added], 500):
+                known.update(str(i) for i, in
+                             session.query(FileGroup.primary_path).filter(FileGroup.primary_path.in_(paths)))
+        for key, row in list(diff.added.items()):
+            absolute_path = media_directory / key
+            if str(absolute_path) in known:
+                continue
+            if absolute_path.is_file():
+                row['files'] = glob_shared_stem(absolute_path)
+            else:
+                logger.debug(f'Not importing recently viewed file which does not exist: {absolute_path}')
+                del diff.added[key]
 
 
 RECENTLY_VIEWED_CONFIG: RecentlyViewedConfig = RecentlyViewedConfig()

wrolpi/files/test/test_recently_viewed_import.py

@@ -0,0 +1,158 @@
+"""recently_viewed.yaml imports by diffing the viewed files against the database, and writing only the differences
+(see wrolpi.config_import).  A file removed from the config has its viewing history cleared, but only within the
+window the config saves: an older view just fell off the end."""
+from datetime import timedelta
+
+import pytest
+import yaml
+
+from wrolpi import dates
+from wrolpi.config_version import get_config_version
+from wrolpi.conftest import config_was_imported
+from wrolpi.files.models import FileGroup
+from wrolpi.files.progress import get_recently_viewed_config
+from wrolpi.test.test_config_import import edit_config, statements, writes
+
+
+@pytest.fixture
+def saved_viewed(test_session, test_directory, async_client, make_files_structure):
+    """`a.txt`, `b.txt` (with progress) and `c.txt`, viewed in that order, saved to the config (which records its
+    version)."""
+    now = dates.now()
+    for name, hours in (('a.txt', 3), ('b.txt', 2), ('c.txt', 1)):
+        file_group = FileGroup.from_paths(test_session, *make_files_structure([name]))
+        file_group.viewed = now - timedelta(hours=hours)
+        if name == 'b.txt':
+            file_group.progress, file_group.position = 0.5, dict(kind='page', page=5)
+    test_session.commit()
+    config = get_recently_viewed_config()
+    config.successful_import = True
+    config.dump_config()
+    config.save()
+    assert get_config_version(test_session, 'recently_viewed.yaml') is not None
+    return config.get_file()
+
+
+def viewed(test_session) -> dict:
+    test_session.expire_all()
+    return {i.primary_path.name: i for i in test_session.query(FileGroup) if i.viewed}
+
+
+def edit_entry(path, name, **values):
+    def change(data):
+        next(i for i in data['files'] if i['path'] == name).update(values)
+
+    edit_config(path, change)
+
+
+@pytest.mark.asyncio
+async def test_unchanged_import_writes_nothing(test_session, saved_viewed):
+    """Importing a config the database already matches opens no write transaction."""
+    config = get_recently_viewed_config()
+    with statements() as collected:
+        config.import_config()
+
+    assert config.successful_import is True
+    assert writes(collected) == []
+
+
+@pytest.mark.asyncio
+async def test_import_writes_only_the_differences(test_session, saved_viewed):
+    """Changed progress is applied; a removed file's history is cleared; the next import writes nothing."""
+    edit_entry(saved_viewed, 'b.txt', progress=0.75, position=dict(kind='page', page=7))
+    edit_config(saved_viewed, lambda data: data.update(files=[i for i in data['files'] if i['path'] != 'c.txt']))
+
+    config = get_recently_viewed_config()
+    config.import_config()
+
+    found = viewed(test_session)
+    assert set(found) == {'a.txt', 'b.txt'}
+    assert (found['b.txt'].progress, found['b.txt'].position) == (0.75, dict(kind='page', page=7))
+    c = test_session.query(FileGroup).filter(FileGroup.primary_path.like('%c.txt')).one()
+    assert (c.viewed, c.progress, c.position) == (None, None, None)
+    with statements() as collected:
+        config.import_config()
+    assert writes(collected) == []
+
+
+@pytest.mark.asyncio
+async def test_older_views_are_not_removed(test_session, test_directory, async_client, make_files_structure,
+                                           monkeypatch):
+    """A view older than the config's oldest entry fell off the end of the saved window; it is kept, and an unchanged
+    import writes nothing."""
+    monkeypatch.setattr('wrolpi.files.progress.RECENTLY_VIEWED_LIMIT', 2)
+    now = dates.now()
+    for name, hours in (('a.txt', 3), ('b.txt', 2), ('c.txt', 1)):
+        FileGroup.from_paths(test_session, *make_files_structure([name])).viewed = now - timedelta(hours=hours)
+    test_session.commit()
+    config = get_recently_viewed_config()
+    config.successful_import = True
+    config.dump_config()
+    config.save()
+    assert [i['path'] for i in yaml.safe_load(config.get_file().read_text())['files']] == ['c.txt', 'b.txt']
+
+    with statements() as collected:
+        config.import_config()
+
+    assert writes(collected) == []
+    assert set(viewed(test_session)) == {'a.txt', 'b.txt', 'c.txt'}
+
+
+@pytest.mark.asyncio
+async def test_malformed_entry_keeps_its_history(test_session, saved_viewed):
+    """A malformed entry keeps its file's history (it is not removed), and does not stop the config from being
+    saved (it is not counted as skipped: recently viewed is saved after every view)."""
+    edit_entry(saved_viewed, 'c.txt', viewed='not a date')
+
+    config = get_recently_viewed_config()
+    config.import_config()
+
+    assert config.import_skipped == 0
+    assert set(viewed(test_session)) == {'a.txt', 'b.txt', 'c.txt'}
+
+
+@pytest.mark.asyncio
+async def test_moved_viewed_file_counts(test_session, test_directory, saved_viewed, make_files_structure):
+    """recently_viewed.yaml saves files by path: moving a viewed FileGroup makes the database newer than the file, so
+    a restart before the save does not clear its history.  Moving a file that was never viewed does not."""
+    before = get_config_version(test_session, 'recently_viewed.yaml')
+    a = test_session.query(FileGroup).filter_by(primary_path=test_directory / 'a.txt').one()
+    a.move(test_directory / 'moved.txt')
+    test_session.commit()
+    assert get_config_version(test_session, 'recently_viewed.yaml') == before + 1
+
+    never = FileGroup.from_paths(test_session, *make_files_structure(['never.txt']))
+    test_session.commit()
+    never.move(test_directory / 'never-moved.txt')
+    test_session.commit()
+    assert get_config_version(test_session, 'recently_viewed.yaml') == before + 1
+
+
+@pytest.mark.asyncio
+async def test_bulk_moved_viewed_file_counts(test_session, test_directory, saved_viewed):
+    """A move written with raw SQL counts too."""
+    from wrolpi.files.lib import _bulk_update_file_groups_db
+    before = get_config_version(test_session, 'recently_viewed.yaml')
+    _bulk_update_file_groups_db(test_session, {test_directory / 'a.txt': test_directory / 'moved/a.txt'})
+    test_session.commit()
+
+    assert get_config_version(test_session, 'recently_viewed.yaml') > before
+
+
+@pytest.mark.asyncio
+async def test_refresh_changing_primary_file_counts(test_session, test_directory, make_files_structure, video_bytes,
+                                                    refresh_files):
+    """A refresh that changes a viewed FileGroup's primary file counts as a change of its path."""
+    make_files_structure({'video.mp4': video_bytes, 'video.jpg': 'poster'})
+    await refresh_files()
+    test_session.query(FileGroup).one().viewed = dates.now()
+    test_session.commit()
+    config_was_imported(test_session, 'recently_viewed.yaml')
+    test_session.commit()
+
+    (test_directory / 'video.mp4').unlink()
+    await refresh_files()
+
+    test_session.expire_all()
+    assert test_session.query(FileGroup).one().primary_path != test_directory / 'video.mp4'
+    assert get_config_version(test_session, 'recently_viewed.yaml') > 0

wrolpi/files/worker.py

@@ -772,6 +772,17 @@ class FileWorker(FileMoveMixin, FileReorganizeMixin):
                         {'updates': updates})
                     if curs.fetchone():
                         count_config_change_curs(curs, 'playlists.yaml')
+                    # recently_viewed.yaml saves viewed files by path too.
+                    curs.execute(
+                        '''SELECT 1
+                           FROM file_group
+                                    JOIN json_each(:updates) AS v ON file_group.id = json_extract(v.value, '$[0]')
+                           WHERE file_group.viewed IS NOT NULL
+                             AND file_group.primary_path != json_extract(v.value, '$[1]')
+                           LIMIT 1''',
+                        {'updates': updates})
+                    if curs.fetchone():
+                        count_config_change_curs(curs, 'recently_viewed.yaml')
                     curs.execute(
                         '''UPDATE file_group
                            SET primary_path = v.new_path