diff --git a/wrolpi/collections/test/test_reorganize.py b/wrolpi/collections/test/test_reorganize.py index fcbe3670..538f27af 100644 --- a/wrolpi/collections/test/test_reorganize.py +++ b/wrolpi/collections/test/test_reorganize.py @@ -784,6 +784,64 @@ async def test_reorganize_preserves_tag_association(async_client, test_directory assert fg.primary_path == new_primary_path, "File should have moved" +@pytest.mark.asyncio +async def test_reorganize_then_refresh_preserves_filegroup_id_and_tags( + async_client, test_directory, video_factory, tag_factory, test_tags_config, refresh_files): + """A refresh after reorganize must keep the FileGroup id (and therefore tags). + + Regression: _bulk_update_file_groups_reorganize left `stem` stale. Compare then + classified the old stem as deleted and the new filename as new, dropping tags. + """ + from wrolpi.files.worker import file_worker, FileTask, FileTaskType + import asyncio + + channel_dir = test_directory / 'videos' / 'refresh_after_reorg' + channel_dir.mkdir(parents=True, exist_ok=True) + + with get_db_session(commit=True) as session: + collection = Collection( + name='Refresh After Reorg', + kind='channel', + directory=channel_dir, + file_format='%(title)s.%(ext)s' + ) + session.add(collection) + session.flush() + channel = Channel(name='Refresh After Reorg', collection_id=collection.id, directory=channel_dir) + session.add(channel) + session.commit() + channel_id = channel.id + + tag = await tag_factory('keepme') + video = video_factory(channel_id=channel_id, title='old title', with_video_file=True) + + with get_db_session(commit=True) as session: + video_obj = session.query(Video).get(video.id) + video_obj.file_group.add_tag(session, tag.id) + fg_id = video_obj.file_group.id + old_primary_path = video_obj.file_group.primary_path + session.commit() + + new_primary_path = channel_dir / '2024-01-01 New Title.mp4' + await file_worker.handle_reorganize(FileTask( + task_type=FileTaskType.reorganize, + paths=[], + move_mappings=[(old_primary_path, new_primary_path)], + job_id='test-reorganize-then-refresh', + )) + await asyncio.sleep(0.1) + + await refresh_files([channel_dir]) + + with get_db_session() as session: + fg = session.query(FileGroup).get(fg_id) + assert fg is not None, 'FileGroup was dropped by refresh after reorganize' + assert fg.primary_path == new_primary_path + assert fg.stem == '2024-01-01 New Title' + assert fg.tag_names == ['keepme'] + assert session.query(FileGroup).count() == 1 + + @pytest.mark.asyncio async def test_reorganize_preserves_multiple_tags(async_client, test_directory, video_factory, tag_factory, test_tags_config): diff --git a/wrolpi/files/lib.py b/wrolpi/files/lib.py index 8d90bdcb..e4e7d650 100644 --- a/wrolpi/files/lib.py +++ b/wrolpi/files/lib.py @@ -842,31 +842,29 @@ def set_stem_algorithm_version(version: int): def ensure_file_group_stems(): - """Recompute file_group.stem when NULL or when STEM_ALGORITHM_VERSION has changed. + """Recompute file_group.stem when STEM_ALGORITHM_VERSION has changed. A stored stem is a cache of split_path_stem_and_suffix. Compare joins on (directory, stem), so an algorithm change without this backfill reports the same files as deleted+new and drops tags, viewed timestamps, and domain rows. Must run before any compare. + + After a successful backfill the kv version matches, and later refreshes skip the table scan. + Upsert, from_paths, move, and reorganize must keep stem current. """ stored = get_stem_algorithm_version() - force = stored != STEM_ALGORITHM_VERSION - with get_db_curs() as curs: - if force: - curs.execute('SELECT id, primary_path FROM file_group') - else: - curs.execute("SELECT id, primary_path FROM file_group WHERE stem IS NULL OR stem = ''") - rows = list(curs.fetchall()) - if not rows: - if stored != STEM_ALGORITHM_VERSION: - set_stem_algorithm_version(STEM_ALGORITHM_VERSION) + if stored == STEM_ALGORITHM_VERSION: return - updates = [] - for row in rows: - stem, _ = split_path_stem_and_suffix(row['primary_path']) - updates.append((stem, row['id'])) - with get_db_curs(commit=True) as curs: - for chunk in chunks(updates, 500): - curs.executemany('UPDATE file_group SET stem = ? WHERE id = ?', chunk) + with get_db_curs() as curs: + curs.execute('SELECT id, primary_path FROM file_group') + rows = list(curs.fetchall()) + if rows: + updates = [] + for row in rows: + stem, _ = split_path_stem_and_suffix(row['primary_path']) + updates.append((stem, row['id'])) + with get_db_curs(commit=True) as curs: + for chunk in chunks(updates, 500): + curs.executemany('UPDATE file_group SET stem = ? WHERE id = ?', chunk) set_stem_algorithm_version(STEM_ALGORITHM_VERSION) @@ -1626,18 +1624,22 @@ def _bulk_update_file_groups_reorganize(updates: List[dict]): with get_db_curs(commit=True) as curs: for update in updates: # Don't change indexed - reorganization only moves files, content unchanged + primary_path = update['primary_path'] + stem, _ = split_path_stem_and_suffix(primary_path) curs.execute(''' UPDATE file_group SET directory = ?, primary_path = ?, files = ?, - data = ? + data = ?, + stem = ? WHERE id = ? ''', ( update['directory'], - update['primary_path'], + primary_path, json.dumps(update['files'], default=_json_serial), json.dumps(update['data'], default=_json_serial) if update.get('data') else None, + stem, update['id'], )) diff --git a/wrolpi/files/test/test_lib.py b/wrolpi/files/test/test_lib.py index becf7379..e831bb53 100644 --- a/wrolpi/files/test/test_lib.py +++ b/wrolpi/files/test/test_lib.py @@ -1712,3 +1712,30 @@ def test_bulk_update_file_groups_reorganize_handles_datetime(test_session, test_ assert fg.files[0]['path'] == 'test_video.mp4' # modification_datetime should be serialized as ISO string assert 'modification_datetime' in fg.files[0] + + +def test_bulk_update_file_groups_reorganize_updates_stem(test_session, test_directory): + """Reorganize must write the new stem so a later refresh does not delete+reinsert the row.""" + from wrolpi.files.lib import _bulk_update_file_groups_reorganize + + old_file = test_directory / 'old title.mp4' + old_file.touch() + fg = FileGroup.from_paths(test_session, old_file) + test_session.commit() + assert fg.stem == 'old title' + fg_id = fg.id + + new_path = test_directory / '2024-01-01 New Title.mp4' + updates = [{ + 'id': fg_id, + 'directory': str(test_directory), + 'primary_path': str(new_path), + 'files': [{'path': new_path.name, 'mimetype': 'video/mp4'}], + 'data': None, + }] + _bulk_update_file_groups_reorganize(updates) + + test_session.expire_all() + fg = test_session.query(FileGroup).filter_by(id=fg_id).one() + assert fg.stem == '2024-01-01 New Title' + assert fg.primary_path == new_path diff --git a/wrolpi/files/test/test_worker.py b/wrolpi/files/test/test_worker.py index 24443833..3e32d6bd 100644 --- a/wrolpi/files/test/test_worker.py +++ b/wrolpi/files/test/test_worker.py @@ -17,6 +17,9 @@ from wrolpi.files.worker import ( FileTaskType, FileWorkerJobFailed, QUEUE_STALL_SECONDS, + _split_nul_field_triples, + _entry_from_printf_fields, + _normalize_roots, ) @@ -1800,6 +1803,47 @@ async def test_refresh_two_directories_does_not_delete_third( 'refreshing A and B deleted a stale FileGroup in C' +def test_split_nul_field_triples_allows_tab_and_newline_in_filename(): + """GNU find -printf uses NUL delimiters so a tab/newline in the name is not a field break.""" + data = ( + b'/media/wrolpi\0file\twith\ttabs.txt\0' + b'17123.45\0' + b'/media/wrolpi\0new\nline.mp4\0' + b'99.5\0leftover' + ) + triples, leftover = _split_nul_field_triples(data) + assert len(triples) == 2 + directory, filename, mtime = triples[0] + assert directory == b'/media/wrolpi' + assert filename == b'file\twith\ttabs.txt' + assert mtime == b'17123.45' + assert triples[1][1] == b'new\nline.mp4' + assert leftover == b'leftover' + + entry = _entry_from_printf_fields(*triples[0]) + assert entry[1] == 'file\twith\ttabs.txt' + assert entry[3] == 17123.45 + + assert _entry_from_printf_fields(b'/tmp', b'bad.txt', b'not-a-float') is None + + +def test_normalize_roots_empty_list_is_media_directory(test_directory): + """An empty roots list must not produce `()` SQL; treat it like a global refresh.""" + assert _normalize_roots([]) == [test_directory] + assert _normalize_roots(None) == [test_directory] + + +@pytest.mark.asyncio +async def test_compare_file_groups_empty_roots_is_global(test_session, test_directory, make_files_structure): + """compare_file_groups([]) scans the media directory, same as omitting roots.""" + make_files_structure(['only.txt']) + via_none = await compare_file_groups(None) + via_empty = await compare_file_groups([]) + assert len(via_none.new) == 1 + assert len(via_empty.new) == 1 + assert via_empty.new[0].stem == 'only' + + @pytest.mark.asyncio async def test_move_planning_tracks_progress(async_client, test_session, test_directory, make_files_structure): """Move planning phase tracks progress.""" diff --git a/wrolpi/files/worker.py b/wrolpi/files/worker.py index 09009f2b..9203bb68 100644 --- a/wrolpi/files/worker.py +++ b/wrolpi/files/worker.py @@ -230,11 +230,13 @@ def classify_file_group_diffs( def _normalize_roots(roots: Path | list[Path] | None) -> list[Path]: + """Always at least the media directory. An empty list is treated like None.""" if roots is None: return [get_media_directory()] if isinstance(roots, (str, Path)): return [Path(roots)] - return [Path(r) for r in roots] + root_list = [Path(r) for r in roots] + return root_list or [get_media_directory()] def _directory_in_roots_sql(roots: list[Path], column: str = 'fg.directory') -> Tuple[str, dict]: @@ -385,6 +387,63 @@ async def _stream_filesystem_paths(root: Path | list[Path]) -> AsyncGenerator[st yield str(Path(directory) / filename) +def _split_nul_field_triples(buf: bytes) -> Tuple[list, bytes]: + """Pull complete NUL-separated (directory, filename, mtime) triples out of `buf`. + + Returns (triples, leftover_bytes). A filename may contain tabs or newlines; only NUL + is the delimiter (GNU find `-printf '%h\\0%f\\0%T@\\0'`). + """ + triples = [] + while True: + parts = buf.split(b'\0', 3) + if len(parts) < 4: + return triples, buf + directory_b, filename_b, mtime_b, buf = parts + triples.append((directory_b, filename_b, mtime_b)) + + +def _entry_from_printf_fields(directory_b: bytes, filename_b: bytes, mtime_b: bytes): + """Decode one find -printf record. Returns None and logs if the record is unusable.""" + try: + directory = directory_b.decode() + filename = filename_b.decode() + mtime = float(mtime_b.decode()) + except (UnicodeDecodeError, ValueError) as e: + logger.warning(f'Skipping unreadable find record {directory_b!r}/{filename_b!r}: {e}') + return None + if not directory or not filename: + return None + stem, _ = split_path_stem_and_suffix(Path(directory) / filename) + return directory, filename, stem, mtime + + +async def _stream_gnu_find_entries(find_args: list[str]) -> AsyncGenerator[Tuple[str, str, str, float], None]: + """Stream (directory, filename, stem, mtime) using GNU find -printf with NUL delimiters.""" + proc = await asyncio.create_subprocess_exec( + 'find', *find_args, '-printf', '%h\0%f\0%T@\0', + stdout=asyncio.subprocess.PIPE, + ) + buf = b'' + try: + while True: + chunk = await proc.stdout.read(65536) + if not chunk: + break + buf += chunk + triples, buf = _split_nul_field_triples(buf) + for fields in triples: + entry = _entry_from_printf_fields(*fields) + if entry: + yield entry + if buf.strip(b'\0'): + logger.warning(f'Incomplete find -printf record discarded: {buf[:200]!r}') + await proc.wait() + finally: + if proc.returncode is None: + proc.kill() + await proc.wait() + + async def _stream_filesystem_entries( roots: Path | list[Path], ) -> AsyncGenerator[Tuple[str, str, str, float], None]: @@ -394,29 +453,12 @@ async def _stream_filesystem_entries( `-printf`, so each path is stat()'d once while streaming. Either way the mtime is stored in the TEMP table and unchanged groups are never stat()'d again. """ - root_list = _normalize_roots(roots) if not isinstance(roots, list) else list(roots) - if not root_list: - return + root_list = _normalize_roots(roots) find_args = _find_args_for_roots(root_list) if not IS_MACOS: - proc = await asyncio.create_subprocess_exec( - 'find', *find_args, '-printf', '%h\t%f\t%T@\n', - stdout=asyncio.subprocess.PIPE, - ) - try: - async for line in proc.stdout: - decoded = line.decode().rstrip('\n') - if not decoded: - continue - directory, filename, mtime_s = decoded.split('\t', 2) - stem, _ = split_path_stem_and_suffix(Path(directory) / filename) - yield directory, filename, stem, float(mtime_s) - await proc.wait() - finally: - if proc.returncode is None: - proc.kill() - await proc.wait() + async for entry in _stream_gnu_find_entries(find_args): + yield entry return proc = await asyncio.create_subprocess_exec( @@ -503,7 +545,7 @@ async def compare_file_groups( if root is not None and roots is None: roots = root root_list = _normalize_roots(roots) - ensure_file_group_stems() + await asyncio.to_thread(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,