files: keep stem current on reorganize; NUL-safe find parse

Reorganize wrote the new primary_path but not stem, so the next refresh
treated the FileGroup as deleted+new and dropped tags.  GNU find now
emits NUL-delimited records so a tab or newline in a filename cannot
abort a global refresh.  Stem backfill skips the table scan once the
algorithm version matches, and runs off the event loop.  An empty roots
list is a global refresh, not invalid SQL.

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):

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'],
                          ))
 

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

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."""

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,