#!/usr/bin/env python3 """ OfflineU - Self-hosted Course Viewer & Tracker Enhanced version with dynamic subdirectory navigation """ import os import json import mimetypes import random import re import sys import argparse import io import time import uuid import zipfile import urllib.request import urllib.error from pathlib import Path from datetime import datetime, timedelta from dataclasses import dataclass, asdict 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__) app.config['SECRET_KEY'] = 'your-secret-key-change-in-production' # Supported file types VIDEO_EXTENSIONS = {'.mp4', '.mkv', '.avi', '.mov', '.webm', '.m4v', '.flv', '.wmv'} AUDIO_EXTENSIONS = {'.mp3', '.wav', '.m4a', '.aac', '.ogg', '.flac'} SUBTITLE_EXTENSIONS = {'.srt', '.vtt', '.ass', '.sub', '.sbv'} TEXT_EXTENSIONS = {'.txt', '.md', '.html', '.htm', '.pdf', '.docx', '.doc', '.rtf'} QUIZ_INDICATORS = {'quiz', 'exam', 'test', 'assessment', 'exercise', 'assignment', 'homework'} IMAGE_EXTENSIONS = {'.jpg', '.jpeg', '.png', '.webp'} THUMBNAIL_BASENAMES = ('cover', 'folder', 'thumbnail', 'thumb', 'poster') # Base directory the "Library" browser scans for courses, so users don't have # to type a full filesystem path. Matches the ./courses volume mount in # docker-compose.yml by default; override with --library-path or the # COURSES_LIBRARY_PATH env var. LIBRARY_PATH = os.environ.get('COURSES_LIBRARY_PATH', '/app/courses') # The library is typically a network-mounted (SMB) directory, where every # iterdir()/rglob() call is a network round-trip - and the same expensive # scans (course listing, per-course media counts, subtitle text) get # recomputed from scratch on every single request with nothing shared # between them. The library itself only changes on human timescales # (someone adds a course), not per-request, so a short-TTL in-process cache # is a safe, high-leverage fix - no external cache needed for a # single-process personal app. _cache_store: Dict[str, Tuple[float, Any]] = {} CACHE_TTL_SECONDS = 300 # 5 minutes def cache_get_or_compute(key: str, compute, ttl: float = CACHE_TTL_SECONDS): now = time.time() cached = _cache_store.get(key) if cached is not None and now - cached[0] < ttl: return cached[1] value = compute() _cache_store[key] = (now, value) return value 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() # App-wide (non per-course) persisted data, e.g. display settings. Matches # the ./data volume mount in docker-compose.yml. DATA_DIR = os.environ.get('OFFLINEU_DATA_DIR', '/app/data') SETTINGS_FILE = os.path.join(DATA_DIR, 'settings.json') DEFAULT_SETTINGS = { 'theme': 'dark', # 'dark' | 'light' 'accent_color': '#007acc', # any #rrggbb hex 'font_family': 'system', # 'system' | 'sans' | 'serif' | 'monospace' 'font_size': 'medium', # 'small' | 'medium' | 'large' | 'xlarge' 'layout_width': 'wide', # 'normal' | 'wide' | 'full' 'density': 'comfortable', # 'comfortable' | 'compact' 'card_style': 'flat', # 'flat' | 'elevated' | 'bordered' 'corner_radius': 'rounded', # 'sharp' | 'rounded' | 'pill' 'library_path': '', # '' = use COURSES_LIBRARY_PATH/--library-path default 'video_width': '', # '' = responsive full-width; else last dragged size, in px 'video_height': '', 'playback_speed': '1', # video/audio playback rate, as a string (see SETTINGS_CHOICES) } # Bounds for the persisted video player size, to reject garbage values # without capping how big/small someone can reasonably drag it. VIDEO_SIZE_BOUNDS = {'video_width': (200, 4000), 'video_height': (120, 3000)} # Allowed values for every setting except accent_color (validated separately # as a hex string). Anything outside these sets is rejected/ignored on save. SETTINGS_CHOICES = { 'theme': { 'dark', 'light', 'dracula', 'tokyo_night', 'catppuccin_mocha', 'ayu_dark', 'github_dark', 'atom_one_dark', 'houston', 'night_owl', 'nord', 'matcha', }, 'font_family': {'system', 'sans', 'serif', 'monospace', 'monaspace'}, 'font_size': {'small', 'medium', 'large', 'xlarge'}, 'layout_width': {'normal', 'wide', 'full'}, 'density': {'comfortable', 'compact'}, 'card_style': {'flat', 'elevated', 'bordered'}, 'corner_radius': {'sharp', 'rounded', 'pill'}, 'playback_speed': {'0.75', '1', '1.25', '1.5', '1.75', '2'}, } # Display names for the theme dropdown - only needed where the raw key # (e.g. 'tokyo_night') isn't already a clean label via .capitalize(). THEME_DISPLAY_NAMES = { 'dark': 'Dark', 'light': 'Light', 'dracula': 'Dracula', 'tokyo_night': 'Tokyo Night', 'catppuccin_mocha': 'Catppuccin Mocha', 'ayu_dark': 'Ayu Dark', 'github_dark': 'GitHub Dark', 'atom_one_dark': 'Atom One Dark', 'houston': 'Houston', 'night_owl': 'Night Owl', 'nord': 'Nord', 'matcha': 'Matcha', } # Full color palette per theme. 'accent' here is only the *default* accent # offered when a theme is first selected - the accent_color setting is what # actually drives --accent afterward, so it stays independently editable. # Sourced from each theme's official palette (Dracula, Tokyo Night, # Catppuccin Mocha, Ayu, GitHub Dark, Atom One Dark, Houston, Nord all # verified against upstream repos/specs). Matcha is sourced from # lucafalasco/matcha's published VS Code theme JSON (editor/sideBar/ # activityBar backgrounds, foreground, panel.border, and the # statusBar/button accent color). # # Dainty used to be here instead of Nord, but it turned out to be a # Lab-space theme *generator* (HotWordland/dainty-vscode) with no fixed # shipped palette - its output depends on whatever base theme you feed # it, so there was no single "official Dainty" hex set to verify our # approximation against. Swapped for Nord, which does have one. THEME_PALETTES = { 'dark': { 'bg-primary': '#1a1a1a', 'bg-secondary': '#2d2d2d', 'bg-tertiary': '#3d3d3d', 'bg-tertiary-hover': '#404040', 'text-primary': '#e0e0e0', 'text-muted': '#999999', 'border-color': '#555555', 'accent': '#007acc', 'accent-hover': '#005a9e', }, 'light': { 'bg-primary': '#f2f2f2', 'bg-secondary': '#ffffff', 'bg-tertiary': '#eeeeee', 'bg-tertiary-hover': '#e2e2e2', 'text-primary': '#222222', 'text-muted': '#666666', 'border-color': '#cccccc', 'accent': '#007acc', 'accent-hover': '#005a9e', }, 'dracula': { 'bg-primary': '#282a36', 'bg-secondary': '#343746', 'bg-tertiary': '#44475a', 'bg-tertiary-hover': '#4d5066', 'text-primary': '#f8f8f2', 'text-muted': '#6272a4', 'border-color': '#44475a', 'accent': '#bd93f9', 'accent-hover': '#a672f0', }, 'tokyo_night': { 'bg-primary': '#1a1b26', 'bg-secondary': '#1f2335', 'bg-tertiary': '#283457', 'bg-tertiary-hover': '#364a73', 'text-primary': '#c0caf5', 'text-muted': '#565f89', 'border-color': '#292e42', 'accent': '#7aa2f7', 'accent-hover': '#6183bb', }, 'catppuccin_mocha': { 'bg-primary': '#1e1e2e', 'bg-secondary': '#313244', 'bg-tertiary': '#45475a', 'bg-tertiary-hover': '#585b70', 'text-primary': '#cdd6f4', 'text-muted': '#a6adc8', 'border-color': '#585b70', 'accent': '#cba6f7', 'accent-hover': '#b48ee0', }, 'ayu_dark': { 'bg-primary': '#0a0e14', 'bg-secondary': '#131721', 'bg-tertiary': '#1f2430', 'bg-tertiary-hover': '#272d3b', 'text-primary': '#b3b1ad', 'text-muted': '#626d7a', 'border-color': '#1f2430', 'accent': '#e6b450', 'accent-hover': '#ffb454', }, 'github_dark': { 'bg-primary': '#0d1117', 'bg-secondary': '#161b22', 'bg-tertiary': '#21262d', 'bg-tertiary-hover': '#30363d', 'text-primary': '#e6edf3', 'text-muted': '#8b949e', 'border-color': '#30363d', 'accent': '#2f81f7', 'accent-hover': '#1f6feb', }, 'atom_one_dark': { 'bg-primary': '#282c34', 'bg-secondary': '#2c313a', 'bg-tertiary': '#3b414d', 'bg-tertiary-hover': '#454b59', 'text-primary': '#abb2bf', 'text-muted': '#5c6370', 'border-color': '#3b414d', 'accent': '#61afef', 'accent-hover': '#528bff', }, 'houston': { 'bg-primary': '#17191e', 'bg-secondary': '#2b2d33', 'bg-tertiary': '#34363d', 'bg-tertiary-hover': '#3d3f47', 'text-primary': '#eef0f9', 'text-muted': '#79808f', 'border-color': '#2b2d33', 'accent': '#acafff', 'accent-hover': '#54b9ff', }, 'night_owl': { 'bg-primary': '#011627', 'bg-secondary': '#001122', 'bg-tertiary': '#0b2942', 'bg-tertiary-hover': '#123a5c', 'text-primary': '#d6deeb', 'text-muted': '#5f7e97', 'border-color': '#102a44', 'accent': '#82aaff', 'accent-hover': '#6690e0', }, 'nord': { 'bg-primary': '#2e3440', 'bg-secondary': '#3b4252', 'bg-tertiary': '#434c5e', 'bg-tertiary-hover': '#4c566a', 'text-primary': '#d8dee9', 'text-muted': '#4c566a', 'border-color': '#434c5e', 'accent': '#88c0d0', 'accent-hover': '#5e81ac', }, 'matcha': { 'bg-primary': '#1c2427', 'bg-secondary': '#273136', 'bg-tertiary': '#323e45', 'bg-tertiary-hover': '#3c4850', 'text-primary': '#d1ded3', 'text-muted': '#7c8885', 'border-color': '#707c4f', 'accent': '#a4b07e', 'accent-hover': '#8b966b', }, } # How each choice resolves to an actual CSS value. Keeping this server-side # (rather than duplicating the mapping in JS) means the client just applies # whatever /api/settings hands back. FONT_FAMILY_CSS = { 'system': "'Segoe UI', Tahoma, Geneva, Verdana, sans-serif", 'sans': "Arial, Helvetica, sans-serif", 'serif': "Georgia, 'Times New Roman', serif", 'monospace': "'Courier New', Consolas, monospace", # Monaspace is GitHub's monospace font superfamily, not a color theme - # falls back to Consolas/monospace if the font isn't installed locally, # same as any other web font declaration. 'monaspace': "'Monaspace Neon', 'Monaspace Argon', 'Cascadia Code', Consolas, monospace", } FONT_SIZE_CSS = {'small': '14px', 'medium': '16px', 'large': '18px', 'xlarge': '20px'} LAYOUT_WIDTH_CSS = {'normal': '1000px', 'wide': '1600px', 'full': '100%'} RADIUS_CSS = {'sharp': '2px', 'rounded': '8px', 'pill': '999px'} HEX_COLOR_RE = re.compile(r'^#[0-9a-fA-F]{6}$') def load_settings() -> Dict[str, Any]: """Load persisted display settings, filling in defaults for anything missing/invalid.""" settings = dict(DEFAULT_SETTINGS) try: if os.path.exists(SETTINGS_FILE): with open(SETTINGS_FILE, 'r') as f: saved = json.load(f) for key, value in saved.items(): if key == 'accent_color' and isinstance(value, str) and HEX_COLOR_RE.match(value): settings[key] = value elif key == 'library_path' and isinstance(value, str): # Lenient on load (directory may be transiently # unavailable at startup) - only validated on save. settings[key] = value elif key in VIDEO_SIZE_BOUNDS: if value == '' or (isinstance(value, int) and not isinstance(value, bool)): settings[key] = value elif key in SETTINGS_CHOICES and value in SETTINGS_CHOICES[key]: settings[key] = value except (json.JSONDecodeError, OSError) as e: print(f"Could not load settings, using defaults: {e}") return settings def save_settings(new_settings: Dict[str, Any]) -> Dict[str, Any]: """Validate and persist display settings; unknown/invalid keys are ignored.""" current = load_settings() for key, value in new_settings.items(): if key == 'accent_color' and isinstance(value, str) and HEX_COLOR_RE.match(value): current[key] = value elif key == 'theme' and value in SETTINGS_CHOICES['theme']: current['theme'] = value # Switching to a named preset also adopts its signature accent, # unless this same request is also setting accent_color # explicitly (in which case that wins). if 'accent_color' not in new_settings: current['accent_color'] = THEME_PALETTES[value]['accent'] elif key == 'library_path' and isinstance(value, str): value = value.strip() if value and not os.path.isdir(value): raise ValueError(f"Directory not found: {value}") current[key] = value elif key in VIDEO_SIZE_BOUNDS: if value in (None, ''): current[key] = '' continue try: num = int(value) except (TypeError, ValueError): raise ValueError(f"{key} must be a number") lo, hi = VIDEO_SIZE_BOUNDS[key] if not (lo <= num <= hi): raise ValueError(f"{key} must be between {lo} and {hi}") current[key] = num elif key in SETTINGS_CHOICES and value in SETTINGS_CHOICES[key]: current[key] = value os.makedirs(DATA_DIR, exist_ok=True) with open(SETTINGS_FILE, 'w') as f: json.dump(current, f, indent=2) if 'library_path' in new_settings: invalidate_cache() # the library root itself changed return current def settings_css_vars(settings: Dict[str, Any]) -> Dict[str, str]: """Resolve a settings dict into the actual CSS custom-property values.""" palette = THEME_PALETTES.get(settings['theme'], THEME_PALETTES['dark']) return { '--accent': settings['accent_color'], '--accent-hover': palette['accent-hover'], '--bg-primary': palette['bg-primary'], '--bg-secondary': palette['bg-secondary'], '--bg-tertiary': palette['bg-tertiary'], '--bg-tertiary-hover': palette['bg-tertiary-hover'], '--text-primary': palette['text-primary'], '--text-muted': palette['text-muted'], '--border-color': palette['border-color'], '--font-family': FONT_FAMILY_CSS[settings['font_family']], '--font-size-base': FONT_SIZE_CSS[settings['font_size']], '--container-max-width': LAYOUT_WIDTH_CSS[settings['layout_width']], '--radius': RADIUS_CSS[settings['corner_radius']], } @dataclass class Lesson: title: str path: str lesson_type: str # 'video', 'audio', 'text', 'quiz', 'mixed' video_file: Optional[str] = None audio_file: Optional[str] = None subtitle_file: Optional[str] = None text_files: List[str] = None completed: bool = False last_accessed: Optional[str] = None progress_seconds: int = 0 duration_seconds: int = 0 order: int = 0 def __post_init__(self): if self.text_files is None: self.text_files = [] @dataclass class DirectoryNode: """Represents a directory in the course structure""" name: str path: str type: str # 'directory' or 'lesson' children: Dict[str, 'DirectoryNode'] = None lessons: List[Lesson] = None completed: bool = False last_accessed: Optional[str] = None order: int = 0 has_content: bool = False # Whether this directory contains actual lesson content def __post_init__(self): if self.children is None: self.children = {} if self.lessons is None: self.lessons = [] @dataclass class Course: name: str path: str root_node: DirectoryNode progress_file: str last_accessed_path: Optional[str] = None completion_percentage: float = 0.0 def __post_init__(self): if self.root_node is None: self.root_node = DirectoryNode("", "", "directory") class DynamicCourseParser: """Enhanced parser that builds a proper directory tree structure""" @staticmethod def scan_directory(course_path: str) -> Course: """Scan directory and build dynamic tree structure""" course_path = Path(course_path) if not course_path.exists() or not course_path.is_dir(): raise ValueError(f"Invalid course path: {course_path}") course_name = course_path.name print(f"Scanning course: {course_name}") # Build the directory tree root_node = DynamicCourseParser._build_directory_tree(course_path, course_path) # Calculate completion statistics stats = DynamicCourseParser._calculate_completion_stats(root_node) progress_file = str(course_path / ".offlineu_progress.json") return Course( name=course_name, path=str(course_path), root_node=root_node, progress_file=progress_file ) @staticmethod def _build_directory_tree(course_path: Path, current_path: Path, depth: int = 0) -> DirectoryNode: """Recursively build directory tree structure""" if depth > 10: # Prevent infinite recursion return DirectoryNode(current_path.name, str(current_path), "directory") node_name = current_path.name if current_path != course_path else "Course Root" node = DirectoryNode( name=node_name, path=str(current_path), type="directory", order=depth ) try: # Get all items in current directory items = sorted(current_path.iterdir(), key=lambda x: (x.is_file(), x.name.lower())) for item in items: if item.name.startswith('.'): continue if item.is_dir(): # Recursively process subdirectory child_node = DynamicCourseParser._build_directory_tree(course_path, item, depth + 1) if child_node.has_content or child_node.children: node.children[child_node.name] = child_node node.has_content = True elif item.is_file(): # Process file as potential lesson content lesson = DynamicCourseParser._create_lesson_from_file(item, course_path) if lesson: # Add lesson directly to this node's lessons list node.lessons.append(lesson) node.has_content = True except (PermissionError, OSError) as e: print(f"Error accessing {current_path}: {e}") return node @staticmethod def _create_lesson_from_file(file_path: Path, course_path: Path) -> Optional[Lesson]: """Create a lesson from a single file""" ext = file_path.suffix.lower() filename = file_path.name.lower() # Skip non-content files if ext in {'.log', '.tmp', '.bak', '.swp', '.DS_Store', '.Thumbs.db'}: return None # Determine lesson type and files video_file = None audio_file = None subtitle_file = None text_files = [] lesson_type = 'text' # Create relative path for file serving - normalize to forward slashes relative_path = str(file_path.relative_to(course_path)).replace('\\', '/') if ext in VIDEO_EXTENSIONS: video_file = relative_path lesson_type = 'video' elif ext in AUDIO_EXTENSIONS: audio_file = relative_path lesson_type = 'audio' elif ext in SUBTITLE_EXTENSIONS: subtitle_file = relative_path return None # Don't create lessons for subtitle files alone elif ext in TEXT_EXTENSIONS: text_files.append(relative_path) if any(indicator in filename for indicator in QUIZ_INDICATORS): lesson_type = 'quiz' else: # Skip unsupported file types return None # Clean up lesson name for display display_name = DynamicCourseParser._clean_lesson_name(file_path.stem) return Lesson( title=display_name, path=str(file_path), # Store the actual file path, not just parent lesson_type=lesson_type, video_file=video_file, audio_file=audio_file, subtitle_file=subtitle_file, text_files=text_files, order=0 ) @staticmethod def _clean_lesson_name(name: str) -> str: """Clean up lesson name for display""" # Remove common patterns name = re.sub(r'^\d+[\.\-_\s]*', '', name) # Remove leading numbers name = re.sub(r'[-_]+', ' ', name) # Replace dashes/underscores with spaces name = ' '.join(word.capitalize() for word in name.split() if word) return name if name.strip() else "Untitled Lesson" @staticmethod def _calculate_completion_stats(node: DirectoryNode) -> Dict[str, Any]: """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, 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), 'total_duration_seconds': total_duration_seconds, 'remaining_seconds': max(0, total_duration_seconds - watched_seconds) } # Exposed to templates so a section header can show "X/Y watched" for any # DirectoryNode without a separate per-node data-loading pass - the course # tree is walked once per render regardless, and this method already # recurses on whatever node it's given. app.jinja_env.globals['section_stats'] = DynamicCourseParser._calculate_completion_stats def get_course_tree(course_path: str) -> Course: """ Cached DynamicCourseParser.scan_directory() - safe to cache the tree *structure* this way because progress (watched/completed) is never baked into it at scan time: ProgressTracker.apply_progress_to_tree() always re-reads the live progress file and overwrites the Lesson fields fresh on every render, same as before this cache existed. """ return cache_get_or_compute(f'course_tree:{course_path}', lambda: DynamicCourseParser.scan_directory(course_path)) def _has_direct_media(directory: Path) -> bool: """Check whether a directory contains media files directly (not recursively)""" try: for f in directory.iterdir(): if f.is_file() and f.suffix.lower() in VIDEO_EXTENSIONS | AUDIO_EXTENSIONS: return True except (PermissionError, OSError): pass return False def find_course_thumbnail(course_path: str) -> Optional[str]: """ Look for a cover image directly inside a course folder (not recursive - this only needs to catch the common 'cover.jpg next to the sections' layout, not go hunting through every subfolder). """ try: names = {f.name.lower(): f for f in Path(course_path).iterdir() if f.is_file()} except (PermissionError, OSError): return None for base in THUMBNAIL_BASENAMES: for ext in IMAGE_EXTENSIONS: match = names.get(f'{base}{ext}') if match: return str(match) return None _SECTION_NAME_RE = re.compile( # (?![a-z]) rather than \b after the keyword - \b won't fire between # "Chapter" and an immediately-following "_" since underscore counts # as a word character too (e.g. "Chapter_1-Introduction"). r'^(section|module|chapter|part|unit|lesson)(?![a-z])|^\d+[\s_.\-]', re.IGNORECASE ) def _looks_like_course(directory: Path) -> bool: """ Heuristic for 'this folder is a course, stop recursing into it': it has media files directly, or its media-holding immediate subfolders all read as chapter/section labels ("Section 1", "Module 2", "01 - Introduction", ...) rather than course titles - covers the common Section 1/, Section 2/... (or numbered-chapter) layout. Requiring *every* media-holding subfolder to match, not just one, is what keeps a publisher/category folder holding several unrelated courses - e.g. Claude/Pluralsight.X.../*.mp4 next to Claude/Linkedin.Learning.Y.../*.mp4 - from being mistaken for a single course just because more than one of its subfolders happens to store episodes directly rather than nested under their own chapter folder. Real course titles ("Pluralsight.Docker.Deep.Dive...") don't read as chapter labels, so this doesn't reopen that ambiguity. """ if _has_direct_media(directory): return True try: children = [ child for child in directory.iterdir() if child.is_dir() and not child.name.startswith('.') ] except (PermissionError, OSError): return False children_with_media = [child for child in children if _has_direct_media(child)] if not children_with_media: return False return all(_SECTION_NAME_RE.match(child.name.strip()) for child in children_with_media) HIDDEN_PATHS_FILE = os.path.join(DATA_DIR, 'hidden_paths.json') def get_hidden_paths() -> List[str]: """Load the set of course/directory paths curated out of the Library browser.""" try: if os.path.exists(HIDDEN_PATHS_FILE): with open(HIDDEN_PATHS_FILE, 'r') as f: data = json.load(f) if isinstance(data, list): return data except (json.JSONDecodeError, OSError) as e: print(f"Could not load hidden paths: {e}") return [] def set_path_hidden(path: str, hidden: bool) -> List[str]: """Add or remove a path from the hidden set; returns the updated list.""" paths = set(get_hidden_paths()) normalized = os.path.abspath(path) if hidden: paths.add(normalized) else: paths.discard(normalized) result = sorted(paths) os.makedirs(DATA_DIR, exist_ok=True) with open(HIDDEN_PATHS_FILE, 'w') as f: json.dump(result, f, indent=2) invalidate_cache() # hidden set affects which courses get_all_course_dirs() yields return result NEXT_UP_FILE = os.path.join(DATA_DIR, 'next_up.json') def get_next_up_paths() -> List[str]: """Load the ordered 'Next Up' course queue - order matters here, unlike hidden_paths, so it's a list, not a set.""" try: if os.path.exists(NEXT_UP_FILE): with open(NEXT_UP_FILE, 'r') as f: data = json.load(f) if isinstance(data, list): return data except (json.JSONDecodeError, OSError) as e: print(f"Could not load Next Up queue: {e}") return [] def _save_next_up_paths(paths: List[str]) -> None: os.makedirs(DATA_DIR, exist_ok=True) with open(NEXT_UP_FILE, 'w') as f: json.dump(paths, f, indent=2) def set_path_queued(path: str, queued: bool) -> List[str]: """Add (to the end) or remove a course from the Next Up queue; returns the updated ordered list.""" paths = get_next_up_paths() normalized = os.path.abspath(path) if queued: if normalized not in paths: paths.append(normalized) else: paths = [p for p in paths if p != normalized] _save_next_up_paths(paths) return paths def reorder_next_up(paths: List[str]) -> List[str]: """Replace the Next Up order wholesale - the client computes the new order (via up/down buttons) and posts it back.""" normalized = [os.path.abspath(p) for p in paths] _save_next_up_paths(normalized) return normalized def get_next_up_courses() -> List[Dict[str, Any]]: """Next Up queue resolved to displayable course summaries, silently dropping any path no longer on disk.""" results = [] for path in get_next_up_paths(): p = Path(path) if p.is_dir(): results.append(_course_summary(p)) return results return result def _rebase_prefix(value: str, old_abs: str, new_abs: str) -> str: """If `value` equals or is nested under `old_abs`, rewrite that prefix to `new_abs`.""" if value == old_abs: return new_abs if value.startswith(old_abs + os.sep): return new_abs + value[len(old_abs):] return value def rebase_library_path(old_abs: str, new_abs: str) -> None: """ After a directory in the library is renamed/moved on disk, rewrite any stored absolute paths that pointed inside it (hidden_paths.json, recent_views.json's course_path) so curation and Recently Viewed don't silently go stale. Lesson-level progress needs no rebasing - it lives inside the directory itself, keyed by paths relative to it, so it moves with the rename automatically. """ hidden = get_hidden_paths() rebased_hidden = [_rebase_prefix(p, old_abs, new_abs) for p in hidden] if rebased_hidden != hidden: os.makedirs(DATA_DIR, exist_ok=True) with open(HIDDEN_PATHS_FILE, 'w') as f: json.dump(sorted(set(rebased_hidden)), f, indent=2) views = get_recent_views() changed = False for entry in views: course_path = entry.get('course_path', '') rebased = _rebase_prefix(course_path, old_abs, new_abs) if rebased != course_path: entry['course_path'] = rebased changed = True if changed: try: os.makedirs(DATA_DIR, exist_ok=True) with open(RECENT_VIEWS_FILE, 'w') as f: json.dump(views, f, indent=2) except OSError as e: print(f"Could not rebase recent views after rename: {e}") def _contains_visible_course(directory: Path, hidden_set: set) -> bool: """ Whether `directory` leads to at least one course that isn't curated out - directly, or via a hidden ancestor folder within this subtree. Used by the normal (skip_hidden=True) Library browser so a folder whose every course has been individually hidden doesn't still show up as a drillable directory that dead-ends empty once opened. Hiding a folder hides everything inside it, so a hidden folder short-circuits the walk rather than counting anything beneath it as visible. """ if os.path.abspath(str(directory)) in hidden_set: return False if _looks_like_course(directory): return True try: children = [ c for c in directory.iterdir() if c.is_dir() and not c.name.startswith('.') ] except (PermissionError, OSError): return False return any(_contains_visible_course(child, hidden_set) for child in children) def list_library_directory(dir_path: str, skip_hidden: bool = True) -> Dict[str, Any]: """ List only the immediate children of dir_path for the lazy-loading Library browser: subdirectories to drill into, and course folders to load directly - without recursively scanning the whole tree. This is what lets the UI start collapsed at the top level and expand on demand instead of rendering hundreds of courses at once. Directories that don't lead to any course anywhere inside them are left out entirely, so drilling down never dead-ends on an empty folder. Items the user has curated out (see get_hidden_paths) are excluded when skip_hidden=True (normal browsing). When skip_hidden=False, they are included but flagged with 'hidden': True instead - used by the Settings-page curation UI, which needs to show hidden items so they can be un-hidden. """ directory = Path(dir_path) items: List[Dict[str, Any]] = [] hidden_set = set(get_hidden_paths()) if not directory.exists() or not directory.is_dir(): return {'items': items, 'errors': [f"Directory not found: {dir_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) as e: return {'items': items, 'errors': [f"Could not read {dir_path}: {e}"]} for entry in entries: entry_path_str = str(entry) is_hidden = os.path.abspath(entry_path_str) in hidden_set if skip_hidden and is_hidden: continue if _looks_like_course(entry): item = _course_summary(entry) item['hidden'] = is_hidden items.append(item) else: if skip_hidden: # Only count courses that aren't themselves curated out - # otherwise a folder whose courses are all hidden # individually would still show up, empty, once opened. has_course_inside = _contains_visible_course(entry, hidden_set) else: has_course_inside = any( f.is_file() and f.suffix.lower() in VIDEO_EXTENSIONS | AUDIO_EXTENSIONS for f in entry.rglob('*') ) if has_course_inside: items.append({ 'type': 'directory', 'name': entry.name, 'path': entry_path_str, 'hidden': is_hidden }) return {'items': items, 'errors': []} def get_library_root() -> str: """ The effective root the Library browser scans: the persisted 'library_path' setting if it's set and exists, otherwise the COURSES_LIBRARY_PATH/--library-path default. """ settings = load_settings() override = (settings.get('library_path') or '').strip() if override and os.path.isdir(override): return override return LIBRARY_PATH def iter_all_courses(dir_path: str) -> Iterator[Path]: """ 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. """ hidden_set = set(get_hidden_paths()) def walk(directory: Path) -> Iterator[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: if os.path.abspath(str(entry)) in hidden_set: continue if _looks_like_course(entry): yield entry else: yield from walk(entry) yield from walk(Path(dir_path)) def get_all_course_dirs() -> List[Path]: """ Cached, materialized iter_all_courses(get_library_root()) - every caller (search, transcript search, Recently Added, library stats, Notes Hub, backup export) needs the exact same recursive directory walk over the library root, so compute it once and share it instead of every caller re-walking the filesystem independently. """ return cache_get_or_compute('course_dirs', lambda: list(iter_all_courses(get_library_root()))) 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.""" def compute(): 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 } return dict(cache_get_or_compute(f'course_summary:{course_dir}', compute)) def search_library_courses(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 get_all_course_dirs. """ query_lower = query.lower() return [ _course_summary(course) for course in get_all_course_dirs() if query_lower in course.name.lower() ] def _extract_subtitle_snippet(text: str, query_lower: str, context_chars: int = 80) -> str: """A short excerpt around the first match, with subtitle sequence-number/timestamp lines stripped.""" lower = text.lower() idx = lower.find(query_lower) if idx == -1: return '' start = max(0, idx - context_chars) end = min(len(text), idx + len(query_lower) + context_chars) excerpt_lines = text[start:end].splitlines() cleaned = [ line.strip() for line in excerpt_lines if line.strip() and '-->' not in line and not line.strip().isdigit() and line.strip().upper() != 'WEBVTT' ] snippet = ' '.join(cleaned) return ('…' if start > 0 else '') + snippet + ('…' if end < len(text) else '') def _course_subtitle_index(course_dir: Path) -> List[Tuple[Path, str, str]]: """ Cached (subtitle_file, text, text_lower) triples for one course - reading and lowercasing every subtitle file is identical work across every search query, so do it once per cache window and just search the already-read text in memory for each new query instead of re-reading every file from disk (typically a network-mounted NAS) every time. """ def compute(): pairs = [] for f in course_dir.rglob('*'): if f.is_file() and f.suffix.lower() in SUBTITLE_EXTENSIONS: try: text = f.read_text(encoding='utf-8', errors='ignore') except OSError: continue pairs.append((f, text, text.lower())) return pairs return cache_get_or_compute(f'subtitle_index:{course_dir}', compute) def search_transcripts(query: str, limit: int = 30) -> List[Dict[str, Any]]: """ Search inside lesson subtitle files (.srt/.vtt/etc.) for `query`, case-insensitive - see _course_subtitle_index for how the expensive per-course file reads are cached and shared across queries. Subtitle files aren't wired into the Lesson tree today (see DynamicCourseParser._create_lesson_from_file - a subtitle file never becomes part of a Lesson, so lesson.subtitle_file is always empty). Rather than depend on that, this matches a subtitle file to its lesson by finding a same-named video/audio file next to it, which is how these files are conventionally paired on disk regardless. """ query_lower = query.lower() results: List[Dict[str, Any]] = [] for course_dir in get_all_course_dirs(): for subtitle_file, text, text_lower in _course_subtitle_index(course_dir): if len(results) >= limit: return results if query_lower not in text_lower: continue media_match = next( (f for f in subtitle_file.parent.iterdir() if f.is_file() and f.stem == subtitle_file.stem and f.suffix.lower() in VIDEO_EXTENSIONS | AUDIO_EXTENSIONS), None ) if not media_match: continue # no lesson to navigate to - skip rather than dead-end lesson_title = DynamicCourseParser._clean_lesson_name(media_match.stem) lesson_relative = media_match.relative_to(course_dir).as_posix() results.append({ 'course_path': str(course_dir), 'course_name': course_dir.name, 'lesson_path': f"{lesson_relative}/{lesson_title.replace(' ', '_')}", 'lesson_title': lesson_title, 'snippet': _extract_subtitle_snippet(text, query_lower) }) return results 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). """ dated = [] for course_dir in get_all_course_dirs(): 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 _scan_library_activity() -> Dict[str, Any]: """ One walk over every course's progress file, computing everything the dashboard's stats card / stale-course nudges / activity heatmap need - library_stats/stale_courses/activity_heatmap in index() all derive from a single call to this rather than each re-scanning every course's progress.json independently. Reads each course's small progress JSON file directly, not by 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() activity_by_date: Dict[str, int] = {} courses: List[Dict[str, Any]] = [] heatmap_cutoff = datetime.now().date() - timedelta(days=89) for course_dir in get_all_course_dirs(): total_courses += 1 try: with open(course_dir / '.offlineu_progress.json', 'r') as f: progress = json.load(f) except (FileNotFoundError, json.JSONDecodeError, OSError): continue course_total = 0 course_completed = 0 course_last_activity: Optional[datetime] = None for key, entry in progress.items(): if key == 'last_accessed_path' or not isinstance(entry, dict): continue lessons_tracked += 1 course_total += 1 if entry.get('completed'): completed_lessons += 1 course_completed += 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 not last_accessed: continue try: accessed_dt = datetime.fromisoformat(last_accessed) except ValueError: continue accessed_date = accessed_dt.date() active_dates.add(accessed_date) if course_last_activity is None or accessed_dt > course_last_activity: course_last_activity = accessed_dt if accessed_date >= heatmap_cutoff: iso = accessed_date.isoformat() activity_by_date[iso] = activity_by_date.get(iso, 0) + 1 if course_total > 0: courses.append({ 'path': str(course_dir), 'total': course_total, 'completed': course_completed, 'last_activity': course_last_activity }) 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_seconds': watched_seconds, 'streak_days': streak_days, 'activity_by_date': activity_by_date, 'courses': courses, } def get_random_incomplete_course() -> Optional[Path]: """A random course for the dashboard's "Surprise Me" pick - excludes courses that are already fully watched where possible, falling back to the whole library if everything's done.""" all_dirs = get_all_course_dirs() if not all_dirs: return None scan = _scan_library_activity() fully_completed = {c['path'] for c in scan['courses'] if c['total'] > 0 and c['completed'] >= c['total']} candidates = [d for d in all_dirs if str(d) not in fully_completed] return random.choice(candidates or all_dirs) def format_library_stats(scan: Dict[str, Any]) -> Dict[str, Any]: """Library-wide overview for the dashboard's stats card, from a _scan_library_activity() result.""" return { 'total_courses': scan['total_courses'], 'lessons_tracked': scan['lessons_tracked'], 'completed_lessons': scan['completed_lessons'], 'watched_display': format_duration(scan['watched_seconds']), 'streak_days': scan['streak_days'] } def format_stale_courses(scan: Dict[str, Any], threshold_days: int = 14, limit: int = 5) -> List[Dict[str, Any]]: """ Courses with some progress that haven't been touched in a while, oldest first - courses with zero progress (never started) and fully completed ones are both excluded, since neither is something to "pick back up." """ cutoff = datetime.now() - timedelta(days=threshold_days) candidates = [ c for c in scan['courses'] if c['completed'] < c['total'] and c['last_activity'] and c['last_activity'] < cutoff ] candidates.sort(key=lambda c: c['last_activity']) results = [] for c in candidates[:limit]: item = _course_summary(Path(c['path'])) days_ago = (datetime.now() - c['last_activity']).days item['last_touched_display'] = f"{days_ago} day{'s' if days_ago != 1 else ''} ago" results.append(item) return results def format_activity_heatmap(scan: Dict[str, Any]) -> List[Dict[str, Any]]: """ The last 90 days as a flat list (oldest first) for the dashboard's contribution-style calendar, each with an ISO date and that day's activity count. """ activity_by_date = scan['activity_by_date'] today = datetime.now().date() return [ {'date': (today - timedelta(days=offset)).isoformat(), 'count': activity_by_date.get((today - timedelta(days=offset)).isoformat(), 0)} for offset in range(89, -1, -1) ] # ---- Outline integration ---- # Deliberately kept separate from DEFAULT_SETTINGS/load_settings/save_settings: # those all flow through GET /api/settings, which theme.js fetches on every # page load - not where an API token should ever ride along. OUTLINE_CONFIG_FILE = os.path.join(DATA_DIR, 'outline_config.json') def load_outline_config() -> Dict[str, str]: """Load the Outline base URL + API token + default collection. Never exposed via GET beyond 'configured'.""" try: if os.path.exists(OUTLINE_CONFIG_FILE): with open(OUTLINE_CONFIG_FILE, 'r') as f: data = json.load(f) if isinstance(data, dict): return { 'base_url': data.get('base_url', ''), 'api_token': data.get('api_token', ''), 'default_collection_id': data.get('default_collection_id', ''), 'default_collection_name': data.get('default_collection_name', '') } except (json.JSONDecodeError, OSError) as e: print(f"Could not load Outline config: {e}") return {'base_url': '', 'api_token': '', 'default_collection_id': '', 'default_collection_name': ''} def save_outline_config(base_url: str, api_token: str, default_collection_id: Optional[str] = None, default_collection_name: Optional[str] = None) -> None: """ Save the Outline base URL / API token / default collection. A blank api_token keeps the previously stored one, so the URL can be updated without re-pasting it; default_collection_id/name are only touched when explicitly passed (None means 'leave as-is'), so saving the URL/token doesn't clear an already-chosen collection. """ current = load_outline_config() current['base_url'] = base_url.rstrip('/') if api_token: current['api_token'] = api_token if default_collection_id is not None: current['default_collection_id'] = default_collection_id if default_collection_name is not None: current['default_collection_name'] = default_collection_name os.makedirs(DATA_DIR, exist_ok=True) with open(OUTLINE_CONFIG_FILE, 'w') as f: json.dump(current, f, indent=2) def _outline_request(path: str, payload: Dict[str, Any]) -> Dict[str, Any]: """ POST to the Outline API (every Outline endpoint is POST, even 'list' ones) with the stored token. Raises RuntimeError with a readable message on any failure - not configured, unreachable, or a non-2xx response. """ config = load_outline_config() if not config['base_url'] or not config['api_token']: raise RuntimeError('Outline is not configured') url = f"{config['base_url']}/api/{path}" body = json.dumps(payload).encode('utf-8') req = urllib.request.Request(url, data=body, method='POST', headers={ 'Authorization': f"Bearer {config['api_token']}", 'Content-Type': 'application/json', 'Accept': 'application/json', }) try: with urllib.request.urlopen(req, timeout=15) as resp: return json.loads(resp.read().decode('utf-8')) except urllib.error.HTTPError as e: detail = e.read().decode('utf-8', errors='replace') raise RuntimeError(f'Outline returned {e.code}: {detail[:200]}') except urllib.error.URLError as e: raise RuntimeError(f'Could not reach Outline: {e.reason}') def list_outline_topics() -> List[Dict[str, str]]: """ Top-level documents in the configured default collection - each one is a "topic" a lesson note can be filed under. Fetches every document in the collection and filters to parentDocumentId is None locally, rather than relying on documents.list's collectionId/parentDocumentId filter params directly - both are marked deprecated in Outline's own API spec and their exact recursive-vs-top-level behavior isn't documented, so filtering the response ourselves is the version that can't be wrong. """ config = load_outline_config() collection_id = config['default_collection_id'] if not collection_id: raise RuntimeError('No default Outline collection set - pick one in Settings first') result = _outline_request('documents.list', {'collectionId': collection_id, 'limit': 100}) return [ {'id': d['id'], 'name': d['title']} for d in result.get('data', []) if not d.get('parentDocumentId') ] def resolve_outline_topic(name: str) -> Dict[str, str]: """ Find an existing topic document (a top-level document in the configured default collection) with this exact title, or create one. Matching by name first (rather than always creating) means naming a "new" topic that happens to match one you already made doesn't spawn a duplicate. """ for topic in list_outline_topics(): if topic['name'] == name: return topic config = load_outline_config() created = _outline_request('documents.create', { 'title': name, 'text': '', 'collectionId': config['default_collection_id'], 'publish': True }) return {'id': created['data']['id'], 'name': name} def push_note_to_outline(course: Course, lesson_path: str, lesson_title: str, topic_id: str, new_topic_name: str) -> Dict[str, Any]: """ Push a lesson's saved note to Outline as a child document nested under the chosen topic document (which itself lives in the configured default collection - see list_outline_topics). Creates the note document on the first push and updates the same one (by the id stashed in the progress file) on every push after that. new_topic_name is a fallback for a topic that somehow never got resolved client-side (see resolve_outline_topic) - the normal path resolves a brand-new topic name to a real document id as soon as it's typed (POST /api/outline/resolve-topic), specifically so this fire-and-forget, can't-read-the-response push never has to decide "is this actually a new topic" on its own. A page that fires pagehide more than once for the same view (bfcache restore, for instance) would otherwise re-send the same "new topic" intent every time and create a duplicate topic document per navigation. """ config = load_outline_config() if not config['default_collection_id']: return {'success': False, 'error': 'No default Outline collection set - pick one in Settings first'} progress = ProgressTracker.load_progress(course) entry = progress.get(lesson_path, {}) notes = ProgressTracker._notes_from_entry(entry) if not notes: return {'success': False, 'error': 'No note to push'} note = "\n\n".join( f"**[{format_timestamp(n.get('timestamp_seconds'))}]** {n.get('text', '')}" if n.get('timestamp_seconds') is not None else n.get('text', '') for n in notes ) if not topic_id: if not new_topic_name: return {'success': False, 'error': 'No topic selected'} topic_id = resolve_outline_topic(new_topic_name)['id'] ProgressTracker.set_lesson_outline_topic(course, lesson_path, topic_id, '') title = f"{lesson_title} — {course.name}" document_id = entry.get('outline_document_id') if document_id: _outline_request('documents.update', {'id': document_id, 'title': title, 'text': note}) else: created = _outline_request('documents.create', { 'title': title, 'text': note, 'collectionId': config['default_collection_id'], 'parentDocumentId': topic_id, 'publish': True }) document_id = created['data']['id'] ProgressTracker.set_lesson_outline_document_id(course, lesson_path, document_id) return {'success': True, 'document_id': document_id, 'topic_id': topic_id} RECENT_VIEWS_FILE = os.path.join(DATA_DIR, 'recent_views.json') MAX_RECENT_VIEWS = 20 def record_recent_view(course_name: str, course_path: str, lesson_path: str, lesson_title: str) -> None: """ Record a lesson view for the cross-course 'Recently Viewed' list on the dashboard, so jumping back to where you left off doesn't require re-browsing the library - even for a different course than whichever one happens to be loaded right now. """ entries = get_recent_views() # Drop any existing entry for this exact lesson so it moves to the # front instead of appearing twice. entries = [ e for e in entries if not (e.get('course_path') == course_path and e.get('lesson_path') == lesson_path) ] entries.insert(0, { 'course_name': course_name, 'course_path': course_path, 'lesson_path': lesson_path, 'lesson_title': lesson_title, 'viewed_at': datetime.now().isoformat() }) entries = entries[:MAX_RECENT_VIEWS] try: os.makedirs(DATA_DIR, exist_ok=True) with open(RECENT_VIEWS_FILE, 'w') as f: json.dump(entries, f, indent=2) except OSError as e: print(f"Could not save recent views: {e}") def get_recent_views() -> List[Dict[str, Any]]: """Load the persisted 'Recently Viewed' list, newest first.""" try: if os.path.exists(RECENT_VIEWS_FILE): with open(RECENT_VIEWS_FILE, 'r') as f: return json.load(f) except (json.JSONDecodeError, OSError) as e: print(f"Could not load recent views: {e}") return [] def _read_lesson_progress_entry(course_path: str, lesson_path: str) -> Dict[str, Any]: """ Read a single lesson's progress entry directly from its course's own progress file, without needing that course to be the currently loaded one. Used to show watch progress for 'Recently Viewed' entries that may belong to a different course than whatever's active right now. """ if not course_path or not lesson_path: return {} progress_file = os.path.join(course_path, '.offlineu_progress.json') try: with open(progress_file, 'r') as f: progress = json.load(f) return progress.get(lesson_path, {}) except (FileNotFoundError, json.JSONDecodeError, OSError): return {} def get_recent_views_for_display() -> List[Dict[str, Any]]: """Recent views with a human-friendly timestamp and watch progress added.""" entries = get_recent_views() for entry in entries: try: dt = datetime.fromisoformat(entry['viewed_at']) entry['viewed_display'] = dt.strftime('%b %d, %I:%M %p').replace(' 0', ' ') except (KeyError, ValueError): entry['viewed_display'] = '' lesson_progress = _read_lesson_progress_entry(entry.get('course_path', ''), entry.get('lesson_path', '')) completed = lesson_progress.get('completed', False) progress_seconds = lesson_progress.get('progress_seconds', 0) duration_seconds = lesson_progress.get('duration_seconds', 0) entry['completed'] = completed if completed: entry['percent_watched'] = 100 elif duration_seconds: entry['percent_watched'] = max(0, min(100, round(100 * progress_seconds / duration_seconds))) else: entry['percent_watched'] = 0 entry['has_thumbnail'] = find_course_thumbnail(entry.get('course_path', '')) is not None return entries def _resolve_lesson_progress_key(course: Course, lesson: Lesson, progress: Dict[str, Any]) -> Optional[str]: """ Match a Lesson to its progress-file key. Lessons have historically been keyed two ways - by their relative file path alone, or with the lesson's title suffix appended (see get_lesson_url) - so check both rather than assuming one. """ lesson_path = os.path.relpath(lesson.path, course.path).replace('\\', '/') if lesson_path.startswith('/'): lesson_path = lesson_path[1:] if lesson_path in progress: return lesson_path lesson_path_with_title = f"{lesson_path}/{lesson.title.replace(' ', '_')}" if lesson_path_with_title in progress: return lesson_path_with_title return None class ProgressTracker: """Handles progress tracking and persistence""" @staticmethod def load_progress(course: Course) -> Dict[str, Any]: """Load progress from JSON file""" try: with open(course.progress_file, 'r') as f: return json.load(f) except (FileNotFoundError, json.JSONDecodeError): return {} @staticmethod def save_progress(course: Course, progress_data: Dict[str, Any]): """Save progress to JSON file""" try: with open(course.progress_file, 'w') as f: json.dump(progress_data, f, indent=2) except Exception as e: print(f"Error saving progress: {e}") @staticmethod def update_lesson_progress(course: Course, lesson_path: str, completed: bool = False, progress_seconds: int = 0, duration_seconds: Optional[int] = None): """Update progress for specific lesson by path""" progress = ProgressTracker.load_progress(course) existing = progress.get(lesson_path, {}) entry = { 'completed': completed, 'progress_seconds': progress_seconds, 'last_accessed': datetime.now().isoformat() } # Preserve a previously-known duration if this particular save # didn't report one, rather than clobbering it back to unknown. if duration_seconds: entry['duration_seconds'] = duration_seconds elif existing.get('duration_seconds'): entry['duration_seconds'] = existing['duration_seconds'] # Same for notes - this call has no opinion on them, so don't let a # routine playback-progress save wipe them out. if existing.get('notes'): entry['notes'] = existing['notes'] elif existing.get('note'): entry['note'] = existing['note'] progress[lesson_path] = entry # Update last accessed path progress['last_accessed_path'] = lesson_path ProgressTracker.save_progress(course, progress) @staticmethod def touch_lesson_accessed(course: Course, lesson_path: str): """ Record that a lesson was opened, without touching its saved completed/progress_seconds/duration_seconds - update_lesson_progress is for the client reporting real playback progress, and calling it (with its completed=False, progress_seconds=0 defaults) just for viewing a page would silently reset an already-watched lesson back to 0% every time it's opened, before the client gets a chance to report anything real. """ progress = ProgressTracker.load_progress(course) entry = progress.setdefault(lesson_path, {}) entry['last_accessed'] = datetime.now().isoformat() progress['last_accessed_path'] = lesson_path ProgressTracker.save_progress(course, progress) @staticmethod def _notes_from_entry(entry: Dict[str, Any]) -> List[Dict[str, Any]]: """ This entry's timestamped notes, transparently upgrading a legacy single-string `note` field into a one-item list so every caller can treat notes as a list without caring which shape is on disk. """ if entry.get('notes'): return entry['notes'] legacy = entry.get('note') if legacy: return [{ 'id': 'legacy', 'timestamp_seconds': None, 'text': legacy, 'created_at': entry.get('last_accessed', '') }] return [] @staticmethod def get_lesson_notes(course: Course, lesson_path: str) -> List[Dict[str, Any]]: """A lesson's timestamped notes, newest-shape-or-migrated-legacy.""" progress = ProgressTracker.load_progress(course) return ProgressTracker._notes_from_entry(progress.get(lesson_path, {})) @staticmethod def add_lesson_note(course: Course, lesson_path: str, text: str, timestamp_seconds: Optional[int]) -> List[Dict[str, Any]]: """Append a new timestamped note, migrating a legacy single-note entry to the list shape if needed.""" progress = ProgressTracker.load_progress(course) entry = progress.setdefault(lesson_path, {}) notes = ProgressTracker._notes_from_entry(entry) notes.append({ 'id': uuid.uuid4().hex[:8], 'text': text, 'timestamp_seconds': timestamp_seconds, 'created_at': datetime.now().isoformat() }) entry['notes'] = notes entry.pop('note', None) ProgressTracker.save_progress(course, progress) return notes @staticmethod def update_lesson_note_text(course: Course, lesson_path: str, note_id: str, text: str) -> List[Dict[str, Any]]: """Edit one timestamped note's text in place.""" progress = ProgressTracker.load_progress(course) entry = progress.setdefault(lesson_path, {}) notes = ProgressTracker._notes_from_entry(entry) for note in notes: if note.get('id') == note_id: note['text'] = text break entry['notes'] = notes entry.pop('note', None) ProgressTracker.save_progress(course, progress) return notes @staticmethod def delete_lesson_note(course: Course, lesson_path: str, note_id: str) -> List[Dict[str, Any]]: """Remove one timestamped note.""" progress = ProgressTracker.load_progress(course) entry = progress.setdefault(lesson_path, {}) notes = [n for n in ProgressTracker._notes_from_entry(entry) if n.get('id') != note_id] entry['notes'] = notes entry.pop('note', None) ProgressTracker.save_progress(course, progress) return notes @staticmethod def set_lesson_outline_topic(course: Course, lesson_path: str, topic_id: str, topic_name: str): """ Remember which Outline topic (collection) a lesson's note should push to - an existing collection id, or a not-yet-created topic name (see push_note_to_outline). Clearing the pick (both blank) removes it so pagehide stops firing a push for this lesson. """ progress = ProgressTracker.load_progress(course) entry = progress.setdefault(lesson_path, {}) if topic_id: entry['outline_topic_id'] = topic_id entry.pop('outline_topic_name', None) elif topic_name: entry['outline_topic_name'] = topic_name entry.pop('outline_topic_id', None) else: entry.pop('outline_topic_id', None) entry.pop('outline_topic_name', None) ProgressTracker.save_progress(course, progress) @staticmethod def set_lesson_outline_document_id(course: Course, lesson_path: str, document_id: str): """Remember the Outline document a lesson's note was pushed to, so the next push updates it instead of creating a duplicate.""" progress = ProgressTracker.load_progress(course) entry = progress.setdefault(lesson_path, {}) entry['outline_document_id'] = document_id ProgressTracker.save_progress(course, progress) @staticmethod def mark_all_completed(course: Course): """Mark every lesson in the course as completed, in one save.""" progress = ProgressTracker.load_progress(course) def mark_node(node: DirectoryNode): for lesson in node.lessons: lesson_path = os.path.relpath(lesson.path, course.path).replace('\\', '/') if lesson_path.startswith('/'): lesson_path = lesson_path[1:] entry = progress.setdefault(lesson_path, {}) entry['completed'] = True entry['last_accessed'] = datetime.now().isoformat() entry.setdefault('progress_seconds', entry.get('duration_seconds', 0)) for child in node.children.values(): mark_node(child) mark_node(course.root_node) ProgressTracker.save_progress(course, progress) @staticmethod def apply_progress_to_tree(course: Course): """Apply saved progress to the course tree""" progress = ProgressTracker.load_progress(course) def apply_to_node(node: DirectoryNode): # Apply progress to lessons in this node for lesson in node.lessons: key = _resolve_lesson_progress_key(course, lesson, progress) if key: entry = progress[key] lesson.completed = entry.get('completed', False) lesson.last_accessed = entry.get('last_accessed') lesson.progress_seconds = entry.get('progress_seconds', 0) lesson.duration_seconds = entry.get('duration_seconds', 0) # Recursively apply to children for child in node.children.values(): apply_to_node(child) apply_to_node(course.root_node) course.last_accessed_path = progress.get('last_accessed_path') @staticmethod def get_completion_stats(course: Course) -> Dict[str, Any]: """Calculate completion statistics""" return DynamicCourseParser._calculate_completion_stats(course.root_node) def course_has_any_notes(course: Course) -> bool: """Whether any lesson in the course has a saved note - used to decide whether to offer a study-guide download.""" progress = ProgressTracker.load_progress(course) return any( isinstance(entry, dict) and ProgressTracker._notes_from_entry(entry) for key, entry in progress.items() if key != 'last_accessed_path' ) def format_timestamp(seconds: Optional[float]) -> Optional[str]: """Render a note's captured playback position as "MM:SS", or None if it wasn't timestamped.""" if seconds is None: return None seconds = int(seconds) return f"{seconds // 60}:{seconds % 60:02d}" def get_all_notes() -> List[Dict[str, Any]]: """ Every timestamped note across the whole library, newest first - a note is otherwise only visible from its own lesson page (or Outline, once pushed), so this is the one place to see everything you've written. A lesson with several notes contributes one row per note. """ notes = [] for course_dir in get_all_course_dirs(): try: with open(course_dir / '.offlineu_progress.json', 'r') as f: progress = json.load(f) except (FileNotFoundError, json.JSONDecodeError, OSError): continue for lesson_path, entry in progress.items(): if lesson_path == 'last_accessed_path' or not isinstance(entry, dict): continue for note in ProgressTracker._notes_from_entry(entry): created_at = note.get('created_at') or entry.get('last_accessed', '') notes.append({ 'course_path': str(course_dir), 'course_name': course_dir.name, 'lesson_path': lesson_path, 'lesson_title': lesson_path.rsplit('/', 1)[-1].replace('_', ' '), 'note_id': note.get('id'), 'text': note.get('text', ''), 'timestamp_seconds': note.get('timestamp_seconds'), 'timestamp_label': format_timestamp(note.get('timestamp_seconds')), 'created_at': created_at, 'pushed_to_outline': bool(entry.get('outline_document_id')) }) notes.sort(key=lambda n: n['created_at'], reverse=True) return notes def build_study_guide_markdown(course: Course) -> str: """ Compile every note written for a course into one markdown document, following the course's own section/lesson structure - only sections and lessons that actually have a note appear; everything else is skipped rather than padding the guide with empty headings. """ progress = ProgressTracker.load_progress(course) def node_has_notes(node: DirectoryNode) -> bool: for lesson in node.lessons: key = _resolve_lesson_progress_key(course, lesson, progress) if key and ProgressTracker._notes_from_entry(progress[key]): return True return any(node_has_notes(child) for child in node.children.values()) lines = [f"# {course.name}", ""] def walk(node: DirectoryNode, heading_level: int): if node.name and node.name != "Course Root": if not node_has_notes(node): return lines.append(f"{'#' * min(heading_level, 6)} {node.name}") lines.append("") for lesson in node.lessons: key = _resolve_lesson_progress_key(course, lesson, progress) lesson_notes = ProgressTracker._notes_from_entry(progress[key]) if key else [] if not lesson_notes: continue lines.append(f"{'#' * min(heading_level + 1, 6)} {lesson.title}") lines.append("") for note in lesson_notes: ts = format_timestamp(note.get('timestamp_seconds')) prefix = f"**[{ts}]** " if ts else "" lines.append(f"- {prefix}{note.get('text', '')}") lines.append("") for child in node.children.values(): walk(child, heading_level + 1) walk(course.root_node, 2) return "\n".join(lines) # Global course storage current_course = None def _split_continue_watching(all_views: List[Dict[str, Any]]) -> Tuple[List[Dict[str, Any]], List[Dict[str, Any]]]: """ Split the recent-views list (already carrying percent_watched/completed, see get_recent_views_for_display) into 'Continue Watching' - genuinely in-progress lessons - and the remaining 'Recently Viewed' entries, each capped at 5 for the dashboard. Reuses the same underlying history rather than a separate library-wide progress index (see plan notes). """ continue_watching = [v for v in all_views if 0 < v['percent_watched'] < 100][:5] continue_ids = {(v.get('course_path'), v.get('lesson_path')) for v in continue_watching} recent_views = [ v for v in all_views if (v.get('course_path'), v.get('lesson_path')) not in continue_ids ][:5] 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""" global current_course continue_watching, recent_views = _split_continue_watching(get_recent_views_for_display()) if current_course is None: # One scan covers stats/stale-courses/heatmap - see _scan_library_activity. scan = _scan_library_activity() return render_template('course_dashboard.html', course=None, stats={'total_lessons': 0, 'completed_lessons': 0, 'completion_percentage': 0}, continue_watching=continue_watching, recent_views=recent_views, recently_added=get_recently_added_courses(), library_stats=format_library_stats(scan), stale_courses=format_stale_courses(scan), activity_heatmap=format_activity_heatmap(scan), next_up=get_next_up_courses()) # 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']) resume_lesson = None if current_course.last_accessed_path: lesson, _ = find_lesson_in_tree(current_course.root_node, current_course.last_accessed_path) if lesson: resume_lesson = {'title': lesson.title, 'url': get_lesson_url(lesson, current_course.path)} return render_template('course_dashboard.html', course=current_course, stats=stats, continue_watching=continue_watching, recent_views=recent_views, has_note=bool(course_has_any_notes(current_course)), is_queued=os.path.abspath(current_course.path) in get_next_up_paths(), resume_lesson=resume_lesson) @app.route('/browse') def browse_directories(): """Browse directories for course selection""" path = request.args.get('path', '') try: # If no path specified, start with available drives on Windows if not path: import platform if platform.system() == 'Windows': # Get available drives on Windows import string drives = [] for letter in string.ascii_uppercase: drive = f"{letter}:\\" if os.path.exists(drive): drives.append({ 'name': f"Drive {letter}:", 'path': drive, 'media_files': 0, 'is_course_candidate': False }) print(f"Returning {len(drives)} drives") return jsonify({ 'current_path': 'Select a Drive', 'parent_path': None, 'directories': drives }) else: # On other systems, start from home directory path = str(Path.home()) current_path = Path(path) if not current_path.exists() or not current_path.is_dir(): # Fallback to home directory if path doesn't exist current_path = Path.home() print(f"Browsing directory: {current_path}") # Get directories and basic info directories = [] try: for item in sorted(current_path.iterdir()): if item.is_dir() and not item.name.startswith('.'): try: # Check if this looks like a course directory media_count = len([f for f in item.rglob('*') if f.suffix.lower() in VIDEO_EXTENSIONS | AUDIO_EXTENSIONS]) directories.append({ 'name': item.name, 'path': str(item), 'media_files': media_count, 'is_course_candidate': media_count > 0 }) except (PermissionError, OSError): directories.append({ 'name': item.name + " (Access Denied)", 'path': str(item), 'media_files': 0, 'is_course_candidate': False }) except (PermissionError, OSError) as e: print(f"Access denied to {current_path}: {str(e)}") return jsonify({'error': f'Access denied to {current_path}: {str(e)}'}), 403 # Determine parent path parent = None if current_path.parent != current_path: try: parent = str(current_path.parent) except (PermissionError, OSError): pass print(f"Found {len(directories)} directories") return jsonify({ 'current_path': str(current_path), 'parent_path': parent, 'directories': directories }) except Exception as e: print(f"Error in browse_directories: {str(e)}") return jsonify({'error': str(e)}), 500 @app.route('/library') def browse_library(): """ Lazily list one directory level of the courses library at a time, so the UI can start collapsed at the top level and drill down on click instead of scanning/rendering the whole tree upfront. Defaults to the effective library root (persisted setting, or the COURSES_LIBRARY_PATH/--library-path default); the 'path' query param lets the browser descend, but is restricted to stay within that root. """ library_root = os.path.abspath(get_library_root()) requested_path = request.args.get('path', library_root) target_path = os.path.abspath(requested_path) if not (target_path == library_root or target_path.startswith(library_root + os.sep)): return jsonify({'error': 'Path outside library root', 'items': [], 'errors': ['Access denied']}), 403 result = list_library_directory(target_path) return jsonify({ 'library_path': library_root, 'current_path': target_path, 'items': result['items'], 'errors': result['errors'] }) @app.route('/library/manage') def browse_library_manage(): """ Same as /library, but includes items the user has hidden (flagged 'hidden': true rather than filtered out), so the Settings-page curation UI can browse the whole tree and toggle visibility. """ library_root = os.path.abspath(get_library_root()) requested_path = request.args.get('path', library_root) target_path = os.path.abspath(requested_path) if not (target_path == library_root or target_path.startswith(library_root + os.sep)): return jsonify({'error': 'Path outside library root', 'items': [], 'errors': ['Access denied']}), 403 result = list_library_directory(target_path, skip_hidden=False) return jsonify({ 'library_path': library_root, 'current_path': target_path, 'items': result['items'], 'errors': result['errors'] }) @app.route('/library/thumbnail') def library_thumbnail(): """Serve a course's cover image (see find_course_thumbnail), if it has one.""" library_root = os.path.abspath(get_library_root()) course_path = request.args.get('path', '') target_path = os.path.abspath(course_path) if not (target_path == library_root or target_path.startswith(library_root + os.sep)): return '', 403 thumbnail = find_course_thumbnail(target_path) if not thumbnail: return '', 404 return send_file(thumbnail) @app.route('/library/search') def library_search(): """Recursively search course names in the library (respects hidden paths).""" query = request.args.get('q', '').strip() library_root = os.path.abspath(get_library_root()) if not query: return jsonify({'library_path': library_root, 'results': []}) results = search_library_courses(query) return jsonify({'library_path': library_root, 'results': results}) @app.route('/api/library/refresh', methods=['POST']) def refresh_library_api(): """Clear cached library-scan results - for when files were added/removed directly on disk (e.g. on the NAS) rather than through the app, which the cache's TTL would otherwise take up to 5 minutes to notice on its own.""" invalidate_cache() return jsonify({'success': True}) @app.route('/api/hidden-paths', methods=['GET']) def get_hidden_paths_api(): """List currently-hidden course/directory paths, with display names.""" paths = get_hidden_paths() return jsonify({ 'hidden_paths': [ {'path': p, 'name': os.path.basename(p.rstrip(os.sep)) or p} for p in paths ] }) @app.route('/api/hidden-paths', methods=['POST']) def set_hidden_path_api(): """Hide or un-hide a course/directory from the Library browser.""" data = request.json or {} path = data.get('path', '') hidden = bool(data.get('hidden', True)) if not path: return jsonify({'error': 'path is required'}), 400 library_root = os.path.abspath(get_library_root()) target = os.path.abspath(path) if not (target == library_root or target.startswith(library_root + os.sep)): return jsonify({'error': 'Path outside library root'}), 403 updated = set_path_hidden(target, hidden) return jsonify({'success': True, 'hidden_paths': updated}) @app.route('/api/hidden-paths/bulk', methods=['POST']) def bulk_set_hidden_paths_api(): """Hide or un-hide several courses/directories at once.""" data = request.json or {} paths = data.get('paths') or [] hidden = bool(data.get('hidden', True)) if not isinstance(paths, list) or not paths: return jsonify({'error': 'paths is required'}), 400 library_root = os.path.abspath(get_library_root()) updated = get_hidden_paths() for path in paths: target = os.path.abspath(path) if target == library_root or target.startswith(library_root + os.sep): updated = set_path_hidden(target, hidden) return jsonify({'success': True, 'hidden_paths': updated}) 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 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(old_path) if not (old_abs == library_root or old_abs.startswith(library_root + os.sep)): return {'success': False, 'error': 'Path outside library root', 'status': 403} if old_abs == library_root: return {'success': False, 'error': 'Cannot rename the library root itself', 'status': 400} if not os.path.isdir(old_abs): 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 {'success': False, 'error': f'"{new_name}" already exists here', 'status': 409} try: os.rename(old_abs, new_abs) except OSError as e: return {'success': False, 'error': f'Rename failed: {e}', 'status': 500} invalidate_cache() # renamed path invalidates any cached listing/summary/tree keyed by the old path rebase_library_path(old_abs, new_abs) active_course_reset = False if current_course is not None: course_abs = os.path.abspath(current_course.path) if course_abs == old_abs or course_abs.startswith(old_abs + os.sep): current_course = None active_course_reset = True 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') def settings_page(): """Render the display-settings page.""" current = load_settings() named_presets = sorted( (t for t in SETTINGS_CHOICES['theme'] if t not in ('dark', 'light')), key=lambda t: THEME_DISPLAY_NAMES[t] ) return render_template( 'settings.html', settings=current, default_library_path=LIBRARY_PATH, theme_choices=['dark', 'light'] + named_presets, theme_display_names=THEME_DISPLAY_NAMES, font_family_choices=['system', 'sans', 'serif', 'monospace', 'monaspace'], font_size_choices=['small', 'medium', 'large', 'xlarge'], layout_width_choices=['normal', 'wide', 'full'], density_choices=['comfortable', 'compact'], card_style_choices=['flat', 'elevated', 'bordered'], corner_radius_choices=['sharp', 'rounded', 'pill'], ) @app.route('/api/settings', methods=['GET']) def get_settings_api(): """Return current settings plus their resolved CSS values, for the shared theme script.""" current = load_settings() return jsonify({ 'settings': current, 'css_vars': settings_css_vars(current) }) @app.route('/api/settings', methods=['POST']) def save_settings_api(): """Persist updated display settings.""" try: new_settings = request.get_json(force=True) or {} saved = save_settings(new_settings) return jsonify({ 'success': True, 'settings': saved, 'css_vars': settings_css_vars(saved) }) except Exception as e: return jsonify({'success': False, 'error': str(e)}), 400 @app.route('/api/settings/reset', methods=['POST']) def reset_settings_api(): """Reset settings back to defaults.""" os.makedirs(DATA_DIR, exist_ok=True) with open(SETTINGS_FILE, 'w') as f: json.dump(DEFAULT_SETTINGS, f, indent=2) return jsonify({ 'success': True, 'settings': DEFAULT_SETTINGS, 'css_vars': settings_css_vars(DEFAULT_SETTINGS) }) @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 get_all_course_dirs(): 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""" global current_course data = request.json course_path = data.get('course_path') if not course_path or not os.path.exists(course_path): return jsonify({'error': 'Invalid course path'}), 400 try: current_course = get_course_tree(course_path) return jsonify({'success': True, 'course_name': current_course.name}) except Exception as e: return jsonify({'error': str(e)}), 500 @app.route('/random-pick') def random_pick(): """"Surprise Me" - load a random not-yet-fully-watched course and land on its dashboard.""" global current_course course_dir = get_random_incomplete_course() if not course_dir: return redirect(url_for('index')) current_course = get_course_tree(str(course_dir)) return redirect(url_for('index')) @app.route('/recent/open') def open_recent(): """ Jump straight to a "Recently Viewed" entry from the dashboard: load its course if it isn't already the active one, then go to that lesson. """ global current_course course_path = request.args.get('course_path', '') lesson_path = request.args.get('lesson_path', '') seek_seconds = request.args.get('t', '') if not course_path or not lesson_path or not os.path.exists(course_path): return redirect(url_for('index')) try: if not current_course or current_course.path != course_path: current_course = get_course_tree(course_path) except Exception as e: print(f"Could not load course for recent view: {e}") return redirect(url_for('index')) lesson_url = url_for('view_lesson', lesson_path=lesson_path) if seek_seconds: lesson_url = f"{lesson_url}?t={seek_seconds}" return redirect(lesson_url) @app.route('/help') def help_page(): """Static help page: how to use the app, supported file types.""" return render_template('help.html') @app.route('/notes') def notes_hub(): """Every lesson note across the whole library in one place.""" return render_template('notes_hub.html', notes=get_all_notes()) @app.route('/library/search-transcripts') def search_transcripts_api(): """Search inside lesson subtitle files across the library.""" query = request.args.get('q', '').strip() if not query: return jsonify({'results': []}) return jsonify({'results': search_transcripts(query)}) @app.route('/course/study-guide') def download_study_guide(): """Download every note written for the currently loaded course as one markdown file.""" global current_course if not current_course: return "No course loaded", 404 markdown = build_study_guide_markdown(current_course) buffer = io.BytesIO(markdown.encode('utf-8')) safe_name = re.sub(r'[^\w\-. ]', '_', current_course.name).strip() or 'course' return send_file(buffer, mimetype='text/markdown', as_attachment=True, download_name=f"{safe_name} - Study Guide.md") @app.route('/api/next-up', methods=['POST']) def set_next_up_api(): """Add or remove a course from the Next Up queue.""" data = request.json or {} path = data.get('path', '') queued = bool(data.get('queued', True)) if not path: return jsonify({'error': 'path is required'}), 400 library_root = os.path.abspath(get_library_root()) target = os.path.abspath(path) if not (target == library_root or target.startswith(library_root + os.sep)): return jsonify({'error': 'Path outside library root'}), 403 updated = set_path_queued(target, queued) return jsonify({'success': True, 'next_up': updated}) @app.route('/api/next-up/bulk', methods=['POST']) def bulk_set_next_up_api(): """Add several courses to the Next Up queue at once.""" data = request.json or {} paths = data.get('paths') or [] if not isinstance(paths, list) or not paths: return jsonify({'error': 'paths is required'}), 400 library_root = os.path.abspath(get_library_root()) updated = get_next_up_paths() for path in paths: target = os.path.abspath(path) if target == library_root or target.startswith(library_root + os.sep): updated = set_path_queued(target, True) return jsonify({'success': True, 'next_up': updated}) @app.route('/api/next-up/reorder', methods=['POST']) def reorder_next_up_api(): """Replace the Next Up order wholesale, from the client's up/down-reordered list.""" data = request.json or {} paths = data.get('paths') if not isinstance(paths, list): return jsonify({'error': 'paths must be a list'}), 400 updated = reorder_next_up(paths) return jsonify({'success': True, 'next_up': updated}) @app.route('/lesson/') def view_lesson(lesson_path: str): """View specific lesson by path""" global current_course if not current_course: return redirect(url_for('index')) # Find the lesson in the tree, and the section (DirectoryNode) it belongs to lesson, section_node = find_lesson_in_tree(current_course.root_node, lesson_path) if not lesson: return redirect(url_for('index')) # Get all lessons for navigation all_lessons = get_all_lessons(current_course.root_node) current_index = -1 # Find current lesson index for i, (path, lesson_obj) in enumerate(all_lessons): if lesson_obj == lesson: current_index = i break # Get next and previous lessons prev_lesson = None next_lesson = None if current_index > 0: prev_lesson = all_lessons[current_index - 1][0] if current_index < len(all_lessons) - 1: next_lesson = all_lessons[current_index + 1][0] # Update last accessed without touching saved progress/completed state ProgressTracker.touch_lesson_accessed(current_course, lesson_path) # Record for the cross-course "Recently Viewed" list on the dashboard record_recent_view(current_course.name, current_course.path, lesson_path, lesson.title) # Read notes and progress directly from the progress file rather than # the Lesson object - apply_progress_to_tree (which populates Lesson # fields) isn't called on this code path, only on the dashboard's tree # render. lesson_progress = ProgressTracker.load_progress(current_course).get(lesson_path, {}) seek_seconds = request.args.get('t', type=int) return render_template('lesson_view.html', course=current_course, lesson=lesson, lesson_path=lesson_path, lesson_notes=ProgressTracker._notes_from_entry(lesson_progress), lesson_progress_seconds=lesson_progress.get('progress_seconds', 0), initial_seek_seconds=seek_seconds, outline_topic_id=lesson_progress.get('outline_topic_id', ''), outline_topic_name=lesson_progress.get('outline_topic_name', ''), section_lessons=get_section_lessons(current_course, section_node, lesson), prev_lesson=prev_lesson, next_lesson=next_lesson) def get_lesson_url(lesson: Lesson, course_path: str) -> str: """Generate the URL for a lesson""" # Create relative path from course root lesson_file_path = os.path.relpath(lesson.path, course_path) lesson_file_path = lesson_file_path.replace('\\', '/') if lesson_file_path.startswith('/'): lesson_file_path = lesson_file_path[1:] # Append lesson title for uniqueness lesson_url = f"{lesson_file_path}/{lesson.title.replace(' ', '_')}" return lesson_url def find_lesson_in_tree(node: DirectoryNode, target_path: str) -> Tuple[Optional[Lesson], Optional[DirectoryNode]]: """ Find a lesson in the tree by path, along with the DirectoryNode that directly owns it - DirectoryNode has no parent pointer, so this is the only way to get "the section this lesson belongs to" (its `.lessons` list is exactly that section's sibling set, used for the lesson page's "up next in this section" list). """ # Check lessons in current node for lesson in node.lessons: lesson_url = get_lesson_url(lesson, current_course.path) # Check multiple possible path formats lesson_file_path = os.path.relpath(lesson.path, current_course.path) lesson_file_path = lesson_file_path.replace('\\', '/') if lesson_file_path.startswith('/'): lesson_file_path = lesson_file_path[1:] # Also check with lesson title appended lesson_path_with_title = f"{lesson_file_path}/{lesson.title.replace(' ', '_')}" if (lesson_url == target_path or lesson_file_path == target_path or lesson_path_with_title == target_path): return lesson, node # Recursively search children for child in node.children.values(): result_lesson, result_node = find_lesson_in_tree(child, target_path) if result_lesson: return result_lesson, result_node return None, None def get_section_lessons(course: Course, section_node: DirectoryNode, current_lesson: Lesson) -> List[Dict[str, Any]]: """ The sibling lessons in current_lesson's own section, for the lesson page's "up next in this section" list. view_lesson() isn't on the apply_progress_to_tree() code path, so - like that route already does for the current lesson's own notes/progress - this reads progress directly from the progress file rather than relying on Lesson fields. """ progress = ProgressTracker.load_progress(course) siblings = [] for sibling in section_node.lessons: key = _resolve_lesson_progress_key(course, sibling, progress) entry = progress.get(key, {}) if key else {} completed = entry.get('completed', False) progress_seconds = entry.get('progress_seconds', 0) duration_seconds = entry.get('duration_seconds', 0) percent_watched = 100 if completed else ( round(100 * progress_seconds / duration_seconds) if duration_seconds and progress_seconds else 0 ) siblings.append({ 'title': sibling.title, 'url': get_lesson_url(sibling, course.path), 'lesson_type': sibling.lesson_type, 'completed': completed, 'percent_watched': percent_watched, 'is_current': sibling is current_lesson, }) return siblings def get_all_lessons(node: DirectoryNode) -> List[Tuple[str, Lesson]]: """Get all lessons from the tree with their paths""" lessons = [] def collect_lessons(n: DirectoryNode, current_path: str = ""): # Add lessons from this node for lesson in n.lessons: lesson_url = get_lesson_url(lesson, current_course.path) lessons.append((lesson_url, lesson)) # Recursively collect from children for child in n.children.values(): collect_lessons(child, current_path) collect_lessons(node) return lessons @app.route('/api/progress', methods=['POST']) def update_progress(): """API endpoint to update lesson progress""" global current_course if not current_course: return jsonify({'error': 'No course loaded'}), 400 data = request.json lesson_path = data.get('lesson_path') completed = data.get('completed', False) progress_seconds = data.get('progress_seconds', 0) duration_seconds = data.get('duration_seconds') or None try: ProgressTracker.update_lesson_progress( current_course, lesson_path, completed, progress_seconds, duration_seconds ) return jsonify({'success': True}) except Exception as e: return jsonify({'error': str(e)}), 500 @app.route('/api/lesson-note/add', methods=['POST']) def add_lesson_note_api(): """Append a new timestamped note to a lesson.""" global current_course if not current_course: return jsonify({'error': 'No course loaded'}), 400 data = request.json or {} lesson_path = data.get('lesson_path') text = (data.get('text') or '').strip() if not lesson_path: return jsonify({'error': 'lesson_path is required'}), 400 if not text: return jsonify({'error': 'text is required'}), 400 timestamp_seconds = data.get('timestamp_seconds') if timestamp_seconds is not None: try: timestamp_seconds = int(timestamp_seconds) except (TypeError, ValueError): timestamp_seconds = None notes = ProgressTracker.add_lesson_note(current_course, lesson_path, text, timestamp_seconds) return jsonify({'success': True, 'notes': notes}) @app.route('/api/lesson-note/update', methods=['POST']) def update_lesson_note_api(): """Edit one of a lesson's timestamped notes.""" global current_course if not current_course: return jsonify({'error': 'No course loaded'}), 400 data = request.json or {} lesson_path = data.get('lesson_path') note_id = data.get('note_id') text = (data.get('text') or '').strip() if not lesson_path or not note_id: return jsonify({'error': 'lesson_path and note_id are required'}), 400 if not text: return jsonify({'error': 'text is required'}), 400 notes = ProgressTracker.update_lesson_note_text(current_course, lesson_path, note_id, text) return jsonify({'success': True, 'notes': notes}) @app.route('/api/lesson-note/delete', methods=['POST']) def delete_lesson_note_api(): """Remove one of a lesson's timestamped notes.""" global current_course if not current_course: return jsonify({'error': 'No course loaded'}), 400 data = request.json or {} lesson_path = data.get('lesson_path') note_id = data.get('note_id') if not lesson_path or not note_id: return jsonify({'error': 'lesson_path and note_id are required'}), 400 notes = ProgressTracker.delete_lesson_note(current_course, lesson_path, note_id) return jsonify({'success': True, 'notes': notes}) @app.route('/api/lesson-note/topic', methods=['POST']) def update_lesson_note_topic_api(): """Remember which Outline topic (collection) a lesson's note should push to.""" global current_course if not current_course: return jsonify({'error': 'No course loaded'}), 400 data = request.json or {} lesson_path = data.get('lesson_path') if not lesson_path: return jsonify({'error': 'lesson_path is required'}), 400 topic_id = (data.get('topic_id') or '').strip() topic_name = (data.get('topic_name') or '').strip() ProgressTracker.set_lesson_outline_topic(current_course, lesson_path, topic_id, topic_name) return jsonify({'success': True}) @app.route('/api/outline/config', methods=['GET']) def get_outline_config_api(): """Whether Outline is configured, the (non-secret) base URL, and the default collection - never the token.""" config = load_outline_config() return jsonify({ 'configured': bool(config['base_url'] and config['api_token']), 'base_url': config['base_url'], 'default_collection_id': config['default_collection_id'], 'default_collection_name': config['default_collection_name'] }) @app.route('/api/outline/config', methods=['POST']) def set_outline_config_api(): """Save the Outline base URL / API token / default collection.""" data = request.json or {} base_url = (data.get('base_url') or '').strip() api_token = (data.get('api_token') or '').strip() if not base_url: return jsonify({'error': 'Outline URL is required'}), 400 save_outline_config( base_url, api_token, default_collection_id=data.get('default_collection_id'), default_collection_name=data.get('default_collection_name') ) return jsonify({'success': True}) @app.route('/api/outline/test') def test_outline_connection_api(): """Ping Outline with the stored config, for the Settings page's 'Test Connection' button.""" try: _outline_request('collections.list', {'limit': 1}) return jsonify({'success': True}) except RuntimeError as e: return jsonify({'success': False, 'error': str(e)}), 502 @app.route('/api/outline/collections') def list_outline_collections_api(): """Every Outline collection - used by the Settings page's default-collection picker.""" try: result = _outline_request('collections.list', {'limit': 100}) except RuntimeError as e: return jsonify({'collections': [], 'error': str(e)}) collections = [{'id': c['id'], 'name': c['name']} for c in result.get('data', [])] return jsonify({'collections': collections}) @app.route('/api/outline/topics') def list_outline_topics_api(): """The lesson page's topic chooser option list - top-level documents in the configured default collection.""" try: topics = list_outline_topics() except RuntimeError as e: return jsonify({'topics': [], 'error': str(e)}) return jsonify({'topics': topics}) @app.route('/api/outline/resolve-topic', methods=['POST']) def resolve_outline_topic_api(): """ Resolve a freshly-typed topic name to a real Outline document id right away (find-or-create), rather than deferring to the fire-and-forget pagehide push - see push_note_to_outline's docstring for why that matters (a page that fires pagehide more than once, e.g. via bfcache, would otherwise send the same 'create a new topic' intent repeatedly). """ data = request.json or {} name = (data.get('name') or '').strip() if not name: return jsonify({'error': 'name is required'}), 400 try: topic = resolve_outline_topic(name) except RuntimeError as e: return jsonify({'error': str(e)}), 502 return jsonify({'success': True, 'id': topic['id'], 'name': topic['name']}) @app.route('/api/outline/push', methods=['POST']) def push_outline_note_api(): """ Push a lesson's note to Outline. Called via navigator.sendBeacon() on pagehide, so this has no interactive caller to report errors back to in the common case - failures are logged server-side and otherwise silent, matching the fire-and-forget nature of the trigger. """ global current_course if not current_course: return jsonify({'error': 'No course loaded'}), 400 data = request.json or {} lesson_path = data.get('lesson_path') lesson_title = data.get('lesson_title', lesson_path) topic_id = (data.get('topic_id') or '').strip() new_topic_name = (data.get('new_topic_name') or '').strip() if not lesson_path: return jsonify({'error': 'lesson_path is required'}), 400 try: result = push_note_to_outline(current_course, lesson_path, lesson_title, topic_id, new_topic_name) except RuntimeError as e: print(f"Outline push failed: {e}") return jsonify({'success': False, 'error': str(e)}), 502 if not result['success']: print(f"Outline push skipped: {result['error']}") return jsonify(result) @app.route('/api/course/mark-watched', methods=['POST']) def mark_course_watched_api(): """Mark every lesson in the currently loaded course as completed.""" global current_course if not current_course: return jsonify({'error': 'No course loaded'}), 400 ProgressTracker.mark_all_completed(current_course) return jsonify({'success': True}) @app.route('/files/') def serve_file(filepath): """Serve course files""" global current_course if not current_course: return "No course loaded", 404 # Security: ensure file is within course directory try: # URL decode the filepath and normalize it from urllib.parse import unquote decoded_filepath = unquote(filepath) # Construct the full path relative to the course directory full_path = os.path.join(current_course.path, decoded_filepath) full_path = os.path.abspath(full_path) course_path = os.path.abspath(current_course.path) print(f"File request: {filepath}") print(f"Decoded filepath: {decoded_filepath}") print(f"Full path: {full_path}") print(f"Course path: {course_path}") # Security check: ensure file is within course directory if not full_path.startswith(course_path): print(f"Access denied: {full_path} not in {course_path}") return "Access denied", 403 if not os.path.exists(full_path): print(f"File not found: {full_path}") return "File not found", 404 print(f"Serving file: {full_path}") # Determine MIME type. Don't rely solely on mimetypes.guess_type() - # its results can vary by OS/container base image depending on what # system mime databases are present. Force the types that matter for # inline preview (PDF above all) so this can't silently regress. ext = os.path.splitext(full_path)[1].lower() KNOWN_MIME_TYPES = { '.pdf': 'application/pdf', '.html': 'text/html', '.htm': 'text/html', '.txt': 'text/plain', '.md': 'text/plain', } if ext in KNOWN_MIME_TYPES: mime_type = KNOWN_MIME_TYPES[ext] else: mime_type, _ = mimetypes.guess_type(full_path) if mime_type is None: mime_type = 'application/octet-stream' # as_attachment=False (the default) sends Content-Disposition: inline # so the browser renders PDFs/HTML in the iframe instead of prompting # a download. Being explicit here so this can't drift. return send_file(full_path, mimetype=mime_type, as_attachment=False) except Exception as e: print(f"Error serving file: {str(e)}") return f"Error serving file: {str(e)}", 500 @app.route('/health') def healthcheck(): """Healthcheck endpoint for Docker""" return jsonify({"status": "healthy"}), 200 @app.route('/reset_course') def reset_course(): """Reset current course selection""" global current_course current_course = None return redirect(url_for('index')) def create_templates(): """Create basic template files if they don't exist""" templates_dir = Path('templates') templates_dir.mkdir(exist_ok=True) # Basic select course template select_template = ''' OfflineU - Select Course

OfflineU - Course Selection

''' # Basic course dashboard template dashboard_template = ''' OfflineU - {{ course.name }}

{{ course.name }}

Progress: {{ stats.completed_lessons }}/{{ stats.total_lessons }} lessons ({{ stats.completion_percentage }}%)

{% for module_idx, module in course.modules|enumerate %}

{{ module.title }}

{% for lesson_idx, lesson in module.lessons|enumerate %}
{{ lesson.title }} ({{ lesson.lesson_type }}) {% if lesson.completed %}✓{% endif %}
{% endfor %}
{% endfor %}

Select Different Course

''' # Basic lesson view template lesson_template = ''' {{ lesson.title }} - {{ course.name }}

{{ lesson.title }}

← Back to Course
{% if lesson.video_file %}

Video

{% endif %} {% if lesson.audio_file %}

Audio

{% endif %} {% if lesson.text_files %}

Additional Resources

{% for text_file in lesson.text_files %} 📄 {{ text_file.split('/')[-1] }} {% endfor %} {% endif %}
''' # Write templates to files template_files = { 'select_course.html': select_template, 'course_dashboard.html': dashboard_template, 'lesson_view.html': lesson_template } for filename, content in template_files.items(): template_path = templates_dir / filename if not template_path.exists(): with open(template_path, 'w', encoding='utf-8') as f: f.write(content) print(f"Created template: {template_path}") if __name__ == '__main__': parser = argparse.ArgumentParser(description='OfflineU Course Viewer & Tracker') parser.add_argument('--host', default='0.0.0.0', help='Host to bind to') parser.add_argument('--port', type=int, default=5000, help='Port to bind to') parser.add_argument('--debug', action='store_true', help='Enable debug mode') parser.add_argument('--create-templates', action='store_true', help='Create basic templates') parser.add_argument('--library-path', default=None, help='Base directory to scan for courses in the Library browser ' '(default: $COURSES_LIBRARY_PATH or /app/courses)') parser.add_argument('course_path', nargs='?', help='Path to course directory') args = parser.parse_args() if args.library_path: LIBRARY_PATH = args.library_path # Create templates if requested if args.create_templates: create_templates() print("Templates created successfully!") if not args.course_path: sys.exit(0) # Auto-load course if provided if args.course_path or os.environ.get('AUTO_LOAD_COURSE'): course_path = args.course_path or os.environ.get('AUTO_LOAD_COURSE') if not os.path.exists(course_path): print(f"Error: Course path does not exist: {course_path}", file=sys.stderr) sys.exit(1) try: current_course = DynamicCourseParser.scan_directory(course_path) print(f"Auto-loaded course: {current_course.name}") print(f"Built dynamic directory tree with {len(current_course.root_node.children)} top-level items") except Exception as e: print(f"Error loading course: {e}", file=sys.stderr) if args.debug: import traceback traceback.print_exc() sys.exit(1) # Create templates directory if it doesn't exist if not Path('templates').exists(): print("Templates directory not found. Creating basic templates...") create_templates() print(f"Starting OfflineU on http://{args.host}:{args.port}") print("Use --create-templates to regenerate template files") try: app.run(debug=args.debug, host=args.host, port=args.port) except KeyboardInterrupt: print("\nShutting down OfflineU...") except Exception as e: print(f"Error starting server: {e}") if args.debug: import traceback traceback.print_exc() sys.exit(1)