#!/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 itertools import shutil import subprocess import threading 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, Set 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' def _load_build_version() -> str: """ Reads the repo-committed VERSION file - lets Settings and /health show which version is actually running, the only way to confirm a push made it into a rebuilt container, since the webhook does a full image rebuild with no other visible confirmation. Deliberately a plain file checked into the repo (updated by hand alongside each commit) rather than baked from `git rev-parse` at Docker build time: Dockhand's build context doesn't reliably have .git available, which silently produced "unknown" instead of a real commit. Falls back to 'dev' if the file is ever missing (shouldn't happen once committed, since it ships via the same COPY . . as everything else). """ version_file = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'VERSION') try: with open(version_file, 'r') as f: content = f.read().strip() if content: return content except OSError: pass return 'dev' BUILD_VERSION = _load_build_version() def load_changelog_entries(limit: int = 10) -> List[str]: """ Reads CHANGELOG.md for the Settings page's "What's New" list - one line per commit (same timestamp + description format as VERSION), newest first. VERSION itself is overwritten each commit and only ever shows the current build; this accumulates that same line over time so you can see what changed recently without digging through git log. Falls back to an empty list if the file is missing (shouldn't happen once committed, same reasoning as _load_build_version). """ changelog_file = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'CHANGELOG.md') try: with open(changelog_file, 'r') as f: lines = [line.strip() for line in f if line.strip()] return lines[:limit] except OSError: return [] # 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') AUTO_THUMBNAIL_FILENAME = '.offlineu_thumbnail.jpg' # 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 # ---- Undo history for move/rename operations ---- # # Only move and rename are ever recorded here - both are trivially # reversible (move/rename the result back). Hide/show is already a # one-click toggle with no need for "undo," and delete is permanent by # design, so it's never in this history no matter how it's framed in the # UI. Recorded as batches (not one entry per file) since the operations # that populate this - Sort Unsorted apply, Bulk Rename apply, a bulk # move - already act on several items as a single user action; undoing # should mirror that, not require clicking "undo" N times. _UNDO_HISTORY_MAX = 20 _undo_history: List[Dict[str, Any]] = [] def _record_undo_batch(label: str, entries: List[Dict[str, str]]) -> None: """Push one reversible batch - entries are {'old_path', 'new_path'} pairs; the label is shown in the UI's Undo button. No-ops if entries is empty (e.g. every item in a batch failed).""" if not entries: return _undo_history.append({ 'label': label, 'entries': entries, 'timestamp': datetime.now().isoformat(), }) del _undo_history[:-_UNDO_HISTORY_MAX] def invalidate_cache(): """Clear all cached library-scan results - call after anything that changes what's on disk (rename, hide/show, library path change).""" _cache_store.clear() # 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', # 34 palettes derived from Figma's "53 Unique Website Color Schemes" # resource page, all converted to dark/colored backgrounds (the # source page's own schemes were mostly light-mode marketing # mockups) - see WEBSITE_SCHEME_CATEGORIES below for the dropdown # grouping, and tools/generate_website_scheme_themes.py (a one-off # content-generation script, not part of the running app) for how # the kept ones were built: dominant colors extracted per scheme, # mapped into this app's bg/text/accent shape with automatic # contrast adjustment against both the background and the white # button text, then deduplicated against each other by actual # color distance (not eyeballed) once everything converged to dark. 'ink_wash', 'jade_pebble_morning', 'woodland', 'graphite', 'pearl', 'yacht_club', 'amber_walnut_morning', 'copper_aquamarine_dream', 'sandstone_aquamarine_serenity', 'honey_opal_sunset', 'seashell_garnet_afternoon', 'rose_quartz_evening', 'calcite', 'fireside', 'terrazzo', 'sapphire_nightfall_whisper', 'marina', 'emerald_lavender_lake', 'sage_peridot_morning', 'amethyst_dawn_haze', 'royal_glimmer', 'neptune', 'tropical_heat', 'celestial', 'festive_eve', 'freshly_squeezed', 'jelly_shoes', 'lemon_granite_morning', 'arctic_reflection', 'slate', 'autumn_luxe', 'inked', 'wraith', 'urban_nocturne', }, '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', 'ink_wash': 'Ink Wash', 'jade_pebble_morning': 'Jade Pebble Morning', 'woodland': 'Woodland', 'graphite': 'Graphite', 'pearl': 'Pearl', 'yacht_club': 'Yacht Club', 'amber_walnut_morning': 'Amber Walnut Morning', 'copper_aquamarine_dream': 'Copper Aquamarine Dream', 'sandstone_aquamarine_serenity': 'Sandstone Aquamarine Serenity', 'honey_opal_sunset': 'Honey Opal Sunset', 'seashell_garnet_afternoon': 'Seashell Garnet Afternoon', 'rose_quartz_evening': 'Rose Quartz Evening', 'calcite': 'Calcite', 'fireside': 'Fireside', 'terrazzo': 'Terrazzo', 'sapphire_nightfall_whisper': 'Sapphire Nightfall Whisper', 'marina': 'Marina', 'emerald_lavender_lake': 'Emerald Lavender Lake', 'sage_peridot_morning': 'Sage Peridot Morning', 'amethyst_dawn_haze': 'Amethyst Dawn Haze', 'royal_glimmer': 'Royal Glimmer', 'neptune': 'Neptune', 'tropical_heat': 'Tropical Heat', 'celestial': 'Celestial', 'festive_eve': 'Festive Eve', 'freshly_squeezed': 'Freshly Squeezed', 'jelly_shoes': 'Jelly Shoes', 'lemon_granite_morning': 'Lemon Granite Morning', 'arctic_reflection': 'Arctic Reflection', 'slate': 'Slate', 'autumn_luxe': 'Autumn Luxe', 'inked': 'Inked', 'wraith': 'Wraith', 'urban_nocturne': 'Urban Nocturne', } # 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', }, # --- Minimal & Neutral (from Figma's "53 Unique Website Color Schemes", converted to dark backgrounds) --- 'ink_wash': { 'bg-primary': '#202020', 'bg-secondary': '#323232', 'bg-tertiary': '#444444', 'bg-tertiary-hover': '#4b4b4b', 'text-primary': '#e6e6e6', 'text-muted': '#9e9e9e', 'border-color': '#494949', 'accent': '#2990d6', 'accent-hover': '#2173ab', }, 'jade_pebble_morning': { 'bg-primary': '#242d24', 'bg-secondary': '#344134', 'bg-tertiary': '#445544', 'bg-tertiary-hover': '#4b5e4b', 'text-primary': '#e5e6e5', 'text-muted': '#9ba19b', 'border-color': '#495b49', 'accent': '#9dca73', 'accent-hover': '#83bc4e', }, 'woodland': { 'bg-primary': '#323220', 'bg-secondary': '#48482e', 'bg-tertiary': '#5d5d3c', 'bg-tertiary-hover': '#676742', 'text-primary': '#e7e7e4', 'text-muted': '#a4a498', 'border-color': '#646440', 'accent': '#acc927', 'accent-hover': '#879e1f', }, 'graphite': { 'bg-primary': '#2c2c26', 'bg-secondary': '#3f3f37', 'bg-tertiary': '#525247', 'bg-tertiary-hover': '#5a5a4f', 'text-primary': '#e6e6e5', 'text-muted': '#a0a09c', 'border-color': '#57574c', 'accent': '#e07b7b', 'accent-hover': '#d65252', }, 'pearl': { 'bg-primary': '#292929', 'bg-secondary': '#3b3b3b', 'bg-tertiary': '#4d4d4d', 'bg-tertiary-hover': '#545454', 'text-primary': '#e6e6e6', 'text-muted': '#9e9e9e', 'border-color': '#525252', 'accent': '#d7ad84', 'accent-hover': '#ca935e', }, 'yacht_club': { 'bg-primary': '#322720', 'bg-secondary': '#48382e', 'bg-tertiary': '#5d493c', 'bg-tertiary-hover': '#675042', 'text-primary': '#e7e5e4', 'text-muted': '#a49d98', 'border-color': '#644e40', 'accent': '#c67139', 'accent-hover': '#9e5a2e', }, # --- Warm (from Figma's "53 Unique Website Color Schemes", converted to dark backgrounds) --- 'amber_walnut_morning': { 'bg-primary': '#292929', 'bg-secondary': '#3b3b3b', 'bg-tertiary': '#4d4d4d', 'bg-tertiary-hover': '#545454', 'text-primary': '#e6e6e6', 'text-muted': '#9e9e9e', 'border-color': '#525252', 'accent': '#d09070', 'accent-hover': '#c3724a', }, 'copper_aquamarine_dream': { 'bg-primary': '#202c32', 'bg-secondary': '#2e3f48', 'bg-tertiary': '#3c525d', 'bg-tertiary-hover': '#425a67', 'text-primary': '#e4e6e7', 'text-muted': '#98a0a4', 'border-color': '#405864', 'accent': '#d65d29', 'accent-hover': '#ab4a21', }, 'sandstone_aquamarine_serenity': { 'bg-primary': '#322420', 'bg-secondary': '#48342e', 'bg-tertiary': '#5d443c', 'bg-tertiary-hover': '#674b42', 'text-primary': '#e7e5e4', 'text-muted': '#a49b98', 'border-color': '#644940', 'accent': '#76c9e5', 'accent-hover': '#4bb8dd', }, 'honey_opal_sunset': { 'bg-primary': '#322920', 'bg-secondary': '#483b2e', 'bg-tertiary': '#5d4d3c', 'bg-tertiary-hover': '#675442', 'text-primary': '#e7e6e4', 'text-muted': '#a49e98', 'border-color': '#645240', 'accent': '#d6b12a', 'accent-hover': '#ac8e21', }, 'seashell_garnet_afternoon': { 'bg-primary': '#203232', 'bg-secondary': '#2e4848', 'bg-tertiary': '#3c5d5d', 'bg-tertiary-hover': '#426767', 'text-primary': '#e4e7e7', 'text-muted': '#98a4a4', 'border-color': '#406464', 'accent': '#f36878', 'accent-hover': '#ef394e', }, 'rose_quartz_evening': { 'bg-primary': '#322024', 'bg-secondary': '#482e34', 'bg-tertiary': '#5d3c44', 'bg-tertiary-hover': '#67424b', 'text-primary': '#e7e4e5', 'text-muted': '#a4989b', 'border-color': '#644049', 'accent': '#f36868', 'accent-hover': '#ef3939', }, 'calcite': { 'bg-primary': '#292929', 'bg-secondary': '#3b3b3b', 'bg-tertiary': '#4d4d4d', 'bg-tertiary-hover': '#545454', 'text-primary': '#e6e6e6', 'text-muted': '#9e9e9e', 'border-color': '#525252', 'accent': '#f1854e', 'accent-hover': '#ed641f', }, 'fireside': { 'bg-primary': '#322420', 'bg-secondary': '#48342e', 'bg-tertiary': '#5d443c', 'bg-tertiary-hover': '#674b42', 'text-primary': '#e7e5e4', 'text-muted': '#a49b98', 'border-color': '#644940', 'accent': '#d67929', 'accent-hover': '#ab6121', }, 'terrazzo': { 'bg-primary': '#203232', 'bg-secondary': '#2e4848', 'bg-tertiary': '#3c5d5d', 'bg-tertiary-hover': '#426767', 'text-primary': '#e4e7e7', 'text-muted': '#98a4a4', 'border-color': '#406464', 'accent': '#ecad6f', 'accent-hover': '#e69342', }, # --- Cool (from Figma's "53 Unique Website Color Schemes", converted to dark backgrounds) --- 'sapphire_nightfall_whisper': { 'bg-primary': '#202932', 'bg-secondary': '#2e3b48', 'bg-tertiary': '#3c4c5d', 'bg-tertiary-hover': '#425467', 'text-primary': '#e4e6e7', 'text-muted': '#989ea4', 'border-color': '#405264', 'accent': '#297fd6', 'accent-hover': '#2166ab', }, 'marina': { 'bg-primary': '#322720', 'bg-secondary': '#48382e', 'bg-tertiary': '#5d493c', 'bg-tertiary-hover': '#675042', 'text-primary': '#e7e5e4', 'text-muted': '#a49d98', 'border-color': '#644e40', 'accent': '#76ade5', 'accent-hover': '#4b93dd', }, 'emerald_lavender_lake': { 'bg-primary': '#203228', 'bg-secondary': '#2e4839', 'bg-tertiary': '#3c5d4a', 'bg-tertiary-hover': '#426752', 'text-primary': '#e4e7e5', 'text-muted': '#98a49d', 'border-color': '#40644f', 'accent': '#2ed174', 'accent-hover': '#25a75d', }, 'sage_peridot_morning': { 'bg-primary': '#203220', 'bg-secondary': '#2e482e', 'bg-tertiary': '#3c5d3c', 'bg-tertiary-hover': '#426742', 'text-primary': '#e4e7e4', 'text-muted': '#98a498', 'border-color': '#406440', 'accent': '#20d1af', 'accent-hover': '#19a58a', }, 'amethyst_dawn_haze': { 'bg-primary': '#242032', 'bg-secondary': '#342e48', 'bg-tertiary': '#443c5d', 'bg-tertiary-hover': '#4b4267', 'text-primary': '#e5e4e7', 'text-muted': '#9b98a4', 'border-color': '#494064', 'accent': '#ccba27', 'accent-hover': '#a1931f', }, 'royal_glimmer': { 'bg-primary': '#1f312d', 'bg-secondary': '#2d4741', 'bg-tertiary': '#3b5c56', 'bg-tertiary-hover': '#41665e', 'text-primary': '#e4e7e7', 'text-muted': '#98a4a2', 'border-color': '#3f635b', 'accent': '#29d6b4', 'accent-hover': '#21ab90', }, 'neptune': { 'bg-primary': '#203232', 'bg-secondary': '#2e4848', 'bg-tertiary': '#3c5d5d', 'bg-tertiary-hover': '#426767', 'text-primary': '#e4e7e7', 'text-muted': '#98a4a4', 'border-color': '#406464', 'accent': '#68ccf3', 'accent-hover': '#39bcef', }, # --- Vibrant & Bold (from Figma's "53 Unique Website Color Schemes", converted to dark backgrounds) --- 'tropical_heat': { 'bg-primary': '#203232', 'bg-secondary': '#2e4848', 'bg-tertiary': '#3c5d5d', 'bg-tertiary-hover': '#426767', 'text-primary': '#e4e7e7', 'text-muted': '#98a4a4', 'border-color': '#406464', 'accent': '#d65729', 'accent-hover': '#ab4621', }, 'celestial': { 'bg-primary': '#323220', 'bg-secondary': '#48482e', 'bg-tertiary': '#5d5d3c', 'bg-tertiary-hover': '#676742', 'text-primary': '#e7e7e4', 'text-muted': '#a4a498', 'border-color': '#646440', 'accent': '#cbc026', 'accent-hover': '#a0971e', }, 'festive_eve': { 'bg-primary': '#202b32', 'bg-secondary': '#2e3e48', 'bg-tertiary': '#3c515d', 'bg-tertiary-hover': '#425a67', 'text-primary': '#e4e6e7', 'text-muted': '#98a0a4', 'border-color': '#405764', 'accent': '#8f5cf2', 'accent-hover': '#6f2dee', }, 'freshly_squeezed': { 'bg-primary': '#322d20', 'bg-secondary': '#48412e', 'bg-tertiary': '#5d553c', 'bg-tertiary-hover': '#675e42', 'text-primary': '#e7e6e4', 'text-muted': '#a4a198', 'border-color': '#645b40', 'accent': '#d6ab29', 'accent-hover': '#ab8921', }, 'jelly_shoes': { 'bg-primary': '#242032', 'bg-secondary': '#342e48', 'bg-tertiary': '#443c5d', 'bg-tertiary-hover': '#4b4267', 'text-primary': '#e5e4e7', 'text-muted': '#9b98a4', 'border-color': '#494064', 'accent': '#bc68f3', 'accent-hover': '#a739ef', }, # --- Modern (from Figma's "53 Unique Website Color Schemes", converted to dark backgrounds) --- 'lemon_granite_morning': { 'bg-primary': '#202c32', 'bg-secondary': '#2e3f48', 'bg-tertiary': '#3c525d', 'bg-tertiary-hover': '#425a67', 'text-primary': '#e4e6e7', 'text-muted': '#98a0a4', 'border-color': '#405864', 'accent': '#c9bf27', 'accent-hover': '#9e961f', }, 'arctic_reflection': { 'bg-primary': '#202c32', 'bg-secondary': '#2e3f48', 'bg-tertiary': '#3c525d', 'bg-tertiary-hover': '#425a67', 'text-primary': '#e4e6e7', 'text-muted': '#98a0a4', 'border-color': '#405864', 'accent': '#4693b9', 'accent-hover': '#387694', }, 'slate': { 'bg-primary': '#262c26', 'bg-secondary': '#373f37', 'bg-tertiary': '#475247', 'bg-tertiary-hover': '#4f5a4f', 'text-primary': '#e5e6e5', 'text-muted': '#9ca09c', 'border-color': '#4c574c', 'accent': '#1ad236', 'accent-hover': '#14a52a', }, 'autumn_luxe': { 'bg-primary': '#303020', 'bg-secondary': '#45452e', 'bg-tertiary': '#5b5b3d', 'bg-tertiary-hover': '#646443', 'text-primary': '#e7e7e4', 'text-muted': '#a4a498', 'border-color': '#616141', 'accent': '#c08040', 'accent-hover': '#9a6633', }, 'inked': { 'bg-primary': '#171717', 'bg-secondary': '#292929', 'bg-tertiary': '#3b3b3b', 'bg-tertiary-hover': '#424242', 'text-primary': '#e6e6e6', 'text-muted': '#9e9e9e', 'border-color': '#404040', 'accent': '#27c9c9', 'accent-hover': '#1f9e9e', }, 'wraith': { 'bg-primary': '#1c1c12', 'bg-secondary': '#323220', 'bg-tertiary': '#48482e', 'bg-tertiary-hover': '#515134', 'text-primary': '#e7e7e4', 'text-muted': '#a4a498', 'border-color': '#4e4e32', 'accent': '#d69c29', 'accent-hover': '#ab7d21', }, 'urban_nocturne': { 'bg-primary': '#171717', 'bg-secondary': '#292929', 'bg-tertiary': '#3b3b3b', 'bg-tertiary-hover': '#424242', 'text-primary': '#e6e6e6', 'text-muted': '#9e9e9e', 'border-color': '#404040', 'accent': '#bfc927', 'accent-hover': '#969e1f', }, } # Groups the 53 Figma-derived theme keys above for the Settings dropdown # (rendered as s) - matches that source page's own "Minimal and # neutral / Warm / Cool / Vibrant and bold / Modern" section headings, so # browsing the dropdown still mirrors how the original schemes were curated. WEBSITE_SCHEME_CATEGORIES = { 'Minimal & Neutral': ['ink_wash', 'jade_pebble_morning', 'woodland', 'graphite', 'pearl', 'yacht_club'], 'Warm': ['amber_walnut_morning', 'copper_aquamarine_dream', 'sandstone_aquamarine_serenity', 'honey_opal_sunset', 'seashell_garnet_afternoon', 'rose_quartz_evening', 'calcite', 'fireside', 'terrazzo'], 'Cool': ['sapphire_nightfall_whisper', 'marina', 'emerald_lavender_lake', 'sage_peridot_morning', 'amethyst_dawn_haze', 'royal_glimmer', 'neptune'], 'Vibrant & Bold': ['tropical_heat', 'celestial', 'festive_eve', 'freshly_squeezed', 'jelly_shoes'], 'Modern': ['lemon_granite_morning', 'arctic_reflection', 'slate', 'autumn_luxe', 'inked', 'wraith', 'urban_nocturne'], } # 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 # duration_seconds comes from actual playback once a lesson's been # watched, falling back to the ffprobe-derived cache otherwise (see # ProgressTracker.apply_progress_to_tree) - so "remaining time" # reflects the whole course, not just what's been played so far. 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]: """ A course's cover image: a manually-placed one directly inside the folder (cover/folder/thumbnail/thumb/poster.*, checked non-recursively - just the common 'cover.jpg next to the sections' layout, not every subfolder) if there is one, otherwise an auto-generated frame grabbed from the course's first video. Almost none of this library's courses ship with real cover art, so without the fallback the grid view would be mostly bare initials. """ course_dir = Path(course_path) try: names = {f.name.lower(): f for f in course_dir.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) auto_thumb = names.get(AUTO_THUMBNAIL_FILENAME) if auto_thumb: return str(auto_thumb) # Generating (or determining there's nothing to generate from) is the # one part of this that isn't a cheap directory listing - cache the # outcome briefly so repeat image requests for a doc-only course don't # re-walk its files on every load. return cache_get_or_compute( f'auto_thumbnail:{course_dir}', lambda: _generate_course_thumbnail(course_dir) ) def _generate_course_thumbnail(course_dir: Path) -> Optional[str]: """ Grab a single frame from the course's first video as stand-in cover art, via ffmpeg, cached to .offlineu_thumbnail.jpg so this only ever runs once per course - the next call finds that file directly through the fast path above instead of hitting this function again. Returns None (nothing cached to disk) if the course has no video at all, so a video added later is still picked up on the next request past the short in-memory cache above. """ video_files = sorted( f for f in course_dir.rglob('*') if f.is_file() and f.suffix.lower() in VIDEO_EXTENSIONS ) if not video_files: return None source = video_files[0] duration = _probe_media_duration_seconds(source) or 0 offset = min(30.0, duration * 0.1) if duration else 5.0 output_path = course_dir / AUTO_THUMBNAIL_FILENAME try: result = subprocess.run( ['ffmpeg', '-y', '-ss', str(offset), '-i', str(source), '-frames:v', '1', '-vf', 'scale=480:-1', '-q:v', '3', str(output_path)], capture_output=True, timeout=20 ) except (OSError, subprocess.SubprocessError): return None if result.returncode != 0 or not output_path.exists(): return None return str(output_path) _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 FAVORITES_FILE = os.path.join(DATA_DIR, 'favorites.json') def get_favorite_paths() -> List[str]: """Load the set of favorited course paths - order doesn't matter here (favorites always render alphabetically), unlike Next Up's queue.""" try: if os.path.exists(FAVORITES_FILE): with open(FAVORITES_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 favorites: {e}") return [] def _save_favorite_paths(paths: List[str]) -> None: os.makedirs(DATA_DIR, exist_ok=True) with open(FAVORITES_FILE, 'w') as f: json.dump(paths, f, indent=2) def set_path_favorited(path: str, favorited: bool) -> List[str]: """Add or remove a course from favorites; returns the updated list.""" paths = get_favorite_paths() normalized = os.path.abspath(path) if favorited: if normalized not in paths: paths.append(normalized) else: paths = [p for p in paths if p != normalized] _save_favorite_paths(paths) return paths def get_favorite_courses() -> List[Dict[str, Any]]: """Favorited courses resolved to displayable summaries (name order, since there's no queue order to preserve), silently dropping any path no longer on disk.""" results = [] for path in sorted(get_favorite_paths(), key=lambda p: os.path.basename(p.rstrip(os.sep)).lower()): p = Path(path) if p.is_dir(): results.append(_course_summary(p)) return results 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, next_up.json, recent_views.json's course_path) so curation, the Next Up queue, 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) queued = get_next_up_paths() rebased_queue = [_rebase_prefix(p, old_abs, new_abs) for p in queued] if rebased_queue != queued: _save_next_up_paths(rebased_queue) favorited = get_favorite_paths() rebased_favorites = [_rebase_prefix(p, old_abs, new_abs) for p in favorited] if rebased_favorites != favorited: _save_favorite_paths(rebased_favorites) 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()) favorite_set = set(get_favorite_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 item['favorited'] = os.path.abspath(entry_path_str) in favorite_set item['completion_percentage'] = _course_completion_percentage(entry, item['media_files']) 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()))) # ---- Unsorted intake: propose where a newly-dropped course belongs ---- # # Workflow: new/incoming courses get dropped into a folder named Unsorted # directly under the library root, sitting alongside the real category # tree (e.g. IT/, IT/AWS/, Business/). Scanning proposes a destination for # each item by keyword overlap against that *existing* tree - read fresh # from disk on every scan, never hardcoded - so it adapts automatically to # whatever categories actually exist. Nothing gets moved until the moves # are explicitly applied (see /api/unsorted/apply). UNSORTED_FOLDER_NAME = 'Unsorted' _TOKEN_SPLIT_RE = re.compile(r'[^a-z0-9]+') _TOKEN_STOPWORDS = { 'a', 'an', 'and', 'for', 'from', 'in', 'is', 'of', 'on', 'or', 'the', 'to', 'with', 'course', 'courses', 'training', 'class', 'complete', 'full', 'guide', 'tutorial', 'part', 'vol', 'volume', 'edition', # Skill-level/marketing filler - describes the packaging, not the # subject, same spirit as 'complete'/'guide'/'tutorial' above. 'advanced', 'intermediate', 'beginner', 'beginners', 'basics', 'basic', 'essentials', 'fundamentals', 'mastering', 'master', 'crash', 'level', 'levels', # E-learning platforms/marketplaces - whichever site a course was bought # or ripped from says nothing about its subject, and it shows up in # almost every course name from that source. 'udemy', 'pluralsight', 'lynda', 'linkedin', 'linkedinlearning', 'coursera', 'skillshare', 'packt', 'packtpub', 'oreilly', 'edx', 'educative', 'egghead', 'frontendmasters', 'codecademy', 'udacity', 'skillsoft', 'cbtnuggets', 'cbt', 'itprotv', 'cybrary', 'tutsplus', 'masterclass', 'treehouse', 'datacamp', # Scene/distribution-group tags and rip metadata that piggyback on # downloaded course names - never a subject, always noise. 'bookware', 'tutsnode', 'ftu', 'ftuforum', 'freecoursesonline', 'freetutorials', 'coursedrive', 'downtem', 'glodls', 'garr', 'ridware', 'illiterate', 'zh', 'nfo', 'readnfo', 'repack', 'reup', 'reupload', 'repost', 'release', 'released', 'rip', 'x264', 'x265', 'www', 'com', 'net', 'org', 'info', # "Freshness" marketing words - describe the release, not the subject. 'update', 'updated', 'new', 'newest', 'latest', 'final', 'retail', # Calendar words - dates say nothing about what a course is about. 'january', 'february', 'march', 'april', 'june', 'july', 'august', 'september', 'october', 'november', 'december', 'jan', 'feb', 'mar', 'apr', 'jun', 'jul', 'aug', 'sep', 'sept', 'oct', 'nov', 'dec', } def _tokenize(name: str) -> Set[str]: """ Break a folder/course name into lowercase keyword tokens for matching. Splits camelCase/PascalCase boundaries before the general split, since course names are commonly dot- or dash-joined with no spaces at all (e.g. "AWSCertifiedSolutionsArchitect" as much as "AWS.Certified..."). Keeps short tokens (unlike most tokenizers) because real category names here are often exactly that short - "IT", "PM" - and drops pure numbers (section/edition numbers) and common filler words instead, which is what actually carries no signal in this context. """ return set(_tokenize_ordered(name)) def _tokenize_ordered(name: str) -> List[str]: """Same tokens as _tokenize, but as a de-duplicated list in the order they first appear - used when a token subset needs to read back as a sensible phrase (a suggested new folder name) rather than just being compared as a set.""" spaced = re.sub(r'(?<=[a-z0-9])(?=[A-Z])', ' ', name) raw = _TOKEN_SPLIT_RE.split(spaced.lower()) seen: Set[str] = set() ordered = [] for t in raw: if len(t) >= 2 and not t.isdigit() and t not in _TOKEN_STOPWORDS and t not in seen: seen.add(t) ordered.append(t) return ordered def _is_unrecognized_leaf_item(directory: Path) -> bool: """ A folder with files directly inside but no subfolders at all - virtually always a single downloaded item (an ebook, an audiobook in a format this app doesn't play, or some other misc file drop) rather than an organizational category, even when _looks_like_course's video/audio check doesn't recognize it as a course (e.g. an EPUB-only folder has no media files at all, so _has_direct_media never fires). Scoped to _build_category_index only - not a general "is this a course" replacement, since this app still can't do anything useful with the file itself, so it'd be wrong to surface it as a course elsewhere (Library browsing, stats, search, ...). """ try: entries = list(directory.iterdir()) except (PermissionError, OSError): return False has_subdir = any(e.is_dir() and not e.name.startswith('.') for e in entries) if has_subdir: return False return any(e.is_file() and not e.name.startswith('.') for e in entries) def _subtree_has_any_media(directory: Path, max_depth: int = 8) -> bool: """Whether `directory` or anything under it (bounded depth, so a huge non-course tree can't make a scan slow) contains at least one recognized video/audio file anywhere. False for a folder that's really a course's bundled extras - downloaded source code, a Python virtualenv/dependency dump, a project's asset tree, etc. - which can have plenty of subfolders (so _is_unrecognized_leaf_item doesn't catch it) but no lecture content anywhere in them.""" try: entries = list(directory.iterdir()) except (PermissionError, OSError): return False for e in entries: if e.is_file() and not e.name.startswith('.') and e.suffix.lower() in VIDEO_EXTENSIONS | AUDIO_EXTENSIONS: return True if max_depth <= 0: return False for e in entries: if e.is_dir() and not e.name.startswith('.') and _subtree_has_any_media(e, max_depth - 1): return True return False def _is_media_free_subtree(directory: Path) -> bool: """A folder that has subfolders (so it isn't just an intentionally empty, freshly-created category) but no video/audio anywhere beneath it - excluded from the destination picker for the same reason as _is_unrecognized_leaf_item, just for a deeper non-course tree instead of a single stray file.""" try: has_subdir = any(e.is_dir() and not e.name.startswith('.') for e in directory.iterdir()) except (PermissionError, OSError): return False if not has_subdir: return False return not _subtree_has_any_media(directory) def _is_course_wrapper(directory: Path, min_similarity: float = 0.5) -> bool: """ A release-bundle folder whose name is really just a course title with extra decoration (platform name, distribution tag, ...), wrapping a single actual course folder one level down alongside unrelated junk at the same level - e.g. a "Udemy...BOOKWARE-XYZ" folder containing both "Course Title Here/Chapter 1/video.mp4" *and* a pile of bundled source code/empty leftover folders from extraction as siblings. _looks_like_course only checks one level of nesting, so this reads as a bare category with a course buried inside it - and unlike _is_media_free_subtree, the junk siblings here can be perfectly ordinary empty folders (indistinguishable from an intentionally empty, freshly-created category), so that check alone doesn't catch it. Detected by token similarity (the same Jaccard measure _find_duplicate_courses uses) between the directory's own name and its one course child's name: high overlap means they're naming the same thing, not a deliberate parent/child relationship like "IT/AWS" holding an "AWS Certified..." course, whose names barely overlap. """ try: children = [c for c in directory.iterdir() if c.is_dir() and not c.name.startswith('.')] except (PermissionError, OSError): return False course_children = [c for c in children if _looks_like_course(c)] if len(course_children) != 1: return False dir_tokens = _tokenize(directory.name) course_tokens = _tokenize(course_children[0].name) union_tokens = dir_tokens | course_tokens if not union_tokens: return False similarity = len(dir_tokens & course_tokens) / len(union_tokens) return similarity >= min_similarity def _build_category_index() -> List[Dict[str, Any]]: """ Every category/subcategory folder currently in the library (e.g. IT, IT/AWS) as a possible destination, each with keyword tokens split into two tiers of trust: - 'path_tokens': from the folder's own path components - a deliberate name the user chose, so a match here is always taken at face value. - 'bonus_tokens': from the titles of courses already filed directly under it - lets a folder match on subjects its name alone doesn't mention (an "IT/AWS" folder full of courses titled "...Solutions Architect..." now also matches on "solutions" and "architect"), without hardcoding any subject list. Since these come from whatever the courses happen to be named, they're incidental rather than deliberate - _propose_destination downweights them by how many *different* categories share the same bonus word, so a word that turns out to be common clutter (recurring across many categories' course titles) can't outweigh a specific match elsewhere just by sharing more of it. Stops recursing into a folder once it reads as a course itself (_looks_like_course, the same rule the Library browser uses), an unrecognized leaf item (_is_unrecognized_leaf_item - an ebook/ audiobook/misc file that isn't video or audio), a subtree with no video/audio anywhere in it at all (_is_media_free_subtree - a course's bundled source code/project files, a Python virtualenv, etc.), or a release-bundle wrapper around a single course one level down (_is_course_wrapper - the video content sits under an extra "course display name" folder, one level deeper than _looks_like_course checks, alongside unrelated - possibly empty - junk siblings), so individual courses and their non-lecture contents never show up as if they were categories to file things under. The Unsorted folder itself is excluded - it's the source, never a valid destination. This walk visits every directory in the library and, over a NAS-mounted (SMB) library, each of those is a network round-trip - the same reasoning behind get_all_course_dirs()'s cache, so this result is cached the same way (same 5-minute TTL, same invalidate_cache() call sites already busting it - no separate invalidation needed): otherwise this ran fresh on every File Management page load and Sort Unsorted scan, neither of which is a manual "this may take a moment" action like Duplicate Courses or Storage Usage, so the cost was invisible until it was already slow. """ def compute() -> List[Dict[str, Any]]: library_root = Path(get_library_root()) try: unsorted_root = (library_root / UNSORTED_FOLDER_NAME).resolve() except OSError: unsorted_root = None categories: List[Dict[str, Any]] = [] def walk(directory: Path, path_parts: List[str]): 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: try: if unsorted_root is not None and entry.resolve() == unsorted_root: continue except OSError: pass if (_looks_like_course(entry) or _is_unrecognized_leaf_item(entry) or _is_media_free_subtree(entry) or _is_course_wrapper(entry)): continue new_parts = path_parts + [entry.name] path_tokens: Set[str] = set() for part in new_parts: path_tokens |= _tokenize(part) bonus_tokens: Set[str] = set() try: for child in entry.iterdir(): if child.is_dir() and not child.name.startswith('.') and _looks_like_course(child): bonus_tokens |= _tokenize(child.name) except (PermissionError, OSError): pass bonus_tokens -= path_tokens categories.append({ 'path': str(entry), 'relative': '/'.join(new_parts), 'depth': len(new_parts), 'path_tokens': path_tokens, 'bonus_tokens': bonus_tokens, }) walk(entry, new_parts) walk(library_root, []) return categories return cache_get_or_compute('category_index', compute) def _bonus_token_frequency(categories: List[Dict[str, Any]]) -> Dict[str, int]: """How many distinct categories each bonus token shows up in, across the whole library - used to tell a word that's actually specific to one place ("cloudformation") from one that's just common clutter recurring in course titles everywhere ("advanced", a creator's name, etc.), so the latter can be downweighted instead of needing to be individually named as a stopword ahead of time.""" freq: Dict[str, int] = {} for cat in categories: for token in cat['bonus_tokens']: freq[token] = freq.get(token, 0) + 1 return freq def _list_unsorted_items() -> List[Path]: """Direct children of the Unsorted folder - each one is a course waiting to be filed into the real category tree.""" unsorted_root = Path(get_library_root()) / UNSORTED_FOLDER_NAME if not unsorted_root.is_dir(): return [] return sorted( (p for p in unsorted_root.iterdir() if p.is_dir() and not p.name.startswith('.')), key=lambda p: p.name.lower() ) def _propose_destination(item_name: str, categories: List[Dict[str, Any]], bonus_df: Dict[str, int]) -> Dict[str, Any]: """ Best-effort destination for one Unsorted item, by keyword overlap against the existing category tree. Three outcomes: - 'existing': a real folder already matches confidently - move there. - 'new_folder': a broader category matches, but nothing that specific already exists under it - propose creating a new subfolder there, named from whichever of the item's own keywords the category didn't already cover. - 'manual': nothing matches with any confidence - needs a human. Categories are ranked by evidence quality, not raw shared-word count - a match on a *path* token (the folder's own deliberate name, e.g. the "AWS" in IT/AWS) always outranks any number of *bonus* tokens (words only borrowed from a sibling course's title - see _build_category_index), since a folder's own name is deliberate and a sibling course's title is incidental. Among bonus-only matches, the single least-diluted word wins rather than the total - a bonus word shared by only one category counts fully, one recurring across N categories counts for 1/N - so several incidental, recurring words (a prolific creator's name, a repeated marketing phrase, etc. showing up across many categories' course titles) can't outvote one genuinely specific word just by there being more of them, and don't need to be individually named as stopwords ahead of time to be discounted. "Confident" then means the best match is either a path-token hit or an essentially-unique bonus word (not diluted across other categories), AND it's either at a fairly specific existing folder (2+ levels deep, e.g. IT/AWS) or corroborated by a second shared word - one weak, common word alone isn't enough to drop something straight into a bare top-level category ("IT"); that's exactly the case a new subfolder proposal is for. """ item_tokens_ordered = _tokenize_ordered(item_name) item_tokens = set(item_tokens_ordered) if not item_tokens: return {'type': 'manual', 'destination': None, 'reason': "Couldn't extract any usable keywords from this name."} scored = [] for cat in categories: overlap = item_tokens & (cat['path_tokens'] | cat['bonus_tokens']) if not overlap: continue path_hits = overlap & cat['path_tokens'] bonus_overlap = overlap - cat['path_tokens'] bonus_best = max((1.0 / bonus_df.get(t, 1) for t in bonus_overlap), default=0.0) scored.append({ 'cat': cat, 'overlap': overlap, 'path_hits': len(path_hits), 'bonus_best': bonus_best, 'count': len(overlap), }) if not scored: return {'type': 'manual', 'destination': None, 'reason': "No existing folder shares any keywords with this name."} scored.sort(key=lambda s: (s['path_hits'], s['bonus_best'], s['count'], s['cat']['depth']), reverse=True) best = scored[0] best_cat = best['cat'] overlap_words = ', '.join(w for w in item_tokens_ordered if w in best['overlap']) best_all_tokens = best_cat['path_tokens'] | best_cat['bonus_tokens'] is_strong_match = best['path_hits'] >= 1 or best['bonus_best'] >= 1.0 if is_strong_match and (best_cat['depth'] >= 2 or best['count'] >= 2): return { 'type': 'existing', 'destination': best_cat['relative'], 'reason': f'Matches existing folder "{best_cat["relative"]}" on: {overlap_words}', } leftover = item_tokens - best_all_tokens if leftover: # preserve the item's own word order rather than the set's # arbitrary one, so "Docker Compose Basics" suggests "Compose # Basics" (dropping the already-covered "docker"), not a # scrambled "Basics Compose" leftover_ordered = [w for w in item_tokens_ordered if w in leftover] new_name = ' '.join(w.upper() if len(w) <= 3 else w.capitalize() for w in leftover_ordered) return { 'type': 'new_folder', 'destination': f"{best_cat['relative']}/{new_name}", 'reason': f'"{best_cat["relative"]}" matches on {overlap_words}, but no existing subfolder covers this - suggesting a new one.', } return { 'type': 'existing', 'destination': best_cat['relative'], 'reason': f'Matches existing folder "{best_cat["relative"]}" on: {overlap_words}', } IGNORED_DUPLICATES_FILE = os.path.join(DATA_DIR, 'ignored_duplicates.json') def get_ignored_duplicate_pairs() -> Set[Tuple[str, str]]: """Pairs of course paths the user has confirmed are NOT duplicates of each other, despite matching the tokenizer - excluded from future duplicate scans (see _find_duplicate_courses) so a wrong call doesn't keep resurfacing. Each pair is stored path-order-independent.""" try: if os.path.exists(IGNORED_DUPLICATES_FILE): with open(IGNORED_DUPLICATES_FILE, 'r') as f: data = json.load(f) if isinstance(data, list): return {tuple(sorted(pair)) for pair in data if isinstance(pair, list) and len(pair) == 2} except (json.JSONDecodeError, OSError) as e: print(f"Could not load ignored duplicates: {e}") return set() def _save_ignored_duplicate_pairs(pairs: Set[Tuple[str, str]]) -> None: os.makedirs(DATA_DIR, exist_ok=True) with open(IGNORED_DUPLICATES_FILE, 'w') as f: json.dump([list(p) for p in sorted(pairs)], f, indent=2) def add_ignored_duplicate_pairs(pairs: List[Tuple[str, str]]) -> None: """Mark one or more course-path pairs as confirmed-not-duplicates.""" existing = get_ignored_duplicate_pairs() existing |= {tuple(sorted(p)) for p in pairs} _save_ignored_duplicate_pairs(existing) def remove_ignored_duplicate_pair(path_a: str, path_b: str) -> None: """Undo add_ignored_duplicate_pairs for one pair - lets a wrong ignore call be reversed, matching the reversible-curation pattern hidden paths/Next Up already follow elsewhere in the app.""" existing = get_ignored_duplicate_pairs() existing.discard(tuple(sorted((path_a, path_b)))) _save_ignored_duplicate_pairs(existing) def _find_duplicate_courses(min_similarity: float = 0.6) -> List[Dict[str, Any]]: """ Group courses across the whole library (Unsorted included, so a just-dropped item that duplicates something already filed gets caught too) whose names look like the same thing filed twice - a re-download that landed in a different category, or a course that's both sorted and still sitting in Unsorted. Reuses Sort Unsorted's own tokenizer/stopword list, so platform names, release-group tags, and dates are already ignored - "AWS Certified Solutions Architect" and "AWS.Certified.Solutions.Architect.2024.BOOKWARE-XYZ" tokenize down to the same core words despite looking nothing alike as strings. Pairwise Jaccard similarity (shared tokens / all tokens) at or above min_similarity gets union-found into groups, so if A matches B and B matches C, all three land in one group instead of two overlapping pairs - except any pair the user has explicitly marked as not-actually- duplicates (get_ignored_duplicate_pairs), which never unions regardless of how similar the names look. Nothing here touches disk - it's read-only, for the user to review and hide/delete/rename/move manually. """ ignored_pairs = get_ignored_duplicate_pairs() entries = [] for course in get_all_course_dirs(): tokens = _tokenize(course.name) if tokens: entries.append({'path': str(course), 'name': course.name, 'tokens': tokens}) n = len(entries) parent = list(range(n)) def find(i: int) -> int: while parent[i] != i: parent[i] = parent[parent[i]] i = parent[i] return i def union(i: int, j: int) -> None: ri, rj = find(i), find(j) if ri != rj: parent[ri] = rj pair_similarity: Dict[Tuple[int, int], float] = {} for i in range(n): for j in range(i + 1, n): if tuple(sorted((entries[i]['path'], entries[j]['path']))) in ignored_pairs: continue a, b = entries[i]['tokens'], entries[j]['tokens'] union_tokens = a | b if not union_tokens: continue similarity = len(a & b) / len(union_tokens) if similarity >= min_similarity: pair_similarity[(i, j)] = similarity union(i, j) clusters: Dict[int, List[int]] = {} for i in range(n): clusters.setdefault(find(i), []).append(i) groups = [] for members in clusters.values(): if len(members) < 2: continue best_similarity = max( (sim for (i, j), sim in pair_similarity.items() if i in members and j in members), default=0.0 ) groups.append({ 'courses': sorted( ({'path': entries[i]['path'], 'name': entries[i]['name']} for i in members), key=lambda c: c['name'].lower() ), 'similarity': round(best_similarity, 2), }) groups.sort(key=lambda g: g['similarity'], reverse=True) return groups def _find_stale_references() -> Dict[str, List[Dict[str, str]]]: """ Entries in hidden_paths.json / next_up.json / recent_views.json that point at a path no longer on disk - typically because a course was renamed, moved, or deleted directly on the NAS rather than through the app (a rename/move made through the app already rebases these, see rebase_library_path). Read-only; nothing is removed until /api/library/stale-references/clean is called with the exact items reviewed here. """ stale: Dict[str, List[Dict[str, str]]] = {'hidden': [], 'next_up': [], 'recent_views': [], 'favorites': []} for path in get_hidden_paths(): if not os.path.isdir(path): stale['hidden'].append({'path': path, 'name': os.path.basename(path.rstrip(os.sep)) or path}) for path in get_next_up_paths(): if not os.path.isdir(path): stale['next_up'].append({'path': path, 'name': os.path.basename(path.rstrip(os.sep)) or path}) for path in get_favorite_paths(): if not os.path.isdir(path): stale['favorites'].append({'path': path, 'name': os.path.basename(path.rstrip(os.sep)) or path}) seen: Set[str] = set() for entry in get_recent_views(): path = entry.get('course_path', '') if path and path not in seen and not os.path.isdir(path): seen.add(path) name = entry.get('course_name') or os.path.basename(path.rstrip(os.sep)) or path stale['recent_views'].append({'path': path, 'name': name}) return stale def _probe_media_duration_seconds(file_path: Path) -> Optional[float]: """Read a single media file's duration via ffprobe. None if ffprobe is missing, the file isn't readable, or the output can't be parsed.""" try: result = subprocess.run( ['ffprobe', '-v', 'error', '-show_entries', 'format=duration', '-of', 'default=noprint_wrappers=1:nokey=1', str(file_path)], capture_output=True, text=True, timeout=15 ) except (OSError, subprocess.SubprocessError): return None if result.returncode != 0: return None try: return float(result.stdout.strip()) except ValueError: return None DURATION_CACHE_FILENAME = '.offlineu_duration_cache.json' def _ensure_media_durations_cached(course_dir: Path, media_files: List[Path]) -> Dict[str, Any]: """ Make sure every given media file has a known duration in the persistent per-course cache file, keyed by (relative path, size, mtime) - ffprobe only ever runs for a file that's new or has changed since it was last cached, so a container restart or the library scan's 5-minute in-memory cache expiring never re-probes an unchanged file. Returns the up-to-date {relative_path: {size, mtime, duration_seconds}} cache, so callers needing a single lesson's duration (not just the course total) don't have to re-read the cache file themselves. """ cache_file = course_dir / DURATION_CACHE_FILENAME try: with open(cache_file, 'r') as f: cache = json.load(f) except (FileNotFoundError, json.JSONDecodeError, OSError): cache = {} changed = False seen_keys = set() for media_path in media_files: try: rel_key = str(media_path.relative_to(course_dir)) stat = media_path.stat() except OSError: continue seen_keys.add(rel_key) cached = cache.get(rel_key) if not (cached and cached.get('size') == stat.st_size and cached.get('mtime') == stat.st_mtime): duration = _probe_media_duration_seconds(media_path) cache[rel_key] = {'size': stat.st_size, 'mtime': stat.st_mtime, 'duration_seconds': duration} changed = True # Drop entries for files that no longer exist, so a renamed/deleted # course's cache doesn't grow stale entries forever. stale_keys = set(cache.keys()) - seen_keys if stale_keys: for key in stale_keys: del cache[key] changed = True if changed: try: with open(cache_file, 'w') as f: json.dump(cache, f, indent=2) except OSError as e: print(f"Could not save duration cache: {e}") return cache def _course_total_duration_seconds(course_dir: Path, media_files: List[Path]) -> Optional[float]: """Total runtime across a course's video/audio files, for the library browser's "how long is this" display.""" if not media_files: return None cache = _ensure_media_durations_cached(course_dir, media_files) total = 0.0 have_any = False for media_path in media_files: try: rel_key = str(media_path.relative_to(course_dir)) except ValueError: continue duration = cache.get(rel_key, {}).get('duration_seconds') if duration: total += duration have_any = True return total if have_any else None def _cached_course_duration_seconds(course_dir: Path) -> float: """ Sum of whatever's already in a course's persistent duration cache file, without walking its files or ever touching ffprobe - for the library-wide "time remaining" stat, which needs to stay cheap enough to compute on every dashboard load (it runs inside the same per-course loop as _scan_library_activity). A course that's never been viewed, searched, or prewarmed yet just contributes 0 here; run "Precompute Lengths & Cover Art" in Settings for a complete total. """ try: with open(course_dir / DURATION_CACHE_FILENAME, 'r') as f: cache = json.load(f) except (FileNotFoundError, json.JSONDecodeError, OSError): return 0.0 return sum(entry.get('duration_seconds') or 0 for entry in cache.values()) def _course_summary(course_dir: Path) -> Dict[str, Any]: """Build the {type, name, path, media_files, hidden, has_thumbnail, duration_display} shape shared by the Library browser, search results, and Recently Added.""" def compute(): media_files = [ f for f in course_dir.rglob('*') if f.is_file() and f.suffix.lower() in VIDEO_EXTENSIONS | AUDIO_EXTENSIONS ] total_duration = _course_total_duration_seconds(course_dir, media_files) return { 'type': 'course', 'name': course_dir.name, 'path': str(course_dir), 'media_files': len(media_files), 'hidden': False, 'has_thumbnail': find_course_thumbnail(str(course_dir)) is not None, # under a minute isn't a meaningful "how long will this take" signal 'duration_display': format_duration(total_duration) if total_duration and total_duration >= 60 else None, } return dict(cache_get_or_compute(f'course_summary:{course_dir}', compute)) def _course_completion_percentage(course_dir: Path, media_files: int) -> Optional[float]: """ Cheap, uncached completion percentage for a single course card - reads only that course's own small progress JSON directly (same pattern as _scan_library_activity), never the cached _course_summary result, so a lesson finished a second ago shows up immediately instead of waiting out the library-scan cache TTL. None if nothing's completed yet, so untouched courses don't show a noisy "0%" badge. """ if media_files <= 0: return None try: with open(course_dir / '.offlineu_progress.json', 'r') as f: progress = json.load(f) except (FileNotFoundError, json.JSONDecodeError, OSError): return None completed = sum( 1 for key, entry in progress.items() if key != 'last_accessed_path' and isinstance(entry, dict) and entry.get('completed') ) if completed == 0: return None return round(min(100.0, 100 * completed / media_files), 1) 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() favorite_set = set(get_favorite_paths()) results = [] for course in get_all_course_dirs(): if query_lower not in course.name.lower(): continue item = _course_summary(course) item['favorited'] = os.path.abspath(str(course)) in favorite_set item['completion_percentage'] = _course_completion_percentage(course, item['media_files']) results.append(item) return results 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 total_duration_seconds = 0.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 total_duration_seconds += _cached_course_duration_seconds(course_dir) 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, 'total_duration_seconds': total_duration_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.""" remaining_seconds = max(0, scan['total_duration_seconds'] - scan['watched_seconds']) return { 'total_courses': scan['total_courses'], 'lessons_tracked': scan['lessons_tracked'], 'completed_lessons': scan['completed_lessons'], 'watched_display': format_duration(scan['watched_seconds']), # None until at least some course's duration cache has been # populated (via viewing/searching the library or running # "Precompute Lengths & Cover Art") - a remaining estimate from a # library that's mostly uncached would just be misleadingly low. 'remaining_display': format_duration(remaining_seconds) if scan['total_duration_seconds'] else None, '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 remove_recent_views_for_path(path: str) -> List[Dict[str, Any]]: """Drop every Recently Viewed entry for one course path - there can be several, one per lesson. Used to clear a stale reference left behind when a course was deleted/moved outside the app.""" entries = get_recent_views() normalized = os.path.abspath(path) remaining = [e for e in entries if e.get('course_path') != normalized] if remaining != entries: os.makedirs(DATA_DIR, exist_ok=True) with open(RECENT_VIEWS_FILE, 'w') as f: json.dump(remaining, f, indent=2) return remaining 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) # A lesson's duration is normally only known once it's been played # (the player reports it back from the