mirror of
https://github.com/lrnselfreliance/wrolpi.git
synced 2026-10-26 10:51:36 -04:00
Progress: write as a writer, confine and merge the import, keep FTS on downgrade
- The progress endpoints begin as writers (request.ctx.db.write()), so a heartbeat during a refresh waits instead of failing a lock upgrade. The config switch fires after the commit. - `..` is collapsed before the ignored-directory check. Clear only resets an existing FileGroup; it never indexes a file. - Import confines config paths to the media directory. A newer plain view keeps its `viewed` but takes the config's position; only progress written after the config wins. Files are globbed before the write transaction. - The dump skips a row outside the media directory instead of failing. - The downgrade drops the columns with ALTER TABLE, which keeps file_group's FTS triggers. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
parent
c8201700a8
commit
e21c6d9f0a
6 changed files with 232 additions and 47 deletions
|
|
@ -19,6 +19,6 @@ def upgrade():
|
|||
|
||||
|
||||
def downgrade():
|
||||
with op.batch_alter_table('file_group') as batch_op:
|
||||
batch_op.drop_column('position')
|
||||
batch_op.drop_column('progress')
|
||||
# Not batch_alter_table: rebuilding file_group drops its FTS triggers (see 2026_09_20_1200).
|
||||
op.execute('ALTER TABLE file_group DROP COLUMN position')
|
||||
op.execute('ALTER TABLE file_group DROP COLUMN progress')
|
||||
|
|
|
|||
|
|
@ -14,6 +14,7 @@ from wrolpi.common import get_media_directory, wrol_mode_check, get_relative_to_
|
|||
background_task, walk, timer, TRACE_LEVEL, unique_by_predicate
|
||||
from wrolpi.errors import InvalidFile, UnknownDirectory, FileUploadFailed, FileConflict
|
||||
from wrolpi.events import Events
|
||||
from wrolpi.files.progress import save_recently_viewed_config
|
||||
from . import lib, schema
|
||||
from .worker import file_worker
|
||||
from ..api_utils import json_response, api_app
|
||||
|
|
@ -65,7 +66,12 @@ async def get_file(request: Request, body: schema.FileRequest):
|
|||
@validate(schema.FileProgressRequest)
|
||||
async def post_file_progress(request: Request, body: schema.FileProgressRequest):
|
||||
path = get_media_directory() / body.file
|
||||
progress, position = lib.set_file_progress(request.ctx.session, path, body.progress, body.position, body.final)
|
||||
# Begin as a writer; a playback heartbeat often overlaps a refresh or download.
|
||||
with request.ctx.db.write() as session:
|
||||
progress, position = lib.set_file_progress(session, path, body.progress, body.position)
|
||||
if body.final and progress is not None:
|
||||
# The User stopped (paused, closed). Heartbeats during playback only write the DB.
|
||||
save_recently_viewed_config.activate_switch()
|
||||
return json_response({'progress': progress, 'position': position})
|
||||
|
||||
|
||||
|
|
@ -76,7 +82,10 @@ async def post_file_progress(request: Request, body: schema.FileProgressRequest)
|
|||
)
|
||||
@validate(schema.FileProgressClearRequest)
|
||||
async def post_file_progress_clear(request: Request, body: schema.FileProgressClearRequest):
|
||||
lib.clear_file_progress(request.ctx.session, get_media_directory() / body.file)
|
||||
with request.ctx.db.write() as session:
|
||||
changed = lib.clear_file_progress(session, get_media_directory() / body.file)
|
||||
if changed:
|
||||
save_recently_viewed_config.activate_switch()
|
||||
return response.empty(HTTPStatus.NO_CONTENT)
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -36,7 +36,7 @@ from wrolpi.errors import InvalidFile, UnknownDirectory, UnknownFile, UnknownTag
|
|||
NoPrimaryFile, InvalidDirectory, IgnoredDirectoryError, UnsupportedArchive, InvalidArchiveMember
|
||||
from wrolpi.events import Events
|
||||
from wrolpi.files.models import FileGroup, Directory
|
||||
from wrolpi.files.progress import normalize_progress, save_recently_viewed_config
|
||||
from wrolpi.files.progress import normalize_progress
|
||||
from wrolpi.lang import ISO_639_CODES, ISO_3166_CODES
|
||||
from wrolpi.tags import TagFile, Tag, tag_append_sub_select_where, save_tags_config, sync_tags_directory
|
||||
from wrolpi.vars import PYTEST, IS_MACOS
|
||||
|
|
@ -160,51 +160,62 @@ async def set_file_viewed(session: Session, file: pathlib.Path):
|
|||
session.commit()
|
||||
|
||||
|
||||
def _get_or_create_file_group(session: Session, file: pathlib.Path) -> FileGroup:
|
||||
def _trackable_path(file: pathlib.Path) -> pathlib.Path | None:
|
||||
"""Return `file` with any `..` collapsed, or None if it is in an ignored directory.
|
||||
|
||||
`..` is collapsed first so it cannot step into an ignored directory past the check.
|
||||
|
||||
@raise InvalidFile: The file is not in the media directory.
|
||||
"""
|
||||
file = pathlib.Path(os.path.normpath(file))
|
||||
if not get_paths_in_media_directory([file]):
|
||||
raise InvalidFile('File must be in the media directory')
|
||||
if not remove_files_in_ignored_directories([file]):
|
||||
return None
|
||||
return file
|
||||
|
||||
|
||||
def set_file_progress(session: Session, file: pathlib.Path, progress: float, position: dict = None) \
|
||||
-> Tuple[float | None, dict | None]:
|
||||
"""Record how far the User is through a file (see wrolpi.files.progress). This is also a view.
|
||||
|
||||
The caller owns the (write) transaction, and saves the recently viewed config after it commits.
|
||||
|
||||
@return: The (progress, position) that was stored, (None, None) if the file is not tracked.
|
||||
"""
|
||||
progress, position = normalize_progress(progress, position)
|
||||
|
||||
if not (file := _trackable_path(file)):
|
||||
return None, None
|
||||
|
||||
try:
|
||||
return FileGroup.find_by_path(session, file)
|
||||
fg = FileGroup.find_by_path(session, file)
|
||||
except UnknownFile:
|
||||
if not file.is_file():
|
||||
raise InvalidFile(f'File does not exist: {get_relative_to_media_directory(file)}')
|
||||
fg = FileGroup.from_paths(session, file)
|
||||
fg.do_model(session)
|
||||
return fg
|
||||
|
||||
|
||||
def set_file_progress(session: Session, file: pathlib.Path, progress: float, position: dict = None,
|
||||
final: bool = False) -> Tuple[float | None, dict | None]:
|
||||
"""Record how far the User is through a file (see wrolpi.files.progress). This is also a view.
|
||||
|
||||
@param final: The User stopped (paused, closed); save the recently viewed config.
|
||||
@return: The (progress, position) that was stored.
|
||||
"""
|
||||
progress, position = normalize_progress(progress, position)
|
||||
|
||||
# Files in ignored directories are not tracked.
|
||||
if not remove_files_in_ignored_directories([file]):
|
||||
return None, None
|
||||
|
||||
fg = _get_or_create_file_group(session, file)
|
||||
# Not `set_viewed`, which saves the config; a heartbeat during playback only writes the DB.
|
||||
fg.viewed = now()
|
||||
fg.progress = progress
|
||||
fg.position = position
|
||||
session.commit()
|
||||
|
||||
if final:
|
||||
save_recently_viewed_config.activate_switch()
|
||||
return progress, position
|
||||
|
||||
|
||||
def clear_file_progress(session: Session, file: pathlib.Path):
|
||||
"""Forget how far the User is through a file, so it starts from the beginning. It remains viewed."""
|
||||
fg = _get_or_create_file_group(session, file)
|
||||
def clear_file_progress(session: Session, file: pathlib.Path) -> bool:
|
||||
"""Forget how far the User is through a file, so it starts from the beginning. It remains viewed.
|
||||
|
||||
A file that was never tracked is left alone. The caller owns the (write) transaction.
|
||||
|
||||
@return: True if a FileGroup was changed.
|
||||
"""
|
||||
if not (file := _trackable_path(file)):
|
||||
return False
|
||||
if not (fg := FileGroup.get_by_path(session, file)):
|
||||
return False
|
||||
fg.progress = None
|
||||
fg.position = None
|
||||
session.commit()
|
||||
save_recently_viewed_config.activate_switch()
|
||||
return True
|
||||
|
||||
|
||||
@cachetools.func.ttl_cache(10_000, 30.0)
|
||||
|
|
|
|||
|
|
@ -14,6 +14,8 @@ The database is the working store. `recently_viewed.yaml` holds the most recent
|
|||
history and positions survive a database rebuild.
|
||||
"""
|
||||
import math
|
||||
import os
|
||||
import pathlib
|
||||
from dataclasses import dataclass, field
|
||||
from datetime import datetime
|
||||
from typing import List, Optional, Tuple
|
||||
|
|
@ -21,7 +23,7 @@ from typing import List, Optional, Tuple
|
|||
import pytz
|
||||
|
||||
from wrolpi import dates
|
||||
from wrolpi.common import ConfigFile, logger, get_media_directory
|
||||
from wrolpi.common import ConfigFile, logger, get_media_directory, get_paths_in_media_directory
|
||||
from wrolpi.db import get_db_session
|
||||
from wrolpi.errors import ValidationError, NoPrimaryFile
|
||||
from wrolpi.events import Events
|
||||
|
|
@ -155,6 +157,11 @@ class RecentlyViewedConfig(ConfigFile):
|
|||
.limit(RECENTLY_VIEWED_LIMIT)
|
||||
files = list()
|
||||
for file_group in file_groups:
|
||||
if not get_paths_in_media_directory([file_group.primary_path]):
|
||||
# One bad row must not stop every later save.
|
||||
logger.warning(f'Not saving recently viewed file outside the media directory:'
|
||||
f' {file_group.primary_path}')
|
||||
continue
|
||||
entry = dict(
|
||||
path=str(file_group.primary_path.relative_to(media_directory)),
|
||||
viewed=file_group.viewed.isoformat(),
|
||||
|
|
@ -172,8 +179,8 @@ class RecentlyViewedConfig(ConfigFile):
|
|||
Events.send_config_save_failed(message)
|
||||
|
||||
def import_config(self, file=None, send_events=False):
|
||||
"""Apply the config to the database. The newer `viewed` wins, so a file viewed while WROLPi was
|
||||
starting keeps its newer position."""
|
||||
"""Apply the config to the database. Progress written after the config (e.g. while WROLPi was starting)
|
||||
is kept; a newer plain view keeps its `viewed`, but takes the config's position."""
|
||||
super().import_config(file, send_events)
|
||||
try:
|
||||
# wrolpi.files.models imports this module.
|
||||
|
|
@ -181,33 +188,51 @@ class RecentlyViewedConfig(ConfigFile):
|
|||
from wrolpi.files.models import FileGroup
|
||||
|
||||
media_directory = get_media_directory()
|
||||
entries = [i for i in map(_parse_entry, self.files) if i]
|
||||
entries = list()
|
||||
for path, viewed, progress, position in filter(None, map(_parse_entry, self.files)):
|
||||
# An absolute `path` replaces the media directory in the join, and `..` can leave it.
|
||||
absolute_path = pathlib.Path(os.path.normpath(media_directory / path))
|
||||
if not get_paths_in_media_directory([absolute_path]):
|
||||
logger.warning(f'Not importing recently viewed file outside the media directory: {path}')
|
||||
continue
|
||||
entries.append((absolute_path, viewed, progress, position))
|
||||
absolute_paths = [i[0] for i in entries]
|
||||
|
||||
# Glob the files which are not yet in the DB before the write transaction; stat'ing a thousand files on
|
||||
# a slow disk would block every other writer.
|
||||
with get_db_session() as session:
|
||||
known = {i for (i,) in session.query(FileGroup.primary_path)
|
||||
.filter(FileGroup.primary_path.in_(absolute_paths))}
|
||||
new_files = {i: glob_shared_stem(i) for i in absolute_paths if i not in known and i.is_file()}
|
||||
|
||||
with get_db_session(commit=True) as session:
|
||||
absolute_paths = [media_directory / path for path, *_ in entries]
|
||||
file_groups = session.query(FileGroup).filter(FileGroup.primary_path.in_(absolute_paths))
|
||||
file_groups_by_path = {i.primary_path: i for i in file_groups}
|
||||
|
||||
applied = 0
|
||||
for path, viewed, progress, position in entries:
|
||||
absolute_path = media_directory / path
|
||||
for absolute_path, viewed, progress, position in entries:
|
||||
file_group = file_groups_by_path.get(absolute_path)
|
||||
if not file_group:
|
||||
if not absolute_path.is_file():
|
||||
if not (files := new_files.pop(absolute_path, None)):
|
||||
# The file was deleted; its history goes with it.
|
||||
logger.debug(f'Not importing recently viewed file which does not exist: {path}')
|
||||
logger.debug(f'Not importing recently viewed file which does not exist: {absolute_path}')
|
||||
continue
|
||||
try:
|
||||
file_group = FileGroup.from_paths(session, *glob_shared_stem(absolute_path))
|
||||
file_group = FileGroup.from_paths(session, *files)
|
||||
except NoPrimaryFile:
|
||||
logger.warning(f'Cannot import recently viewed file: {path}')
|
||||
logger.warning(f'Cannot import recently viewed file: {absolute_path}')
|
||||
continue
|
||||
session.add(file_group)
|
||||
session.flush([file_group])
|
||||
file_groups_by_path[absolute_path] = file_group
|
||||
|
||||
if file_group.viewed and file_group.viewed >= viewed:
|
||||
db_is_newer = file_group.viewed is not None and file_group.viewed >= viewed
|
||||
if db_is_newer and file_group.progress is not None:
|
||||
# Progress was written after the config (e.g. while WROLPi was starting).
|
||||
continue
|
||||
file_group.viewed = viewed
|
||||
# A newer plain view is kept, but it is not a newer place to resume.
|
||||
if not db_is_newer:
|
||||
file_group.viewed = viewed
|
||||
file_group.progress = progress
|
||||
file_group.position = position
|
||||
applied += 1
|
||||
|
|
|
|||
|
|
@ -301,3 +301,117 @@ async def test_recently_viewed_follows_move(test_session, async_client, test_dir
|
|||
entry, = read_recently_viewed(test_directory)
|
||||
assert entry['path'] == 'watched/movie.mp4'
|
||||
assert entry['progress'] == 0.4
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_progress_survives_concurrent_writer(async_client, test_session, test_directory, make_files_structure):
|
||||
"""A progress save waits for another writer (a refresh, a download) instead of failing."""
|
||||
from wrolpi.conftest import production_like_sessions, write_lock_held_briefly
|
||||
movie, = make_files_structure(['videos/movie.mp4'])
|
||||
FileGroup.from_paths(test_session, movie)
|
||||
test_session.commit()
|
||||
db_file = test_session.get_bind().url.database
|
||||
|
||||
body = json.dumps(dict(file='videos/movie.mp4', progress=0.4, position=dict(kind='time', seconds=480)))
|
||||
with production_like_sessions(test_session) as maker:
|
||||
with write_lock_held_briefly(db_file):
|
||||
request, response = await async_client.post('/api/files/progress', content=body)
|
||||
assert response.status_code == HTTPStatus.OK, response.body
|
||||
|
||||
with write_lock_held_briefly(db_file):
|
||||
request, response = await async_client.post('/api/files/progress/clear',
|
||||
content=json.dumps(dict(file='videos/movie.mp4')))
|
||||
assert response.status_code == HTTPStatus.NO_CONTENT, response.body
|
||||
|
||||
session = maker()
|
||||
try:
|
||||
fg, = session.query(FileGroup).all()
|
||||
assert fg.viewed and fg.progress is None
|
||||
finally:
|
||||
session.close()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_progress_ignored_directory_dot_dot(test_session, async_client, test_directory, make_files_structure,
|
||||
test_wrolpi_config, await_switches):
|
||||
"""`..` cannot step into an ignored directory."""
|
||||
from wrolpi.common import get_wrolpi_config
|
||||
make_files_structure(['private/movie.mp4', 'videos/other.mp4'])
|
||||
get_wrolpi_config().ignored_directories = [str(test_directory / 'private')]
|
||||
await await_switches()
|
||||
|
||||
body = dict(file='videos/../private/movie.mp4', progress=0.4, position=dict(kind='time', seconds=480))
|
||||
await async_client.post('/api/files/progress', content=json.dumps(body))
|
||||
assert test_session.query(FileGroup).count() == 0
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_progress_clear_creates_nothing(test_session, async_client, test_directory, make_files_structure,
|
||||
test_wrolpi_config, await_switches):
|
||||
"""Starting over a file that was never tracked (or is ignored) does not index it."""
|
||||
from wrolpi.common import get_wrolpi_config
|
||||
make_files_structure(['private/movie.mp4', 'videos/movie.mp4'])
|
||||
get_wrolpi_config().ignored_directories = [str(test_directory / 'private')]
|
||||
await await_switches()
|
||||
|
||||
for file in ('private/movie.mp4', 'videos/movie.mp4'):
|
||||
request, response = await async_client.post('/api/files/progress/clear', content=json.dumps(dict(file=file)))
|
||||
assert response.status_code == HTTPStatus.NO_CONTENT
|
||||
assert test_session.query(FileGroup).count() == 0
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_recently_viewed_import_after_plain_view(test_session, async_client, test_directory,
|
||||
make_files_structure):
|
||||
"""A plain view during startup is newer, but it is not a newer resume point; the config's position applies."""
|
||||
movie, = make_files_structure(['videos/movie.mp4'])
|
||||
fg = FileGroup.from_paths(test_session, movie)
|
||||
viewed_now = fg.set_viewed()
|
||||
test_session.commit()
|
||||
|
||||
config_file = test_directory / 'config/recently_viewed.yaml'
|
||||
config_file.parent.mkdir(exist_ok=True)
|
||||
config_file.write_text(yaml.dump(dict(version=1, files=[
|
||||
dict(path='videos/movie.mp4', viewed=(viewed_now - timedelta(days=1)).isoformat(), progress=0.5,
|
||||
position=dict(kind='time', seconds=600))])))
|
||||
|
||||
get_recently_viewed_config().import_config()
|
||||
|
||||
test_session.expire_all()
|
||||
fg, = test_session.query(FileGroup).all()
|
||||
assert fg.progress == 0.5
|
||||
assert fg.position == dict(kind='time', seconds=600.0)
|
||||
assert fg.viewed == viewed_now, 'The newer view is kept'
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_recently_viewed_import_outside_media_directory(test_session, async_client, test_directory,
|
||||
make_files_structure):
|
||||
"""A hand-edited entry cannot create a FileGroup outside the media directory, and one bad row in the DB
|
||||
does not stop the config from being saved."""
|
||||
make_files_structure(['videos/movie.mp4'])
|
||||
outside = test_directory.parent / f'{test_directory.name}-outside.mp4'
|
||||
outside.touch()
|
||||
try:
|
||||
now = dates.now()
|
||||
config_file = test_directory / 'config/recently_viewed.yaml'
|
||||
config_file.parent.mkdir(exist_ok=True)
|
||||
config_file.write_text(yaml.dump(dict(version=1, files=[
|
||||
dict(path=str(outside), viewed=now.isoformat()),
|
||||
dict(path=f'../{outside.name}', viewed=now.isoformat()),
|
||||
dict(path='videos/movie.mp4', viewed=(now - timedelta(hours=1)).isoformat()),
|
||||
])))
|
||||
config = get_recently_viewed_config()
|
||||
config.import_config()
|
||||
test_session.expire_all()
|
||||
assert [i.primary_path.name for i in test_session.query(FileGroup)] == ['movie.mp4']
|
||||
|
||||
# A row outside the media directory (however it got there) is skipped by the dump.
|
||||
bad = FileGroup.from_paths(test_session, outside)
|
||||
bad.viewed = now
|
||||
test_session.commit()
|
||||
config.dump_config()
|
||||
config.save()
|
||||
assert [i['path'] for i in read_recently_viewed(test_directory)] == ['videos/movie.mp4']
|
||||
finally:
|
||||
outside.unlink()
|
||||
|
|
|
|||
|
|
@ -173,6 +173,32 @@ def test_drop_file_group_idempotency_preserves_fts(test_directory):
|
|||
assert len(hits) == 1
|
||||
|
||||
|
||||
def test_file_group_progress_downgrade_preserves_fts(test_directory):
|
||||
"""Dropping file_group.progress/position must not rebuild file_group, which would drop its FTS triggers."""
|
||||
from alembic.config import Config
|
||||
from alembic import command
|
||||
from wrolpi.db import get_db_uri
|
||||
from wrolpi.vars import PROJECT_DIR
|
||||
|
||||
(test_directory / 'config').mkdir(parents=True, exist_ok=True)
|
||||
config = Config(str(PROJECT_DIR / 'alembic.ini'))
|
||||
config.set_main_option('sqlalchemy.url', get_db_uri())
|
||||
command.upgrade(config, '2026_10_08_1200')
|
||||
command.downgrade(config, '2026_10_06_1500')
|
||||
|
||||
with sqlite3.connect(get_db_file()) as conn:
|
||||
cols = {r[1] for r in conn.execute('PRAGMA table_info(file_group)')}
|
||||
assert not {'progress', 'position'} & cols
|
||||
triggers = {r[0] for r in conn.execute("SELECT name FROM sqlite_master WHERE type='trigger'")}
|
||||
assert {'file_group_fts_ai', 'file_group_fts_ad', 'file_group_fts_au'} <= triggers
|
||||
conn.execute(
|
||||
'INSERT INTO file_group (directory, primary_path, files, stem, indexed, a_text) VALUES (?,?,?,?,0,?)',
|
||||
(str(test_directory), str(test_directory / 'searchable.txt'), '[]', 'searchable', 'alpha uniquephrase'),
|
||||
)
|
||||
hits = conn.execute("SELECT rowid FROM file_group_fts WHERE file_group_fts MATCH 'uniquephrase'").fetchall()
|
||||
assert len(hits) == 1
|
||||
|
||||
|
||||
def test_domain_collection_summary_migration_backfills(test_directory):
|
||||
"""The domain-summary migration installs the archive triggers and backfills the summary
|
||||
columns of existing Domain Collections from their Archives."""
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue