rmsitz/offlineu
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add Recently Added, bulk rename, time remaining, library stats, backup export

Browse Source 
Five more usability features on top of the search/thumbnails batch:

- Recently Added: dashboard card of courses by folder mtime, separate from
  Recently Viewed (watched vs. just showed up on disk).
- Bulk find/replace rename across every course/folder name at once, with a
  mandatory preview step before anything touches disk. Refactored the
  single-item rename route to share the same validate/apply logic.
- Estimated time remaining on the loaded course's stats card, computed only
  from lessons that have actually reported a duration.
- Library-wide stats overview: total courses, lessons tracked, time
  watched, daily streak - read from each course's small progress file
  rather than re-scanning course contents.
- One-click backup/export zip of settings, hidden-paths, recently-viewed,
  and every course's progress/notes.

Also fixes a real performance bug found along the way: search was routing
through list_library_directory, which computes a full recursive file count
and thumbnail lookup for every course at every level regardless of match -
turning a search into an O(every file in the library) scan. Gave search its
own lightweight directory-only walk (iter_all_courses), now shared by
Recently Added and the stats overview too, so the expensive per-course work
only runs for courses that actually match.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
rmsitz
2026-08-22 11:36:05 -04:00
co-authored by Claude Sonnet 5
parent dbf9528fea
commit a6c49d3762
4 changed files with 611 additions and 66 deletions
Download Patch File Download Diff File Expand all files Collapse all files
offlineu_core.py
+329 -64
View File
@@ -10,10 +10,12 @@ import mimetypes
import re
import sys
import argparse
import io
import zipfile
from pathlib import Path
from datetime import datetime
from datetime import datetime, timedelta
from dataclasses import dataclass, asdict
from typing import List, Dict, Optional, Any, Tuple
from typing import List, Dict, Optional, Any, Tuple, Iterator
from flask import Flask, render_template, request, jsonify, send_file, redirect, url_for
app = Flask(__name__)
@@ -453,28 +455,41 @@ class DynamicCourseParser:
"""Calculate completion statistics for a directory node"""
total_lessons = 0
completed_lessons = 0
# Only lessons that have actually been played report a duration, so
# "remaining time" is only ever an estimate over what's known so far
# - there's no way to know the length of a lesson nobody's opened yet.
total_duration_seconds = 0
watched_seconds = 0
def count_lessons_recursive(n: DirectoryNode):
nonlocal total_lessons, completed_lessons
nonlocal total_lessons, completed_lessons, total_duration_seconds, watched_seconds
# Count lessons in this node
for lesson in n.lessons:
total_lessons += 1
if lesson.completed:
completed_lessons += 1
if lesson.duration_seconds:
total_duration_seconds += lesson.duration_seconds
watched_seconds += (
lesson.duration_seconds if lesson.completed
else min(lesson.progress_seconds, lesson.duration_seconds)
)
# Recursively count in children
for child in n.children.values():
count_lessons_recursive(child)
count_lessons_recursive(node)
completion_percentage = (completed_lessons / total_lessons * 100) if total_lessons > 0 else 0
return {
'total_lessons': total_lessons,
'completed_lessons': completed_lessons,
'completion_percentage': round(completion_percentage, 1)
'completion_percentage': round(completion_percentage, 1),
'total_duration_seconds': total_duration_seconds,
'remaining_seconds': max(0, total_duration_seconds - watched_seconds)
}
@@ -732,25 +747,19 @@ def get_library_root() -> str:
return LIBRARY_PATH
def search_library_courses(dir_path: str, query: str) -> List[Dict[str, Any]]:
def iter_all_courses(dir_path: str) -> Iterator[Path]:
"""
Recursively search the library for courses whose name contains `query`
(case-insensitive), respecting hidden paths exactly like normal browsing.
Deliberately doesn't reuse list_library_directory for the walk itself:
that function computes media_count (a full recursive file count via
rglob) and thumbnail presence for *every* course at each level, which is
fine for showing one directory's worth of courses but turns a full-
library search into an O(every file in every course) scan. Here, the
directory walk only touches directory entries (cheap - iterdir, no file
stats), and the expensive per-course lookups only run for the handful of
courses whose name actually matches.
Recursively yield every course directory under dir_path, respecting
hidden paths exactly like normal browsing - without touching file
contents/counts. This is the cheap directory-only walk (iterdir +
_looks_like_course) that search, "Recently Added," and library stats
all need; the expensive per-course work (media_count via rglob,
thumbnail lookup, progress-file reads) is left to each caller to do
only for the courses it actually ends up using.
"""
query_lower = query.lower()
hidden_set = set(get_hidden_paths())
results: List[Dict[str, Any]] = []
def walk(directory: Path):
def walk(directory: Path) -> Iterator[Path]:
try:
entries = sorted(
(p for p in directory.iterdir() if p.is_dir() and not p.name.startswith('.')),
@@ -762,28 +771,136 @@ def search_library_courses(dir_path: str, query: str) -> List[Dict[str, Any]]:
for entry in entries:
if os.path.abspath(str(entry)) in hidden_set:
continue
if _looks_like_course(entry):
if query_lower in entry.name.lower():
media_count = len([
f for f in entry.rglob('*')
if f.is_file() and f.suffix.lower() in VIDEO_EXTENSIONS | AUDIO_EXTENSIONS
])
results.append({
'type': 'course',
'name': entry.name,
'path': str(entry),
'media_files': media_count,
'hidden': False,
'has_thumbnail': find_course_thumbnail(str(entry)) is not None
})
yield entry
else:
walk(entry)
yield from walk(entry)
walk(Path(dir_path))
yield from walk(Path(dir_path))
def _course_summary(course_dir: Path) -> Dict[str, Any]:
"""Build the {type, name, path, media_files, hidden, has_thumbnail} shape
shared by the Library browser, search results, and Recently Added."""
media_count = len([
f for f in course_dir.rglob('*')
if f.is_file() and f.suffix.lower() in VIDEO_EXTENSIONS | AUDIO_EXTENSIONS
])
return {
'type': 'course',
'name': course_dir.name,
'path': str(course_dir),
'media_files': media_count,
'hidden': False,
'has_thumbnail': find_course_thumbnail(str(course_dir)) is not None
}
def search_library_courses(dir_path: str, query: str) -> List[Dict[str, Any]]:
"""
Search the library for courses whose name contains `query`
(case-insensitive). The expensive per-course lookups (media_count,
thumbnail) only run for courses whose name actually matches - see
iter_all_courses.
"""
query_lower = query.lower()
return [
_course_summary(course)
for course in iter_all_courses(dir_path)
if query_lower in course.name.lower()
]
def _humanize_days_ago(dt: datetime) -> str:
"""'Today' / 'Yesterday' / 'N days ago' / a plain date once it's old enough to matter less."""
days = (datetime.now().date() - dt.date()).days
if days <= 0:
return 'Today'
if days == 1:
return 'Yesterday'
if days < 14:
return f'{days} days ago'
return dt.strftime('%b %d')
def get_recently_added_courses(limit: int = 5) -> List[Dict[str, Any]]:
"""
Courses whose folder was most recently created/modified on disk, for
the dashboard's "Recently Added" card - separate from Recently Viewed
(which tracks what you've *watched*, not what showed up in the library).
"""
library_root = get_library_root()
dated = []
for course_dir in iter_all_courses(library_root):
try:
mtime = course_dir.stat().st_mtime
except OSError:
continue
dated.append((mtime, course_dir))
dated.sort(key=lambda pair: pair[0], reverse=True)
results = []
for mtime, course_dir in dated[:limit]:
item = _course_summary(course_dir)
item['added_display'] = _humanize_days_ago(datetime.fromtimestamp(mtime))
results.append(item)
return results
def get_library_stats() -> Dict[str, Any]:
"""
Library-wide overview for the dashboard: total courses, lessons with
any progress record, completed lessons, total time watched, and the
current daily streak. Reads each course's small progress JSON file
directly rather than re-scanning course directory contents, so this
stays cheap regardless of how many files are inside each course.
"""
total_courses = 0
lessons_tracked = 0
completed_lessons = 0
watched_seconds = 0
active_dates = set()
for course_dir in iter_all_courses(get_library_root()):
total_courses += 1
try:
with open(course_dir / '.offlineu_progress.json', 'r') as f:
progress = json.load(f)
except (FileNotFoundError, json.JSONDecodeError, OSError):
continue
for key, entry in progress.items():
if key == 'last_accessed_path' or not isinstance(entry, dict):
continue
lessons_tracked += 1
if entry.get('completed'):
completed_lessons += 1
watched_seconds += entry.get('duration_seconds') or 0
else:
watched_seconds += entry.get('progress_seconds') or 0
last_accessed = entry.get('last_accessed')
if last_accessed:
try:
active_dates.add(datetime.fromisoformat(last_accessed).date())
except ValueError:
pass
streak_days = 0
day = datetime.now().date()
while day in active_dates:
streak_days += 1
day -= timedelta(days=1)
return {
'total_courses': total_courses,
'lessons_tracked': lessons_tracked,
'completed_lessons': completed_lessons,
'watched_display': format_duration(watched_seconds),
'streak_days': streak_days
}
RECENT_VIEWS_FILE = os.path.join(DATA_DIR, 'recent_views.json')
MAX_RECENT_VIEWS = 20
@@ -1021,6 +1138,13 @@ def _split_continue_watching(all_views: List[Dict[str, Any]]) -> Tuple[List[Dict
return continue_watching, recent_views
def format_duration(seconds: int) -> str:
"""Render a duration in seconds as e.g. '3h 24m' or '45m' (under an hour)."""
total_minutes = max(0, int(seconds)) // 60
hours, minutes = divmod(total_minutes, 60)
return f"{hours}h {minutes}m" if hours else f"{minutes}m"
@app.route('/')
def index():
"""Main dashboard"""
@@ -1034,11 +1158,15 @@ def index():
course=None,
stats={'total_lessons': 0, 'completed_lessons': 0, 'completion_percentage': 0},
continue_watching=continue_watching,
recent_views=recent_views)
recent_views=recent_views,
recently_added=get_recently_added_courses(),
library_stats=get_library_stats())
# Apply progress data to tree
ProgressTracker.apply_progress_to_tree(current_course)
stats = ProgressTracker.get_completion_stats(current_course)
if stats.get('total_duration_seconds'):
stats['remaining_display'] = format_duration(stats['remaining_seconds'])
return render_template('course_dashboard.html',
course=current_course,
@@ -1241,41 +1369,48 @@ def set_hidden_path_api():
return jsonify({'success': True, 'hidden_paths': updated})
@app.route('/api/rename-path', methods=['POST'])
def rename_path_api():
"""Rename a course/folder directory in the library, in place on disk."""
def validate_new_name(new_name: str) -> Optional[str]:
"""Validate a proposed directory name; returns an error message, or None if valid."""
if not new_name:
return 'New name cannot be empty'
if '/' in new_name or '\\' in new_name or '\x00' in new_name or new_name in ('.', '..'):
return 'New name cannot contain path separators'
if new_name.startswith('.'):
return 'New name cannot start with a dot'
return None
def perform_rename(old_path: str, new_name: str) -> Dict[str, Any]:
"""
Validate and apply a single directory rename within the library root:
rebases any hidden-path/recent-view references that pointed inside it,
and resets the active course if it (or an ancestor) was the thing
renamed. Shared by the single-item and bulk rename routes, always
returning a dict with a 'status' key for the caller to respond with.
"""
global current_course
data = request.json or {}
path = data.get('path', '')
new_name = (data.get('new_name') or '').strip()
if not path:
return jsonify({'error': 'path is required'}), 400
if not new_name:
return jsonify({'error': 'New name cannot be empty'}), 400
if '/' in new_name or '\\' in new_name or '\x00' in new_name or new_name in ('.', '..'):
return jsonify({'error': 'New name cannot contain path separators'}), 400
if new_name.startswith('.'):
return jsonify({'error': 'New name cannot start with a dot'}), 400
error = validate_new_name(new_name)
if error:
return {'success': False, 'error': error, 'status': 400}
library_root = os.path.abspath(get_library_root())
old_abs = os.path.abspath(path)
old_abs = os.path.abspath(old_path)
if not (old_abs == library_root or old_abs.startswith(library_root + os.sep)):
return jsonify({'error': 'Path outside library root'}), 403
return {'success': False, 'error': 'Path outside library root', 'status': 403}
if old_abs == library_root:
return jsonify({'error': 'Cannot rename the library root itself'}), 400
return {'success': False, 'error': 'Cannot rename the library root itself', 'status': 400}
if not os.path.isdir(old_abs):
return jsonify({'error': 'Directory not found'}), 404
return {'success': False, 'error': 'Directory not found', 'status': 404}
new_abs = os.path.join(os.path.dirname(old_abs), new_name)
if os.path.exists(new_abs):
return jsonify({'error': f'"{new_name}" already exists here'}), 409
return {'success': False, 'error': f'"{new_name}" already exists here', 'status': 409}
try:
os.rename(old_abs, new_abs)
except OSError as e:
return jsonify({'error': f'Rename failed: {e}'}), 500
return {'success': False, 'error': f'Rename failed: {e}', 'status': 500}
rebase_library_path(old_abs, new_abs)
@@ -1286,7 +1421,107 @@ def rename_path_api():
current_course = None
active_course_reset = True
return jsonify({'success': True, 'new_path': new_abs, 'active_course_reset': active_course_reset})
return {
'success': True,
'new_path': new_abs,
'active_course_reset': active_course_reset,
'status': 200
}
@app.route('/api/rename-path', methods=['POST'])
def rename_path_api():
"""Rename a course/folder directory in the library, in place on disk."""
data = request.json or {}
path = data.get('path', '')
new_name = (data.get('new_name') or '').strip()
if not path:
return jsonify({'error': 'path is required'}), 400
result = perform_rename(path, new_name)
status = result.pop('status')
return jsonify(result), status
def _iter_all_directories(directory: Path, hidden_set: set) -> Iterator[Path]:
"""
Yield every directory in the library tree - both course folders and the
category/group folders above them - skipping hidden ones and not
descending into a course's own internal sections (Section 1, etc. -
those aren't independently manageable library items anywhere else in
the app either, so bulk rename shouldn't touch them).
"""
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:
if os.path.abspath(str(entry)) in hidden_set:
continue
yield entry
if not _looks_like_course(entry):
yield from _iter_all_directories(entry, hidden_set)
def find_bulk_rename_matches(library_root: str, pattern: str, replacement: str) -> List[Dict[str, str]]:
"""Every directory (course or folder) in the library whose name contains `pattern`."""
hidden_set = set(get_hidden_paths())
return [
{
'path': str(directory),
'old_name': directory.name,
'new_name': directory.name.replace(pattern, replacement)
}
for directory in _iter_all_directories(Path(library_root), hidden_set)
if pattern in directory.name
]
@app.route('/api/bulk-rename/preview')
def bulk_rename_preview_api():
"""Preview a find/replace rename across the whole library, without touching disk."""
pattern = request.args.get('pattern', '')
replacement = request.args.get('replacement', '')
if not pattern:
return jsonify({'error': 'pattern is required'}), 400
library_root = os.path.abspath(get_library_root())
matches = find_bulk_rename_matches(library_root, pattern, replacement)
return jsonify({'matches': matches})
@app.route('/api/bulk-rename/apply', methods=['POST'])
def bulk_rename_apply_api():
"""
Apply a bulk rename. Takes the *exact* {path, new_name} list the client
got back from /api/bulk-rename/preview, rather than re-deriving matches
from the pattern again - what gets renamed is provably what the user
saw and approved, and can't drift if the library changed in between.
Continues past individual failures (e.g. a collision) instead of
aborting the whole batch, reporting a per-item result.
"""
data = request.json or {}
items = data.get('items') or []
if not isinstance(items, list) or not items:
return jsonify({'error': 'items is required'}), 400
results = []
for item in items:
path = item.get('path', '')
new_name = item.get('new_name', '')
outcome = perform_rename(path, new_name)
outcome.pop('status', None)
outcome['path'] = path
outcome['old_name'] = item.get('old_name', '')
outcome['new_name'] = new_name
results.append(outcome)
return jsonify({'results': results})
@app.route('/settings')
@@ -1350,6 +1585,36 @@ def reset_settings_api():
})
@app.route('/api/backup')
def download_backup():
"""
Bundle everything that isn't recoverable from the course files
themselves - settings, hidden-path curation, recently-viewed history,
and every course's progress/notes file - into a single downloadable
zip. Cheap insurance before a NAS migration or a docker volume mistake.
"""
buffer = io.BytesIO()
with zipfile.ZipFile(buffer, 'w', zipfile.ZIP_DEFLATED) as zf:
for name, path in (
('settings.json', SETTINGS_FILE),
('hidden_paths.json', HIDDEN_PATHS_FILE),
('recent_views.json', RECENT_VIEWS_FILE),
):
if os.path.exists(path):
zf.write(path, name)
library_root = get_library_root()
for course_dir in iter_all_courses(library_root):
progress_file = course_dir / '.offlineu_progress.json'
if progress_file.exists():
relative = os.path.relpath(str(course_dir), library_root)
zf.write(progress_file, os.path.join('progress', relative, '.offlineu_progress.json'))
buffer.seek(0)
filename = f"offlineu-backup-{datetime.now().strftime('%Y-%m-%d')}.zip"
return send_file(buffer, mimetype='application/zip', as_attachment=True, download_name=filename)
@app.route('/load_course', methods=['POST'])
def load_course():
"""Load course from selected directory"""