From bd15b85c86b171cf3e8fe9f470cc59cbb583dba5 Mon Sep 17 00:00:00 2001 From: Roland Date: Wed, 16 Sep 2026 20:20:42 -0600 Subject: [PATCH] files: SQL FileGroup compare, stem column, drop unused discovery path Refresh walked the tree twice (find, then stat every unchanged group), and a two-directory refresh scanned the whole media directory. Compare now diffs in SQL on a TEMP table that already has mtime, accepts multiple roots, and groups by a stored stem that is versioned so a suffix-rule change cannot delete-and-reinsert tagged files. Deletes refresh_discover_paths and /api/files/refresh_progress, which nothing in production called. Video/doc/zim modelers use the same (no domain row OR indexed != True) gate. --- .../2026_09_16_1200_file_group_stem.py | 37 ++ modules/archive/__init__.py | 5 +- modules/docs/__init__.py | 2 +- modules/docs/test/test_doc_modeler.py | 7 +- modules/videos/__init__.py | 3 +- modules/videos/schema.py | 2 - modules/zim/lib.py | 19 +- scripts/api_perf.py | 1 - wrolpi/common.py | 8 +- wrolpi/conftest.py | 7 +- wrolpi/contexts.py | 2 - wrolpi/files/api.py | 11 - wrolpi/files/indexers.py | 16 +- wrolpi/files/lib.py | 286 ++++------ wrolpi/files/models.py | 29 +- wrolpi/files/test/test_lib.py | 120 +++-- wrolpi/files/test/test_worker.py | 76 ++- wrolpi/files/worker.py | 510 +++++++++++------- 18 files changed, 634 insertions(+), 507 deletions(-) create mode 100644 alembic/versions/2026_09_16_1200_file_group_stem.py diff --git a/alembic/versions/2026_09_16_1200_file_group_stem.py b/alembic/versions/2026_09_16_1200_file_group_stem.py new file mode 100644 index 00000000..f5d06136 --- /dev/null +++ b/alembic/versions/2026_09_16_1200_file_group_stem.py @@ -0,0 +1,37 @@ +"""Add file_group.stem for (directory, stem) grouping. + +A stored stem is a cache of split_path_stem_and_suffix. Compare joins on it. +NULL stems (and algorithm-version mismatches) are backfilled at refresh by +ensure_file_group_stems() before any compare is allowed. + +The unique (directory, stem) index is intentionally omitted: duplicates are +collapsed at refresh time. A later release can add UNIQUE after a full refresh. + +Revision ID: 2026_09_16_1200 +Revises: 2026_07_14_0900 +""" +from alembic import op + +# revision identifiers, used by Alembic. +revision = '2026_09_16_1200' +down_revision = '2026_07_14_0900' +branch_labels = None +depends_on = None + + +def upgrade(): + op.execute('ALTER TABLE file_group ADD COLUMN stem VARCHAR') + op.execute('CREATE INDEX IF NOT EXISTS file_group_directory_stem_idx ' + 'ON file_group (directory, stem)') + op.execute(''' + CREATE TABLE IF NOT EXISTS wrolpi_kv ( + key TEXT PRIMARY KEY, + value TEXT NOT NULL + ) + ''') + + +def downgrade(): + op.execute('DROP INDEX IF EXISTS file_group_directory_stem_idx') + # SQLite cannot DROP COLUMN without rebuilding the table; leave stem in place. + op.execute('DROP TABLE IF EXISTS wrolpi_kv') diff --git a/modules/archive/__init__.py b/modules/archive/__init__.py index 78eaa984..3ad5554b 100644 --- a/modules/archive/__init__.py +++ b/modules/archive/__init__.py @@ -369,7 +369,10 @@ async def archive_modeler(progress_callback: Callable[[int], None] = None): with get_db_session(commit=True) as session: results = session.query(FileGroup, Archive) \ .filter( - # Get all groups that contain an HTML file that have not been indexed. + # Archive shares text/html with generic HTML. Unlike video/doc/zim, do NOT + # select Archive.id IS NULL: that would re-probe every leftover HTML file + # every refresh. Non-SingleFile HTML stays indexed=0 so apply_indexers + # can claim it. FileGroup.indexed != True, FileGroup.mimetype == 'text/html', ).filter(not_(FileGroup.id.in_(list(invalid_archives)))) \ diff --git a/modules/docs/__init__.py b/modules/docs/__init__.py index a22530fa..e7b15bce 100644 --- a/modules/docs/__init__.py +++ b/modules/docs/__init__.py @@ -60,7 +60,7 @@ async def doc_modeler(progress_callback: Callable[[int], None] = None): # permanently unmodeled and invisible in /api/docs. or_( Doc.id.is_(None), - FileGroup.indexed == False, + FileGroup.indexed != True, ), ) if failed_ids: diff --git a/modules/docs/test/test_doc_modeler.py b/modules/docs/test/test_doc_modeler.py index a219fb73..315c91df 100644 --- a/modules/docs/test/test_doc_modeler.py +++ b/modules/docs/test/test_doc_modeler.py @@ -10,7 +10,6 @@ from sqlalchemy.exc import OperationalError from modules.docs import doc_modeler from modules.docs.models import Doc, DocSection -from wrolpi.dates import now from wrolpi.files import lib as files_lib from wrolpi.files.models import FileGroup from wrolpi.files.worker import file_worker @@ -105,7 +104,7 @@ async def test_doc_modeler_replaces_sections(async_client, test_session, test_di @pytest.mark.asyncio async def test_doc_do_model_sets_model(test_session, example_pdf): """Doc.do_model sets file_group.model so the FileGroup is recognized as a doc.""" - files_lib._upsert_files([example_pdf], now()) + files_lib._upsert_files([example_pdf]) file_group = test_session.query(FileGroup).one() # Reset to simulate a newly discovered file. file_group.model = None @@ -394,7 +393,7 @@ async def test_doc_modeler_pdf_max_size(test_session, example_pdf): """The contents of a large PDF are not indexed.""" example_pdf.write_bytes(example_pdf.read_bytes() * 5000) - files_lib._upsert_files([example_pdf], now()) + files_lib._upsert_files([example_pdf]) await doc_modeler() file_group = test_session.query(FileGroup).one() @@ -407,7 +406,7 @@ async def test_doc_modeler_pdf_poster(test_session, example_pdf): poster_path = example_pdf.with_suffix('.jpg') Image.new('RGB', (25, 25), color='grey').save(poster_path) - files_lib._upsert_files([example_pdf, poster_path], now()) + files_lib._upsert_files([example_pdf, poster_path]) await doc_modeler() file_group = test_session.query(FileGroup).one() diff --git a/modules/videos/__init__.py b/modules/videos/__init__.py index 4a376848..a2d1f27a 100644 --- a/modules/videos/__init__.py +++ b/modules/videos/__init__.py @@ -57,7 +57,8 @@ async def video_modeler(progress_callback: Callable[[int], None] = None): .filter( or_(FileGroup.mimetype.like('video/%'), FileGroup.mimetype.like('audio/%')), FileGroup.mimetype.notin_(AUDIO_PLAYLIST_MIMETYPES), - # Also model indexed FileGroups that never got a Video row. + # Exclusive-mimetype modeler: claim unmodeled rows even if apply_indexers + # already set indexed=True, and re-model when files changed (indexed != True). or_(Video.id.is_(None), FileGroup.indexed != True), ) if failed_ids: diff --git a/modules/videos/schema.py b/modules/videos/schema.py index 7296d8db..3eac9834 100644 --- a/modules/videos/schema.py +++ b/modules/videos/schema.py @@ -122,7 +122,6 @@ class Video: video_path: str name: str caption_path: str - idempotency: str info_json_path: str channel_id: int @@ -138,7 +137,6 @@ class VideoWithChannel: video_path: str name: str caption_path: str - idempotency: str info_json_path: str channel_id: int diff --git a/modules/zim/lib.py b/modules/zim/lib.py index 3d32328c..84b5297a 100644 --- a/modules/zim/lib.py +++ b/modules/zim/lib.py @@ -8,6 +8,7 @@ from typing import Callable, List, Tuple, Dict import cachetools.func from libzim import Entry +from sqlalchemy import or_ from sqlalchemy.orm import Session from sqlalchemy.orm.exc import NoResultFound # noqa @@ -75,15 +76,20 @@ def model_zim(file_group: FileGroup, session: Session) -> Zim: @register_modeler async def zim_modeler(progress_callback: Callable[[int], None] = None): total_processed = 0 + failed_ids: set = set() while True: with get_db_session(commit=True) as session: - file_groups = session.query(FileGroup, Zim) \ - .filter(FileGroup.indexed != True, # noqa - FileGroup.primary_path.ilike('%.zim'), - ) \ + query = session.query(FileGroup, Zim) \ .outerjoin(Zim, Zim.file_group_id == FileGroup.id) \ - .limit(10) - file_groups: List[Tuple[FileGroup, Zim]] = list(file_groups) + .filter(FileGroup.primary_path.ilike('%.zim')) \ + .filter( + # Exclusive-mimetype modeler: claim unmodeled rows even if apply_indexers + # already set indexed=True, and re-model when files changed. + or_(Zim.id.is_(None), FileGroup.indexed != True), + ) + if failed_ids: + query = query.filter(FileGroup.id.notin_(failed_ids)) + file_groups: List[Tuple[FileGroup, Zim]] = list(query.limit(10)) processed = 0 for file_group, zim in file_groups: @@ -101,6 +107,7 @@ async def zim_modeler(progress_callback: Callable[[int], None] = None): file_group.indexed = True file_group.model = 'zim' except Exception as e: + failed_ids.add(file_group.id) if PYTEST: raise logger.error(f'Unable to model Zim {zim_id=} {file_group.primary_path=}', exc_info=e) diff --git a/scripts/api_perf.py b/scripts/api_perf.py index 579aca57..67c74431 100644 --- a/scripts/api_perf.py +++ b/scripts/api_perf.py @@ -95,7 +95,6 @@ CATALOG = [ dict(name='config/backups', method='GET', path='/api/config/backups'), # --- files --- - dict(name='files/refresh_progress', method='GET', path='/api/files/refresh_progress', page='files'), dict(name='files/worker_status', method='GET', path='/api/files/worker_status'), dict(name='files/bulk_tag/progress', method='GET', path='/api/files/bulk_tag/progress'), dict(name='files (root listing)', method='POST', path='/api/files/', body={'directories': []}, page='files'), diff --git a/wrolpi/common.py b/wrolpi/common.py index 377afa11..7769567c 100644 --- a/wrolpi/common.py +++ b/wrolpi/common.py @@ -2270,6 +2270,7 @@ modelers = [] def register_modeler(modeler: callable): + """Register a FileGroup modeler. `modeler` must accept `progress_callback=None`.""" modelers.append(modeler) return modeler @@ -2297,10 +2298,9 @@ async def apply_modelers(progress_callback: callable = None): for modeler in modelers: logger_.info(f'Applying modeler {modeler.__name__}') try: - # Check if this modeler accepts a progress_callback parameter - sig = inspect.signature(modeler) - if 'progress_callback' in sig.parameters and progress_callback and initial_unindexed > 0: - # Create a callback that updates the cumulative progress + # Modelers must accept progress_callback= (see register_modeler). The leftover + # indexer (apply_indexers) runs after all of these. + if progress_callback and initial_unindexed > 0: def make_modeler_callback(current_cumulative): def modeler_progress_callback(batch_processed: int): nonlocal cumulative_processed diff --git a/wrolpi/conftest.py b/wrolpi/conftest.py index 746794dc..da6fbdcb 100644 --- a/wrolpi/conftest.py +++ b/wrolpi/conftest.py @@ -1048,14 +1048,17 @@ def insert_file_group(test_session, test_directory): # Store relative filenames in files (new schema) files = [dict(path=i.name, mimetype='fake') for i in paths] primary_path = paths[0] + from wrolpi.files.lib import split_path_stem_and_suffix + stem, _ = split_path_stem_and_suffix(primary_path) params = dict( directory=str(primary_path.parent), primary_path=str(primary_path), files=json.dumps(files), + stem=stem, ) test_session.execute('INSERT INTO file_group ' - '(indexed, directory, primary_path, files) VALUES ' - '(true, :directory, :primary_path, :files)', params) + '(indexed, directory, primary_path, files, stem) VALUES ' + '(true, :directory, :primary_path, :files, :stem)', params) return _ diff --git a/wrolpi/contexts.py b/wrolpi/contexts.py index 9e94a7af..5aece80a 100644 --- a/wrolpi/contexts.py +++ b/wrolpi/contexts.py @@ -36,7 +36,6 @@ def attach_shared_contexts(app: Sanic): app.shared_ctx.map_pins_config = manager.dict() app.shared_ctx.flasher_config = manager.dict() # Shared dicts. - app.shared_ctx.refresh = manager.dict() app.shared_ctx.uploaded_files = manager.dict() app.shared_ctx.status = manager.dict() app.shared_ctx.cache = manager.dict() @@ -131,7 +130,6 @@ def reset_shared_contexts(app: Sanic): app.shared_ctx.archive_downloader_config.clear() app.shared_ctx.download_cache_config.clear() # Shared dicts. - app.shared_ctx.refresh.clear() app.shared_ctx.uploaded_files.clear() app.shared_ctx.status.clear() app.shared_ctx.speedtest_active.value = 0 diff --git a/wrolpi/files/api.py b/wrolpi/files/api.py index ea7654cd..5cef5593 100644 --- a/wrolpi/files/api.py +++ b/wrolpi/files/api.py @@ -134,17 +134,6 @@ async def refresh(request: Request): return response.empty() -@files_bp.get('/refresh_progress') -@openapi.definition( - summary='Get the progress of the file refresh' -) -async def refresh_progress(request: Request): - progress = lib.get_refresh_progress() - return json_response(dict( - progress=progress, - )) - - @files_bp.get('/worker_status') @openapi.definition( summary='Get the FileWorker status with detailed progress' diff --git a/wrolpi/files/indexers.py b/wrolpi/files/indexers.py index abaeec75..2701ed5d 100644 --- a/wrolpi/files/indexers.py +++ b/wrolpi/files/indexers.py @@ -40,18 +40,18 @@ indexer_map = defaultdict(lambda: DefaultIndexer) def find_indexer(mimetype: str) -> Type[Indexer]: - """Find the Indexer for a given File model.""" + """Find the Indexer for a given File model. Longest registered prefix wins.""" if not mimetype: return DefaultIndexer - # Get the indexer that matches the mimetype of this file. - indexer = next((v for k, v in indexer_map.items() if mimetype.startswith(k)), None) + matches = [(key, indexer) for key, indexer in indexer_map.items() if mimetype.startswith(key)] + if matches: + # text/html must beat a later-inserted 'text' DefaultIndexer fallback. + _key, indexer = max(matches, key=lambda item: len(item[0])) + return indexer - if not indexer: - # Use the broad indexer. - indexer = indexer_map[mimetype.split('/')[0]] - - return indexer + # dict.get does not insert into the defaultdict, so a miss does not pollute the map. + return indexer_map.get(mimetype.split('/')[0], DefaultIndexer) def register_indexer(*mimetypes: str): diff --git a/wrolpi/files/lib.py b/wrolpi/files/lib.py index 5ebfbb40..8d90bdcb 100644 --- a/wrolpi/files/lib.py +++ b/wrolpi/files/lib.py @@ -23,11 +23,9 @@ from sqlalchemy import asc, or_, text as sa_text from sqlalchemy.exc import IntegrityError from sqlalchemy.orm import Session -from wrolpi import flags from wrolpi.cmd import which from wrolpi.common import get_media_directory, wrol_mode_check, logger, \ - partition, \ - get_files_and_directories, chunks, chunks_by_stem, walk, \ + chunks, chunks_by_stem, walk, \ get_wrolpi_config, \ unique_by_predicate, get_paths_in_media_directory, TRACE_LEVEL, get_relative_to_media_directory, strip_surrogates from wrolpi.dates import now, from_timestamp, months_selector_to_where, date_range_to_where @@ -58,7 +56,8 @@ __all__ = ['list_directories_contents', 'delete', 'split_path_stem_and_suffix', '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'] + 'cached_search_total', 'search_filter_cache_key', 'count_file_groups', + 'STEM_ALGORITHM_VERSION', 'ensure_file_group_stems'] def get_file_tag_names(session: Session, file: pathlib.Path) -> List[str]: @@ -488,6 +487,13 @@ _EXTRA_SUFFIXES_LOWER = frozenset(s.lower() for s in EXTRA_SUFFIXES) _HARDCODED_SUFFIXES = frozenset( {'.info.json', '.ffprobe.json', '.live_chat.json', '.readability.html', '.readability.json', '.readability.txt'}) +# Bump when SUFFIXES / split_path_stem_and_suffix grouping rules change. A stored +# file_group.stem is a cache of this algorithm; compare joins on it, so a mismatch +# looks like delete+insert and would drop tags. ensure_file_group_stems() must run +# before any compare when this version does not match the value in wrolpi_kv. +STEM_ALGORITHM_VERSION = 1 +STEM_ALGORITHM_VERSION_KEY = 'stem_algorithm_version' + PART_PARSER = re.compile(r'(.+?)(\.f[\d]{2,3})?(\.info)?(\.\w{3,4})(\.part)', re.IGNORECASE) @@ -674,7 +680,7 @@ def get_primary_file(files: Union[Tuple[pathlib.Path], Iterable[pathlib.Path]]) raise NoPrimaryFile(f'Cannot find primary file for group: {files}') -def _upsert_files(files: List[pathlib.Path], idempotency: datetime.datetime, +def _upsert_files(files: List[pathlib.Path], progress_callback: Callable[[int, int], None] = None): """Insert/update all records of the provided files. @@ -683,7 +689,6 @@ def _upsert_files(files: List[pathlib.Path], idempotency: datetime.datetime, Args: files: List of file paths to upsert - idempotency: Timestamp used to track which files were processed in this refresh progress_callback: Optional callback(processed, total) called after each chunk is processed It is assumed all files exist.""" @@ -709,10 +714,11 @@ def _upsert_files(files: List[pathlib.Path], idempotency: datetime.datetime, # 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.dumps(_paths_to_files_dict(group)) - suffix = (split_path_stem_and_suffix(primary_path)[1] or '').lower() or None - values[primary_path] = (modification_datetime, mimetype, size, files, suffix) - non_primary_files = {i for i in group if i != primary_path} + 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: @@ -721,9 +727,10 @@ def _upsert_files(files: List[pathlib.Path], idempotency: datetime.datetime, mimetype = get_mimetype(primary_path) modification_datetime = from_timestamp(primary_path.stat().st_mtime) size = primary_path.stat().st_size - files = json.dumps(_paths_to_files_dict([primary_path])) - suffix = (split_path_stem_and_suffix(primary_path)[1] or '').lower() or None - values[primary_path] = (modification_datetime, mimetype, size, files, suffix) + 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) 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) @@ -732,7 +739,7 @@ def _upsert_files(files: List[pathlib.Path], idempotency: datetime.datetime, values = [(str(primary_path), str(primary_path.parent), # directory False, # `indexed` is false to force indexing by default. - idempotency, *i) for primary_path, i in values.items()] + *i) for primary_path, i in values.items()] with get_db_curs(commit=True) as curs: # SQLite's RETURNING only exposes the post-upsert row; find the previously-indexed # paths first so re-indexing invalidations can be counted. @@ -744,18 +751,18 @@ def _upsert_files(files: List[pathlib.Path], idempotency: datetime.datetime, placeholders, params = values_clause(values) stmt = f''' INSERT INTO file_group - (primary_path, directory, indexed, idempotency, modification_datetime, mimetype, size, files, - suffix) + (primary_path, directory, indexed, modification_datetime, mimetype, size, files, + suffix, stem) VALUES {placeholders} ON CONFLICT (primary_path) DO UPDATE SET directory=EXCLUDED.directory, - idempotency=EXCLUDED.idempotency, modification_datetime=EXCLUDED.modification_datetime, mimetype=EXCLUDED.mimetype, size=EXCLUDED.size, - -- Always keep suffix current so suffix search works even for already-indexed files. + -- Always keep suffix/stem current so grouping and suffix search stay correct. suffix=EXCLUDED.suffix, + stem=EXCLUDED.stem, files=( -- Do not overwrite files unless the files have changed and need to be re-indexed. CASE @@ -767,7 +774,7 @@ def _upsert_files(files: List[pathlib.Path], idempotency: datetime.datetime, ELSE EXCLUDED.files END ), - -- Preseve TRUE `indexed` only if the files have not changed. + -- Preserve TRUE `indexed` only if the files have not changed. indexed=( file_group.indexed = true AND file_group.modification_datetime = EXCLUDED.modification_datetime @@ -791,6 +798,7 @@ def _upsert_files(files: List[pathlib.Path], idempotency: datetime.datetime, # 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 processed_count += len(chunk) @@ -810,105 +818,88 @@ def remove_files_in_ignored_directories(files: List[pathlib.Path]) -> List[pathl return files -async def refresh_discover_paths(paths: List[pathlib.Path], idempotency: datetime.datetime = None): - """Discover all files in the directories provided in paths, as well as all files in paths. +def _ensure_kv_table(curs): + curs.execute('CREATE TABLE IF NOT EXISTS wrolpi_kv (key TEXT PRIMARY KEY, value TEXT NOT NULL)') - All records for files in `paths` that do not exist will be deleted. - Will refuse to refresh when the media directory is empty.""" - try: - next(get_media_directory().iterdir()) - except StopIteration: - # We don't want to delete a bunch of files which would exist if the drive was mounted. - raise UnknownDirectory(f'Refusing to refresh because media directory is empty or does not exist.') - except FileNotFoundError: - raise UnknownDirectory(f'Refusing to refresh because media directory is empty or does not exist.') +def get_stem_algorithm_version() -> int: + """The grouping-algorithm version last applied to file_group.stem, or 0 if never stored.""" + with get_db_curs() as curs: + _ensure_kv_table(curs) + curs.execute('SELECT value FROM wrolpi_kv WHERE key = ?', (STEM_ALGORITHM_VERSION_KEY,)) + row = curs.fetchone() + return int(row['value']) if row else 0 - if not paths: - raise ValueError('Must provide some paths to refresh.') - - idempotency = idempotency or now() - - exists, deleted = partition(lambda i: i.exists(), paths) - - # DISCOVER all files, upsert their records. - - if exists: - # Recursively upsert all files that exist, and files in the existing directories. - directories, files = partition(lambda i: i.is_dir(), exists) - while directories: - directory = directories.pop(0) - try: - new_files, new_directories = get_files_and_directories(directory) - except FileNotFoundError as e: - # Directory may have been deleted during refresh. - logger.warning(f'Cannot refresh directory because it is missing: {directory}', exc_info=e) - continue - except PermissionError as e: - # Directory may have been deleted during refresh. - logger.warning(f'Do not have permission to refresh directory: {directory}', exc_info=e) - continue - directories.extend(new_directories) - files.extend(new_files) - - # Remove any files in ignored directories. - files = remove_files_in_ignored_directories(files) - - refresh_logger.debug(f'Discovered {len(new_files)} files in {directory}') - if len(files) >= 100: - # Wait until there are enough files to perform the upsert. - try: - _upsert_files(files, idempotency) - except Exception as e: - refresh_logger.error(f'Failed to upsert files', exc_info=e) - files = list() - # Sleep to catch cancel. - await asyncio.sleep(0) - if files: - # Not enough files for the chunks above, finish what is left. - try: - _upsert_files(files, idempotency) - except Exception as e: - refresh_logger.error(f'Failed to upsert files', exc_info=e) - - refresh_logger.info('Finished discovering files. Will now remove deleted files from DB...') +def set_stem_algorithm_version(version: int): with get_db_curs(commit=True) as curs: - params = [] - deleted_files = '' - if deleted: - # Delete any paths that do not exist. - deleted_files = ' OR '.join(['primary_path = ?'] * len(deleted)) - params.extend(str(i) for i in deleted) + _ensure_kv_table(curs) + curs.execute( + 'INSERT INTO wrolpi_kv (key, value) VALUES (?, ?) ' + 'ON CONFLICT(key) DO UPDATE SET value = excluded.value', + (STEM_ALGORITHM_VERSION_KEY, str(version)), + ) - wheres = '' - if paths: - # Use indexed directory column to delete any children of directories that are deleted. - # Match exact directory or subdirectories - conditions = [] - where_params = [] - for p in paths: - p_str = str(p) - conditions.append('(directory = ? OR directory LIKE ?)') - where_params.extend([p_str, f'{p_str}/%']) - wheres = ' ( (idempotency != ? OR idempotency is null) AND (' + ' OR '.join(conditions) + '))' - params.append(idempotency) - params.extend(where_params) - stmt = f''' - DELETE FROM file_group - WHERE - -- Delete all known-deleted files. - {deleted_files} - -- Delete any files in the refreshed paths that were not updated. - {" OR " + wheres if deleted_files and wheres else wheres} - ''' - refresh_logger.debug(stmt) - curs.execute(stmt, params) +def ensure_file_group_stems(): + """Recompute file_group.stem when NULL or 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. + """ + 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) + 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) + set_stem_algorithm_version(STEM_ALGORITHM_VERSION) async def apply_indexers(progress_callback: Callable[[int, int], None] = None): - """Indexes any Files that have not yet been indexed by Modelers, or by previous calls of this function. + """Index leftover FileGroups that no modeler claimed. + + File indexing state machine + --------------------------- + A FileGroup is discovered (a row exists), then modeled and/or content-indexed. + + `indexed` means "a modeler or this leftover indexer has finished with this row + for the current files." It is not "a domain model exists." `model` is the + domain claim (`video` / `archive` / `doc` / `zim`). + + 1. Discover/upsert: new or changed groups are inserted with indexed=0. + Unchanged groups keep indexed=1. + 2. Modelers run first. A modeler that exclusively owns its mimetype + (video, doc, zim) selects: + (domain_row IS NULL OR indexed != 1) AND mimetype_matches + so a FileGroup the leftover indexer already marked indexed=1 still + gets a Video/Doc/Zim row. Failures are skipped for the rest of the + run (failed_ids) and retried on the next refresh. + 3. Archive shares text/html with generic HTML. It selects only + indexed != 1. HTML that is not a SingleFile stays indexed=0 so this + function can claim it. HTML that is an Archive is indexed=1 with + model='archive'. + 4. This function claims every remaining indexed != 1 row. It writes + a/b/c/d_text via the Indexer registry and sets indexed=1 even on + extractor failure so poison files do not loop forever. + 5. Cleanup (channel claims, empty collections, tag config) runs last. + + FileGroup.do_model() is the single-file equivalent of steps 2-4, used by + uploads. Batch modelers are used by FileWorker refresh. Args: progress_callback: Optional callback(processed, total) called after each batch. @@ -1481,85 +1472,6 @@ async def remove_file_group_tag(session: Session, file_group_id: int, file_group file_group.untag(session, tag.id) -@dataclasses.dataclass -class RefreshProgress: - counted_files: int = 0 - counting: bool = False - discovery: bool = False - indexed: int = 0 - indexing: bool = False - modeled: int = 0 - modeling: bool = False - cleanup: bool = False - refreshing: bool = False - total_file_groups: int = 0 - unindexed: int = 0 - - def __json__(self) -> dict: - d = dict( - counted_files=self.counted_files, - counting=self.counting, - discovery=self.discovery, - indexed=self.indexed, - indexing=self.indexing, - modeled=self.modeled, - modeling=self.modeling, - cleanup=self.cleanup, - refreshing=self.refreshing, - total_file_groups=self.total_file_groups, - unindexed=self.unindexed, - ) - return d - - -def get_refresh_progress() -> RefreshProgress: - from wrolpi.api_utils import api_app - - idempotency = api_app.shared_ctx.refresh.get('idempotency') - if idempotency: - stmt = ''' - SELECT - -- Sum all the files in each FileGroup. - SUM(json_array_length(files)) FILTER (WHERE idempotency = :idempotency) AS "total_file_groups", - COUNT(id) FILTER (WHERE indexed IS TRUE AND idempotency = :idempotency) AS "indexed", - COUNT(id) FILTER (WHERE indexed IS FALSE AND idempotency = :idempotency) AS "unindexed", - COUNT(id) FILTER (WHERE model IS NOT NULL AND idempotency = :idempotency) AS "modeled" - FROM file_group \ - ''' - else: - # Idempotency has not yet been declared. - stmt = ''' - SELECT - -- Sum all the files in each FileGroup. - SUM(json_array_length(files)) AS "total_file_groups", - COUNT(id) FILTER (WHERE indexed IS TRUE) AS "indexed", - COUNT(id) FILTER (WHERE indexed IS FALSE) AS "unindexed", - COUNT(id) FILTER (WHERE model IS NOT NULL) AS "modeled" - FROM file_group \ - ''' - - with get_db_curs() as curs: - curs.execute(stmt, dict(idempotency=idempotency)) - results = dict(curs.fetchone()) - # TODO counts are wrong if we are not refreshing all files. - - progress = RefreshProgress( - counted_files=api_app.shared_ctx.refresh.get('counted_files', 0), - counting=flags.file_worker_counting.is_set(), - discovery=flags.file_worker_discovery.is_set(), - indexed=int(results['indexed'] or 0), - indexing=flags.file_worker_indexing.is_set(), - modeled=int(results['modeled'] or 0), - modeling=flags.file_worker_modeling.is_set(), - cleanup=flags.file_worker_cleanup.is_set(), - refreshing=flags.file_worker_busy.is_set(), - total_file_groups=int(results['total_file_groups'] or 0), - unindexed=int(results['unindexed'] or 0), - ) - - return progress - - def mimetypes_to_sql_wheres(wheres: List[str], params: dict, mimetypes: List[str]) -> Tuple[List[str], dict]: if mimetypes: local_wheres = list() diff --git a/wrolpi/files/models.py b/wrolpi/files/models.py index 81b54984..e13d710c 100644 --- a/wrolpi/files/models.py +++ b/wrolpi/files/models.py @@ -83,6 +83,8 @@ 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'), ) # SQLite requires exactly "INTEGER PRIMARY KEY" for the rowid alias (FTS5 content_rowid). id: int = Column(BigInteger().with_variant(Integer, 'sqlite'), primary_key=True) @@ -96,8 +98,14 @@ class FileGroup(ModelHelper, Base): data = Column(FancyJSON) # populated by the modeler download_datetime = Column(TZDateTime) # the date WROLPi downloaded this file. files = Column(FancyJSON) # populated during discovery - idempotency = Column(TZDateTime) # used to track which files need to be deleted during refresh - indexed = Column(Boolean, default=lambda: False) # wrolpi.files.lib.apply_indexers + # Leftover from the idempotency-stamp refresh. FileWorker no longer writes this. + # directory.idempotency is still used. Do not drop: file_group is the FTS content table. + idempotency = Column(TZDateTime) + indexed = Column(Boolean, default=lambda: False) # see indexing state machine on apply_indexers + # Cached output of split_path_stem_and_suffix(primary_path). Identity of a group is + # (directory, stem); primary_path is the preferred file inside the group. Recomputed on + # upsert and when STEM_ALGORITHM_VERSION changes (see ensure_file_group_stems). + stem = Column(String) length = Column(BigInteger) # video duration, article words, etc. mimetype = Column(String) # wrolpi.files.lib.get_mimetype model = Column(String) # "video", "archive", "doc", etc. @@ -546,7 +554,9 @@ class FileGroup(ModelHelper, Base): file_group.size = sum(i.stat().st_size for i in paths) file_group.mimetype = get_mimetype(file_group.primary_path) # Store the lowercased suffix so searches can filter by file type using the indexed column. - file_group.suffix = (split_path_stem_and_suffix(primary_path)[1] or '').lower() or None + stem, suffix = split_path_stem_and_suffix(primary_path) + file_group.suffix = (suffix or '').lower() or None + file_group.stem = stem logger.trace(f'FileGroup.from_paths: {file_group}') return file_group @@ -712,11 +722,14 @@ class FileGroup(ModelHelper, Base): self.directory = new_directory self.primary_path = new_primary_path - # Update title and a_text only if filename changed + # Update title, a_text, and stem only if filename changed if filename_changed: if self.title == old_name: self.title = new_primary_path.name self.a_text = split_file_name_words(new_primary_path.name) + stem, suffix = split_path_stem_and_suffix(new_primary_path) + self.stem = stem + self.suffix = (suffix or '').lower() or None # Note: `data` now contains relative paths that don't need updating! # The filenames stay the same, only the directory changes. @@ -824,3 +837,11 @@ class Directory(ModelHelper, Base): def __repr__(self): return f'' + + +class WrolpiKV(Base): + """Internal key/value store. Not user config. Holds schema-adjacent stamps such as + STEM_ALGORITHM_VERSION so compare can refuse to run on a stale stem cache.""" + __tablename__ = 'wrolpi_kv' + key = Column(String, primary_key=True) + value = Column(String, nullable=False) diff --git a/wrolpi/files/test/test_lib.py b/wrolpi/files/test/test_lib.py index 9792f1c8..becf7379 100644 --- a/wrolpi/files/test/test_lib.py +++ b/wrolpi/files/test/test_lib.py @@ -15,7 +15,6 @@ import wrolpi.common from modules.videos import Video from wrolpi.common import timer, get_wrolpi_config from wrolpi.conftest import await_switches -from wrolpi.dates import now from wrolpi.errors import InvalidFile, UnknownDirectory, FileGroupIsTagged, NoPrimaryFile from wrolpi.files import lib, indexers from wrolpi.files.models import FileGroup, Directory @@ -340,19 +339,18 @@ async def test__upsert_files(test_session, make_files_structure, test_directory, srt_file3 = srt_file3.rename(test_directory / 'video.en.srt') # All files are found because they are in this refresh request, or in the `dir1` directory. - idempotency = now() - lib._upsert_files([video_file, srt_file3, bar, baz], idempotency) + lib._upsert_files([video_file, srt_file3, bar, baz]) # Note: files now store relative filenames only (not absolute paths) assert_file_groups([ - {'primary_path': video_file, 'idempotency': idempotency, 'indexed': False, + {'primary_path': video_file, 'indexed': False, 'stem': 'video', 'files': [ {'path': srt_file3.name, 'size': 951, 'suffix': '.en.srt', 'mimetype': 'text/srt'}, {'path': video_file.name, 'size': 1056318, 'suffix': '.mp4', 'mimetype': 'video/mp4'}, ]}, - {'primary_path': bar, 'idempotency': idempotency, 'indexed': False, + {'primary_path': bar, 'indexed': False, 'stem': 'bar', 'files': [{'path': bar.name, 'size': 0, 'suffix': '.txt', 'mimetype': 'inode/x-empty'}] }, - {'primary_path': baz, 'idempotency': idempotency, 'indexed': False, + {'primary_path': baz, 'indexed': False, 'stem': 'baz', 'files': [{'path': baz.name, 'size': 8, 'suffix': '.txt', 'mimetype': 'text/plain'}] }, ]) @@ -364,67 +362,67 @@ async def test__upsert_files(test_session, make_files_structure, test_directory, test_session.query(FileGroup).filter_by(primary_path=video_file).one().indexed = True test_session.commit() assert_file_groups( - [{'primary_path': str(video_file), 'idempotency': idempotency, - 'indexed': True}], + [{'primary_path': str(video_file), 'indexed': True}], assert_count=False) # Only modified files need to be re-indexed. - lib._upsert_files([video_file, srt_file3, bar, baz], idempotency) + lib._upsert_files([video_file, srt_file3, bar, baz]) # Note: files now store relative filenames only (not absolute paths) assert_file_groups([ - {'primary_path': video_file, 'idempotency': idempotency, 'indexed': True, + {'primary_path': video_file, 'indexed': True, 'stem': 'video', 'files': [ {'path': srt_file3.name, 'size': 951, 'suffix': '.en.srt', 'mimetype': 'text/srt'}, {'path': video_file.name, 'size': 1056318, 'suffix': '.mp4', 'mimetype': 'video/mp4'}, ]}, - {'primary_path': bar, 'idempotency': idempotency, 'indexed': False, + {'primary_path': bar, 'indexed': False, 'files': [{'path': bar.name, 'size': 0, 'suffix': '.txt', 'mimetype': 'inode/x-empty'}], }, - {'primary_path': baz, 'idempotency': idempotency, 'indexed': False, + {'primary_path': baz, 'indexed': False, 'files': [{'path': baz.name, 'size': 7, 'suffix': '.txt', 'mimetype': 'text/plain'}], }, ]) # Deleting SRT removes it from the video. srt_file3.unlink() - lib._upsert_files([video_file, bar, baz], idempotency) + lib._upsert_files([video_file, bar, baz]) video_file_group: FileGroup = test_session.query(FileGroup).filter_by(primary_path=str(video_file)).one() assert len(video_file_group.files) == 1, 'SRT file was not removed from files' # Note: files now store relative filenames only (not absolute paths) assert_file_groups([ # Video is no longer indexed because SRT was removed. - {'primary_path': video_file, 'idempotency': idempotency, 'indexed': False, + {'primary_path': video_file, 'indexed': False, 'files': [{'path': video_file.name, 'size': 1056318, 'suffix': '.mp4', 'mimetype': 'video/mp4'}]}, - {'primary_path': bar, 'idempotency': idempotency, 'indexed': False, + {'primary_path': bar, 'indexed': False, 'files': [{'path': bar.name, 'size': 0, 'suffix': '.txt', 'mimetype': 'inode/x-empty'}], }, - {'primary_path': baz, 'idempotency': idempotency, 'indexed': False, + {'primary_path': baz, 'indexed': False, 'files': [{'path': baz.name, 'size': 7, 'suffix': '.txt', 'mimetype': 'text/plain'}], }, ]) @pytest.mark.asyncio -async def test_refresh_discover_paths(test_session, make_files_structure, test_directory, assert_files, - assert_file_groups): +async def test_refresh_paths(test_session, make_files_structure, test_directory, assert_files, + refresh_files): + """Refreshing a file, a directory, or missing files updates FileGroups via FileWorker.""" foo, bar, baz = make_files_structure(['dir1/foo.txt', 'dir1/bar.txt', 'baz.txt']) dir1 = foo.parent - # `refresh_paths` only refreshes the file requested. - await lib.refresh_discover_paths([foo, ]) + # Refreshing one file only discovers that file. + await refresh_files([foo]) assert_files([ {'path': 'dir1/foo.txt'}, ]) - # `refresh_paths` refreshes recursively. - await lib.refresh_discover_paths([foo.parent, ]) + # Refreshing a directory discovers recursively. + await refresh_files([foo.parent]) assert_files([ {'path': 'dir1/foo.txt'}, {'path': 'dir1/bar.txt'}, ]) - # `refresh_paths` finally discovers the file at the top of the media directory. - await lib.refresh_discover_paths([test_directory, ]) + # Refreshing the media directory discovers the file at the top. + await refresh_files([test_directory]) assert_files([ {'path': 'dir1/foo.txt'}, {'path': 'dir1/bar.txt'}, @@ -433,7 +431,7 @@ async def test_refresh_discover_paths(test_session, make_files_structure, test_d # Records for deleted files are deleted. Request a refresh of `dir1` so we indirectly refresh `foo`. foo.unlink() - await lib.refresh_discover_paths([dir1, ]) + await refresh_files([dir1]) assert_files([ {'path': 'dir1/bar.txt'}, {'path': 'baz.txt'}, @@ -442,36 +440,68 @@ async def test_refresh_discover_paths(test_session, make_files_structure, test_d # Records for all children of a directory are deleted. bar.unlink() dir1.rmdir() - await lib.refresh_discover_paths([dir1, ]) + await refresh_files([dir1]) assert_files([ {'path': 'baz.txt'}, ]) @pytest.mark.asyncio -async def test_refresh_discover_paths_groups(test_session, make_files_structure, test_directory, video_bytes): +async def test_refresh_groups(test_session, make_files_structure, test_directory, video_bytes, refresh_files): make_files_structure({'dir1/foo.mp4': video_bytes, 'dir1/foo.info.json': 'hello', 'baz.txt': 'hello'}) - await lib.refresh_discover_paths([test_directory, ], now()) + await refresh_files([test_directory]) # Two "foo" files, one "baz" file. assert test_session.query(FileGroup).count() == 2 baz, foo = test_session.query(FileGroup).order_by(FileGroup.primary_path).all() - # "foo" files are related to the "foo" group. - assert sorted([str(i['path'].relative_to(test_directory)) for i in foo.my_files()]) == [ - 'dir1/foo.info.json', 'dir1/foo.mp4', - ] - # "bar" file is the only file related to the "bar" group. + # "foo" files are related to the "foo" group. Modeling may add sidecars (.ffprobe.json). + foo_names = {i['path'].name for i in foo.my_files()} + assert {'foo.info.json', 'foo.mp4'} <= foo_names + assert foo.stem == 'foo' + # "baz" file is the only file related to the "baz" group. assert len(baz.my_files()) == 1 assert str(baz.my_files()[0]['path'].relative_to(test_directory)) == 'baz.txt' - assert baz.indexed is False + assert baz.indexed is True # leftover indexer claims plain text + assert baz.stem == 'baz' + + +def test_find_indexer_longest_prefix(): + """Longest registered mimetype prefix wins, even if a shorter key was inserted later.""" + # Simulate the old fallback, which did indexer_map[mimetype.split('/')[0]] and inserted + # DefaultIndexer for 'text' into the defaultdict. + indexers.indexer_map['text'] + try: + assert indexers.find_indexer('text/html') is indexers.HTMLIndexer + assert indexers.find_indexer('text/plain') is indexers.TextIndexer + assert indexers.find_indexer('text/csv') is indexers.DefaultIndexer + assert indexers.find_indexer('application/zip') is indexers.ZipIndexer + finally: + indexers.indexer_map.pop('text', None) + + +def test_ensure_file_group_stems_recomputes_on_version_change(test_session, test_directory, make_files_structure): + """A stored stem is a cache of the grouping algorithm; a version bump must backfill.""" + foo, = make_files_structure(['foo.en.srt']) + fg = FileGroup.from_paths(test_session, foo) + test_session.commit() + assert fg.stem == 'foo' + fg.stem = 'stale-stem' + test_session.commit() + + lib.set_stem_algorithm_version(0) + lib.ensure_file_group_stems() + test_session.expire(fg) + assert fg.stem == 'foo' + assert lib.get_stem_algorithm_version() == lib.STEM_ALGORITHM_VERSION @pytest.mark.asyncio -async def test_file_group_tag(test_session, make_files_structure, test_directory, tag_factory, await_switches): +async def test_file_group_tag(test_session, make_files_structure, test_directory, tag_factory, await_switches, + refresh_files): """A FileGroup can be tagged.""" make_files_structure(['foo.mp4']) - await lib.refresh_discover_paths([test_directory, ], now()) + await refresh_files() one = await tag_factory() foo: FileGroup = test_session.query(FileGroup).one() @@ -789,24 +819,6 @@ def test_get_primary_file(test_directory, video_file, srt_file3, example_epub, e assert lib.get_primary_file([foo, bar]) -@pytest.mark.asyncio -async def test_get_refresh_progress(async_client, test_session): - request, response = await async_client.get('/api/files/refresh_progress') - assert response.status_code == HTTPStatus.OK - assert 'progress' in response.json - progress = response.json['progress'] - assert 'cleanup' in progress - assert 'discovery' in progress - assert 'indexed' in progress - assert 'indexing' in progress - assert 'modeled' in progress - assert 'modeling' in progress - assert 'refreshing' in progress - assert 'counted_files' in progress - assert 'total_file_groups' in progress - assert 'unindexed' in progress - - @pytest.mark.asyncio async def test_refresh_files_no_groups(async_client, test_session, test_directory, make_files_structure, zip_file_factory, refresh_files): diff --git a/wrolpi/files/test/test_worker.py b/wrolpi/files/test/test_worker.py index 74b90f05..24443833 100644 --- a/wrolpi/files/test/test_worker.py +++ b/wrolpi/files/test/test_worker.py @@ -326,17 +326,17 @@ async def test_compare_file_groups_survives_concurrent_commit( make_files_structure(['docs/file1.txt', 'docs/file2.txt']) from wrolpi.files import worker as worker_mod - original = worker_mod._stream_filesystem_paths + original = worker_mod._stream_filesystem_entries - async def commit_after_first(root): + async def commit_after_first(roots): first = True - async for p in original(root): - yield p + async for entry in original(roots): + yield entry if first: first = False test_session.commit() - with mock.patch.object(worker_mod, '_stream_filesystem_paths', commit_after_first): + with mock.patch.object(worker_mod, '_stream_filesystem_entries', commit_after_first): result = await compare_file_groups(test_directory) assert len(result.new) == 2 @@ -351,11 +351,11 @@ async def test_compare_file_groups_cleans_up_work_table_on_aborted_tx( make_files_structure(['docs/file1.txt']) from wrolpi.files import worker as worker_mod - original = worker_mod._stream_filesystem_paths + original = worker_mod._stream_filesystem_entries - async def commit_then_abort_then_raise(root): - async for p in original(root): - yield p + async def commit_then_abort_then_raise(roots): + async for entry in original(roots): + yield entry test_session.commit() # A failed statement mid-transaction must not prevent the work-table cleanup. try: @@ -364,7 +364,7 @@ async def test_compare_file_groups_cleans_up_work_table_on_aborted_tx( pass raise RuntimeError('simulated crash mid-scan') - with mock.patch.object(worker_mod, '_stream_filesystem_paths', commit_then_abort_then_raise): + with mock.patch.object(worker_mod, '_stream_filesystem_entries', commit_then_abort_then_raise): with pytest.raises(RuntimeError): await compare_file_groups(test_directory) @@ -1626,7 +1626,6 @@ async def test_file_worker_discovery_flag_is_set(async_client, test_session, tes async def test_upsert_files_progress_callback(test_session, test_directory, make_files_structure): """_upsert_files calls progress callback during processing.""" from wrolpi.files.lib import _upsert_files - from wrolpi.dates import now # Create files files = make_files_structure([f'docs/file{i}.txt' for i in range(10)]) @@ -1636,8 +1635,7 @@ async def test_upsert_files_progress_callback(test_session, test_directory, make def on_progress(processed, total): progress_calls.append((processed, total)) - idempotency = now() - _upsert_files(files, idempotency, on_progress) + _upsert_files(files, on_progress) # Progress callback should have been called assert len(progress_calls) > 0 @@ -1750,6 +1748,58 @@ async def test_compare_file_groups_duplicate_keeps_one(test_session, test_direct assert deleted_ids & {fg1_id, fg2_id} +@pytest.mark.asyncio +async def test_compare_file_groups_multiple_roots_does_not_touch_outside( + test_session, test_directory, make_files_structure): + """Refreshing directories A and B must not delete a stale FileGroup in C.""" + a_file, b_file, c_file = make_files_structure([ + 'dirA/a.txt', + 'dirB/b.txt', + 'dirC/c.txt', + ]) + FileGroup.from_paths(test_session, a_file) + FileGroup.from_paths(test_session, b_file) + c_fg = FileGroup.from_paths(test_session, c_file) + test_session.commit() + c_id = c_fg.id + + c_file.unlink() + + result = await compare_file_groups([test_directory / 'dirA', test_directory / 'dirB']) + + assert all(d.file_group_id != c_id for d in result.deleted), \ + 'stale FileGroup outside the requested roots was reported deleted' + assert test_session.query(FileGroup).filter_by(id=c_id).one() + + full = await compare_file_groups(test_directory) + assert c_id in {d.file_group_id for d in full.deleted} + + +@pytest.mark.asyncio +async def test_refresh_two_directories_does_not_delete_third( + async_client, test_session, test_directory, make_files_structure): + """FileWorker refresh of A+B must not delete a stale row in C (the old root=None bug).""" + from wrolpi.conftest import await_file_worker + + a_file, b_file, c_file = make_files_structure([ + 'dirA/a.txt', + 'dirB/b.txt', + 'dirC/c.txt', + ]) + FileGroup.from_paths(test_session, a_file) + FileGroup.from_paths(test_session, b_file) + c_fg = FileGroup.from_paths(test_session, c_file) + test_session.commit() + c_id = c_fg.id + c_file.unlink() + + file_worker.queue_refresh([test_directory / 'dirA', test_directory / 'dirB']) + await await_file_worker() + + assert test_session.query(FileGroup).filter_by(id=c_id).one_or_none() is not None, \ + 'refreshing A and B deleted a stale FileGroup in C' + + @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 e662f86a..09009f2b 100644 --- a/wrolpi/files/worker.py +++ b/wrolpi/files/worker.py @@ -26,17 +26,17 @@ 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, get_wrolpi_config, logger, walk, chunks, unique_by_predicate -from wrolpi.dates import now +from wrolpi.common import get_media_directory, logger, walk, chunks, unique_by_predicate from wrolpi.db import get_db_session, get_db_curs from wrolpi.errors import NoPrimaryFile from wrolpi.events import Events -from wrolpi.vars import PYTEST +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, _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, ) logger = logger.getChild(__name__) @@ -55,6 +55,7 @@ __all__ = [ 'FileGroupDiff', 'FileComparisonResult', 'FileWorkerJobFailed', + 'classify_file_group_diffs', 'compare_file_groups', 'count_files', 'file_worker', @@ -141,6 +142,113 @@ def _deduplicate_db_groups( return deduplicated, duplicate_diffs +def _filenames_from_files_json(raw) -> Set[str]: + """Filenames from a FileGroup.files JSON value (relative or absolute paths).""" + if not raw: + return set() + if isinstance(raw, str): + raw = json.loads(raw) + if not isinstance(raw, list): + return set() + names = set() + for entry in raw: + path = entry.get('path') if isinstance(entry, dict) else entry + if path: + names.add(Path(path).name) + return names + + +def classify_file_group_diffs( + fs_groups: Dict[Tuple[str, str], Set[str]], + db_groups: Dict[Tuple[str, str], List[Tuple[int, Set[str], float]]], + fs_mtimes: Dict[Tuple[str, str], float] | None = None, +) -> FileComparisonResult: + """Classify filesystem vs DB groups into new/deleted/modified/unchanged. + + `db_groups` may contain duplicates per (directory, stem); extras are marked deleted. + If `fs_mtimes` is provided (max mtime per key), unchanged-by-name groups with a newer + filesystem mtime are classified modified without a second stat(). If omitted, those + groups are stat()'d (used by the small direct-file path). + """ + deduplicated, duplicate_diffs = _deduplicate_db_groups(db_groups) + + unchanged = [] + new = [] + deleted = list(duplicate_diffs) + modified = [] + + all_keys = set(fs_groups.keys()) | set(deduplicated.keys()) + + for key in all_keys: + directory, stem = key + fs_files = fs_groups.get(key, set()) + db_id, db_files, db_mtime = deduplicated.get(key, (None, set(), 0)) + + diff = FileGroupDiff( + directory=Path(directory), + stem=stem, + db_files=db_files, + fs_files=fs_files, + file_group_id=db_id, + ) + + if diff.is_new: + new.append(diff) + elif diff.is_deleted: + deleted.append(diff) + elif diff.needs_update: + modified.append(diff) + elif diff.is_unchanged: + if fs_mtimes is not None: + fs_mtime = fs_mtimes.get(key, 0) + if fs_mtime > float(db_mtime) + MTIME_TOLERANCE_SECONDS: + modified.append(diff) + else: + unchanged.append(diff) + else: + try: + dir_path = Path(directory) + fs_mtime = max( + (dir_path / filename).stat().st_mtime + for filename in fs_files + ) + if fs_mtime > float(db_mtime) + MTIME_TOLERANCE_SECONDS: + modified.append(diff) + else: + unchanged.append(diff) + except (OSError, ValueError): + unchanged.append(diff) + elif db_id is not None and not fs_files and not db_files: + deleted.append(diff) + + return FileComparisonResult( + unchanged=unchanged, + new=new, + deleted=deleted, + modified=modified, + ) + + +def _normalize_roots(roots: Path | list[Path] | None) -> list[Path]: + if roots is None: + return [get_media_directory()] + if isinstance(roots, (str, Path)): + return [Path(roots)] + return [Path(r) for r in roots] + + +def _directory_in_roots_sql(roots: list[Path], column: str = 'fg.directory') -> Tuple[str, dict]: + """SQL fragment: `column` is any of the roots or a subdirectory of one.""" + clauses = [] + params = {} + for i, root in enumerate(roots): + root_str = str(root) + clauses.append(f'({column} = :root_{i} OR {column} LIKE :root_pat_{i})') + params[f'root_{i}'] = root_str + params[f'root_pat_{i}'] = f'{root_str}/%' + return '(' + ' OR '.join(clauses) + ')', params + + def _diff_to_paths(diff: FileGroupDiff) -> list[pathlib.Path]: """Convert a FileGroupDiff to full file paths.""" directory = pathlib.Path(diff.directory) @@ -260,18 +368,56 @@ async def find_directories(directories: list[Path]) -> set[Path]: await proc.wait() -async def _stream_filesystem_paths(root: Path) -> AsyncGenerator[str, None]: - """Stream file paths using find command. Memory efficient and cancellable. - - Excludes hidden files and ignored directories from config. - """ - # Build find arguments with ignored directory exclusions - find_args = [str(root), '-type', 'f', '-not', '-path', '*/.*'] +def _find_args_for_roots(roots: list[Path]) -> list[str]: + """Common `find` arguments: files under roots, no hidden paths, no ignored dirs, no DB file.""" + find_args = [str(r) for r in roots] + find_args.extend(['-type', 'f', '-not', '-path', '*/.*']) for ignored in _get_normalized_ignored_directories(): find_args.extend(['-not', '-path', f'{ignored}/*']) - # Never index the database (or its WAL sidecars), even if `config` is un-ignored. from wrolpi.db import get_db_file find_args.extend(['-not', '-path', f'{get_db_file()}*']) + return find_args + + +async def _stream_filesystem_paths(root: Path | list[Path]) -> AsyncGenerator[str, None]: + """Stream file paths using find. Kept for tests that patch this generator.""" + async for directory, filename, stem, mtime in _stream_filesystem_entries(root): + yield str(Path(directory) / filename) + + +async def _stream_filesystem_entries( + roots: Path | list[Path], +) -> AsyncGenerator[Tuple[str, str, str, float], None]: + """Yield (directory, filename, stem, mtime) for files under roots. + + GNU find `-printf` records mtime in the same walk (Debian/RPi). macOS find has no + `-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 + 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() + return proc = await asyncio.create_subprocess_exec( 'find', *find_args, @@ -280,11 +426,17 @@ async def _stream_filesystem_paths(root: Path) -> AsyncGenerator[str, None]: try: async for line in proc.stdout: path = line.decode().rstrip('\n') - if path: - yield path + if not path: + continue + p = Path(path) + try: + mtime = p.stat().st_mtime + except OSError: + continue + stem, _ = split_path_stem_and_suffix(p) + yield str(p.parent), p.name, stem, mtime await proc.wait() finally: - # Kill subprocess if still running (handles both normal exit and cancellation) if proc.returncode is None: proc.kill() await proc.wait() @@ -310,36 +462,56 @@ def _insert_fs_batch(conn, table_name: str, batch: list): return conn.execute( text(f""" - INSERT INTO {table_name} (directory, filename, stem) - VALUES (:directory, :filename, :stem) + INSERT INTO {table_name} (directory, filename, stem, mtime) + VALUES (:directory, :filename, :stem, :mtime) ON CONFLICT DO NOTHING """), - [{"directory": b[0], "filename": b[1], "stem": b[2]} for b in batch], + [{"directory": b[0], "filename": b[1], "stem": b[2], "mtime": b[3]} for b in batch], + ) + + +def _diff_from_sql_row(row, fs_files: Set[str], db_files: Set[str]) -> FileGroupDiff: + return FileGroupDiff( + directory=Path(row.directory), + stem=row.stem, + db_files=db_files, + fs_files=fs_files, + file_group_id=row.id if getattr(row, 'id', None) else None, ) async def compare_file_groups( - root: Path = None, + roots: Path | list[Path] | None = None, batch_size: int = 10000, progress_callback: Callable[[int], None] = None, + root: Path | None = None, ) -> FileComparisonResult: """ Compare FileGroups in DB to filesystem. Memory efficient for Raspberry Pi. - Uses a uniquely-named SQLite work table to handle millions of files without - loading everything into Python memory. + `roots` may be a single path or a list. Only FileGroups under those directories + are considered; a two-directory refresh must not delete a stale row in a third. - Returns which FileGroups are new, deleted, modified, or unchanged. + Uses a uniquely-named SQLite TEMP table on a dedicated connection so the set-diff + runs in SQL (the join must use this connection; TEMP tables are per-connection). + Unchanged groups are never stat()'d a second time: mtime is recorded during the + find walk. This function is async and cancellable - when cancelled, it will terminate the subprocess scanning the filesystem. """ - root = root or get_media_directory() + if root is not None and roots is None: + roots = root + root_list = _normalize_roots(roots) + 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, # it is invisible to other connections, and it is dropped automatically when the connection # closes — orphaned work tables are impossible. table_name = f'fs_files_{uuid.uuid4().hex[:12]}' + groups_table = f'{table_name}_groups' + roots_sql, root_params = _directory_in_roots_sql(root_list) with get_db_session() as session: work_conn = session.get_bind().connect() @@ -350,21 +522,17 @@ async def compare_file_groups( directory TEXT NOT NULL, filename TEXT NOT NULL, stem TEXT NOT NULL, + mtime REAL NOT NULL DEFAULT 0, PRIMARY KEY (directory, filename) ) """)) - # Stream filesystem files into work table with stems batch = [] file_count = 0 - path_gen = _stream_filesystem_paths(root) + path_gen = _stream_filesystem_entries(root_list) try: - async for path in path_gen: - p = Path(path) - directory = str(p.parent) - filename = p.name - stem, _ = split_path_stem_and_suffix(p) - batch.append((directory, filename, stem)) + async for directory, filename, stem, mtime in path_gen: + batch.append((directory, filename, stem, mtime)) if len(batch) >= batch_size: _insert_fs_batch(work_conn, table_name, batch) @@ -382,117 +550,111 @@ async def compare_file_groups( logger.info(f'Scanned {file_count} files from filesystem') - # Index for fast grouping work_conn.execute( text(f"CREATE INDEX IF NOT EXISTS {table_name}_dir_stem_idx ON {table_name}(directory, stem)")) + work_conn.execute(text(f""" + CREATE TEMP TABLE {groups_table} AS + SELECT directory, stem, + json_group_array(filename) AS files, + COUNT(*) AS file_count, + MAX(mtime) AS mtime + FROM {table_name} + GROUP BY directory, stem + """)) + work_conn.execute(text(f"CREATE INDEX IF NOT EXISTS {groups_table}_idx ON {groups_table}(directory, stem)")) work_conn.execute(text(f"ANALYZE {table_name}")) - # Get filesystem files grouped by (directory, stem) - fs_groups: Dict[Tuple[str, str], Set[str]] = {} - result = work_conn.execute(text(f""" - SELECT directory, stem, json_group_array(filename) as files - FROM {table_name} - GROUP BY directory, stem - """)) - for row in result: - fs_groups[(row.directory, row.stem)] = set(json.loads(row.files)) + # Canonical DB row per (directory, stem). Duplicates (rn > 1) are deleted. + # The join runs on work_conn so it can see the TEMP tables. + db_canonical = f'{table_name}_db' + work_conn.execute(text(f""" + CREATE TEMP TABLE {db_canonical} AS + SELECT id, directory, stem, files, modification_datetime, file_count, rn + FROM ( + SELECT fg.id, + fg.directory, + COALESCE(fg.stem, '') AS stem, + fg.files, + fg.modification_datetime, + CASE + WHEN fg.files IS NULL OR json_type(fg.files) != 'array' + THEN 0 + ELSE json_array_length(fg.files) + END AS file_count, + ROW_NUMBER() OVER ( + PARTITION BY fg.directory, COALESCE(fg.stem, '') + ORDER BY ( + CASE + WHEN fg.files IS NULL OR json_type(fg.files) != 'array' + THEN 0 + ELSE json_array_length(fg.files) + END > 0 + ) DESC, fg.id DESC + ) AS rn + FROM file_group fg + WHERE fg.directory IS NOT NULL + AND {roots_sql} + ) + """), root_params) - logger.info(f'Found {len(fs_groups)} file groups on filesystem') - - # Get all DB FileGroups within root directory (both with files and empty) - # Collect ALL FileGroups per (directory, stem) key — duplicates are deduped after. - db_groups: Dict[Tuple[str, str], List[Tuple[int, Set[str], float]]] = defaultdict(list) - root_str = str(root) - empty_count = 0 - - result = session.execute(text(""" - SELECT fg.id, - fg.directory, - fg.primary_path, - (fg.files IS NULL - OR (json_type(fg.files) = 'array' - AND json_array_length(fg.files) = 0)) as is_empty, - fg.modification_datetime as modification_datetime, - fg.files as files - FROM file_group fg - WHERE fg.directory IS NOT NULL - AND (fg.directory = :root OR fg.directory LIKE :root_pattern) - """), {"root": root_str, "root_pattern": f"{root_str}/%"}) - - for row in result: - # Raw TEXT datetime (UTC, microsecond precision); unixepoch() would truncate to - # whole seconds which breaks the sub-second MTIME_TOLERANCE_SECONDS comparison. - mtime_epoch = _datetime_text_to_epoch(row.modification_datetime) - if row.is_empty: - empty_count += 1 - stem, _ = split_path_stem_and_suffix(Path(row.primary_path)) - key = (row.directory, stem) - db_groups[key].append((row.id, set(), mtime_epoch)) - else: - raw_files = row.files - if isinstance(raw_files, str): - raw_files = json.loads(raw_files) - if not isinstance(raw_files, list) or not raw_files: - # Matches the old Postgres behavior: non-array `files` was skipped. - continue - files = [f['path'] for f in raw_files] - first_file = files[0] - stem, _ = split_path_stem_and_suffix(Path(row.directory) / first_file) - key = (row.directory, stem) - db_groups[key].append((row.id, set(files), mtime_epoch)) - - logger.info(f'Found {len(db_groups)} file groups in database') - if empty_count: - logger.debug(f'Found {empty_count} FileGroups with empty files arrays') - - # Deduplicate: keep one FileGroup per key, mark the rest for deletion - deduplicated, duplicate_diffs = _deduplicate_db_groups(db_groups) - - unchanged = [] new = [] - deleted = list(duplicate_diffs) + deleted = [] modified = [] + unchanged = [] - all_keys = set(fs_groups.keys()) | set(deduplicated.keys()) + # Duplicates: extra FileGroups for the same (directory, stem). + for row in work_conn.execute(text(f""" + SELECT id, directory, stem, files FROM {db_canonical} WHERE rn > 1 + """)): + deleted.append(_diff_from_sql_row(row, set(), _filenames_from_files_json(row.files))) - for key in all_keys: - directory, stem = key - fs_files = fs_groups.get(key, set()) - db_id, db_files, db_mtime = deduplicated.get(key, (None, set(), 0)) + # Deleted: canonical DB group with no filesystem group. + for row in work_conn.execute(text(f""" + SELECT c.id, c.directory, c.stem, c.files + FROM {db_canonical} c + LEFT JOIN {groups_table} fs ON fs.directory = c.directory AND fs.stem = c.stem + WHERE c.rn = 1 AND fs.stem IS NULL + """)): + deleted.append(_diff_from_sql_row(row, set(), _filenames_from_files_json(row.files))) + # New: filesystem group with no DB group. + for row in work_conn.execute(text(f""" + SELECT fs.directory, fs.stem, fs.files, NULL AS id + FROM {groups_table} fs + LEFT JOIN {db_canonical} c ON c.directory = fs.directory AND c.stem = fs.stem AND c.rn = 1 + WHERE c.id IS NULL + """)): + fs_files = set(json.loads(row.files) if isinstance(row.files, str) else (row.files or [])) + new.append(_diff_from_sql_row(row, fs_files, set())) + + # Both sides present: compare names and mtime. Only pull file lists for these rows; + # unchanged file contents stay in SQL. + for row in work_conn.execute(text(f""" + SELECT c.id, c.directory, c.stem, c.files AS db_files, c.modification_datetime, + fs.files AS fs_files, fs.mtime AS fs_mtime, fs.file_count AS fs_count, + c.file_count AS db_count + FROM {db_canonical} c + JOIN {groups_table} fs ON fs.directory = c.directory AND fs.stem = c.stem + WHERE c.rn = 1 + """)): + fs_files = set(json.loads(row.fs_files) if isinstance(row.fs_files, str) else (row.fs_files or [])) + db_files = _filenames_from_files_json(row.db_files) diff = FileGroupDiff( - directory=Path(directory), - stem=stem, + directory=Path(row.directory), + stem=row.stem, db_files=db_files, fs_files=fs_files, - file_group_id=db_id, + file_group_id=row.id, ) - - if diff.is_new: - new.append(diff) - elif diff.is_deleted: - deleted.append(diff) - elif diff.needs_update: + db_mtime = _datetime_text_to_epoch(row.modification_datetime) + if diff.needs_update: modified.append(diff) elif diff.is_unchanged: - # Files match by name - check if content changed via mtime - # Get max mtime from filesystem files - try: - dir_path = Path(directory) - fs_mtime = max( - (dir_path / filename).stat().st_mtime - for filename in fs_files - ) - # If filesystem mtime is newer than DB mtime, content changed - if fs_mtime > float(db_mtime) + MTIME_TOLERANCE_SECONDS: - modified.append(diff) - else: - unchanged.append(diff) - except (OSError, ValueError): - # If we can't stat files, treat as unchanged + if float(row.fs_mtime or 0) > db_mtime + MTIME_TOLERANCE_SECONDS: + modified.append(diff) + else: unchanged.append(diff) - elif db_id is not None and not fs_files and not db_files: - # Empty FileGroup with no files on disk — orphan, delete it + elif not fs_files and not db_files: deleted.append(diff) logger.info( @@ -1302,8 +1464,7 @@ class FileWorker: percent = int((count / task.count) * 100) if task.count > 0 else 0 self.update_status(operation_processed=count, operation_percent=percent) - root = dir_paths[0] if len(dir_paths) == 1 else None - dir_result = await compare_file_groups(root=root, progress_callback=on_compare_progress) + dir_result = await compare_file_groups(roots=dir_paths, progress_callback=on_compare_progress) if is_global_refresh: Events.send_global_refresh_discovery_completed() @@ -1462,8 +1623,7 @@ class FileWorker: self.update_status(operation_processed=processed, operation_percent=percent) # Use existing _upsert_files function which handles grouping and primary file detection - idempotency = now() - await asyncio.to_thread(_upsert_files, all_paths, idempotency, on_upsert_progress) + await asyncio.to_thread(_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]): @@ -1881,20 +2041,12 @@ class FileWorker: file_groups = query.all() for fg in file_groups: - if not fg.files: - stem, _ = split_path_stem_and_suffix(fg.primary_path) - key = (fg.directory, stem) - if key in fs_groups: - mtime = fg.modification_datetime.timestamp() if fg.modification_datetime else 0 - db_groups[key].append((fg.id, set(), mtime)) - else: - first_file = fg.files[0]['path'] - stem, _ = split_path_stem_and_suffix(pathlib.Path(fg.directory) / first_file) - key = (fg.directory, stem) - if key in fs_groups: - fg_filenames = {pathlib.Path(f['path']).name for f in fg.files} - mtime = fg.modification_datetime.timestamp() if fg.modification_datetime else 0 - db_groups[key].append((fg.id, fg_filenames, mtime)) + stem = fg.stem or split_path_stem_and_suffix(fg.primary_path)[0] + key = (str(fg.directory), stem) + if key not in fs_groups: + continue + mtime = fg.modification_datetime.timestamp() if fg.modification_datetime else 0 + db_groups[key].append((fg.id, _filenames_from_files_json(fg.files), mtime)) # Track FG IDs already loaded by the directory query to avoid double-appending seen_fg_ids: set[int] = set() @@ -1912,11 +2064,10 @@ class FileWorker: ).all() for fg in deleted_fgs: seen_fg_ids.add(fg.id) - stem, _ = split_path_stem_and_suffix(fg.primary_path) - key = (fg.directory, stem) - fg_filenames = {pathlib.Path(f['path']).name for f in fg.files} if fg.files else set() + stem = fg.stem or split_path_stem_and_suffix(fg.primary_path)[0] + key = (str(fg.directory), stem) mtime = fg.modification_datetime.timestamp() if fg.modification_datetime else 0 - db_groups[key].append((fg.id, fg_filenames, mtime)) + db_groups[key].append((fg.id, _filenames_from_files_json(fg.files), mtime)) if key not in fs_groups: fs_groups[key] = set() @@ -1932,72 +2083,19 @@ class FileWorker: ).all() for fg in dir_deleted_fgs: seen_fg_ids.add(fg.id) - stem, _ = split_path_stem_and_suffix(fg.primary_path) - key = (fg.directory, stem) - fg_filenames = {pathlib.Path(f['path']).name for f in fg.files} if fg.files else set() + stem = fg.stem or split_path_stem_and_suffix(fg.primary_path)[0] + key = (str(fg.directory), stem) mtime = fg.modification_datetime.timestamp() if fg.modification_datetime else 0 - db_groups[key].append((fg.id, fg_filenames, mtime)) + db_groups[key].append((fg.id, _filenames_from_files_json(fg.files), mtime)) if key not in fs_groups: fs_groups[key] = set() - # Deduplicate: keep one FileGroup per key, mark the rest for deletion - deduplicated, duplicate_diffs = _deduplicate_db_groups(db_groups) - - unchanged = [] - new = [] - deleted = list(duplicate_diffs) - modified = [] - - all_keys = set(fs_groups.keys()) | set(deduplicated.keys()) - - for key in all_keys: - directory, stem = key - fs_files = fs_groups.get(key, set()) - db_id, db_files, db_mtime = deduplicated.get(key, (None, set(), 0)) - - diff = FileGroupDiff( - directory=Path(directory), - stem=stem, - db_files=db_files, - fs_files=fs_files, - file_group_id=db_id, - ) - - if diff.is_new: - new.append(diff) - elif diff.is_deleted: - deleted.append(diff) - elif diff.needs_update: - modified.append(diff) - elif diff.is_unchanged: - # Check mtime for content changes - try: - dir_path = Path(directory) - fs_mtime = max( - (dir_path / filename).stat().st_mtime - for filename in fs_files - ) - if fs_mtime > db_mtime + MTIME_TOLERANCE_SECONDS: - modified.append(diff) - else: - unchanged.append(diff) - except (OSError, ValueError): - unchanged.append(diff) - elif db_id is not None and not fs_files and not db_files: - # Empty FileGroup with no files on disk — orphan, delete it - deleted.append(diff) - + result = classify_file_group_diffs(fs_groups, db_groups) logger.info( - f'Direct file refresh: {len(unchanged)} unchanged, {len(new)} new, ' - f'{len(deleted)} deleted, {len(modified)} modified' - ) - - return FileComparisonResult( - unchanged=unchanged, - new=new, - deleted=deleted, - modified=modified, + f'Direct file refresh: {len(result.unchanged)} unchanged, {len(result.new)} new, ' + f'{len(result.deleted)} deleted, {len(result.modified)} modified' ) + return result async def _apply_post_processing(self, is_global_refresh: bool = False) -> None: """Run modelers, indexers, and cleanup after file operations.