diff --git a/alembic/versions/2026_09_18_1200_unique_directory_stem.py b/alembic/versions/2026_09_18_1200_unique_directory_stem.py new file mode 100644 index 00000000..ad4ee6c3 --- /dev/null +++ b/alembic/versions/2026_09_18_1200_unique_directory_stem.py @@ -0,0 +1,30 @@ +"""Unique (directory, stem) on file_group. + +713 added the stem column and a non-unique index. This release collapses any +remaining duplicate (directory, stem) rows (moving tags and collection +membership onto the winner) and makes the index UNIQUE. + +Revision ID: 2026_09_18_1200 +Revises: 2026_09_16_1200 +""" +from alembic import op + +# revision identifiers, used by Alembic. +revision = '2026_09_18_1200' +down_revision = '2026_09_16_1200' +branch_labels = None +depends_on = None + + +def upgrade(): + from wrolpi.files.lib import backfill_null_file_group_stems, dedupe_file_groups_by_directory_stem + + backfill_null_file_group_stems() + dedupe_file_groups_by_directory_stem() + op.execute('DROP INDEX IF EXISTS file_group_directory_stem_idx') + op.execute('CREATE UNIQUE INDEX file_group_directory_stem_idx ON file_group (directory, stem)') + + +def downgrade(): + op.execute('DROP INDEX IF EXISTS file_group_directory_stem_idx') + op.execute('CREATE INDEX IF NOT EXISTS file_group_directory_stem_idx ON file_group (directory, stem)') diff --git a/wrolpi/cmd.py b/wrolpi/cmd.py index b5317396..76895932 100644 --- a/wrolpi/cmd.py +++ b/wrolpi/cmd.py @@ -299,7 +299,8 @@ async def run_command(cmd: tuple[str | pathlib.Path, ...], cwd: pathlib.Path | s await proc.wait() if reader_task: - reader_task.cancel() + # Drain remaining stdout. Cancelling here races the last lines on a fast process + # (CI often loses "line three" because wait() returns before the reader finishes). try: await reader_task except (CancelledError, asyncio.CancelledError): diff --git a/wrolpi/common.py b/wrolpi/common.py index 7769567c..acfe2999 100644 --- a/wrolpi/common.py +++ b/wrolpi/common.py @@ -2382,23 +2382,24 @@ def chunks_by_stem(it: List[Union[pathlib.Path, str, int]], size: int) -> Genera from wrolpi.files.lib import split_path_stem_and_suffix it = sorted(it.copy()) index = size - last_stem = None + last_key = None while it: if index >= len(it): # Ran out of items, yield what is left. yield it return path = it[index] + path = pathlib.Path(path) stem, _ = split_path_stem_and_suffix(path) - if last_stem and stem != last_stem: - # Found a break in the path names, yield and reset. + key = (str(path.parent), stem) + if last_key and key != last_key: + # Found a break in directory or stem, yield and reset. chunk, it = it[:index], it[index:] index = size - last_stem = None + last_key = None yield chunk continue - # Didn't find a name change, try again. - last_stem = stem + last_key = key index += 1 diff --git a/wrolpi/files/lib.py b/wrolpi/files/lib.py index bfe815b5..ce034a14 100644 --- a/wrolpi/files/lib.py +++ b/wrolpi/files/lib.py @@ -52,13 +52,14 @@ except ImportError: logger = logger.getChild(__name__) __all__ = ['list_directories_contents', 'delete', 'split_path_stem_and_suffix', 'search_files', - 'get_mimetype', 'split_file_name_words', 'get_primary_file', 'get_file_statistics', + 'get_mimetype', 'split_file_name_words', 'get_primary_file', 'choose_primary_file', 'get_file_statistics', 'search_file_suggestion_count', 'glob_shared_stem', 'upsert_file', 'get_unique_files_by_stem', 'rename', 'delete_directory', 'handle_file_group_search_results', 'get_file_location_href', 'get_tagged_file_groups_by_ids', 'delete_file_groups', 'get_special_directories', 'get_normalized_ignored_directories', 'sanitize_ignored_directories', 'cached_search_total', 'search_filter_cache_key', 'count_file_groups', - 'STEM_ALGORITHM_VERSION', 'ensure_file_group_stems'] + 'STEM_ALGORITHM_VERSION', 'ensure_file_group_stems', + 'backfill_null_file_group_stems', 'dedupe_file_groups_by_directory_stem'] def get_file_tag_names(session: Session, file: pathlib.Path) -> List[str]: @@ -684,6 +685,22 @@ def get_primary_file(files: Union[Tuple[pathlib.Path], Iterable[pathlib.Path]]) raise NoPrimaryFile(f'Cannot find primary file for group: {files}') +def choose_primary_file(files: Union[Tuple[pathlib.Path], Iterable[pathlib.Path]]) -> pathlib.Path: + """Primary file for a stem group. Falls back to a sorted path so every caller agrees. + + `_upsert_files` and `_upsert_file_groups` must pick the same path. A set's iteration + order is not stable; using it as the fallback would DELETE the existing tagged row + and INSERT a new untagged one. + """ + paths = [pathlib.Path(p) for p in files] + if not paths: + raise ValueError('Cannot find primary file without any files') + try: + return get_primary_file(paths) + except NoPrimaryFile: + return sorted(paths)[0] + + def _upsert_files(files: List[pathlib.Path], progress_callback: Callable[[int, int], None] = None): """Insert/update all records of the provided files. @@ -708,33 +725,18 @@ def _upsert_files(files: List[pathlib.Path], for group in grouped: # The primary file is the video/SingleFile/epub, etc. try: - try: - primary_path = get_primary_file(group) - # Multiple files in this group. - primary_path: pathlib.Path = primary_path - # The primary mimetype allows modelers to find its file_groups. - mimetype = get_mimetype(primary_path) - # The group uses a common modification_datetime so the group will be re-indexed when any of it's - # files are modified. - modification_datetime = from_timestamp(max(i.stat().st_mtime for i in group)) - size = sum(i.stat().st_size for i in group) - files_json = json.dumps(_paths_to_files_dict(group)) - stem, suffix = split_path_stem_and_suffix(primary_path) - suffix = (suffix or '').lower() or None - values[primary_path] = (modification_datetime, mimetype, size, files_json, suffix, stem) - non_primary_files.update(i for i in group if i != primary_path) - except NoPrimaryFile: - # Cannot find primary path, create a `file_group` for each file. - for primary_path in group: - primary_path: pathlib.Path - # The primary mimetype allows modelers to find its file_groups. - mimetype = get_mimetype(primary_path) - modification_datetime = from_timestamp(primary_path.stat().st_mtime) - size = primary_path.stat().st_size - files_json = json.dumps(_paths_to_files_dict([primary_path])) - stem, suffix = split_path_stem_and_suffix(primary_path) - suffix = (suffix or '').lower() or None - values[primary_path] = (modification_datetime, mimetype, size, files_json, suffix, stem) + primary_path = choose_primary_file(group) + # The primary mimetype allows modelers to find its file_groups. + mimetype = get_mimetype(primary_path) + # The group uses a common modification_datetime so the group will be re-indexed when any of it's + # files are modified. + modification_datetime = from_timestamp(max(i.stat().st_mtime for i in group)) + size = sum(i.stat().st_size for i in group) + files_json = json.dumps(_paths_to_files_dict(group)) + stem, suffix = split_path_stem_and_suffix(primary_path) + suffix = (suffix or '').lower() or None + values[primary_path] = (modification_datetime, mimetype, size, files_json, suffix, stem) + non_primary_files.update(i for i in group if i != primary_path) except FileNotFoundError as e: # A file was deleted between discovery and upsert (temp files, WAL sidecars, etc.). refresh_logger.warning(f'File vanished during refresh, skipping group near {group[0]}', exc_info=e) @@ -752,6 +754,12 @@ def _upsert_files(files: List[pathlib.Path], '(SELECT value FROM json_each(?))', (json.dumps(primary_paths),)) previously_indexed = {i['primary_path'] for i in curs.fetchall()} + if non_primary_files: + # Delete before INSERT so a new primary_path that shares (directory, stem) + # with an old sidecar FileGroup does not hit the unique index. + curs.execute('DELETE FROM file_group WHERE primary_path IN (SELECT value FROM json_each(?))', + (json.dumps(list(map(str, non_primary_files))),)) + placeholders, params = values_clause(values) stmt = f''' INSERT INTO file_group @@ -797,11 +805,6 @@ def _upsert_files(files: List[pathlib.Path], else: refresh_logger.debug(f'Upserted {len(chunk)} files near {chunk[0]}') - if non_primary_files: - # New files may have been added which change what primary paths exist. Delete any file_groups which - # use paths which are not primary. - curs.execute('DELETE FROM file_group WHERE primary_path IN (SELECT value FROM json_each(?))', - (json.dumps(list(map(str, non_primary_files))),)) non_primary_files.clear() # Report progress after each chunk @@ -876,6 +879,84 @@ def ensure_file_group_stems(): set_stem_algorithm_version(STEM_ALGORITHM_VERSION) +def backfill_null_file_group_stems(): + """Fill file_group.stem for rows that never got one. Used before creating the unique index.""" + with get_db_curs() as curs: + curs.execute("SELECT id, primary_path FROM file_group WHERE stem IS NULL OR stem = ''") + rows = list(curs.fetchall()) + if not rows: + return + updates = [(split_path_stem_and_suffix(row['primary_path'])[0], row['id']) for row in rows] + with get_db_curs(commit=True) as curs: + for chunk in chunks(updates, 500): + curs.executemany('UPDATE file_group SET stem = ? WHERE id = ?', chunk) + + +def dedupe_file_groups_by_directory_stem() -> int: + """Collapse FileGroups that share (directory, stem). + + Prefers a row with a domain model, then tags, then non-empty files, then highest id. + Tags and collection membership on losers are moved onto the winner. Returns the + number of FileGroups deleted. + """ + with get_db_curs() as curs: + curs.execute(''' + SELECT directory, stem + FROM file_group + WHERE directory IS NOT NULL + AND stem IS NOT NULL + AND stem != '' + GROUP BY directory, stem + HAVING COUNT(*) > 1 + ''') + keys = [(row['directory'], row['stem']) for row in curs.fetchall()] + + deleted = 0 + for directory, stem in keys: + with get_db_curs() as curs: + curs.execute(''' + SELECT fg.id, + CASE + WHEN fg.files IS NULL OR json_type(fg.files) != 'array' THEN 0 + ELSE json_array_length(fg.files) + END AS file_count, + CASE + WHEN fg.model IS NOT NULL AND fg.model != '' THEN 1 + ELSE 0 + END AS has_model, + (SELECT COUNT(*) FROM tag_file tf WHERE tf.file_group_id = fg.id) AS tag_count + FROM file_group fg + WHERE fg.directory = ? + AND fg.stem = ? + ''', (directory, stem)) + members = list(curs.fetchall()) + members.sort(key=lambda m: (m['has_model'], m['tag_count'], m['file_count'] > 0, m['id']), reverse=True) + winner_id = members[0]['id'] + 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)) + curs.execute('DELETE FROM file_group WHERE id = ?', (loser_id,)) + deleted += 1 + if deleted: + refresh_logger.info(f'Collapsed {deleted} duplicate FileGroups sharing (directory, stem)') + return deleted + + async def apply_indexers(progress_callback: Callable[[int, int], None] = None): """Index leftover FileGroups that no modeler claimed. @@ -1419,16 +1500,16 @@ def group_files_by_stem(files: List[pathlib.Path], pre_sorted: bool = False) -> files = sorted(files) if not pre_sorted else files.copy() file = files.pop(0) group = [file, ] - prev_stem, _ = split_path_stem_and_suffix(file) + prev_dir, prev_stem = file.parent, split_path_stem_and_suffix(file)[0] for file in files: - stem, suffix = split_path_stem_and_suffix(file) - if stem == prev_stem: + stem, _ = split_path_stem_and_suffix(file) + if stem == prev_stem and file.parent == prev_dir: group.append(file) continue - # Stem has changed, group is finished. + # Directory or stem changed, group is finished. yield group group = [file, ] - prev_stem = stem + prev_dir, prev_stem = file.parent, stem yield group diff --git a/wrolpi/files/models.py b/wrolpi/files/models.py index e13d710c..838f41b5 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, Index, Integer, JSON +from sqlalchemy import Column, String, BigInteger, Boolean, event, Index, Integer, JSON, or_ from sqlalchemy import types from sqlalchemy.orm import deferred, relationship, Session @@ -83,8 +83,9 @@ class FileGroup(ModelHelper, Base): Index('file_group_size_ix', 'size'), Index('file_group_url_idx', 'url'), Index('file_group_viewed_idx', 'viewed'), - # Speeds compare joins. Not UNIQUE: duplicates are collapsed at refresh time. - Index('file_group_directory_stem_idx', 'directory', 'stem'), + # Identity of a FileGroup. Duplicates are collapsed by + # dedupe_file_groups_by_directory_stem before this unique index is created. + Index('file_group_directory_stem_idx', 'directory', 'stem', unique=True), ) # SQLite requires exactly "INTEGER PRIMARY KEY" for the rowid alias (FTS5 content_rowid). id: int = Column(BigInteger().with_variant(Integer, 'sqlite'), primary_key=True) @@ -517,18 +518,27 @@ class FileGroup(ModelHelper, Base): @classmethod def from_paths(cls, session: Session, *paths: pathlib.Path) -> 'FileGroup': """Create a new FileGroup which contains the provided file paths.""" - from wrolpi.files.lib import get_primary_file, get_mimetype, sanitize_filename_surrogates, \ + from wrolpi.files.lib import choose_primary_file, get_mimetype, sanitize_filename_surrogates, \ split_path_stem_and_suffix # Sanitize any paths with invalid UTF-8 characters (renames files on disk if needed) paths = tuple(sanitize_filename_surrogates(p) for p in paths) - existing_groups = session.query(FileGroup).filter(FileGroup.primary_path.in_(list(map(str, paths)))).all() + primary_path = choose_primary_file(paths) if paths else None + stem = split_path_stem_and_suffix(primary_path)[0] if primary_path else None + directory = str(primary_path.parent) if primary_path else None + filters = [FileGroup.primary_path.in_(list(map(str, paths)))] + if stem: + filters.append((FileGroup.directory == directory) & (FileGroup.stem == stem)) + existing_groups = unique_by_predicate( + session.query(FileGroup).filter(or_(*filters)).all(), + lambda fg: fg.id, + ) logger.trace(f'FileGroup.from_paths: {len(existing_groups)=}') if len(existing_groups) == 0: # These paths have not been used previously, create a new FileGroup. file_group = FileGroup() - primary_path = get_primary_file(paths) + primary_path = choose_primary_file(paths) file_group.primary_path = primary_path file_group.append_files(*paths) session.add(file_group) @@ -536,10 +546,10 @@ class FileGroup(ModelHelper, Base): # Found one FileGroup with these paths, no need to create a new FileGroup. file_group = existing_groups[0] file_group.append_files(*paths) - primary_path = get_primary_file(file_group.my_paths()) + primary_path = choose_primary_file(file_group.my_paths()) else: # Multiple FileGroups contain these paths as primary. - primary_path = get_primary_file(paths) + primary_path = choose_primary_file(paths) file_group: FileGroup = next(filter(lambda i: i.primary_path == primary_path, existing_groups), None) if not file_group: file_group = FileGroup.from_paths(session, primary_path) @@ -689,15 +699,33 @@ class FileGroup(ModelHelper, Base): new_name, _ = split_path_stem_and_suffix(new_primary_path, full=True) new_directory = new_primary_path.parent + new_stem, _ = split_path_stem_and_suffix(new_primary_path) + + # Unique (directory, stem): merge into the destination group before touching disk. + # Otherwise shutil.move succeeds, then commit hits IntegrityError and disk/DB disagree. + session = Session.object_session(self) + if session is not None: + existing = session.query(FileGroup).filter( + FileGroup.directory == str(new_directory), + FileGroup.stem == new_stem, + FileGroup.id != self.id, + ).one_or_none() + if existing is not None: + self.merge([existing]) + # Flush the DELETE first. Otherwise SQLAlchemy UPDATEs our stem before + # removing the other row and hits UNIQUE (directory, stem). + session.flush() # Get current files with resolved absolute paths for moving current_files = self.my_files() # Ensure that all destination files do not yet exist before move. + # After merging a same-stem group, some dest paths are the files we already own. for file in current_files: - _, suffix = split_path_stem_and_suffix(file['path']) + old_path = file['path'] + _, suffix = split_path_stem_and_suffix(old_path) new_path = pathlib.Path(f'{new_name}{suffix}') - if new_path.exists(): + if new_path.exists() and new_path != old_path: raise FileExistsError(f'Cannot move {self} to {new_path} because it already exists.') # Move physical files and collect new relative file entries @@ -707,7 +735,8 @@ class FileGroup(ModelHelper, Base): _, suffix = split_path_stem_and_suffix(old_path) new_path = pathlib.Path(f'{new_name}{suffix}') if old_path.is_file(): - shutil.move(old_path, new_path) + if old_path != new_path: + shutil.move(old_path, new_path) # Store just the filename (relative path) new_files.append({'path': new_path.name, 'mimetype': file['mimetype']}) diff --git a/wrolpi/files/test/test_lib.py b/wrolpi/files/test/test_lib.py index e831bb53..95f29d1d 100644 --- a/wrolpi/files/test/test_lib.py +++ b/wrolpi/files/test/test_lib.py @@ -15,6 +15,8 @@ import wrolpi.common from modules.videos import Video from wrolpi.common import timer, get_wrolpi_config from wrolpi.conftest import await_switches +from sqlalchemy.exc import IntegrityError + from wrolpi.errors import InvalidFile, UnknownDirectory, FileGroupIsTagged, NoPrimaryFile from wrolpi.files import lib, indexers from wrolpi.files.models import FileGroup, Directory @@ -496,6 +498,143 @@ def test_ensure_file_group_stems_recomputes_on_version_change(test_session, test assert lib.get_stem_algorithm_version() == lib.STEM_ALGORITHM_VERSION +def test_file_group_directory_stem_is_unique(test_session, test_directory, make_files_structure): + """Two FileGroups in the same directory cannot share a stem.""" + foo_mp4, foo_png = make_files_structure(['foo.mp4', 'foo.png']) + FileGroup.from_paths(test_session, foo_mp4) + test_session.commit() + + other = FileGroup( + directory=foo_png.parent, + primary_path=foo_png, + stem='foo', + files=[{'path': foo_png.name, 'mimetype': 'image/png'}], + ) + test_session.add(other) + with pytest.raises(IntegrityError): + test_session.commit() + + +def test_choose_primary_file_is_sorted_and_set_stable(test_directory, make_files_structure): + """Unmodeled groups must pick the same primary regardless of set iteration order.""" + csv, txt = make_files_structure(['n.csv', 'n.txt']) + assert lib.choose_primary_file([txt, csv]) == csv + assert lib.choose_primary_file({txt, csv}) == csv + + +@pytest.mark.asyncio +async def test_refresh_keeps_tag_when_sidecar_added(async_client, test_session, test_directory, + make_files_structure, tag_factory, refresh_files): + """Adding a sidecar to a tagged unmodeled FileGroup must not drop the tag on refresh.""" + txt, = make_files_structure(['note.txt']) + fg = FileGroup.from_paths(test_session, txt) + test_session.commit() + tag = await tag_factory('keep') + fg.add_tag(test_session, tag.id) + test_session.commit() + fg_id = fg.id + + (test_directory / 'note.csv').write_text('a,b\n') + await refresh_files() + test_session.expire_all() + + remaining = test_session.query(FileGroup).filter_by(id=fg_id).one() + assert remaining.tag_names == ['keep'] + names = {pathlib.Path(f['path']).name for f in remaining.files} + assert names == {'note.txt', 'note.csv'} + + +@pytest.mark.asyncio +async def test_rename_into_existing_stem_merges(async_client, test_session, test_directory, make_files_structure): + """Renaming onto another FileGroup's stem merges before moving files, so disk and DB agree.""" + make_files_structure({'foo.jpg': 'x', 'bar.mp4': 'video'}) + jpg = test_directory / 'foo.jpg' + mp4 = test_directory / 'bar.mp4' + FileGroup.from_paths(test_session, jpg) + FileGroup.from_paths(test_session, mp4) + test_session.commit() + assert test_session.query(FileGroup).count() == 2 + assert {fg.primary_path.name for fg in test_session.query(FileGroup)} == {'foo.jpg', 'bar.mp4'} + + await lib.rename_file(mp4, 'foo.mp4') + test_session.expire_all() + + remaining = test_session.query(FileGroup).one() + names = {pathlib.Path(f['path']).name for f in remaining.files} + assert names == {'foo.jpg', 'foo.mp4'} + assert (test_directory / 'foo.mp4').is_file() + assert (test_directory / 'foo.jpg').is_file() + assert not (test_directory / 'bar.mp4').exists() + + +@pytest.mark.asyncio +async def test_dedupe_file_groups_merges_tags_onto_winner(async_client, test_session, test_directory, + make_files_structure, tag_factory): + """Duplicates sharing (directory, stem) collapse onto one row; tags on the loser move.""" + from sqlalchemy import text + + foo, sidecar = make_files_structure(['foo.mp4', 'foo.info.json']) + test_session.execute(text('DROP INDEX IF EXISTS file_group_directory_stem_idx')) + + keeper = FileGroup.from_paths(test_session, foo) + keeper.model = 'video' + loser = FileGroup( + directory=foo.parent, + primary_path=sidecar, + stem='foo', + files=[{'path': sidecar.name, 'mimetype': 'application/json'}], + ) + test_session.add(loser) + test_session.commit() + + tag = await tag_factory('keepme') + loser.add_tag(test_session, tag.id) + test_session.commit() + keeper_id, loser_id = keeper.id, loser.id + assert test_session.query(FileGroup).count() == 2 + + deleted = lib.dedupe_file_groups_by_directory_stem() + test_session.expire_all() + + assert deleted == 1 + remaining = test_session.query(FileGroup).one() + assert remaining.id == keeper_id + assert remaining.tag_names == ['keepme'] + assert test_session.query(FileGroup).filter_by(id=loser_id).one_or_none() is None + + +@pytest.mark.asyncio +async def test_dedupe_file_groups_prefers_tagged_row(async_client, test_session, test_directory, + make_files_structure, tag_factory): + """The tagged FileGroup wins even when it has the lower id and no model.""" + from sqlalchemy import text + + a, b = make_files_structure(['bar.mp4', 'bar.png']) + test_session.execute(text('DROP INDEX IF EXISTS file_group_directory_stem_idx')) + first = FileGroup.from_paths(test_session, a) + second = FileGroup( + directory=b.parent, + primary_path=b, + stem='bar', + files=[{'path': b.name, 'mimetype': 'image/png'}], + model='video', + ) + test_session.add(second) + test_session.commit() + assert first.id < second.id + tag = await tag_factory('tagged-first') + first.add_tag(test_session, tag.id) + test_session.commit() + + lib.dedupe_file_groups_by_directory_stem() + test_session.expire_all() + + remaining = test_session.query(FileGroup).one() + # has_model ranks above tags, so the modeled later row wins; tags move onto it. + assert remaining.id == second.id + assert remaining.tag_names == ['tagged-first'] + + @pytest.mark.asyncio async def test_file_group_tag(test_session, make_files_structure, test_directory, tag_factory, await_switches, refresh_files): @@ -791,6 +930,13 @@ def test_group_files_by_stem(make_files_structure, test_directory): test_directory / 'foo.txt'], ] + make_files_structure(['dir_a/s3.bin', 'dir_b/s3.bin']) + same_stem = [test_directory / 'dir_a/s3.bin', test_directory / 'dir_b/s3.bin'] + assert list(lib.group_files_by_stem(same_stem)) == [ + [test_directory / 'dir_a/s3.bin'], + [test_directory / 'dir_b/s3.bin'], + ] + def test_get_primary_file(test_directory, video_file, srt_file3, example_epub, example_mobi, example_pdf, singlefile_contents_factory, make_files_structure): @@ -822,7 +968,7 @@ def test_get_primary_file(test_directory, video_file, srt_file3, example_epub, e @pytest.mark.asyncio async def test_refresh_files_no_groups(async_client, test_session, test_directory, make_files_structure, zip_file_factory, refresh_files): - """Files that share a name, but cannot be grouped into a FileGroup have their own FileGroups.""" + """Files that share a stem are one FileGroup even when neither is a modeled primary.""" foo_txt, foo_zip = make_files_structure({ 'foo.txt': 'text', 'foo.zip': zip_file_factory(), @@ -831,11 +977,12 @@ async def test_refresh_files_no_groups(async_client, test_session, test_director await refresh_files() - # Two distinct FileGroups. - assert test_session.query(FileGroup).count() == 2 - txt, zip_ = test_session.query(FileGroup) - assert txt.primary_path == foo_txt and txt.size == foo_txt.stat().st_size - assert zip_.primary_path == foo_zip and zip_.size == foo_zip.stat().st_size + # Same stem is one FileGroup even when neither file is a modeled primary. + assert test_session.query(FileGroup).count() == 1 + fg = test_session.query(FileGroup).one() + names = {pathlib.Path(f['path']).name for f in fg.files} + assert names == {'foo.txt', 'foo.zip'} + assert fg.size == foo_txt.stat().st_size + foo_zip.stat().st_size @pytest.mark.asyncio @@ -890,53 +1037,34 @@ async def test_upsert_directories_chunks_large_batches(test_session, test_direct @pytest.mark.asyncio async def test_file_group_merge(async_client, test_session, test_directory, make_files_structure, tag_factory, video_bytes, srt_file3): - """A FileGroup can be created from multiple existing FileGroups. Any Tags applied to the existing groups will be - migrated.""" + """from_paths on a sidecar finds the existing (directory, stem) FileGroup instead of inserting a duplicate.""" vid, srt = make_files_structure({ 'vid.mp4': video_bytes, 'vid.srt': (PROJECT_DIR / 'test/example3.en.srt').read_text(), }) one, two = await tag_factory(), await tag_factory() vid_group = FileGroup.from_paths(test_session, vid) - srt_group = FileGroup.from_paths(test_session, srt) - test_session.add_all([vid_group, srt_group]) - test_session.flush([vid_group, srt_group]) + test_session.flush([vid_group]) vid_tag_file = vid_group.add_tag(test_session, one.name) - srt_tag_file = srt_group.add_tag(test_session, two.name) - test_session.flush([vid_tag_file, srt_tag_file]) tag_file_created_at = vid_tag_file.created_at + test_session.commit() + + srt_group = FileGroup.from_paths(test_session, srt) + assert srt_group.id == vid_group.id + srt_tag_file = srt_group.add_tag(test_session, two.name) srt_file_created_at = srt_tag_file.created_at test_session.commit() - assert vid_group.mimetype == 'video/mp4' - assert srt_group.mimetype == 'text/srt' - - # Add srt to vid_group so both groups have it (simulates race condition scenario) - vid_group.append_files(srt) + merged = FileGroup.from_paths(test_session, vid, srt) test_session.commit() - - # Verify both groups have the srt file - assert 'vid.srt' in [f['path'] for f in vid_group.files] - assert 'vid.srt' in [f['path'] for f in srt_group.files] - - # Both FileGroups are merged. - vid = FileGroup.from_paths(test_session, vid, srt) - test_session.commit() - # files now stores relative filenames as strings, use my_files() to get resolved Paths - assert {i['path'].name for i in vid.my_files()} == {'vid.mp4', 'vid.srt'} - + assert merged.id == vid_group.id assert test_session.query(FileGroup).count() == 1 - assert set(vid.tag_names) == {'one', 'two'} - assert {i['path'].name for i in vid.my_files()} == {'vid.mp4', 'vid.srt'} - # TagFile.created_at is preserved. - assert [i for i in vid.tag_files if i.tag.name == 'one'][0].created_at == tag_file_created_at - assert [i for i in vid.tag_files if i.tag.name == 'two'][0].created_at == srt_file_created_at - # Size is combined - assert vid.size > len(video_bytes) - assert vid.mimetype == 'video/mp4' - - # Verify no duplicate files were created during merge - filenames = [f['path'] for f in vid.files] + assert set(merged.tag_names) == {'one', 'two'} + assert {i['path'].name for i in merged.my_files()} == {'vid.mp4', 'vid.srt'} + assert [i for i in merged.tag_files if i.tag.name == 'one'][0].created_at == tag_file_created_at + assert [i for i in merged.tag_files if i.tag.name == 'two'][0].created_at == srt_file_created_at + assert merged.mimetype == 'video/mp4' + filenames = [f['path'] for f in merged.files] assert len(filenames) == len(set(filenames)), f"Duplicate files found: {filenames}" diff --git a/wrolpi/files/test/test_worker.py b/wrolpi/files/test/test_worker.py index 5d7cdb9d..4883bb28 100644 --- a/wrolpi/files/test/test_worker.py +++ b/wrolpi/files/test/test_worker.py @@ -1,6 +1,7 @@ """Tests for the file comparison worker.""" import asyncio import os +import pathlib from unittest import mock import pytest @@ -874,9 +875,10 @@ async def test_file_worker_deletes_removed_file_groups( file_worker.private_queue.put_nowait(task) await file_worker.process_queue() - # Verify FileGroup was deleted + # Verify FileGroup was deleted (do not key off id: SQLite may reuse it for the baseline upsert). test_session.expire_all() - assert test_session.query(FileGroup).filter_by(id=fg_id).one_or_none() is None + remaining = {pathlib.Path(str(fg.primary_path)).name for fg in test_session.query(FileGroup)} + assert 'file1.txt' not in remaining @pytest.mark.asyncio @@ -912,10 +914,13 @@ async def test_file_worker_deletes_tagged_file_groups_auto_removes_tags( file_worker.private_queue.put_nowait(task) await file_worker.process_queue() - # Verify FileGroup and TagFile were deleted (cascade) + # Verify FileGroup and TagFile were deleted (cascade). Do not key off recycled SQLite ids. test_session.expire_all() - assert test_session.query(FileGroup).filter_by(id=fg_id).one_or_none() is None - assert test_session.query(TagFile).filter_by(file_group_id=fg_id).count() == 0 + remaining = {pathlib.Path(str(fg.primary_path)).name for fg in test_session.query(FileGroup)} + assert 'file1.txt' not in remaining + assert test_session.query(TagFile).join(FileGroup).filter( + FileGroup.primary_path.like('%/file1.txt') + ).count() == 0 @pytest.mark.asyncio @@ -1120,11 +1125,10 @@ async def test_file_worker_deletes_multiple_file_groups( file_worker.private_queue.put_nowait(task) await file_worker.process_queue() - # Verify all FileGroups were deleted + # Verify all FileGroups were deleted (do not key off recycled SQLite ids). test_session.expire_all() - assert test_session.query(FileGroup).filter( - FileGroup.id.in_([fg1_id, fg2_id, fg3_id]) - ).count() == 0 + remaining = {pathlib.Path(str(fg.primary_path)).name for fg in test_session.query(FileGroup)} + assert remaining.isdisjoint({'file1.txt', 'file2.txt', 'file3.txt'}) @pytest.mark.asyncio @@ -1681,78 +1685,15 @@ async def test_upsert_file_groups_tracks_actual_db_progress(async_client, test_s @pytest.mark.asyncio -async def test_compare_file_groups_duplicate_deleted_filegroups(test_session, test_directory, make_files_structure): - """When multiple FileGroups share the same (directory, stem) and all files are deleted, - all duplicates should appear in the deleted diffs — not just the last one processed.""" - make_files_structure([ - 'docs/file1.txt', - 'docs/file1.json', - ]) - - docs_dir = str(test_directory / 'docs') - - # Create 3 separate FileGroups for the same (directory, stem) — simulating a botched move. - # Use different primary_paths (unique constraint) but same directory and stem. - fg1 = FileGroup(directory=docs_dir, primary_path=str(test_directory / 'docs/file1.txt'), - files=[{'path': 'file1.txt', 'mimetype': 'text/plain', 'size': 0}]) - fg2 = FileGroup(directory=docs_dir, primary_path=str(test_directory / 'docs/file1.json'), - files=[{'path': 'file1.json', 'mimetype': 'application/json', 'size': 0}]) - # Third ghost with empty files list — use a unique primary_path - (test_directory / 'docs/file1.bak').touch() - fg3 = FileGroup(directory=docs_dir, primary_path=str(test_directory / 'docs/file1.bak'), - files=[]) - test_session.add_all([fg1, fg2, fg3]) - test_session.commit() - - fg1_id, fg2_id, fg3_id = fg1.id, fg2.id, fg3.id - - # Delete all files from disk - for f in (test_directory / 'docs').iterdir(): - f.unlink() +async def test_compare_file_groups_unique_stem_is_one_group(test_session, test_directory, make_files_structure): + """Files that share a stem are one FileGroup; compare reports a single group.""" + make_files_structure(['docs/file1.txt', 'docs/file1.json']) + from wrolpi.files.lib import _upsert_files + _upsert_files(list((test_directory / 'docs').iterdir())) result = await compare_file_groups(test_directory) - - # ALL 3 FileGroup IDs must appear in deleted diffs - deleted_ids = {d.file_group_id for d in result.deleted} - assert fg1_id in deleted_ids, f'fg1 ({fg1_id}) missing from deleted: {deleted_ids}' - assert fg2_id in deleted_ids, f'fg2 ({fg2_id}) missing from deleted: {deleted_ids}' - assert fg3_id in deleted_ids, f'fg3 ({fg3_id}) missing from deleted: {deleted_ids}' - assert len(result.new) == 0 - - -@pytest.mark.asyncio -async def test_compare_file_groups_duplicate_keeps_one(test_session, test_directory, make_files_structure): - """When duplicate FileGroups exist and files are on disk, one should be kept (modified/unchanged) - and the others should be marked for deletion.""" - make_files_structure([ - 'docs/file1.txt', - ]) - - docs_dir = str(test_directory / 'docs') - - # Create 2 FileGroups with same (directory, stem) - fg1 = FileGroup(directory=docs_dir, primary_path=str(test_directory / 'docs/file1.txt'), - files=[{'path': 'file1.txt', 'mimetype': 'text/plain', 'size': 0}]) - (test_directory / 'docs/file1.json').touch() - fg2 = FileGroup(directory=docs_dir, primary_path=str(test_directory / 'docs/file1.json'), - files=[{'path': 'file1.json', 'mimetype': 'application/json', 'size': 0}]) - test_session.add_all([fg1, fg2]) - test_session.commit() - - fg1_id, fg2_id = fg1.id, fg2.id - # Remove the .json so only .txt remains - (test_directory / 'docs/file1.json').unlink() - - result = await compare_file_groups(test_directory) - - # One FG should have the file on disk, the other should be deleted as duplicate - deleted_ids = {d.file_group_id for d in result.deleted} - all_diff_ids = {d.file_group_id for d in result.deleted + result.modified + result.unchanged} - assert fg1_id in all_diff_ids - assert fg2_id in all_diff_ids - # At least one must be in deleted (the duplicate) - assert len(deleted_ids) >= 1 - assert deleted_ids & {fg1_id, fg2_id} + assert len(result.new) + len(result.unchanged) + len(result.modified) == 1 + assert test_session.query(FileGroup).count() == 1 @pytest.mark.asyncio diff --git a/wrolpi/files/worker.py b/wrolpi/files/worker.py index 764ed0dd..a0076328 100644 --- a/wrolpi/files/worker.py +++ b/wrolpi/files/worker.py @@ -33,7 +33,7 @@ from wrolpi.events import Events from wrolpi.vars import PYTEST, IS_MACOS from wrolpi.files.lib import ( split_path_stem_and_suffix, _upsert_files, get_unique_files_by_stem, glob_shared_stem, - group_files_by_stem, get_primary_file, delete_directory, apply_indexers, + group_files_by_stem, get_primary_file, choose_primary_file, delete_directory, apply_indexers, _move_file_group_files, _bulk_update_file_groups_db, MOVE_CHUNK_SIZE, _bulk_update_file_groups_reorganize, get_normalized_ignored_directories, remove_files_in_ignored_directories, ensure_file_group_stems, @@ -41,6 +41,20 @@ from wrolpi.files.lib import ( logger = logger.getChild(__name__) + +async def _await_db(func, *args): + """Run a function that uses get_db_curs. + + Tests share one SQLite connection via get_db_curs. sqlite3 is not safe for concurrent + use of that connection, so asyncio.to_thread segfaults in CI (xdist + session commit + on the event loop while the worker thread is still in get_stem_algorithm_version / + _upsert_files). Production opens a dedicated raw connection and can use a thread. + """ + if PYTEST: + return func(*args) + return await asyncio.to_thread(func, *args) + + # Update status every N items to avoid excessive overhead PROGRESS_UPDATE_INTERVAL = 100 @@ -550,7 +564,7 @@ async def compare_file_groups( if root is not None and roots is None: roots = root root_list = _normalize_roots(roots) - await asyncio.to_thread(ensure_file_group_stems) + await _await_db(ensure_file_group_stems) # Per-call table name so concurrent scans don't collide. The work table is a TEMP table on # its own dedicated connection: it never shares a transaction (or locks) with the session, @@ -1259,8 +1273,8 @@ class FileWorker: try: result = await self._refresh_files_directly(paths) self._cleanup_modified_models(result.modified) - await self._upsert_file_groups(result.new + result.modified) await self._delete_file_groups(result.deleted) + await self._upsert_file_groups(result.new + result.modified) if post_processing and (result.new or result.modified or result.deleted): # Skip global modelers/indexers when the caller opts out (e.g. `lib.delete` from the # upload path defers that to the post-upload `upsert_file`) or when nothing changed. @@ -1492,10 +1506,11 @@ class FileWorker: logger.info(f'Refreshing {len(file_paths)} files directly') file_result = await self._refresh_files_directly(file_paths, expand_stems=task.expand_stems) - # Process file results + # Delete first so a new primary_path cannot collide on (directory, stem) + # with a duplicate FileGroup that compare already marked deleted. self._cleanup_modified_models(file_result.modified) - await self._upsert_file_groups(file_result.new + file_result.modified) await self._delete_file_groups(file_result.deleted) + await self._upsert_file_groups(file_result.new + file_result.modified) # Process directories with full scan (existing path) if dir_paths: @@ -1520,10 +1535,11 @@ class FileWorker: f'{len(dir_result.deleted)} deleted, {len(dir_result.unchanged)} unchanged' ) - # Process directory results + # Delete first so a new primary_path cannot collide on (directory, stem) + # with a duplicate FileGroup that compare already marked deleted. self._cleanup_modified_models(dir_result.modified) - await self._upsert_file_groups(dir_result.new + dir_result.modified) await self._delete_file_groups(dir_result.deleted) + await self._upsert_file_groups(dir_result.new + dir_result.modified) # Run indexers, modelers, and cleanup once for all changes await self._apply_post_processing(is_global_refresh=is_global_refresh) @@ -1594,17 +1610,12 @@ class FileWorker: # For modified diffs where file_group_id exists, we need to handle primary_path changes. # If the primary file changed (e.g., video deleted, leaving only poster), the ON CONFLICT # won't match, so we need to update the existing FileGroup's primary_path first. - from wrolpi.files.lib import NoPrimaryFile modified_ids_to_update = [] for diff in diffs: if diff.file_group_id and diff.fs_files: # This is a modified FileGroup - may need to update primary_path new_paths = [diff.directory / f for f in diff.fs_files] - try: - new_primary = get_primary_file(new_paths) - except NoPrimaryFile: - # No clear primary file, just use the first one - new_primary = new_paths[0] if new_paths else None + new_primary = choose_primary_file(new_paths) if new_primary: modified_ids_to_update.append((diff.file_group_id, str(new_primary))) @@ -1670,7 +1681,7 @@ class FileWorker: self.update_status(operation_processed=processed, operation_percent=percent) # Use existing _upsert_files function which handles grouping and primary file detection - await asyncio.to_thread(_upsert_files, all_paths, on_upsert_progress) + await _await_db(_upsert_files, all_paths, on_upsert_progress) logger.info(f'Upserted {len(diffs)} FileGroups ({len(all_paths)} files)') async def _delete_file_groups(self, diffs: list[FileGroupDiff]): diff --git a/wrolpi/jobs.py b/wrolpi/jobs.py index 44acb5c0..a2508240 100644 --- a/wrolpi/jobs.py +++ b/wrolpi/jobs.py @@ -300,8 +300,19 @@ class JobLogHandler(logging.Handler): _log_handler = JobLogHandler() _log_handler.setFormatter(logging.Formatter('%(levelname)s:%(name)s:%(message)s')) -if not any(isinstance(i, JobLogHandler) for i in logging.getLogger().handlers): - logging.getLogger().addHandler(_log_handler) + + +def _ensure_log_handler(): + """Sanic/pytest dictConfig replaces root handlers and would drop JobLogHandler. + + Attach to this module's logger (not only root) so capture survives that reset. + """ + for target in (logging.getLogger(), logger): + if not any(isinstance(i, JobLogHandler) for i in target.handlers): + target.addHandler(_log_handler) + + +_ensure_log_handler() def _json_safe(value): @@ -314,6 +325,7 @@ def _json_safe(value): async def _run_job(job_id: str): """Run one Job to completion (or failure/cancellation) in this process.""" + _ensure_log_handler() record = _claim_job(job_id) if not record: # Cancelled while it was waiting in the queue. @@ -360,8 +372,13 @@ async def _run_job(job_id: str): raise except Exception as e: logger.error(f'Job {job_id} failed', exc_info=e) + # Do not rely on JobLogHandler for this line: Sanic/pytest dictConfig can + # strip root handlers (especially under xdist on CI), so the traceback + # never reaches the Job log even though `error` is set. + context.log(f'{type(e).__name__}: {e}') context.flush() - _update_job(job_id, status=FAILED, error=str(e) or repr(e), finished_at=now().isoformat()) + _update_job(job_id, status=FAILED, error=str(e) or repr(e), + log=list(context._log), finished_at=now().isoformat()) Events.send_user_notify(f'Failed: {record["description"]}') return diff --git a/wrolpi/test/test_db_bootstrap.py b/wrolpi/test/test_db_bootstrap.py index 056c7eb4..045a050d 100644 --- a/wrolpi/test/test_db_bootstrap.py +++ b/wrolpi/test/test_db_bootstrap.py @@ -66,6 +66,57 @@ def test_compare_db_version_uninitialized_file(test_directory): assert db_bootstrap.compare_db_version() == 'current' +def test_unique_stem_migration_collapses_duplicates(test_directory): + """The unique-index migration merges duplicate (directory, stem) rows and keeps tags.""" + import sqlite3 + from alembic.config import Config + from alembic import command + from wrolpi.db import get_db_uri, get_db_file + from wrolpi.vars import PROJECT_DIR + + (test_directory / 'config').mkdir(parents=True, exist_ok=True) + config = Config(str(PROJECT_DIR / 'alembic.ini')) + config.set_main_option('sqlalchemy.url', get_db_uri()) + command.upgrade(config, '2026_09_16_1200') + + db_file = get_db_file() + directory = str(test_directory) + with sqlite3.connect(db_file) as conn: + conn.execute('PRAGMA foreign_keys=ON') + conn.execute( + 'INSERT INTO file_group (directory, primary_path, files, stem, indexed) VALUES (?,?,?,?,0)', + (directory, str(test_directory / 'dup.mp4'), '[]', 'dup'), + ) + conn.execute( + 'INSERT INTO file_group (directory, primary_path, files, stem, indexed) VALUES (?,?,?,?,0)', + (directory, str(test_directory / 'dup.info.json'), '[]', 'dup'), + ) + first_id, second_id = [r[0] for r in conn.execute('SELECT id FROM file_group ORDER BY id').fetchall()] + conn.execute("INSERT INTO tag (name, color) VALUES ('migrated', '#111111')") + tag_id = conn.execute('SELECT id FROM tag').fetchone()[0] + # Tag the lower-id row so it wins over the later untagged duplicate. + conn.execute( + 'INSERT INTO tag_file (tag_id, file_group_id) VALUES (?, ?)', + (tag_id, first_id), + ) + conn.commit() + assert conn.execute('SELECT COUNT(*) FROM file_group').fetchone()[0] == 2 + assert first_id < second_id + + command.upgrade(config, 'head') + + with sqlite3.connect(db_file) as conn: + rows = list(conn.execute('SELECT id, stem FROM file_group')) + assert len(rows) == 1 + assert rows[0][0] == first_id + tagged = conn.execute('SELECT file_group_id FROM tag_file').fetchone()[0] + assert tagged == first_id + sql = conn.execute( + "SELECT sql FROM sqlite_master WHERE type='index' AND name='file_group_directory_stem_idx'" + ).fetchone()[0] + assert 'UNIQUE' in sql.upper() + + def test_bootstrap_lock(test_directory): """Only one process can hold the bootstrap lock.""" with db_bootstrap.bootstrap_lock() as acquired: