tags.yaml imports by diffing (#868)

* tags.yaml imports by diffing

tags.yaml declares three sections (Tags, tagged files, tagged Zim
entries), so its import reads the file once, diffs it against a snapshot,
and writes only the differences, with the file's version, in one write.
An unchanged import writes nothing.

A removed Tag is deleted with its tagged files and Zim entries; a
Collection using it loses only the Tag.  A removed tagged file or Zim
entry is untagged.  A tagged file whose Tag the config does not list is
skipped.  prepare finds files without a FileGroup on disk, skips missing
ones before the write, moves Zim entries of a replaced Zim to the newest
one, and maps a file named by a secondary member to its group; the
config is then saved with the database's paths.

Moving a tagged FileGroup counts as a tags.yaml change (ORM, the bulk
move, and a refresh that changes the primary file), so a restart before
the save does not untag it.

Sections can define `same` (a tagged file without a date matches any
date), and prepare can count skipped rows.

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

* A removed Tag leaves no Download that would apply it

The tags import removes a deleted Tag's name from every Download's
tag_names and from the tag_names and channel_tag_name its settings pass
on (a change to download_manager.yaml, which is then saved).  Applying a
Tag that does not exist fails the download, and it is retried.  Once
tags.yaml is imported, a newer download_manager.yaml drops names no Tag
has, and is saved without them.

A replaced Zim's entry tags move to the newest Zim of the same parsed
name, by the date in its name (as migrate_outdated_zim_tags picks it),
not by name prefix and path order.

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

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

wrolpi/collections/config.py

@@ -4,10 +4,11 @@ from dataclasses import dataclass, field
 from typing import List
 
 from wrolpi.common import ConfigFile, get_media_directory, get_relative_to_media_directory, logger
-from wrolpi.config_version import IMPORT, SAVE
+from wrolpi.config_version import IMPORT, SAVE, count_config_change
 from wrolpi.db import get_db_session
 from wrolpi.events import Events
 from wrolpi.switches import ActivateSwitchMethod, register_switch_handler
+from wrolpi.tags import TagFile, TagsConfig, save_tags_config
 
 logger = logger.getChild(__name__)
 
@@ -301,10 +302,15 @@ 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 its change.
 
     `file_group_ids` may be an iterable of ids or a select of ids."""
     from .models import Collection, CollectionItem
 
+    if session.query(TagFile.file_group_id).filter(TagFile.file_group_id.in_(file_group_ids)).first():
+        count_config_change(session, TagsConfig.file_name)
+        save_tags_config.activate_switch()
+
     in_playlist = session.query(CollectionItem.id) \
         .join(Collection, Collection.id == CollectionItem.collection_id) \
         .filter(Collection.kind == 'playlist', CollectionItem.file_group_id.in_(file_group_ids)) \

wrolpi/common.py

@@ -844,8 +844,10 @@ class ConfigFile:
         file_rows = {i.name: i.file_rows(self._config) for i in self.sections}
         with get_db_session() as session:
             db_rows = {i.name: i.db_rows(session) for i in self.sections}
-        diffs = {i.name: diff_section(*file_rows[i.name], db_rows[i.name], deletes=decision == IMPORT)
+        diffs = {i.name: diff_section(*file_rows[i.name], db_rows[i.name], deletes=decision == IMPORT, same=i.same)
                  for i in self.sections}
+        # `prepare` may skip rows it cannot apply.
+        self.import_skipped = sum(len(invalid) for _, invalid in file_rows.values())
         if any(diffs.values()):
             self.prepare(diffs)
 
@@ -874,7 +876,6 @@ class ConfigFile:
                 logger_.info(f'{self.file_name} changed in the database while importing; deciding again')
                 return False
 
-        self.import_skipped = sum(len(invalid) for _, invalid in file_rows.values())
         changed = {name: diff for name, diff in diffs.items() if diff}
         logger_.info(f'Imported {self.file_name} ({decision}): '
                      + (', '.join(f'{name}: {len(diff.added)} added, {len(diff.changed)} changed, '
@@ -887,7 +888,8 @@ class ConfigFile:
 
     def prepare(self, diffs: dict):
         """Slow work the write needs (files, network), run with no transaction open, and only when there are
-        differences.  `diffs` maps each section's name to its `SectionDiff`.
+        differences.  `diffs` maps each section's name to its `SectionDiff`; it may drop rows it cannot apply
+        (count them in `import_skipped`), or add what it found to a row for `upsert`.
 
         It must be safe to repeat and to abandon: when the database changes before the write, the import decides
         again (up to three passes), and the diffs it was given may never be written (the write fails, or the next

wrolpi/config_import.py

@@ -5,8 +5,9 @@ A config declares a `ConfigSection` for each kind of row it saves (tags.yaml has
 Rows are compared as plain data keyed the way the file names them (a URL, a name, a path), so the snapshot needs no
 ORM objects and the read can finish before any slow preparation starts.
 """
+import operator
 from dataclasses import dataclass, field
-from typing import Any, Dict, Hashable, Iterable, List, Set, Tuple
+from typing import Any, Callable, Dict, Hashable, Iterable, List, Set, Tuple
 
 from sqlalchemy.orm import Session
 
@@ -45,6 +46,10 @@ class ConfigSection:
         deleted since the snapshot."""
         raise NotImplementedError()
 
+    def same(self, file_row: dict, db_row: dict) -> bool:
+        """Whether a file row is already in the database."""
+        return file_row == db_row
+
     def delete(self, session: Session, keys: Set[Hashable]):
         """Delete the rows with these keys.  A row deleted since the snapshot, or no longer saved by the config, is
         left alone."""
@@ -52,14 +57,14 @@ class ConfigSection:
 
 
 def diff_section(file_rows: Dict[Hashable, dict], invalid_keys: Set[Hashable], db_rows: Dict[Hashable, dict],
-                 deletes: bool) -> SectionDiff:
+                 deletes: bool, same: Callable[[dict, dict], bool] = operator.eq) -> SectionDiff:
     """Compare a section's file rows with its database rows.  `deletes`: rows the file no longer lists are removed.
     An empty section never removes anything: it cannot be told apart from a lost or blanked file."""
     diff = SectionDiff()
     for key, row in file_rows.items():
         if key not in db_rows:
             diff.added[key] = row
-        elif db_rows[key] != row:
+        elif not same(row, db_rows[key]):
             diff.changed[key] = row
     if deletes and (file_rows or invalid_keys):
         diff.removed = db_rows.keys() - file_rows.keys() - invalid_keys

wrolpi/downloader.py

@@ -2232,6 +2232,36 @@ class DownloadManagerConfig(ConfigFile):
     def downloads(self, value: List[dict]):
         self.update({'downloads': value})
 
+    def prepare(self, diffs: dict):
+        """Drop Tag names that no Tag has (the Tag was removed) from the Downloads to write: applying one fails the
+        download.  Only once tags.yaml was imported; until then its Tags may be missing."""
+        from wrolpi.tags import Tag, get_tags_config, without_tag_names
+        if not get_tags_config().successful_import:
+            return
+        rows = [*diffs['downloads'].added.values(), *diffs['downloads'].changed.values()]
+        names = set()
+        for row in rows:
+            names.update(row['tag_names'] or [])
+            for key in ('tag_names', 'channel_tag_name'):
+                names.update((row['settings'] or dict()).get(key) or [])
+        if not names:
+            return
+        with get_db_session() as session:
+            unknown = names - {i for i, in session.query(Tag.name).filter(Tag.name.in_(list(names)))}
+        if not unknown:
+            return
+        logger.warning(f'Removing unknown Tags from Downloads: {sorted(unknown)}')
+        for row in rows:
+            tag_names, row['settings'] = without_tag_names(row['tag_names'], row['settings'], unknown)
+            row['tag_names'] = tag_names or None
+        # The file is saved without them.
+        self._config_outdated = True
+
+    def after_import(self, diffs: dict):
+        if getattr(self, '_config_outdated', False):
+            self._config_outdated = False
+            self.background_dump.activate_switch()
+
     @staticmethod
     def _filter_settings_for_yaml(settings: dict, global_defaults: dict = None) -> dict:
         """Remove inheritable video settings that match current global defaults.

wrolpi/files/models.py

@@ -6,7 +6,7 @@ import urllib.parse
 from datetime import datetime
 from typing import List, Type, Optional, Iterable
 
-from sqlalchemy import Column, String, BigInteger, Boolean, event, Float, Index, Integer, JSON, or_, Text
+from sqlalchemy import Column, String, BigInteger, Boolean, event, Float, Index, Integer, JSON, or_, Text, inspect, text
 from sqlalchemy import types
 from sqlalchemy.orm import Mapped, deferred, relationship, Session, selectinload
 
@@ -892,6 +892,19 @@ def update_effective_datetime(mapper, connection, target):
 register_ownership('recently_viewed.yaml', FileGroup, columns={'viewed', 'progress', 'position'})
 
 
+def _tagged_file_group_moved(file_group: FileGroup) -> bool:
+    """An existing FileGroup with TagFiles changed path (tags.yaml saves tagged files by path).  A new FileGroup,
+    or a deleted one (its TagFiles go with it), did not move."""
+    if not inspect(file_group).attrs.primary_path.history.deleted:
+        return False
+    session = Session.object_session(file_group)
+    return session.connection().execute(
+        text('SELECT 1 FROM tag_file WHERE file_group_id = :id LIMIT 1'), {'id': file_group.id}).first() is not None
+
+
+register_ownership('tags.yaml', FileGroup, columns={'primary_path'}, row=_tagged_file_group_moved)
+
+
 class Directory(ModelHelper, Base):
     """A representation of a file directory in the media directory."""
     __tablename__ = 'directory'

wrolpi/files/worker.py

@@ -22,6 +22,7 @@ from sqlalchemy import text, or_
 from wrolpi import flags
 from wrolpi.common import apply_modelers, apply_refresh_cleanup
 from wrolpi.common import get_media_directory, logger, wrol_mode_enabled
+from wrolpi.config_version import count_config_change_curs
 from wrolpi.db import get_db_session, get_db_curs, session_curs
 from wrolpi.events import Events
 from wrolpi.vars import PYTEST
@@ -746,6 +747,18 @@ class FileWorker(FileMoveMixin, FileReorganizeMixin):
 
                 # Bulk UPDATE using UPDATE...FROM over a JSON array of [id, new_path] pairs
                 if valid_updates:
+                    updates = json.dumps([[fg_id, new_path] for fg_id, new_path in valid_updates])
+                    # tags.yaml saves tagged files by path: a tagged FileGroup's new path is its change.
+                    curs.execute(
+                        '''SELECT 1
+                           FROM tag_file
+                                    JOIN file_group ON file_group.id = tag_file.file_group_id
+                                    JOIN json_each(:updates) AS v ON file_group.id = json_extract(v.value, '$[0]')
+                           WHERE file_group.primary_path != json_extract(v.value, '$[1]')
+                           LIMIT 1''',
+                        {'updates': updates})
+                    if curs.fetchone():
+                        count_config_change_curs(curs, 'tags.yaml')
                     curs.execute(
                         '''UPDATE file_group
                            SET primary_path = v.new_path
@@ -754,7 +767,7 @@ class FileWorker(FileMoveMixin, FileReorganizeMixin):
                                  FROM json_each(:updates)) AS v
                            WHERE file_group.id = v.id
                              AND file_group.primary_path != v.new_path''',
-                        {'updates': json.dumps([[fg_id, new_path] for fg_id, new_path in valid_updates])}
+                        {'updates': updates}
                     )
 
         # Collect all file paths from the diffs (fast operation, no progress tracking needed)

wrolpi/tags.py

@@ -2,18 +2,20 @@ import contextlib
 import pathlib
 from dataclasses import dataclass, field
 from datetime import datetime
-from typing import List, Dict, Tuple, Optional, Union
+from typing import List, Dict, Tuple, Optional, Union, Set
 
 import cachetools
+import pytz
 from cachetools.keys import hashkey
-from sqlalchemy import Column, Integer, String, ForeignKey, BigInteger, event, UniqueConstraint
+from sqlalchemy import Column, Integer, String, ForeignKey, BigInteger, event, UniqueConstraint, insert
 from sqlalchemy.exc import IntegrityError
-from sqlalchemy.orm import Mapped, relationship, Session
+from sqlalchemy.orm import Mapped, relationship, Session, load_only
 
 from wrolpi import dates, flags
 from wrolpi.common import loaded, ModelHelper, Base, logger, ConfigFile, get_media_directory, background_task, \
     get_relative_to_media_directory, is_valid_hex_color, walk, INVALID_FILE_CHARS, get_wrolpi_config
-from wrolpi.config_version import IMPORT, SAVE, register_ownership
+from wrolpi.config_import import ConfigSection, SectionDiff, chunks
+from wrolpi.config_version import count_config_change, register_ownership
 from wrolpi.dates import TZDateTime
 from wrolpi.db import get_db_session, named_placeholders, session_curs, serializer
 from wrolpi.downloader import save_downloads_config
@@ -296,8 +298,272 @@ class TagsConfigValidator:
     version: int = None
 
 
+def _utc_or_none(value) -> Optional[datetime]:
+    """A config's date (an ISO string, or a datetime YAML parsed) or the database's, in UTC."""
+    if not value:
+        return None
+    if isinstance(value, str):
+        return dates.strptime_ms(value)
+    return value.astimezone(pytz.UTC) if value.tzinfo else value.replace(tzinfo=pytz.UTC)
+
+
+def _relative_path(path) -> str:
+    """A path as the config names it: relative to the media directory."""
+    path = pathlib.Path(path)
+    return str(path.relative_to(get_media_directory()) if path.is_absolute() else path)
+
+
+def without_tag_names(download_tag_names: Optional[List[str]], settings: Optional[dict], names: Set[str]) \
+        -> Tuple[Optional[List[str]], Optional[dict]]:
+    """A Download's tag_names and settings without these Tag names, in the settings' tag_names and channel_tag_name
+    (passed on to the Downloads and Channels it creates) too.  New objects, so an ORM change to the JSON is seen."""
+    if download_tag_names:
+        download_tag_names = [i for i in download_tag_names if i not in names]
+    if settings:
+        settings = {key: [i for i in value if i not in names] if key in ('tag_names', 'channel_tag_name') and value
+                    else value for key, value in settings.items()}
+    return download_tag_names, settings
+
+
+def remove_tag_names_from_downloads(session: Session, names: Set[str]):
+    """Remove these Tag names from every Download that would apply them; applying a Tag that does not exist fails
+    the download.  (Changes download_manager.yaml.)"""
+    from wrolpi.downloader import Download
+    changed = False
+    for download in session.query(Download).options(load_only(Download.id, Download.tag_names, Download.settings)):
+        tag_names, settings = without_tag_names(download.tag_names, download.settings, names)
+        if tag_names != download.tag_names or settings != download.settings:
+            download.tag_names, download.settings = tag_names, settings
+            changed = True
+    if changed:
+        save_downloads_config.activate_switch()
+
+
+def _tag_ids(session: Session, names) -> Dict[str, int]:
+    tag_ids = dict()
+    for chunk in chunks(names):
+        tag_ids.update(session.query(Tag.name, Tag.id).filter(Tag.name.in_(chunk)))
+    return tag_ids
+
+
+class TagsSection(ConfigSection):
+    """tags.yaml's Tags, keyed by name."""
+    name = 'tags'
+
+    def file_rows(self, config: dict) -> Tuple[Dict[str, dict], Set[str]]:
+        rows, invalid = dict(), set()
+        for name, attrs in (config.get('tags') or dict()).items():
+            try:
+                rows[name] = dict(color=attrs.get('color'))
+            except Exception as e:
+                logger.error(f'Skipping invalid Tag in config: {name}', exc_info=e)
+                invalid.add(name)
+        return rows, invalid
+
+    def db_rows(self, session: Session) -> Dict[str, dict]:
+        return {name: dict(color=color) for name, color in session.query(Tag.name, Tag.color)}
+
+    def upsert(self, session: Session, rows: Dict[str, dict]):
+        existing = {i.name: i for i in session.query(Tag).filter(Tag.name.in_(list(rows)))}
+        for name, row in rows.items():
+            if tag := existing.get(name):
+                tag.color = row['color']
+            else:
+                logger.info(f'Creating Tag {repr(name)}')
+                session.add(Tag(name=name, color=row['color']))
+        session.flush()
+
+    def delete(self, session: Session, keys: Set[str]):
+        """A removed Tag is deleted with its tagged files and Zim entries.  A Collection using it keeps its
+        directory, and loses only the Tag (its config is then saved)."""
+        from wrolpi.collections import Collection
+        remove_tag_names_from_downloads(session, set(keys))
+        for tag in session.query(Tag).filter(Tag.name.in_(list(keys))):
+            for collection in session.query(Collection).filter(Collection.tag_id == tag.id):
+                logger.warning(f'Removing Tag {repr(tag.name)} from {collection}')
+                collection.tag = None
+                collection.tag_id = None
+            logger.warning(f'Deleting {tag} because it is not in the config')
+            session.delete(tag)
+
+
+class TagFilesSection(ConfigSection):
+    """tags.yaml's tagged files, keyed by (Tag name, the FileGroup's primary path relative to the media
+    directory).  A tagged file whose Tag the config does not list is skipped (no FileGroup is created for it)."""
+    name = 'tag_files'
+
+    def file_rows(self, config: dict) -> Tuple[Dict[tuple, dict], Set[tuple]]:
+        tag_names = set(config.get('tags') or dict())
+        rows, invalid = dict(), set()
+        for item in config.get('tag_files') or []:
+            try:
+                tag_name, path, *rest = item
+                key = (tag_name, _relative_path(path))
+            except Exception as e:
+                logger.error(f'Skipping invalid tagged file in config: {item}', exc_info=e)
+                continue
+            if tag_names and tag_name not in tag_names:
+                logger.warning(f'Cannot find Tag for {repr(key[1])}')
+                invalid.add(key)
+                continue
+            try:
+                rows[key] = dict(created_at=_utc_or_none(rest[0] if rest else None))
+            except Exception as e:
+                logger.error(f'Skipping invalid tagged file in config: {item}', exc_info=e)
+                invalid.add(key)
+        return rows, invalid
+
+    def db_rows(self, session: Session) -> Dict[tuple, dict]:
+        from wrolpi.files.models import FileGroup
+        media_directory = get_media_directory()
+        query = session.query(Tag.name, FileGroup.primary_path, TagFile.created_at) \
+            .join(TagFile, TagFile.tag_id == Tag.id) \
+            .join(FileGroup, FileGroup.id == TagFile.file_group_id)
+        return {(name, str(primary_path.relative_to(media_directory))): dict(created_at=_utc_or_none(created_at))
+                for name, primary_path, created_at in query}
+
+    def same(self, file_row: dict, db_row: dict) -> bool:
+        # A tagged file without a date keeps the database's.
+        return file_row['created_at'] is None or file_row == db_row
+
+    def upsert(self, session: Session, rows: Dict[tuple, dict]):
+        from wrolpi.files.models import FileGroup
+        media_directory = get_media_directory()
+        tag_ids = _tag_ids(session, {tag_name for tag_name, _ in rows})
+        paths = {key: str(media_directory / key[1]) for key in rows}
+        file_group_ids = dict()
+        for chunk in chunks(set(paths.values())):
+            file_group_ids.update((str(path), id_) for path, id_ in
+                                  session.query(FileGroup.primary_path, FileGroup.id)
+                                  .filter(FileGroup.primary_path.in_(chunk)))
+
+        # FileGroups for files found on disk by `prepare`, flushed together.  Every member path is cached, so a
+        # group named by two of its files is created once.
+        new_groups = dict()
+        with session.no_autoflush:
+            for key, row in rows.items():
+                if not row.get('files') or paths[key] in file_group_ids or paths[key] in new_groups:
+                    continue
+                file_group = FileGroup.from_paths(session, *row['files'])
+                for member in row['files']:
+                    new_groups[str(member)] = file_group
+        if new_groups:
+            session.flush(list({id(i): i for i in new_groups.values()}.values()))
+            file_group_ids.update({path: file_group.id for path, file_group in new_groups.items()})
+
+        existing = dict()
+        for chunk in chunks(set(file_group_ids.values())):
+            existing.update({(i.tag_id, i.file_group_id): i for i in
+                             session.query(TagFile).filter(TagFile.file_group_id.in_(chunk))})
+        new = dict()
+        for key, row in rows.items():
+            tag_id, file_group_id = tag_ids.get(key[0]), file_group_ids.get(paths[key])
+            if not tag_id or not file_group_id:
+                logger.warning(f'Cannot tag {repr(key[1])} with {repr(key[0])}')
+                continue
+            if tag_file := existing.get((tag_id, file_group_id)):
+                if row['created_at']:
+                    tag_file.created_at = row['created_at']
+            else:
+                new[(tag_id, file_group_id)] = dict(tag_id=tag_id, file_group_id=file_group_id,
+                                                    created_at=row['created_at'] or dates.now())
+        if new:
+            # One statement for all new TagFiles.  Not flushed: count it.
+            session.execute(insert(TagFile), list(new.values()))
+            count_config_change(session, TagsConfig.file_name)
+        session.flush()
+
+    def delete(self, session: Session, keys: Set[tuple]):
+        from wrolpi.files.models import FileGroup
+        media_directory = get_media_directory()
+        for chunk in chunks(keys):
+            chunk = set(chunk)
+            query = session.query(TagFile, Tag.name, FileGroup.primary_path) \
+                .join(Tag, Tag.id == TagFile.tag_id) \
+                .join(FileGroup, FileGroup.id == TagFile.file_group_id) \
+                .filter(FileGroup.primary_path.in_({str(media_directory / path) for _, path in chunk}))
+            for tag_file, name, primary_path in query:
+                if (name, str(primary_path.relative_to(media_directory))) in chunk:
+                    session.delete(tag_file)
+
+
+class TagZimsSection(ConfigSection):
+    """tags.yaml's tagged Zim entries, keyed by (Tag name, Zim path relative to the media directory, entry)."""
+    name = 'tag_zims'
+
+    def file_rows(self, config: dict) -> Tuple[Dict[tuple, dict], Set[tuple]]:
+        tag_names = set(config.get('tags') or dict())
+        rows, invalid = dict(), set()
+        for item in config.get('tag_zims') or []:
+            try:
+                tag_name, zim_path, zim_entry, *rest = item
+                key = (tag_name, _relative_path(zim_path), zim_entry)
+            except Exception as e:
+                logger.error(f'Skipping invalid tagged Zim entry in config: {item}', exc_info=e)
+                continue
+            if tag_names and tag_name not in tag_names:
+                logger.warning(f'Cannot find Tag for {repr(key[1])}')
+                invalid.add(key)
+                continue
+            try:
+                rows[key] = dict(created_at=_utc_or_none(rest[0] if rest else None))
+            except Exception as e:
+                logger.error(f'Skipping invalid tagged Zim entry in config: {item}', exc_info=e)
+                invalid.add(key)
+        return rows, invalid
+
+    def db_rows(self, session: Session) -> Dict[tuple, dict]:
+        from modules.zim.models import TagZimEntry, Zim
+        media_directory = get_media_directory()
+        query = session.query(Tag.name, Zim.path, TagZimEntry.zim_entry, TagZimEntry.created_at) \
+            .join(TagZimEntry, TagZimEntry.tag_id == Tag.id) \
+            .join(Zim, Zim.id == TagZimEntry.zim_id)
+        return {(name, str(path.relative_to(media_directory)), zim_entry): dict(created_at=_utc_or_none(created_at))
+                for name, path, zim_entry, created_at in query}
+
+    def same(self, file_row: dict, db_row: dict) -> bool:
+        return file_row['created_at'] is None or file_row == db_row
+
+    def upsert(self, session: Session, rows: Dict[tuple, dict]):
+        from modules.zim.models import TagZimEntry, Zim
+        media_directory = get_media_directory()
+        tag_ids = _tag_ids(session, {tag_name for tag_name, _, _ in rows})
+        zim_ids = {str(path.relative_to(media_directory)): id_ for id_, path in session.query(Zim.id, Zim.path)}
+        existing = {(i.tag_id, i.zim_id, i.zim_entry): i for i in
+                    session.query(TagZimEntry).filter(TagZimEntry.tag_id.in_(list(tag_ids.values())))}
+        new = dict()
+        for (tag_name, zim_path, zim_entry), row in rows.items():
+            tag_id, zim_id = tag_ids.get(tag_name), zim_ids.get(zim_path)
+            if not tag_id or not zim_id:
+                logger.warning(f'Cannot tag {repr(zim_entry)} of {repr(zim_path)} with {repr(tag_name)}')
+                continue
+            if tag_zim_entry := existing.get((tag_id, zim_id, zim_entry)):
+                if row['created_at']:
+                    tag_zim_entry.created_at = row['created_at']
+            else:
+                new[(tag_id, zim_id, zim_entry)] = dict(tag_id=tag_id, zim_id=zim_id, zim_entry=zim_entry,
+                                                        created_at=row['created_at'] or dates.now())
+        if new:
+            # Not flushed: count it.
+            session.execute(insert(TagZimEntry), list(new.values()))
+            count_config_change(session, TagsConfig.file_name)
+        session.flush()
+
+    def delete(self, session: Session, keys: Set[tuple]):
+        from modules.zim.models import TagZimEntry, Zim
+        media_directory = get_media_directory()
+        query = session.query(TagZimEntry, Tag.name, Zim.path) \
+            .join(Tag, Tag.id == TagZimEntry.tag_id) \
+            .join(Zim, Zim.id == TagZimEntry.zim_id) \
+            .filter(Tag.name.in_({tag_name for tag_name, _, _ in keys}))
+        for tag_zim_entry, name, path in query:
+            if (name, str(path.relative_to(media_directory)), tag_zim_entry.zim_entry) in keys:
+                session.delete(tag_zim_entry)
+
+
 class TagsConfig(ConfigFile):
     file_name = 'tags.yaml'
+    sections = (TagsSection(), TagFilesSection(), TagZimsSection())
     width = 500
     validator = TagsConfigValidator
     default_config = dict(
@@ -393,196 +659,105 @@ class TagsConfig(ConfigFile):
                 Events.send_config_save_failed(message)
 
     def import_config(self, file: pathlib.Path = None, send_events=False):
-        from modules.zim import lib as zim_lib
-        from modules.zim.models import Zim, TagZimEntry
-        from wrolpi.files.lib import split_path_stem_and_suffix
-
         if PYTEST and not TEST_TAGS_CONFIG:
             logger.warning('Refusing to import tags without test tags config.  '
                            'Use `test_tags_config` fixture if you would like to call this.')
             return
 
-        config = get_tags_config()
-        logger.info('Importing tags config')
-        super().import_config(file)
+        self._config_outdated = False
+        super().import_config(file, send_events)
 
-        decision = self.import_decision()
-        if decision == SAVE:
-            self.save_database_instead()
+    def prepare(self, diffs: dict):
+        self._prepare_tag_files(diffs['tag_files'])
+        self._prepare_tag_zims(diffs['tag_zims'])
+
+    def _prepare_tag_files(self, diff: SectionDiff):
+        """Find the files of added tagged files that have no FileGroup yet (a rebuilt database), so `upsert` can
+        create them.  A file that is neither in the database nor on disk is skipped.  A file named by one of its
+        group's other files (an old config) tags the group's primary file."""
+        from wrolpi.files.lib import choose_primary_file, split_path_stem_and_suffix
+        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 / path) for _, path in diff.added}):
+                known.update(str(i) for i, in
+                             session.query(FileGroup.primary_path).filter(FileGroup.primary_path.in_(paths)))
+
+        listings: Dict[pathlib.Path, List[pathlib.Path]] = dict()
+        for key, row in list(diff.added.items()):
+            tag_name, path = key
+            absolute = media_directory / path
+            if str(absolute) in known:
+                continue
+            del diff.added[key]
+            if not absolute.is_file():
+                logger.warning(f'Cannot find FileGroup for {repr(path)}')
+                self.import_skipped += 1
+                continue
+            # Each directory is listed once (a Channel's directory can hold thousands of files).
+            if absolute.parent not in listings:
+                listings[absolute.parent] = list(absolute.parent.iterdir())
+            stem, _ = split_path_stem_and_suffix(absolute)
+            files = [i for i in listings[absolute.parent] if i == absolute or i.name.startswith(f'{stem}.')]
+            try:
+                primary = choose_primary_file(files)
+            except NoPrimaryFile:
+                logger.error(f'Failed to tag {absolute}')
+                self.import_skipped += 1
+                continue
+            primary_key = (tag_name, str(primary.relative_to(media_directory)))
+            if primary_key != key:
+                # The config is saved with the primary file.
+                self._config_outdated = True
+                diff.removed.discard(primary_key)
+            diff.added.setdefault(primary_key, dict(row, files=files))
+
+    def _prepare_tag_zims(self, diff: SectionDiff):
+        """A Zim entry tagged in a Zim file that is gone (replaced by a newer one) moves to the newest Zim of the
+        same name; one with no such Zim is skipped."""
+        from modules.zim import lib as zim_lib
+        from modules.zim.models import Zim
+        if not diff.added:
             return
+        media_directory = get_media_directory()
+        with get_db_session() as session:
+            zim_paths = [i for i, in session.query(Zim.path)]
+        known = {str(i.relative_to(media_directory)) for i in zim_paths}
+        # {name: (date, path)} of the newest Zim of each name, like `migrate_outdated_zim_tags`.
+        newest = dict()
+        for path in zim_paths:
+            try:
+                name, date = zim_lib.parse_name(path)
+            except ValueError:
+                continue
+            if name not in newest or date > newest[name][0]:
+                newest[name] = (date, path)
 
-        try:
-            need_commit = False
+        for key, row in list(diff.added.items()):
+            tag_name, zim_path, zim_entry = key
+            if zim_path in known:
+                continue
+            del diff.added[key]
+            try:
+                name, _ = zim_lib.parse_name(media_directory / zim_path)
+            except ValueError:
+                name = None
+            if name not in newest:
+                logger.warning(f'Cannot find Zim for {repr(zim_path)}')
+                self.import_skipped += 1
+                continue
+            new_key = (tag_name, str(newest[name][1].relative_to(media_directory)), zim_entry)
+            logger.warning(f'Migrating Zim entry tag {repr(tag_name)} from {zim_path} to {new_key[1]}')
+            self._config_outdated = True
+            diff.removed.discard(new_key)
+            diff.added.setdefault(new_key, row)
 
-            with get_db_session() as session:
-                if config.tags:
-                    # Tags have been saved to config, import them
-                    tags_by_name: Dict[str, Tag] = {i.name: i for i in session.query(Tag)}
-                    new_tags = list()
-                    for name, attrs in config.tags.items():
-                        tag = tags_by_name.get(name)
-                        if not tag:
-                            # Maintainer added a Tag to the config manually, or DB was wiped.
-                            tag = Tag(name=name, color=attrs['color'])
-                            new_tags.append(tag)
-                            logger.info(f'Creating new {tag}')
-                        tag.color = attrs['color']
-
-                    if new_tags:
-                        session.add_all(new_tags)
-                        need_commit = True
-
-                media_directory = get_media_directory()
-
-                # Get all Tags again because new ones may exist.
-                tags_by_name: Dict[str, Tag] = {i.name: i for i in session.query(Tag)}
-
-                # Tag all FileGroups.
-                if config.tag_files:
-                    from wrolpi.files.models import FileGroup
-
-                    primary_paths = [str(media_directory / i[1]) for i in config.tag_files]
-                    file_groups = session.query(FileGroup).filter(FileGroup.primary_path.in_(primary_paths))
-                    file_groups_by_primary_path = {i.primary_path: i for i in file_groups}
-                    # Get all TagFiles so we can create new ones.
-                    tag_files = {(i.tag_id, i.file_group_id): i for i in session.query(TagFile)}
-
-                    # Cache each directory's listing so we scan it at most once, rather than globbing per file
-                    # (which is O(files²) for large Channel/Domain directories).
-                    dir_listings: Dict[pathlib.Path, List[pathlib.Path]] = dict()
-
-                    def shared_stem_files(path: pathlib.Path) -> List[pathlib.Path]:
-                        # Equivalent to `glob_shared_stem`, but reuses a cached directory listing.
-                        parent = path.parent
-                        if parent not in dir_listings:
-                            dir_listings[parent] = list(parent.iterdir()) if parent.is_dir() else []
-                        stem, _ = split_path_stem_and_suffix(path)
-                        return [i for i in dir_listings[parent] if i == path or i.name.startswith(f'{stem}.')]
-
-                    # First pass: resolve (and create) FileGroups.  Newly-created FileGroups are flushed in a single
-                    # batch below so we don't pay a DB round-trip per file.  `no_autoflush` keeps the pending
-                    # FileGroups from being flushed one-at-a-time by `from_paths`'s own query.
-                    resolved = list()  # [(tag, file_group, created_at), ...]
-                    new_file_groups = list()
-                    with session.no_autoflush:
-                        for tag_name, primary_path, created_at in config.tag_files:
-                            tag: Tag = tags_by_name.get(tag_name)
-                            # Paths are absolute in the DB, relative in config.
-                            absolute_path = media_directory / primary_path
-                            file_group: FileGroup = file_groups_by_primary_path.get(absolute_path)
-                            if not file_group and absolute_path.is_file():
-                                # File exists, but is not yet in DB.
-                                files = shared_stem_files(absolute_path)
-                                try:
-                                    file_group = FileGroup.from_paths(session, *files)
-                                    session.add(file_group)
-                                    # Cache every member path (not just this one) so another config entry for a
-                                    # different member of the same shared-stem group reuses this FileGroup.  Without
-                                    # this, `no_autoflush` would hide the pending group from `from_paths`, creating a
-                                    # duplicate primary_path that aborts the batch flush.
-                                    for member_path in files:
-                                        file_groups_by_primary_path[member_path] = file_group
-                                    new_file_groups.append(file_group)
-                                except NoPrimaryFile:
-                                    logger.error(f'Failed to tag {absolute_path}')
-                                    self.import_skipped += 1
-                                    continue
-
-                            if tag and file_group:
-                                resolved.append((tag, file_group, created_at))
-                            elif not file_group:
-                                logger.warning(f'Cannot find FileGroup for {repr(str(primary_path))}')
-                                self.import_skipped += 1
-                            elif not tag:
-                                logger.warning(f'Cannot find Tag for {repr(str(primary_path))}')
-                                self.import_skipped += 1
-
-                    if new_file_groups:
-                        # Assign IDs to all newly-created FileGroups in a single flush.
-                        session.flush(new_file_groups)
-
-                    # Second pass: create the TagFiles now that every FileGroup has an ID.
-                    for tag, file_group, created_at in resolved:
-                        tag_file: TagFile = tag_files.get((tag.id, file_group.id))
-                        if not tag_file:
-                            # This FileGroup has not been tagged with the Tag, add it.
-                            logger.debug(f'Creating TagFile for tag_id={tag.id} file_group_id={file_group.id}')
-                            tag_file = TagFile(file_group_id=file_group.id, tag_id=tag.id)
-                            tag_files[(tag_file.tag_id, tag_file.file_group_id)] = tag_file
-                            session.add(tag_file)
-                        tag_file.created_at = dates.strptime_ms(created_at) if created_at else dates.now()
-                        need_commit = True
-
-                # Tag all Zim entries.
-                if config.tag_zims:
-                    # { pathlib.Path('/media/wrolpi/...'): <Zim>, ... }
-                    zims_by_path = {i.path: i for i in session.query(Zim)}
-
-                    # { (pathlib.Path('relative/path'), 'entry path'): <TagZimEntry>, ... }
-                    tag_zim_entries = {(get_relative_to_media_directory(i.zim.path), i.zim_entry): i for i in
-                                       session.query(TagZimEntry)}
-
-                    for tag_name, zim_path, zim_entry, created_at in config.tag_zims:
-                        zim_path = pathlib.Path(zim_path)
-                        tag: Tag = tags_by_name.get(tag_name)
-                        absolute_path = media_directory / zim_path
-                        zim: Zim = zims_by_path.get(absolute_path)
-                        if not zim:
-                            # No Zim matches the path.  It's likely that the old Zim file was deleted.  Attempt to migrate
-                            # the entry.
-                            name, date = zim_lib.parse_name(absolute_path)
-                            # Find any Zims that match the old Zim's name (wikipedia_en_all_maxi_*)
-                            possible_zims = {i: j for i, j in zims_by_path.items() if i.name.startswith(name)}
-                            if not possible_zims:
-                                logger.warning(f'Cannot find Zim for {repr(str(zim_path))}')
-                                self.import_skipped += 1
-                                continue
-                            zim = zims_by_path[sorted(possible_zims.keys())[-1]]
-                            new_zim_path = get_relative_to_media_directory(media_directory / zim.path)
-                            logger.warning(
-                                f'Migrating Zim entry tag {repr(str(tag_name))} from {zim_path} to {new_zim_path}')
-                            zim_path = new_zim_path
-
-                        if tag:
-                            tag_zim_entry: TagZimEntry = tag_zim_entries.get((zim_path, zim_entry))
-                            if not tag_zim_entry:
-                                tag_zim_entry = TagZimEntry(tag=tag, zim=zim, zim_entry=zim_entry)
-                                session.add(tag_zim_entry)
-                                # Track this new TagZimEntry because migration may cause duplicates.
-                                tag_zim_entries[(zim_path, zim_entry)] = tag_zim_entry
-                                logger.info(f'Created TagZimEntry: {tag_zim_entry}')
-                            tag_zim_entry.created_at = datetime.fromisoformat(created_at) if created_at else dates.now()
-                            need_commit = True
-                        elif not tag:
-                            logger.warning(f'Cannot find Tag for {repr(str(zim_path))}')
-                            self.import_skipped += 1
-
-                # Delete missing Tags last in case they are used above.
-                if config.tags and decision == IMPORT:
-                    config_tag_names = set(config.tags.keys())
-                    for tag in session.query(Tag):
-                        if tag.name not in config_tag_names:
-                            if tag.has_relations():
-                                logger.warning(f'Refusing to delete {tag} because it is used.')
-                            else:
-                                logger.warning(f'Deleting {tag} because it is not in the config.')
-                                session.delete(tag)
-                                need_commit = True
-
-                if need_commit:
-                    session.commit()
-
-            self.record_import_version()
-            self.successful_import = True
-            logger.info('Importing tags config complete')
-        except Exception as e:
-            self.successful_import = False
-            message = f'Failed to import {self.file_name}'
-            logger.error(message, exc_info=e)
-            if send_events:
-                from wrolpi.events import Events
-                Events.send_config_import_failed(message)
-            raise
+    def after_import(self, diffs: dict):
+        if diffs['tags']:
+            Tag.invalidate_cache()
+        if self._config_outdated:
+            self.background_dump.activate_switch()
 
     def preview_backup_import(self, backup_date: str, mode: str) -> dict:
         backup_file = self._get_backup_file(backup_date)

wrolpi/test/test_config_api.py

@@ -7,9 +7,6 @@ from wrolpi.tags import Tag
 
 
 @pytest.mark.asyncio
-# tags.import_config reads, then writes, in a deferred session; splitting its reads (and directory walk) from a
-# short write is its own change.
-@pytest.mark.allow_deferred_request_writes
 async def test_configs_crud(async_client, test_session, test_tags_config, tag_factory, test_wrolpi_config,
                             test_channels_config, await_switches):
     """Configs can be imported or saved on demand."""

wrolpi/test/test_config_backup_import.py

@@ -603,9 +603,6 @@ class TestBackupAPIEndpoints:
         assert 'preview' in response.json
         assert len(response.json['preview']['add']) == 1
 
-    # tags.import_config reads, then writes, in a deferred session; splitting its reads (and directory walk) from a
-    # short write is its own change.
-    @pytest.mark.allow_deferred_request_writes
     async def test_import_endpoint(self, async_client, test_directory, test_tags_config, test_session):
         from wrolpi.tags import get_tags_config
         config = get_tags_config()

wrolpi/test/test_config_import.py

@@ -384,3 +384,35 @@ async def test_nothing_to_write_skips_prepare(test_session, saved_downloads, mon
     get_download_manager_config().import_config()
 
     assert prepared == []
+
+
+@pytest.mark.asyncio
+async def test_unknown_tag_names_are_dropped(test_session, saved_downloads, test_tags_config, tag_factory,
+                                             await_switches):
+    """Once tags.yaml is imported, a Tag name the file gives a Download that no Tag has (the Tag was removed) is
+    dropped, and the file is saved without it: the next import writes nothing."""
+    from wrolpi.tags import get_tags_config
+    await tag_factory('one')
+    test_session.commit()
+    get_tags_config().dump_config()
+    get_tags_config().import_config()
+    assert get_tags_config().successful_import is True
+
+    def change(data):
+        for download in data['downloads']:
+            if download['url'] == 'https://example.com/recurring':
+                download['tag_names'] = ['one', 'gone']
+                download['settings'] = dict(download.get('settings') or {}, channel_tag_name=['gone'])
+
+    edit_config(saved_downloads, change)
+    config = get_download_manager_config()
+    config.import_config()
+    await await_switches()
+
+    test_session.expire_all()
+    download = Download.get_by_url(test_session, 'https://example.com/recurring')
+    assert download.tag_names == ['one']
+    assert download.settings['channel_tag_name'] == []
+    with statements() as collected:
+        config.import_config()
+    assert writes(collected) == []

wrolpi/test/test_config_version.py

@@ -129,26 +129,30 @@ async def test_unknown_database_version_imports_without_deletes(test_session, te
 
 
 @pytest.mark.asyncio
-async def test_failed_import_leaves_version_unknown(test_session, test_tags_config, monkeypatch, async_client):
-    """An import's own writes do not count: one that fails partway leaves the version unknown, not newer than the
-    file it did not finish applying (which would save that partial database over the file)."""
-    from wrolpi.tags import TagsConfig
+async def test_failed_import_changes_nothing(test_session, test_tags_config, monkeypatch, async_client):
+    """An import that fails partway writes nothing: neither its rows nor the file's version (which would mark a
+    partial database as up to date)."""
+    from wrolpi.tags import TagsConfig, TagZimsSection
 
     test_session.add(Tag(name='kept', color='#000000'))
     test_session.commit()
     set_config_version(test_session, 'tags.yaml', 3)
     test_session.commit()
-    write_tags_config(test_tags_config, 3, ['kept', 'added'])
+    write_tags_config(test_tags_config, 4, ['kept', 'added'])
 
-    def fail(self):
+    def fail(*args, **kwargs):
         raise RuntimeError('Import failed')
 
-    monkeypatch.setattr(TagsConfig, 'record_import_version', fail)
+    # Fails after the Tags were written.
+    monkeypatch.setattr(TagZimsSection, 'upsert', fail)
+    monkeypatch.setattr(TagZimsSection, 'file_rows', lambda self, config: ({('kept', 'a.zim', 'home'): {}}, set()))
+    monkeypatch.setattr(TagsConfig, 'prepare', lambda self, diffs: None)
     with pytest.raises(RuntimeError):
         import_tags_config()
 
-    assert {i.name for i in test_session.query(Tag)} == {'kept', 'added'}
-    assert db_version(test_session) is None
+    test_session.expire_all()
+    assert {i.name for i in test_session.query(Tag)} == {'kept'}
+    assert db_version(test_session) == 3
 
 
 @pytest.mark.asyncio

wrolpi/test/test_tags.py

@@ -349,7 +349,7 @@ async def test_import_tags_config_missing_file(test_session, test_directory, tes
 @pytest.mark.asyncio
 async def test_import_tags_delete_missing(test_session, test_directory, test_tags_config, make_files_structure,
                                           tag_factory, test_zim, await_switches):
-    """Tags not in the config are deleted on import.  Tags that are used will not be deleted."""
+    """Tags not in the config are deleted on import, with the files and Zim entries they tag."""
     from modules.zim import lib as zim_lib
 
     # Create some tags, use tag1.
@@ -382,23 +382,17 @@ async def test_import_tags_delete_missing(test_session, test_directory, test_tag
     assert len(config.tag_zims) == 1
     assert len(config.tags) == 1
 
-    # Import the config.  Only unused Tags not in the config are deleted.
+    # Import the config.  Every Tag not in the config is deleted, used or not.
     config_was_imported(test_session, 'tags.yaml')
     tags.import_tags_config()
-    assert {i.name for i in test_session.query(tags.Tag)} == {
-        tag1.name,  # tag1 is used by FileGroup.
-        # tag2.name,  tag2 is unused, and deleted.
-        tag3.name,  # tag3 was not deleted.
-        tag4.name,  # tag4 is used by ZimEntry.
-    }
+    assert {i.name for i in test_session.query(tags.Tag)} == {tag3.name}
 
-    # DB is written to config with only those Tags left.
+    # DB is written to config with only tag3 left.
     tags.save_tags_config.activate_switch()
     await await_switches()
-    # Tags that are used are restored.
-    assert len(config.tag_files) == 1
-    assert len(config.tag_zims) == 1
-    assert len(config.tags) == 3
+    assert len(config.tag_files) == 0
+    assert len(config.tag_zims) == 0
+    assert len(config.tags) == 1
 
 
 @pytest.mark.asyncio

wrolpi/test/test_tags_config_import.py

@@ -0,0 +1,305 @@
+"""tags.yaml imports by diffing its Tags, tagged files, and tagged Zim entries against the database, and writing only
+the differences (see wrolpi.config_import)."""
+import pytest
+import yaml
+
+from modules.zim.models import TagZimEntry
+from wrolpi import tags
+from wrolpi.conftest import config_was_imported
+from wrolpi.collections import Collection
+from wrolpi.config_version import get_config_version
+from wrolpi.files.models import FileGroup
+from wrolpi.tags import Tag, TagFile
+from wrolpi.test.test_config_import import edit_config, statements, writes
+
+
+@pytest.fixture
+async def saved_tags(test_session, test_directory, test_tags_config, make_files_structure, tag_factory, test_zim,
+                     await_switches):
+    """Tags `one` and `two` tag files, `three` tags a Zim entry; saved to the config (which records its version)."""
+    one, two, three = await tag_factory('one'), await tag_factory('two'), await tag_factory('three')
+    a, b = make_files_structure({'a.txt': 'a', 'b.txt': 'b'})
+    a, b = FileGroup.from_paths(test_session, a), FileGroup.from_paths(test_session, b)
+    test_session.flush([a, b])
+    a.add_tag(test_session, one.id)
+    a.add_tag(test_session, two.id)
+    b.add_tag(test_session, one.id)
+    test_zim.tag_entry(test_session, three.name, 'home')
+    test_session.commit()
+
+    tags.get_tags_config().dump_config()
+    await await_switches()
+    assert get_config_version(test_session, 'tags.yaml') is not None
+    return test_tags_config
+
+
+def tag_files(test_session):
+    test_session.expire_all()
+    return {(i.tag.name, i.file_group.primary_path.name) for i in test_session.query(TagFile)}
+
+
+@pytest.mark.asyncio
+async def test_unchanged_import_writes_nothing(test_session, saved_tags):
+    """Importing a config the database already matches opens no write transaction."""
+    config = tags.get_tags_config()
+    with statements() as collected:
+        config.import_config()
+
+    assert config.successful_import is True
+    assert writes(collected) == []
+    # The version, then one snapshot per section.
+    assert len(collected) <= 6, collected
+
+
+@pytest.mark.asyncio
+async def test_import_writes_only_the_differences(test_session, saved_tags):
+    """A changed color is saved, a removed tagged file is untagged, an added one is tagged; the rest is untouched."""
+    def change(data):
+        data['tags']['one']['color'] = '#abcdef'
+        data['tag_files'] = [i for i in data['tag_files'] if (i[0], i[1]) != ('two', 'a.txt')]
+        data['tag_files'].append(['three', 'b.txt', None])
+
+    edit_config(saved_tags, change)
+    config = tags.get_tags_config()
+    config.import_config()
+
+    assert config.successful_import is True
+    assert Tag.get_by_name(test_session, 'one').color == '#abcdef'
+    assert tag_files(test_session) == {('one', 'a.txt'), ('one', 'b.txt'), ('three', 'b.txt')}
+    # A tagged file without a date is not rewritten by the next import.
+    with statements() as collected:
+        config.import_config()
+    assert writes(collected) == []
+
+
+@pytest.mark.asyncio
+async def test_removed_tag_is_deleted(test_session, saved_tags):
+    """A Tag removed from the config is deleted with its tagged files and Zim entries; a Collection using it keeps
+    its directory and loses only the Tag."""
+    three = Tag.get_by_name(test_session, 'three')
+    test_session.add(Collection(name='example.com', kind='domain', tag_id=Tag.get_by_name(test_session, 'one').id))
+    test_session.commit()
+    tags.get_tags_config().dump_config()
+
+    def change(data):
+        del data['tags']['one']
+        del data['tags']['three']
+
+    edit_config(saved_tags, change)
+    config = tags.get_tags_config()
+    config.import_config()
+
+    assert config.successful_import is True
+    test_session.expire_all()
+    assert {i.name for i in test_session.query(Tag)} == {'two'}
+    assert tag_files(test_session) == {('two', 'a.txt')}
+    assert test_session.query(TagZimEntry).filter_by(tag_id=three.id).count() == 0
+    collection = test_session.query(Collection).one()
+    assert (collection.name, collection.tag_id) == ('example.com', None)
+
+
+@pytest.mark.asyncio
+async def test_unknown_tag_is_skipped(test_session, test_directory, saved_tags, make_files_structure):
+    """A tagged file whose Tag the config does not list is skipped: no FileGroup is created for it, and the file is
+    not "removed"."""
+    make_files_structure({'c.txt': 'c'})
+    edit_config(saved_tags, lambda data: data['tag_files'].append(['unknown', 'c.txt', None]))
+
+    config = tags.get_tags_config()
+    config.import_config()
+
+    assert config.successful_import is True
+    assert config.import_skipped == 1
+    assert test_session.query(FileGroup).filter_by(primary_path=test_directory / 'c.txt').count() == 0
+    assert tag_files(test_session) == {('one', 'a.txt'), ('two', 'a.txt'), ('one', 'b.txt')}
+
+
+@pytest.mark.asyncio
+async def test_empty_tags_delete_nothing(test_session, saved_tags):
+    """An empty section cannot be told apart from a lost or blanked one: it deletes nothing."""
+    edit_config(saved_tags, lambda data: data.update(tags={}, tag_files=[], tag_zims=[]))
+
+    config = tags.get_tags_config()
+    config.import_config()
+
+    assert config.successful_import is True
+    assert {i.name for i in test_session.query(Tag)} == {'one', 'two', 'three'}
+    assert len(tag_files(test_session)) == 3
+    assert test_session.query(TagZimEntry).count() == 1
+
+
+@pytest.mark.asyncio
+async def test_missing_file_group_is_created(test_session, test_directory, saved_tags, make_files_structure):
+    """A tagged file on disk without a FileGroup (a rebuilt database) gets one, and is tagged."""
+    make_files_structure({'c.txt': 'c'})
+    edit_config(saved_tags, lambda data: data['tag_files'].append(['two', 'c.txt', '2000-01-01T00:00:00+00:00']))
+
+    config = tags.get_tags_config()
+    config.import_config()
+
+    assert ('two', 'c.txt') in tag_files(test_session)
+    with statements() as collected:
+        config.import_config()
+    assert writes(collected) == []
+
+
+@pytest.mark.asyncio
+async def test_missing_file_is_skipped_without_a_write(test_session, saved_tags):
+    """A tagged file that is neither in the database nor on disk is skipped before the write: an import with nothing
+    else to do writes nothing."""
+    edit_config(saved_tags, lambda data: data['tag_files'].append(['two', 'gone.txt', None]))
+
+    config = tags.get_tags_config()
+    with statements() as collected:
+        config.import_config()
+
+    assert config.successful_import is True
+    assert config.import_skipped == 1
+    assert writes(collected) == []
+
+
+@pytest.mark.asyncio
+async def test_secondary_file_path_is_its_file_group(test_session, test_directory, saved_tags, make_files_structure,
+                                                    video_bytes, await_switches):
+    """A tagged file named by one of its group's other files (an old config) tags that FileGroup; it is not deleted
+    as "removed" from the primary path, and the config is saved with the primary path."""
+    video = FileGroup.from_paths(test_session, *make_files_structure({'video.mp4': video_bytes,
+                                                                       'video.info.json': '{}'}))
+    test_session.flush([video])
+    video.add_tag(test_session, Tag.get_by_name(test_session, 'two').id)
+    test_session.commit()
+    tags.get_tags_config().dump_config()
+    edit_config(saved_tags, lambda data: data.update(tag_files=[
+        i if i[1] != 'video.mp4' else [i[0], 'video.info.json', i[2]] for i in data['tag_files']]))
+
+    config = tags.get_tags_config()
+    config.import_config()
+    await await_switches()
+
+    assert ('two', 'video.mp4') in tag_files(test_session)
+    assert ['two', 'video.mp4'] in [i[:2] for i in yaml.safe_load(saved_tags.read_text())['tag_files']]
+
+
+@pytest.mark.asyncio
+async def test_moved_tagged_file_counts(test_session, test_directory, saved_tags):
+    """tags.yaml saves tagged files by path: moving a tagged FileGroup makes the database newer than the file, so a
+    restart before the save does not untag it.  Moving an untagged one does not."""
+    before = get_config_version(test_session, 'tags.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, 'tags.yaml') == before + 1
+
+    untagged = FileGroup.from_paths(test_session, test_directory / 'b.txt')
+    untagged.untag(test_session, Tag.get_by_name(test_session, 'one').id)
+    test_session.commit()
+    before = get_config_version(test_session, 'tags.yaml')
+    untagged.move(test_directory / 'untagged.txt')
+    test_session.commit()
+    assert get_config_version(test_session, 'tags.yaml') == before
+
+
+@pytest.mark.asyncio
+async def test_zim_entry_migrates_to_the_newest_zim(test_session, test_directory, test_tags_config, zim_factory,
+                                                    tag_factory, await_switches):
+    """A Zim entry tagged in a Zim file that is gone moves to the newest Zim of the same name, and the config is
+    saved with that Zim: the next import writes nothing."""
+    old = zim_factory('wikipedia_en_all_maxi_2020-01.zim')
+    new = zim_factory('wikipedia_en_all_maxi_2020-02.zim')
+    await tag_factory('one')
+    old.tag_entry(test_session, 'one', 'home')
+    test_session.commit()
+    tags.get_tags_config().dump_config()
+    await await_switches()
+    old.delete()
+
+    config = tags.get_tags_config()
+    config.import_config()
+    await await_switches()
+
+    test_session.expire_all()
+    assert [(i.tag.name, i.zim_id, i.zim_entry) for i in test_session.query(TagZimEntry)] == [('one', new.id, 'home')]
+    assert 'wikipedia_en_all_maxi_2020-02.zim' in test_tags_config.read_text()
+    with statements() as collected:
+        config.import_config()
+    assert writes(collected) == []
+
+
+@pytest.mark.asyncio
+async def test_bulk_moved_tagged_file_counts(test_session, test_directory, saved_tags):
+    """A move written with raw SQL counts too."""
+    from wrolpi.files.lib import _bulk_update_file_groups_db
+    before = get_config_version(test_session, 'tags.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, 'tags.yaml') == before + 1
+
+
+@pytest.mark.asyncio
+async def test_refresh_changing_primary_file_counts(test_session, test_directory, test_tags_config, tag_factory,
+                                                    make_files_structure, video_bytes, refresh_files):
+    """A refresh that changes a tagged FileGroup's primary file (its video was deleted, its other files are left)
+    counts as a change of the tagged file's path."""
+    make_files_structure({'video.mp4': video_bytes, 'video.jpg': 'poster'})
+    await refresh_files()
+    tag = await tag_factory('one')
+    file_group = test_session.query(FileGroup).one()
+    file_group.add_tag(test_session, tag.id)
+    test_session.commit()
+    config_was_imported(test_session, 'tags.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, 'tags.yaml') > 0
+
+
+@pytest.mark.asyncio
+async def test_removed_tag_is_removed_from_downloads(test_session, saved_tags, test_download_manager_config):
+    """A removed Tag's name is removed from every Download that would apply it (its tag_names, and the tag_names and
+    channel_tag_name it passes on); download_manager.yaml is then newer than its file, so it is saved."""
+    from wrolpi.config_version import set_config_version
+    from wrolpi.downloader import Download
+    test_session.add(Download(url='https://example.com/channel', downloader='video_channel', frequency=86400,
+                              tag_names=['one', 'two'],
+                              settings=dict(tag_names=['one'], channel_tag_name=['one'], video_resolutions=['720p'])))
+    test_session.commit()
+    set_config_version(test_session, 'download_manager.yaml', 1)
+    test_session.commit()
+    edit_config(saved_tags, lambda data: data['tags'].pop('one'))
+
+    tags.get_tags_config().import_config()
+
+    test_session.expire_all()
+    download = Download.get_by_url(test_session, 'https://example.com/channel')
+    assert download.tag_names == ['two']
+    assert download.settings == dict(tag_names=[], channel_tag_name=[], video_resolutions=['720p'])
+    assert get_config_version(test_session, 'download_manager.yaml') > 1
+
+
+@pytest.mark.asyncio
+async def test_zim_entry_migrates_by_name_and_date(test_session, test_directory, test_tags_config, zim_factory,
+                                                   tag_factory, await_switches):
+    """The newest Zim of the same name is chosen by the date in its name, wherever it is, and a Zim whose name
+    merely starts with the same name is not the same Zim."""
+    (test_directory / 'a').mkdir()
+    (test_directory / 'b').mkdir()
+    old = zim_factory('wikipedia_en_all_maxi_2020-01.zim')
+    newest = zim_factory('a/wikipedia_en_all_maxi_2020-03.zim')
+    zim_factory('b/wikipedia_en_all_maxi_2020-02.zim')
+    zim_factory('wikipedia_en_all_maxi_extra_2021-01.zim')
+    await tag_factory('one')
+    old.tag_entry(test_session, 'one', 'home')
+    test_session.commit()
+    tags.get_tags_config().dump_config()
+    await await_switches()
+    old.delete()
+
+    tags.get_tags_config().import_config()
+
+    test_session.expire_all()
+    assert [(i.tag.name, i.zim_id) for i in test_session.query(TagZimEntry)] == [('one', newest.id)]