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