diff --git a/wrolpi/files/lib.py b/wrolpi/files/lib.py index 68dc8d37..2bebcc8c 100644 --- a/wrolpi/files/lib.py +++ b/wrolpi/files/lib.py @@ -60,7 +60,8 @@ __all__ = ['list_directories_contents', 'delete', 'split_path_stem_and_suffix', 'get_normalized_ignored_directories', 'sanitize_ignored_directories', 'cached_search_total', 'search_filter_cache_key', 'count_file_groups', 'STEM_ALGORITHM_VERSION', 'ensure_file_group_stems', - 'backfill_null_file_group_stems', 'dedupe_file_groups_by_directory_stem'] + 'backfill_null_file_group_stems', 'dedupe_file_groups_by_directory_stem', 'merge_file_group_relations', + 'merge_into_surviving_duplicates'] def get_file_tag_names(session: Session, file: pathlib.Path) -> List[str]: @@ -1018,21 +1019,7 @@ def dedupe_file_groups_by_directory_stem() -> int: loser_ids = [m['id'] for m in members[1:]] with get_db_curs(commit=True) as curs: for loser_id in loser_ids: - curs.execute(''' - INSERT OR IGNORE INTO tag_file (tag_id, file_group_id, created_at) - SELECT tag_id, ?, created_at - FROM tag_file - WHERE file_group_id = ? - ''', (winner_id, loser_id)) - curs.execute(''' - UPDATE collection_item - SET file_group_id = ? - WHERE file_group_id = ? - AND collection_id NOT IN (SELECT collection_id - FROM (SELECT collection_id - FROM collection_item - WHERE file_group_id = ?)) - ''', (winner_id, loser_id, winner_id)) + merge_file_group_relations(curs, loser_id, winner_id) curs.execute('DELETE FROM file_group WHERE id = ?', (loser_id,)) deleted += 1 if deleted: @@ -1040,6 +1027,57 @@ def dedupe_file_groups_by_directory_stem() -> int: return deleted +def merge_file_group_relations(curs, loser_id: int, winner_id: int): + """Copy the loser FileGroup's tags and collection membership onto the winner, so deleting the loser + (which cascades) does not lose them. The caller deletes the loser.""" + curs.execute(''' + INSERT OR IGNORE INTO tag_file (tag_id, file_group_id, created_at) + SELECT tag_id, ?, created_at + FROM tag_file + WHERE file_group_id = ? + ''', (winner_id, loser_id)) + curs.execute(''' + UPDATE collection_item + SET file_group_id = ? + WHERE file_group_id = ? + AND collection_id NOT IN (SELECT collection_id + FROM (SELECT collection_id + FROM collection_item + WHERE file_group_id = ?)) + ''', (winner_id, loser_id, winner_id)) + + +def merge_into_surviving_duplicates(curs, file_group_ids: List[int], deleting_ids: Set[int]) -> int: + """Before `file_group_ids` are deleted, move each one's tags and collection membership onto a FileGroup + that shares its (directory, stem) and is not in `deleting_ids`. Returns how many FileGroups were merged. + + A duplicate FileGroup (same directory and stem) is deleted by refresh; its relations belong to the survivor.""" + params = ','.join('?' * len(file_group_ids)) + curs.execute(f''' + SELECT fg.id, fg.directory, fg.stem + FROM file_group fg + WHERE fg.id IN ({params}) + AND fg.stem IS NOT NULL + AND (EXISTS (SELECT 1 FROM tag_file tf WHERE tf.file_group_id = fg.id) + OR EXISTS (SELECT 1 FROM collection_item ci WHERE ci.file_group_id = fg.id)) + ''', file_group_ids) + merged = 0 + for loser in curs.fetchall(): + curs.execute(''' + SELECT id + FROM file_group + WHERE directory = ? + AND stem = ? + AND id != ? + ORDER BY id DESC + ''', (loser['directory'], loser['stem'], loser['id'])) + survivor_id = next((row['id'] for row in curs.fetchall() if row['id'] not in deleting_ids), None) + if survivor_id is not None: + merge_file_group_relations(curs, loser['id'], survivor_id) + merged += 1 + return merged + + async def apply_indexers(progress_callback: Callable[[int, int], None] = None): """Index leftover FileGroups that no modeler claimed. diff --git a/wrolpi/files/test/test_worker.py b/wrolpi/files/test/test_worker.py index 8c585617..b3345161 100644 --- a/wrolpi/files/test/test_worker.py +++ b/wrolpi/files/test/test_worker.py @@ -950,6 +950,48 @@ async def test_file_worker_deletes_tagged_file_groups_auto_removes_tags( ).count() == 0 +@pytest.mark.asyncio +@pytest.mark.parametrize('refresh_file', [False, True]) +async def test_file_worker_duplicate_file_groups_keep_tags( + async_client, test_session, test_directory, make_files_structure, tag_factory, refresh_files, refresh_file +): + """Refresh collapses duplicate FileGroups sharing (directory, stem) without losing the loser's tags or + collection membership. + + The older, tagged duplicate loses to the newer one; its relations must move onto the survivor.""" + from wrolpi.collections.models import Collection, CollectionItem + from wrolpi.tags import TagFile + + foo, sidecar, _baseline = make_files_structure(['videos/foo.mp4', 'videos/foo.info.json', 'videos/baseline.txt']) + test_session.execute(text('DROP INDEX IF EXISTS file_group_directory_stem_idx')) + tagged = FileGroup(directory=foo.parent, primary_path=sidecar, stem='foo', + files=[{'path': sidecar.name, 'mimetype': 'application/json'}]) + test_session.add(tagged) + test_session.commit() + newer = FileGroup(directory=foo.parent, primary_path=foo, stem='foo', + files=[{'path': foo.name, 'mimetype': 'video/mp4'}]) + test_session.add(newer) + test_session.commit() + assert tagged.id < newer.id + + tag = await tag_factory() + tagged.add_tag(test_session, tag.id) + collection = Collection(name='Playlist', kind='playlist') + test_session.add(collection) + test_session.flush() + collection.add_file_group(tagged, session=test_session) + test_session.commit() + + # Refresh the file directly, or the whole directory. + await refresh_files([foo] if refresh_file else [test_directory]) + + test_session.expire_all() + remaining = test_session.query(FileGroup).filter_by(stem='foo').one() + assert remaining.tag_names == [tag.name] + assert test_session.query(TagFile).count() == 1 + assert [i.file_group_id for i in test_session.query(CollectionItem)] == [remaining.id] + + @pytest.mark.asyncio async def test_file_worker_adds_deleted_file_url_to_skip_list( async_client, test_session, test_directory, make_files_structure, diff --git a/wrolpi/files/worker.py b/wrolpi/files/worker.py index b33ccb0f..98e54b0c 100644 --- a/wrolpi/files/worker.py +++ b/wrolpi/files/worker.py @@ -22,14 +22,14 @@ 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.db import get_db_session, get_db_curs +from wrolpi.db import get_db_session, get_db_curs, session_curs from wrolpi.events import Events from wrolpi.vars import PYTEST from wrolpi.files.progress import save_recently_viewed_config from wrolpi.files.lib import ( split_path_stem_and_suffix, _upsert_files, glob_shared_stem, group_files_by_stem, choose_primary_file, apply_indexers, - get_normalized_ignored_directories, remove_files_in_ignored_directories, + get_normalized_ignored_directories, remove_files_in_ignored_directories, merge_into_surviving_duplicates, ) from wrolpi.files.refresh import ( FileGroupDiff, FileComparisonResult, classify_file_group_diffs, compare_file_groups, @@ -794,6 +794,7 @@ class FileWorker(FileMoveMixin, FileReorganizeMixin): return total = len(file_group_ids) + deleting_ids = set(file_group_ids) self.update_status( status='deleting', operation_total=total, @@ -834,6 +835,10 @@ class FileWorker(FileMoveMixin, FileReorganizeMixin): Download.url == fg.url ).delete(synchronize_session=False) + # A duplicate's tags and collection membership move to the FileGroup that survives it. + with session_curs(session) as curs: + merge_into_surviving_duplicates(curs, batch_ids, deleting_ids) + # Delete FileGroups (TagFiles cascade-deleted automatically) batch_deleted = session.query(FileGroup).filter( FileGroup.id.in_(batch_ids)