rt/wrolpi
1
0
Fork 0
mirror of https://github.com/lrnselfreliance/wrolpi.git synced 2026-10-25 18:40:06 -04:00
Code Issues Projects Releases Packages Wiki Activity Actions 5

files: SQL FileGroup compare, stem column, drop unused discovery path

Browse source 
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.
This commit is contained in:
Roland 2026-09-16 20:20:42 -06:00
parent b37ad92563
commit bd15b85c86
No known key found for this signature in database
GPG key ID: 1BA620CE6C9A4680
18 changed files with 634 additions and 507 deletions
Download patch file Download diff file Expand all files Collapse all files

37
alembic/versions/2026_09_16_1200_file_group_stem.py Normal file
View file

@ -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')

5
modules/archive/__init__.py
View file

@ -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)))) \

2
modules/docs/__init__.py
View file

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

7
modules/docs/test/test_doc_modeler.py
View file

@ -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()

3
modules/videos/__init__.py
View file

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

2
modules/videos/schema.py
View file

@ -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

19
modules/zim/lib.py
View file

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

1
scripts/api_perf.py
View file

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

8
wrolpi/common.py
View file

@ -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

7
wrolpi/conftest.py
View file

@ -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 _

2
wrolpi/contexts.py
View file

@ -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

11
wrolpi/files/api.py
View file

@ -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'

16
wrolpi/files/indexers.py
View file

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

286
wrolpi/files/lib.py
View file

@ -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()

29
wrolpi/files/models.py
View file

@ -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'<Directory path={repr(str(self.path))}>'
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)

120
wrolpi/files/test/test_lib.py
View file

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

76
wrolpi/files/test/test_worker.py
View file

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

510
wrolpi/files/worker.py
View file

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