rmsitzandClaude Sonnet 5
3fedf849d3
view_lesson() called update_lesson_progress() with no arguments on every page load, which defaults completed=False and progress_seconds=0 and silently overwrote whatever was already saved - just opening an already-watched lesson reset its progress/completed state unless the client happened to report real values within the next 15 seconds. Replaced with touch_lesson_accessed(), which only bumps last_accessed. Also fixes the "Resume from Xs" feature this fed into: it read a value that was never populated on this route, and even when given a real one, checked activeMedia.duration synchronously before metadata had loaded, so it silently never applied. Now waits for loadedmetadata (or resolves immediately if already available) and defers to an explicit note-timestamp jump (?t=) when both are present. Notes Hub gets a live search box filtering by note text, lesson title, or course name (client-side, no reload), with a "no notes match" state - useful now that a single category folder can hold dozens of courses' worth of timestamped notes. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
3041 lines
119 KiB
Python
3041 lines
119 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 re
|
|
import sys
|
|
import argparse
|
|
import io
|
|
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')
|
|
|
|
# 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)
|
|
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)
|
|
}
|
|
|
|
|
|
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)
|
|
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):
|
|
media_count = len([
|
|
f for f in entry.rglob('*')
|
|
if f.is_file() and f.suffix.lower() in VIDEO_EXTENSIONS | AUDIO_EXTENSIONS
|
|
])
|
|
items.append({
|
|
'type': 'course',
|
|
'name': entry.name,
|
|
'path': entry_path_str,
|
|
'media_files': media_count,
|
|
'hidden': is_hidden,
|
|
'has_thumbnail': find_course_thumbnail(entry_path_str) is not None
|
|
})
|
|
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 _course_summary(course_dir: Path) -> Dict[str, Any]:
|
|
"""Build the {type, name, path, media_files, hidden, has_thumbnail} shape
|
|
shared by the Library browser, search results, and Recently Added."""
|
|
media_count = len([
|
|
f for f in course_dir.rglob('*')
|
|
if f.is_file() and f.suffix.lower() in VIDEO_EXTENSIONS | AUDIO_EXTENSIONS
|
|
])
|
|
return {
|
|
'type': 'course',
|
|
'name': course_dir.name,
|
|
'path': str(course_dir),
|
|
'media_files': media_count,
|
|
'hidden': False,
|
|
'has_thumbnail': find_course_thumbnail(str(course_dir)) is not None
|
|
}
|
|
|
|
|
|
def search_library_courses(dir_path: str, query: str) -> List[Dict[str, Any]]:
|
|
"""
|
|
Search the library for courses whose name contains `query`
|
|
(case-insensitive). The expensive per-course lookups (media_count,
|
|
thumbnail) only run for courses whose name actually matches - see
|
|
iter_all_courses.
|
|
"""
|
|
query_lower = query.lower()
|
|
return [
|
|
_course_summary(course)
|
|
for course in iter_all_courses(dir_path)
|
|
if query_lower in course.name.lower()
|
|
]
|
|
|
|
|
|
def _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 search_transcripts(query: str, limit: int = 30) -> List[Dict[str, Any]]:
|
|
"""
|
|
Search inside lesson subtitle files (.srt/.vtt/etc.) for `query`,
|
|
case-insensitive. Unlike course-name search, subtitles live inside each
|
|
course's own directory tree rather than at the course boundary, so this
|
|
needs an actual per-course file walk - kept cheap by doing a raw
|
|
substring check as the filter step and only extracting a cleaned-up
|
|
snippet for files that actually match.
|
|
|
|
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 iter_all_courses(get_library_root()):
|
|
for subtitle_file in course_dir.rglob('*'):
|
|
if len(results) >= limit:
|
|
return results
|
|
if not subtitle_file.is_file() or subtitle_file.suffix.lower() not in SUBTITLE_EXTENSIONS:
|
|
continue
|
|
try:
|
|
text = subtitle_file.read_text(encoding='utf-8', errors='ignore')
|
|
except OSError:
|
|
continue
|
|
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).
|
|
"""
|
|
library_root = get_library_root()
|
|
dated = []
|
|
for course_dir in iter_all_courses(library_root):
|
|
try:
|
|
mtime = course_dir.stat().st_mtime
|
|
except OSError:
|
|
continue
|
|
dated.append((mtime, course_dir))
|
|
dated.sort(key=lambda pair: pair[0], reverse=True)
|
|
|
|
results = []
|
|
for mtime, course_dir in dated[:limit]:
|
|
item = _course_summary(course_dir)
|
|
item['added_display'] = _humanize_days_ago(datetime.fromtimestamp(mtime))
|
|
results.append(item)
|
|
return results
|
|
|
|
|
|
def _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 iter_all_courses(get_library_root()):
|
|
total_courses += 1
|
|
try:
|
|
with open(course_dir / '.offlineu_progress.json', 'r') as f:
|
|
progress = json.load(f)
|
|
except (FileNotFoundError, json.JSONDecodeError, OSError):
|
|
continue
|
|
|
|
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 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 iter_all_courses(get_library_root()):
|
|
try:
|
|
with open(course_dir / '.offlineu_progress.json', 'r') as f:
|
|
progress = json.load(f)
|
|
except (FileNotFoundError, json.JSONDecodeError, OSError):
|
|
continue
|
|
|
|
for lesson_path, entry in progress.items():
|
|
if lesson_path == 'last_accessed_path' or not isinstance(entry, dict):
|
|
continue
|
|
for note in ProgressTracker._notes_from_entry(entry):
|
|
created_at = note.get('created_at') or entry.get('last_accessed', '')
|
|
notes.append({
|
|
'course_path': str(course_dir),
|
|
'course_name': course_dir.name,
|
|
'lesson_path': lesson_path,
|
|
'lesson_title': lesson_path.rsplit('/', 1)[-1].replace('_', ' '),
|
|
'note_id': note.get('id'),
|
|
'text': note.get('text', ''),
|
|
'timestamp_seconds': note.get('timestamp_seconds'),
|
|
'timestamp_label': format_timestamp(note.get('timestamp_seconds')),
|
|
'created_at': created_at,
|
|
'pushed_to_outline': bool(entry.get('outline_document_id'))
|
|
})
|
|
|
|
notes.sort(key=lambda n: n['created_at'], reverse=True)
|
|
return notes
|
|
|
|
|
|
def build_study_guide_markdown(course: Course) -> str:
|
|
"""
|
|
Compile every note written for a course into one markdown document,
|
|
following the course's own section/lesson structure - only sections
|
|
and lessons that actually have a note appear; everything else is
|
|
skipped rather than padding the guide with empty headings.
|
|
"""
|
|
progress = ProgressTracker.load_progress(course)
|
|
|
|
def node_has_notes(node: DirectoryNode) -> bool:
|
|
for lesson in node.lessons:
|
|
key = _resolve_lesson_progress_key(course, lesson, progress)
|
|
if key and ProgressTracker._notes_from_entry(progress[key]):
|
|
return True
|
|
return any(node_has_notes(child) for child in node.children.values())
|
|
|
|
lines = [f"# {course.name}", ""]
|
|
|
|
def walk(node: DirectoryNode, heading_level: int):
|
|
if node.name and node.name != "Course Root":
|
|
if not node_has_notes(node):
|
|
return
|
|
lines.append(f"{'#' * min(heading_level, 6)} {node.name}")
|
|
lines.append("")
|
|
|
|
for lesson in node.lessons:
|
|
key = _resolve_lesson_progress_key(course, lesson, progress)
|
|
lesson_notes = ProgressTracker._notes_from_entry(progress[key]) if key else []
|
|
if not lesson_notes:
|
|
continue
|
|
lines.append(f"{'#' * min(heading_level + 1, 6)} {lesson.title}")
|
|
lines.append("")
|
|
for note in lesson_notes:
|
|
ts = format_timestamp(note.get('timestamp_seconds'))
|
|
prefix = f"**[{ts}]** " if ts else ""
|
|
lines.append(f"- {prefix}{note.get('text', '')}")
|
|
lines.append("")
|
|
|
|
for child in node.children.values():
|
|
walk(child, heading_level + 1)
|
|
|
|
walk(course.root_node, 2)
|
|
return "\n".join(lines)
|
|
|
|
|
|
# Global course storage
|
|
current_course = None
|
|
|
|
|
|
def _split_continue_watching(all_views: List[Dict[str, Any]]) -> Tuple[List[Dict[str, Any]], List[Dict[str, Any]]]:
|
|
"""
|
|
Split the recent-views list (already carrying percent_watched/completed,
|
|
see get_recent_views_for_display) into 'Continue Watching' - genuinely
|
|
in-progress lessons - and the remaining 'Recently Viewed' entries, each
|
|
capped at 5 for the dashboard. Reuses the same underlying history rather
|
|
than a separate library-wide progress index (see plan notes).
|
|
"""
|
|
continue_watching = [v for v in all_views if 0 < v['percent_watched'] < 100][:5]
|
|
continue_ids = {(v.get('course_path'), v.get('lesson_path')) for v in continue_watching}
|
|
recent_views = [
|
|
v for v in all_views
|
|
if (v.get('course_path'), v.get('lesson_path')) not in continue_ids
|
|
][:5]
|
|
return continue_watching, recent_views
|
|
|
|
|
|
def format_duration(seconds: int) -> str:
|
|
"""Render a duration in seconds as e.g. '3h 24m' or '45m' (under an hour)."""
|
|
total_minutes = max(0, int(seconds)) // 60
|
|
hours, minutes = divmod(total_minutes, 60)
|
|
return f"{hours}h {minutes}m" if hours else f"{minutes}m"
|
|
|
|
|
|
@app.route('/')
|
|
def index():
|
|
"""Main dashboard"""
|
|
global current_course
|
|
|
|
continue_watching, recent_views = _split_continue_watching(get_recent_views_for_display())
|
|
|
|
if current_course is None:
|
|
# One scan covers stats/stale-courses/heatmap - see _scan_library_activity.
|
|
scan = _scan_library_activity()
|
|
return render_template('course_dashboard.html',
|
|
course=None,
|
|
stats={'total_lessons': 0, 'completed_lessons': 0, 'completion_percentage': 0},
|
|
continue_watching=continue_watching,
|
|
recent_views=recent_views,
|
|
recently_added=get_recently_added_courses(),
|
|
library_stats=format_library_stats(scan),
|
|
stale_courses=format_stale_courses(scan),
|
|
activity_heatmap=format_activity_heatmap(scan),
|
|
next_up=get_next_up_courses())
|
|
|
|
# Apply progress data to tree
|
|
ProgressTracker.apply_progress_to_tree(current_course)
|
|
stats = ProgressTracker.get_completion_stats(current_course)
|
|
if stats.get('total_duration_seconds'):
|
|
stats['remaining_display'] = format_duration(stats['remaining_seconds'])
|
|
|
|
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())
|
|
|
|
|
|
@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(library_root, query)
|
|
return jsonify({'library_path': library_root, 'results': results})
|
|
|
|
|
|
@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})
|
|
|
|
|
|
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}
|
|
|
|
rebase_library_path(old_abs, new_abs)
|
|
|
|
active_course_reset = False
|
|
if current_course is not None:
|
|
course_abs = os.path.abspath(current_course.path)
|
|
if course_abs == old_abs or course_abs.startswith(old_abs + os.sep):
|
|
current_course = None
|
|
active_course_reset = True
|
|
|
|
return {
|
|
'success': True,
|
|
'new_path': new_abs,
|
|
'active_course_reset': active_course_reset,
|
|
'status': 200
|
|
}
|
|
|
|
|
|
@app.route('/api/rename-path', methods=['POST'])
|
|
def rename_path_api():
|
|
"""Rename a course/folder directory in the library, in place on disk."""
|
|
data = request.json or {}
|
|
path = data.get('path', '')
|
|
new_name = (data.get('new_name') or '').strip()
|
|
|
|
if not path:
|
|
return jsonify({'error': 'path is required'}), 400
|
|
|
|
result = perform_rename(path, new_name)
|
|
status = result.pop('status')
|
|
return jsonify(result), status
|
|
|
|
|
|
def _iter_all_directories(directory: Path, hidden_set: set) -> Iterator[Path]:
|
|
"""
|
|
Yield every directory in the library tree - both course folders and the
|
|
category/group folders above them - skipping hidden ones and not
|
|
descending into a course's own internal sections (Section 1, etc. -
|
|
those aren't independently manageable library items anywhere else in
|
|
the app either, so bulk rename shouldn't touch them).
|
|
"""
|
|
try:
|
|
entries = sorted(
|
|
(p for p in directory.iterdir() if p.is_dir() and not p.name.startswith('.')),
|
|
key=lambda p: p.name.lower()
|
|
)
|
|
except (PermissionError, OSError):
|
|
return
|
|
|
|
for entry in entries:
|
|
if os.path.abspath(str(entry)) in hidden_set:
|
|
continue
|
|
yield entry
|
|
if not _looks_like_course(entry):
|
|
yield from _iter_all_directories(entry, hidden_set)
|
|
|
|
|
|
def find_bulk_rename_matches(library_root: str, pattern: str, replacement: str) -> List[Dict[str, str]]:
|
|
"""Every directory (course or folder) in the library whose name contains `pattern`."""
|
|
hidden_set = set(get_hidden_paths())
|
|
return [
|
|
{
|
|
'path': str(directory),
|
|
'old_name': directory.name,
|
|
'new_name': directory.name.replace(pattern, replacement)
|
|
}
|
|
for directory in _iter_all_directories(Path(library_root), hidden_set)
|
|
if pattern in directory.name
|
|
]
|
|
|
|
|
|
@app.route('/api/bulk-rename/preview')
|
|
def bulk_rename_preview_api():
|
|
"""Preview a find/replace rename across the whole library, without touching disk."""
|
|
pattern = request.args.get('pattern', '')
|
|
replacement = request.args.get('replacement', '')
|
|
if not pattern:
|
|
return jsonify({'error': 'pattern is required'}), 400
|
|
|
|
library_root = os.path.abspath(get_library_root())
|
|
matches = find_bulk_rename_matches(library_root, pattern, replacement)
|
|
return jsonify({'matches': matches})
|
|
|
|
|
|
@app.route('/api/bulk-rename/apply', methods=['POST'])
|
|
def bulk_rename_apply_api():
|
|
"""
|
|
Apply a bulk rename. Takes the *exact* {path, new_name} list the client
|
|
got back from /api/bulk-rename/preview, rather than re-deriving matches
|
|
from the pattern again - what gets renamed is provably what the user
|
|
saw and approved, and can't drift if the library changed in between.
|
|
Continues past individual failures (e.g. a collision) instead of
|
|
aborting the whole batch, reporting a per-item result.
|
|
"""
|
|
data = request.json or {}
|
|
items = data.get('items') or []
|
|
if not isinstance(items, list) or not items:
|
|
return jsonify({'error': 'items is required'}), 400
|
|
|
|
results = []
|
|
for item in items:
|
|
path = item.get('path', '')
|
|
new_name = item.get('new_name', '')
|
|
outcome = perform_rename(path, new_name)
|
|
outcome.pop('status', None)
|
|
outcome['path'] = path
|
|
outcome['old_name'] = item.get('old_name', '')
|
|
outcome['new_name'] = new_name
|
|
results.append(outcome)
|
|
|
|
return jsonify({'results': results})
|
|
|
|
|
|
@app.route('/settings')
|
|
def settings_page():
|
|
"""Render the display-settings page."""
|
|
current = load_settings()
|
|
named_presets = sorted(
|
|
(t for t in SETTINGS_CHOICES['theme'] if t not in ('dark', 'light')),
|
|
key=lambda t: THEME_DISPLAY_NAMES[t]
|
|
)
|
|
return render_template(
|
|
'settings.html',
|
|
settings=current,
|
|
default_library_path=LIBRARY_PATH,
|
|
theme_choices=['dark', 'light'] + named_presets,
|
|
theme_display_names=THEME_DISPLAY_NAMES,
|
|
font_family_choices=['system', 'sans', 'serif', 'monospace', 'monaspace'],
|
|
font_size_choices=['small', 'medium', 'large', 'xlarge'],
|
|
layout_width_choices=['normal', 'wide', 'full'],
|
|
density_choices=['comfortable', 'compact'],
|
|
card_style_choices=['flat', 'elevated', 'bordered'],
|
|
corner_radius_choices=['sharp', 'rounded', 'pill'],
|
|
)
|
|
|
|
|
|
@app.route('/api/settings', methods=['GET'])
|
|
def get_settings_api():
|
|
"""Return current settings plus their resolved CSS values, for the shared theme script."""
|
|
current = load_settings()
|
|
return jsonify({
|
|
'settings': current,
|
|
'css_vars': settings_css_vars(current)
|
|
})
|
|
|
|
|
|
@app.route('/api/settings', methods=['POST'])
|
|
def save_settings_api():
|
|
"""Persist updated display settings."""
|
|
try:
|
|
new_settings = request.get_json(force=True) or {}
|
|
saved = save_settings(new_settings)
|
|
return jsonify({
|
|
'success': True,
|
|
'settings': saved,
|
|
'css_vars': settings_css_vars(saved)
|
|
})
|
|
except Exception as e:
|
|
return jsonify({'success': False, 'error': str(e)}), 400
|
|
|
|
|
|
@app.route('/api/settings/reset', methods=['POST'])
|
|
def reset_settings_api():
|
|
"""Reset settings back to defaults."""
|
|
os.makedirs(DATA_DIR, exist_ok=True)
|
|
with open(SETTINGS_FILE, 'w') as f:
|
|
json.dump(DEFAULT_SETTINGS, f, indent=2)
|
|
return jsonify({
|
|
'success': True,
|
|
'settings': DEFAULT_SETTINGS,
|
|
'css_vars': settings_css_vars(DEFAULT_SETTINGS)
|
|
})
|
|
|
|
|
|
@app.route('/api/backup')
|
|
def download_backup():
|
|
"""
|
|
Bundle everything that isn't recoverable from the course files
|
|
themselves - settings, hidden-path curation, recently-viewed history,
|
|
and every course's progress/notes file - into a single downloadable
|
|
zip. Cheap insurance before a NAS migration or a docker volume mistake.
|
|
"""
|
|
buffer = io.BytesIO()
|
|
with zipfile.ZipFile(buffer, 'w', zipfile.ZIP_DEFLATED) as zf:
|
|
for name, path in (
|
|
('settings.json', SETTINGS_FILE),
|
|
('hidden_paths.json', HIDDEN_PATHS_FILE),
|
|
('recent_views.json', RECENT_VIEWS_FILE),
|
|
):
|
|
if os.path.exists(path):
|
|
zf.write(path, name)
|
|
|
|
library_root = get_library_root()
|
|
for course_dir in iter_all_courses(library_root):
|
|
progress_file = course_dir / '.offlineu_progress.json'
|
|
if progress_file.exists():
|
|
relative = os.path.relpath(str(course_dir), library_root)
|
|
zf.write(progress_file, os.path.join('progress', relative, '.offlineu_progress.json'))
|
|
|
|
buffer.seek(0)
|
|
filename = f"offlineu-backup-{datetime.now().strftime('%Y-%m-%d')}.zip"
|
|
return send_file(buffer, mimetype='application/zip', as_attachment=True, download_name=filename)
|
|
|
|
|
|
@app.route('/load_course', methods=['POST'])
|
|
def load_course():
|
|
"""Load course from selected directory"""
|
|
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 = DynamicCourseParser.scan_directory(course_path)
|
|
return jsonify({'success': True, 'course_name': current_course.name})
|
|
except Exception as e:
|
|
return jsonify({'error': str(e)}), 500
|
|
|
|
|
|
@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 = DynamicCourseParser.scan_directory(course_path)
|
|
except Exception as e:
|
|
print(f"Could not load course for recent view: {e}")
|
|
return redirect(url_for('index'))
|
|
|
|
lesson_url = url_for('view_lesson', lesson_path=lesson_path)
|
|
if seek_seconds:
|
|
lesson_url = f"{lesson_url}?t={seek_seconds}"
|
|
return redirect(lesson_url)
|
|
|
|
|
|
@app.route('/help')
|
|
def help_page():
|
|
"""Static help page: how to use the app, supported file types."""
|
|
return render_template('help.html')
|
|
|
|
|
|
@app.route('/notes')
|
|
def notes_hub():
|
|
"""Every lesson note across the whole library in one place."""
|
|
return render_template('notes_hub.html', notes=get_all_notes())
|
|
|
|
|
|
@app.route('/library/search-transcripts')
|
|
def search_transcripts_api():
|
|
"""Search inside lesson subtitle files across the library."""
|
|
query = request.args.get('q', '').strip()
|
|
if not query:
|
|
return jsonify({'results': []})
|
|
return jsonify({'results': search_transcripts(query)})
|
|
|
|
|
|
@app.route('/course/study-guide')
|
|
def download_study_guide():
|
|
"""Download every note written for the currently loaded course as one markdown file."""
|
|
global current_course
|
|
if not current_course:
|
|
return "No course loaded", 404
|
|
|
|
markdown = build_study_guide_markdown(current_course)
|
|
buffer = io.BytesIO(markdown.encode('utf-8'))
|
|
safe_name = re.sub(r'[^\w\-. ]', '_', current_course.name).strip() or 'course'
|
|
return send_file(buffer, mimetype='text/markdown', as_attachment=True,
|
|
download_name=f"{safe_name} - Study Guide.md")
|
|
|
|
|
|
@app.route('/api/next-up', methods=['POST'])
|
|
def set_next_up_api():
|
|
"""Add or remove a course from the Next Up queue."""
|
|
data = request.json or {}
|
|
path = data.get('path', '')
|
|
queued = bool(data.get('queued', True))
|
|
if not path:
|
|
return jsonify({'error': 'path is required'}), 400
|
|
|
|
library_root = os.path.abspath(get_library_root())
|
|
target = os.path.abspath(path)
|
|
if not (target == library_root or target.startswith(library_root + os.sep)):
|
|
return jsonify({'error': 'Path outside library root'}), 403
|
|
|
|
updated = set_path_queued(target, queued)
|
|
return jsonify({'success': True, 'next_up': updated})
|
|
|
|
|
|
@app.route('/api/next-up/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
|
|
lesson = 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', ''),
|
|
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) -> Optional[Lesson]:
|
|
"""Find a lesson in the tree by path"""
|
|
# 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
|
|
|
|
# Recursively search children
|
|
for child in node.children.values():
|
|
result = find_lesson_in_tree(child, target_path)
|
|
if result:
|
|
return result
|
|
|
|
return None
|
|
|
|
|
|
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)
|