From f2dda201169ade0d7abef022e96364af18929377 Mon Sep 17 00:00:00 2001 From: Roland Date: Sat, 24 Oct 2026 19:08:20 -0600 Subject: [PATCH] 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 --- wrolpi/collections/config.py | 8 +- wrolpi/common.py | 8 +- wrolpi/config_import.py | 11 +- wrolpi/files/models.py | 15 +- wrolpi/files/worker.py | 15 +- wrolpi/tags.py | 508 +++++++++++++++-------- wrolpi/test/test_config_api.py | 3 - wrolpi/test/test_config_backup_import.py | 3 - wrolpi/test/test_config_version.py | 22 +- wrolpi/test/test_tags.py | 20 +- wrolpi/test/test_tags_config_import.py | 258 ++++++++++++ 11 files changed, 650 insertions(+), 221 deletions(-) create mode 100644 wrolpi/test/test_tags_config_import.py diff --git a/wrolpi/collections/config.py b/wrolpi/collections/config.py index 91beaa86..d94e94da 100644 --- a/wrolpi/collections/config.py +++ b/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)) \ diff --git a/wrolpi/common.py b/wrolpi/common.py index bcefab5a..961982f3 100644 --- a/wrolpi/common.py +++ b/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 diff --git a/wrolpi/config_import.py b/wrolpi/config_import.py index d05798ff..bd7f36c5 100644 --- a/wrolpi/config_import.py +++ b/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 diff --git a/wrolpi/files/models.py b/wrolpi/files/models.py index 6c3635c1..f78e72db 100644 --- a/wrolpi/files/models.py +++ b/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' diff --git a/wrolpi/files/worker.py b/wrolpi/files/worker.py index 941ee525..ecd3ae09 100644 --- a/wrolpi/files/worker.py +++ b/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) diff --git a/wrolpi/tags.py b/wrolpi/tags.py index f8139426..c24f1340 100644 --- a/wrolpi/tags.py +++ b/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 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,245 @@ 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 _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 + 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 +632,97 @@ 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 = sorted(i for i, in session.query(Zim.path)) + known = {str(i.relative_to(media_directory)) for i in zim_paths} - 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 + newest = [i for i in zim_paths if name and i.name.startswith(name)] + if not newest: + logger.warning(f'Cannot find Zim for {repr(zim_path)}') + self.import_skipped += 1 + continue + new_key = (tag_name, str(newest[-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/...'): , ... } - zims_by_path = {i.path: i for i in session.query(Zim)} - - # { (pathlib.Path('relative/path'), 'entry path'): , ... } - 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) diff --git a/wrolpi/test/test_config_api.py b/wrolpi/test/test_config_api.py index 4a947069..ccb0b6a0 100644 --- a/wrolpi/test/test_config_api.py +++ b/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.""" diff --git a/wrolpi/test/test_config_backup_import.py b/wrolpi/test/test_config_backup_import.py index 8cbf73b4..09ef6dc2 100644 --- a/wrolpi/test/test_config_backup_import.py +++ b/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() diff --git a/wrolpi/test/test_config_version.py b/wrolpi/test/test_config_version.py index b979d4a4..43082bf4 100644 --- a/wrolpi/test/test_config_version.py +++ b/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 diff --git a/wrolpi/test/test_tags.py b/wrolpi/test/test_tags.py index e34f4660..2ca7d715 100644 --- a/wrolpi/test/test_tags.py +++ b/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 diff --git a/wrolpi/test/test_tags_config_import.py b/wrolpi/test/test_tags_config_import.py new file mode 100644 index 00000000..04a024a6 --- /dev/null +++ b/wrolpi/test/test_tags_config_import.py @@ -0,0 +1,258 @@ +"""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