Add search, bulk actions, undo, and storage usage to File Management

- Manage Library search: find any course/folder by name anywhere in
  the tree instead of expanding levels one by one; results carry the
  same Move/Rename/Hide actions.
- Bulk select: every row gets a checkbox, so items found via search or
  expanded across different tree levels can be hidden, shown, or moved
  to the same destination together in one batch.
- Undo last action: a "Last action: ... [Undo]" bar appears after any
  move/rename (Sort Unsorted apply, Bulk Rename apply, a Manage
  Library move/rename, a bulk move) and reverses the whole batch.
  Deliberately never covers Hide/Show (already a one-click toggle) or
  Delete (permanent by design) - only ever move/rename, which are
  trivially reversible. Re-checks each item before reversing it, so a
  partial failure reports exactly what did and didn't reverse.
- Storage Usage: disk usage per top-level library folder, largest
  first, with a simple proportional bar; a manual scan since it reads
  every file's size.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
2026-08-24 14:41:26 -04:00
co-authored by Claude Sonnet 5
parent 71d6158c60
commit ba8d7e2a1b
4 changed files with 780 additions and 54 deletions
+301
View File
@@ -95,6 +95,34 @@ def cache_get_or_compute(key: str, compute, ttl: float = CACHE_TTL_SECONDS):
return value
# ---- Undo history for move/rename operations ----
#
# Only move and rename are ever recorded here - both are trivially
# reversible (move/rename the result back). Hide/show is already a
# one-click toggle with no need for "undo," and delete is permanent by
# design, so it's never in this history no matter how it's framed in the
# UI. Recorded as batches (not one entry per file) since the operations
# that populate this - Sort Unsorted apply, Bulk Rename apply, a bulk
# move - already act on several items as a single user action; undoing
# should mirror that, not require clicking "undo" N times.
_UNDO_HISTORY_MAX = 20
_undo_history: List[Dict[str, Any]] = []
def _record_undo_batch(label: str, entries: List[Dict[str, str]]) -> None:
"""Push one reversible batch - entries are {'old_path', 'new_path'}
pairs; the label is shown in the UI's Undo button. No-ops if entries
is empty (e.g. every item in a batch failed)."""
if not entries:
return
_undo_history.append({
'label': label,
'entries': entries,
'timestamp': datetime.now().isoformat(),
})
del _undo_history[:-_UNDO_HISTORY_MAX]
def invalidate_cache():
"""Clear all cached library-scan results - call after anything that changes what's on disk (rename, hide/show, library path change)."""
_cache_store.clear()
@@ -3015,6 +3043,57 @@ def browse_library_manage():
})
def search_manage_tree(query: str, library_root: str) -> List[Dict[str, Any]]:
"""
Every course or category folder anywhere in the library whose name
contains `query` (case-insensitive) - lets Manage Library jump
straight to something instead of manually expanding the tree level by
level. Includes hidden items (flagged 'hidden': True), unlike the
normal Library browser - finding something specifically to un-hide it
is exactly what this is for. Stops descending once a course is found,
same as everywhere else, so a course's internal chapter folders never
show up as if they were independently manageable.
"""
query_lower = query.lower().strip()
if not query_lower:
return []
hidden_set = set(get_hidden_paths())
results: List[Dict[str, Any]] = []
def walk(directory: Path):
try:
entries = sorted(
(p for p in directory.iterdir() if p.is_dir() and not p.name.startswith('.')),
key=lambda p: p.name.lower()
)
except (PermissionError, OSError):
return
for entry in entries:
is_course = _looks_like_course(entry)
is_hidden = os.path.abspath(str(entry)) in hidden_set
if query_lower in entry.name.lower():
results.append({
'type': 'course' if is_course else 'directory',
'name': entry.name,
'path': str(entry),
'hidden': is_hidden,
})
if not is_course:
walk(entry)
walk(Path(library_root))
return results
@app.route('/api/library/manage-search')
def manage_search_api():
"""Search endpoint backing Manage Library's search box (see
search_manage_tree)."""
query = request.args.get('q', '')
library_root = os.path.abspath(get_library_root())
return jsonify({'results': search_manage_tree(query, library_root)})
@app.route('/library/thumbnail')
def library_thumbnail():
"""Serve a course's cover image (see find_course_thumbnail), if it has one."""
@@ -3143,9 +3222,81 @@ def apply_unsorted_api():
errors.append({'source': source, 'error': str(e)})
invalidate_cache()
undo_entries = [{'old_path': m['source'], 'new_path': m['destination']} for m in moved]
_record_undo_batch(f'Sort Unsorted ({len(moved)} item{"" if len(moved) == 1 else "s"})', undo_entries)
return jsonify({'success': True, 'moved': len(moved), 'errors': errors})
def _directory_size_bytes(directory: Path, max_depth: int = 20) -> int:
"""Total size of every file anywhere under `directory` (bounded depth,
same caution used elsewhere in File Management for deep/unusual
trees). Skips anything it can't read rather than failing the whole
scan - a NAS with mixed permissions shouldn't block a size report for
everything else."""
total = 0
def walk(d: Path, depth: int) -> None:
nonlocal total
try:
entries = list(d.iterdir())
except (PermissionError, OSError):
return
for e in entries:
if e.name.startswith('.'):
continue
try:
if e.is_file():
total += e.stat().st_size
elif e.is_dir() and depth < max_depth:
walk(e, depth + 1)
except OSError:
continue
walk(directory, 0)
return total
def _format_bytes(n: int) -> str:
"""Human-readable size, e.g. 12.3 GB - matches the precision people
actually care about for a course library (no need for KB granularity)."""
size = float(n)
for unit in ('B', 'KB', 'MB', 'GB', 'TB'):
if size < 1024 or unit == 'TB':
return f'{size:.0f} {unit}' if unit == 'B' else f'{size:.1f} {unit}'
size /= 1024
return f'{size:.1f} TB'
@app.route('/api/library/storage-usage')
def storage_usage_api():
"""
Disk usage per top-level library folder (category or a bare course
sitting directly at the root), sorted largest-first, for spotting
what's eating the most space on the NAS. Manually triggered like
Duplicate Courses - a full recursive size scan touches every file in
the library, too expensive to run automatically on page load.
"""
library_root = Path(get_library_root())
try:
top_level = sorted(
(p for p in library_root.iterdir() if p.is_dir() and not p.name.startswith('.')),
key=lambda p: p.name.lower()
)
except (PermissionError, OSError) as e:
return jsonify({'error': str(e)}), 500
entries = []
for entry in top_level:
if entry.name == UNSORTED_FOLDER_NAME:
continue
size = _directory_size_bytes(entry)
entries.append({'name': entry.name, 'path': str(entry), 'bytes': size, 'human': _format_bytes(size)})
entries.sort(key=lambda e: e['bytes'], reverse=True)
total_bytes = sum(e['bytes'] for e in entries)
return jsonify({'entries': entries, 'total_bytes': total_bytes, 'total_human': _format_bytes(total_bytes)})
@app.route('/api/library/categories')
def library_categories_api():
"""Every category/subcategory folder in the library, for destination
@@ -3210,6 +3361,8 @@ def move_library_item_api():
invalidate_cache()
rebase_library_path(source_abs, final_path)
_record_undo_batch(f'Move "{os.path.basename(source_abs)}"',
[{'old_path': source_abs, 'new_path': final_path}])
global current_course
active_course_reset = False
@@ -3222,6 +3375,145 @@ def move_library_item_api():
return jsonify({'success': True, 'new_path': final_path, 'active_course_reset': active_course_reset})
@app.route('/api/library/bulk-move', methods=['POST'])
def bulk_move_library_items_api():
"""
Move several courses/folders already in the library to the same
destination in one batch - the multi-select complement to
/api/library/move for reorganizing many items at once. Same safety
checks as a single move, applied per item; continues past individual
failures and reports each outcome, the same pattern Bulk Rename uses.
"""
data = request.json or {}
sources = data.get('sources') or []
destination_relative = (data.get('destination') or '').strip().strip('/')
if not isinstance(sources, list) or not sources or not destination_relative:
return jsonify({'success': False, 'error': 'sources and destination are required'}), 400
library_root = os.path.abspath(get_library_root())
dest_dir = os.path.abspath(os.path.join(library_root, destination_relative))
if not (dest_dir == library_root or dest_dir.startswith(library_root + os.sep)):
return jsonify({'success': False, 'error': 'Destination is outside the library'}), 403
global current_course
results = []
undo_entries = []
active_course_reset = False
for source in sources:
source_abs = os.path.abspath(source)
result = {'source': source}
if not (source_abs == library_root or source_abs.startswith(library_root + os.sep)) or source_abs == library_root:
result.update(success=False, error='Source is outside the library')
elif dest_dir == source_abs or dest_dir.startswith(source_abs + os.sep):
result.update(success=False, error='Cannot move a folder into itself')
elif not os.path.isdir(source_abs):
result.update(success=False, error='No longer exists')
else:
final_name = os.path.basename(source_abs)
final_path = os.path.join(dest_dir, final_name)
if final_path == source_abs:
result.update(success=False, error='Already there')
elif os.path.exists(final_path):
result.update(success=False, error=f'"{final_name}" already exists at that destination')
else:
try:
os.makedirs(dest_dir, exist_ok=True)
shutil.move(source_abs, final_path)
rebase_library_path(source_abs, final_path)
undo_entries.append({'old_path': source_abs, 'new_path': final_path})
result.update(success=True, new_path=final_path)
if current_course is not None:
course_abs = os.path.abspath(current_course.path)
if course_abs == source_abs or course_abs.startswith(source_abs + os.sep):
current_course = None
active_course_reset = True
except OSError as e:
result.update(success=False, error=str(e))
results.append(result)
invalidate_cache()
moved_count = len(undo_entries)
_record_undo_batch(f'Bulk move ({moved_count} item{"" if moved_count == 1 else "s"}) to "{destination_relative}"',
undo_entries)
return jsonify({'success': True, 'moved': moved_count, 'results': results, 'active_course_reset': active_course_reset})
@app.route('/api/library/undo-last')
def undo_last_status_api():
"""Whether there's a move/rename batch available to undo, and what it
was - lets the UI show a specific "Undo: <label>" instead of a blind
button, or hide it entirely when there's nothing to undo."""
if not _undo_history:
return jsonify({'available': False})
batch = _undo_history[-1]
return jsonify({'available': True, 'label': batch['label'], 'count': len(batch['entries'])})
@app.route('/api/library/undo-last', methods=['POST'])
def undo_last_action_api():
"""
Reverse the most recent move/rename batch (Sort Unsorted apply, Bulk
Rename apply, a Manage Library move/rename, a bulk move) as a single
action, moving/renaming each entry back to where it came from.
Re-checks every entry before touching it - the destination must still
exist and the original location must still be free - since the
filesystem may have changed since the batch was recorded; a partial
failure still pops the batch (retrying the same undo again wouldn't
help) and reports exactly what did and didn't reverse.
"""
if not _undo_history:
return jsonify({'success': False, 'error': 'Nothing to undo'}), 400
batch = _undo_history.pop()
library_root = os.path.abspath(get_library_root())
results = []
global current_course
active_course_reset = False
for entry in reversed(batch['entries']):
old_abs = os.path.abspath(entry['old_path'])
new_abs = os.path.abspath(entry['new_path'])
result = {'old_path': entry['old_path'], 'new_path': entry['new_path']}
if not (new_abs == library_root or new_abs.startswith(library_root + os.sep)):
result.update(success=False, error='Outside the library')
elif not os.path.isdir(new_abs):
result.update(success=False, error='No longer exists - already moved or renamed again?')
elif os.path.exists(old_abs):
result.update(success=False, error='Original location is occupied again')
else:
try:
os.makedirs(os.path.dirname(old_abs), exist_ok=True)
shutil.move(new_abs, old_abs)
rebase_library_path(new_abs, old_abs)
result['success'] = True
if current_course is not None:
course_abs = os.path.abspath(current_course.path)
if course_abs == new_abs or course_abs.startswith(new_abs + os.sep):
current_course = None
active_course_reset = True
except OSError as e:
result.update(success=False, error=str(e))
results.append(result)
invalidate_cache()
reversed_count = sum(1 for r in results if r['success'])
return jsonify({
'success': True,
'label': batch['label'],
'reversed': reversed_count,
'total': len(results),
'results': results,
'active_course_reset': active_course_reset,
})
@app.route('/api/library/duplicates')
def duplicate_courses_api():
"""Scan for courses that look like copies of each other by name (see
@@ -3537,6 +3829,9 @@ def rename_path_api():
return jsonify({'error': 'path is required'}), 400
result = perform_rename(path, new_name)
if result.get('success'):
_record_undo_batch(f'Rename "{os.path.basename(path)}"',
[{'old_path': os.path.abspath(path), 'new_path': result['new_path']}])
status = result.pop('status')
return jsonify(result), status
@@ -3658,6 +3953,7 @@ def bulk_rename_apply_api():
return jsonify({'error': 'items is required'}), 400
results = []
undo_entries = []
for item in items:
path = item.get('path', '')
new_name = item.get('new_name', '')
@@ -3666,8 +3962,13 @@ def bulk_rename_apply_api():
outcome['path'] = path
outcome['old_name'] = item.get('old_name', '')
outcome['new_name'] = new_name
if outcome.get('success'):
undo_entries.append({'old_path': os.path.abspath(path), 'new_path': outcome['new_path']})
results.append(outcome)
succeeded = len(undo_entries)
_record_undo_batch(f'Bulk rename ({succeeded} item{"" if succeeded == 1 else "s"})', undo_entries)
return jsonify({'results': results})