(no message) Changes: import_masterclass.py

This commit is contained in:
Steven (Zen) 2026-08-11 15:02:45 -06:00
parent 20aa5f87bd
commit 4932f397b9

View File

@ -3,6 +3,12 @@
import_masterclass.py import_masterclass.py
Creates a Masterclass.library in Eagle format from /volume1/steve/Courses/masterclass/ Creates a Masterclass.library in Eagle format from /volume1/steve/Courses/masterclass/
Run on the Synology NAS. Run on the Synology NAS.
Eagle requirements confirmed from working library inspection:
- Each .info dir must contain: <original-file>, <stem>_thumbnail.png, metadata.json
- metadata.json "name" must match filename stem EXACTLY
- Library root needs: metadata.json, mtime.json, actions.json, tags.json, saved-filters.json
- mtime.json must have all item IDs -> modificationTime, plus "all" -> count
""" """
import os import os
@ -10,26 +16,25 @@ import json
import time import time
import subprocess import subprocess
import sys import sys
import hashlib
from pathlib import Path from pathlib import Path
from collections import defaultdict
SOURCE = Path("/volume1/steve/Courses/masterclass") SOURCE = Path("/volume1/steve/Courses/masterclass")
LIBRARY = Path("/volume1/steve/Eagle_Libraries/Masterclass.library") LIBRARY = Path("/volume1/steve/Eagle_Libraries/Masterclass.library")
IMAGES = LIBRARY / "images" IMAGES = LIBRARY / "images"
SUPPORTED_EXTS = { SUPPORTED_EXTS = {
"mp4", "mov", "m4v", "avi", "mkv", "webm", # video "mp4", "mov", "m4v", "avi", "mkv", "webm",
"mp3", "m4a", "aac", "wav", "flac", # audio "mp3", "m4a", "aac", "wav", "flac",
"jpg", "jpeg", "png", "gif", "webp", # image "jpg", "jpeg", "png", "gif", "webp",
"pdf", # docs "pdf",
} }
VIDEO_EXTS = {"mp4", "mov", "m4v", "avi", "mkv", "webm"} VIDEO_EXTS = {"mp4", "mov", "m4v", "avi", "mkv", "webm"}
IMAGE_EXTS = {"jpg", "jpeg", "png", "gif", "webp"} IMAGE_EXTS = {"jpg", "jpeg", "png", "gif", "webp"}
# ID counter
_id_counter = 0 _id_counter = 0
_id_prefix = f"MC{int(time.time()) % 10000000:07d}" _id_prefix = f"MC{int(time.time()) % 100000000:08d}"
def gen_id(): def gen_id():
global _id_counter global _id_counter
@ -39,68 +44,44 @@ def gen_id():
def now_ms(): def now_ms():
return int(time.time() * 1000) return int(time.time() * 1000)
def get_video_info(filepath): def generate_thumbnail(filepath, thumb_path):
"""Returns (width, height, duration) via ffprobe.""" """Generate thumbnail PNG. Returns True if successful."""
try:
result = subprocess.run(
["ffprobe", "-v", "quiet", "-print_format", "json", "-show_streams", str(filepath)],
capture_output=True, text=True, timeout=30
)
data = json.loads(result.stdout)
width, height, duration = 0, 0, None
for stream in data.get("streams", []):
if stream.get("codec_type") in ("video", "audio"):
w = stream.get("width", 0)
h = stream.get("height", 0)
if w and h:
width, height = w, h
dur = stream.get("duration")
if dur:
duration = float(dur)
if width and duration:
break
return width, height, duration
except Exception:
return 0, 0, None
def generate_thumbnail(filepath, thumb_path, duration=None):
"""Generate thumbnail PNG from video or image."""
ext = filepath.suffix.lower().lstrip(".") ext = filepath.suffix.lower().lstrip(".")
try: try:
if ext in VIDEO_EXTS: if ext in VIDEO_EXTS:
seek = "00:00:10" # Extract frame at 10 seconds in (or start if short)
if duration and duration > 0: result = subprocess.run(
t = max(0, duration * 0.1) ["ffmpeg", "-v", "quiet", "-ss", "10",
seek = f"{int(t // 3600):02d}:{int((t % 3600) // 60):02d}:{t % 60:05.2f}" "-i", str(filepath),
subprocess.run( "-vframes", "1", "-vf", "scale=320:-1",
["ffmpeg", "-v", "quiet", "-ss", seek, "-i", str(filepath), str(thumb_path)],
"-vframes", "1", "-vf", "scale=320:-1", str(thumb_path)], timeout=30, capture_output=True
timeout=60, capture_output=True
) )
if not thumb_path.exists():
# Try from start
subprocess.run(
["ffmpeg", "-v", "quiet",
"-i", str(filepath),
"-vframes", "1", "-vf", "scale=320:-1",
str(thumb_path)],
timeout=30, capture_output=True
)
elif ext in IMAGE_EXTS: elif ext in IMAGE_EXTS:
subprocess.run( subprocess.run(
["ffmpeg", "-v", "quiet", "-i", str(filepath), ["ffmpeg", "-v", "quiet", "-i", str(filepath),
"-vf", "scale=320:-1", str(thumb_path)], "-vf", "scale=320:-1", str(thumb_path)],
timeout=30, capture_output=True timeout=30, capture_output=True
) )
return thumb_path.exists()
except Exception: except Exception:
pass return False
def clean_name(filename_stem):
"""Turn file stem into a readable display name."""
name = filename_stem.replace("-", " ").replace("_", " ")
# Strip leading numbering like "01 " or "1a "
import re
name = re.sub(r"^\d+[a-z]?\s+", "", name)
return name.strip() or filename_stem
def collect_files(): def collect_files():
"""Walk source dir, return list of (filepath, category, course) tuples.""" """Walk source dir, return list of (filepath, category, course) tuples."""
files = [] files = []
for root, dirs, filenames in os.walk(SOURCE): for root, dirs, filenames in os.walk(SOURCE):
dirs[:] = [d for d in sorted(dirs) if d != "@eaDir"] dirs[:] = sorted([d for d in dirs if d != "@eaDir"])
root_path = Path(root) root_path = Path(root)
# Determine category and course from path depth
rel = root_path.relative_to(SOURCE) rel = root_path.relative_to(SOURCE)
parts = rel.parts parts = rel.parts
category = parts[0] if len(parts) >= 1 else None category = parts[0] if len(parts) >= 1 else None
@ -115,15 +96,13 @@ def collect_files():
return files return files
def build_folder_tree(files): def build_folder_tree(files):
"""Build Eagle folder hierarchy from category/course structure.""" """Build Eagle folder hierarchy. Returns (folders_json, folder_ids dict)."""
# Collect unique category -> set of courses
from collections import defaultdict, OrderedDict
tree = defaultdict(set) tree = defaultdict(set)
for _, category, course in files: for _, category, course in files:
if category: if category:
tree[category].add(course or "") tree[category].add(course or "")
folder_ids = {} # (category, course) -> id folder_ids = {}
folders_json = [] folders_json = []
for category in sorted(tree): for category in sorted(tree):
@ -137,7 +116,7 @@ def build_folder_tree(files):
folder_ids[(category, course)] = course_id folder_ids[(category, course)] = course_id
children.append({ children.append({
"id": course_id, "id": course_id,
"name": course.replace("-", " ").replace("_", " ").title(), "name": course,
"description": "", "description": "",
"children": [], "children": [],
"modificationTime": now_ms(), "modificationTime": now_ms(),
@ -147,7 +126,7 @@ def build_folder_tree(files):
}) })
folders_json.append({ folders_json.append({
"id": cat_id, "id": cat_id,
"name": category.replace("-", " ").replace("_", " ").title(), "name": category,
"description": "", "description": "",
"children": children, "children": children,
"modificationTime": now_ms(), "modificationTime": now_ms(),
@ -159,25 +138,23 @@ def build_folder_tree(files):
return folders_json, folder_ids return folders_json, folder_ids
def main(): def main():
print(f"Source: {SOURCE}") print(f"Source: {SOURCE}", flush=True)
print(f"Library: {LIBRARY}") print(f"Library: {LIBRARY}", flush=True)
if not SOURCE.exists(): if not SOURCE.exists():
print("ERROR: Source directory not found.", file=sys.stderr) print("ERROR: Source directory not found.", file=sys.stderr)
sys.exit(1) sys.exit(1)
# Collect files print("Scanning files...", flush=True)
print("Scanning files...")
files = collect_files() files = collect_files()
print(f"Found {len(files)} files to import.") print(f"Found {len(files)} files to import.", flush=True)
# Build folder tree
folders_json, folder_ids = build_folder_tree(files) folders_json, folder_ids = build_folder_tree(files)
# Create library structure # Create library structure
IMAGES.mkdir(parents=True, exist_ok=True) IMAGES.mkdir(parents=True, exist_ok=True)
# Write library metadata.json # Write all required library root files
meta = { meta = {
"folders": folders_json, "folders": folders_json,
"smartFolders": [], "smartFolders": [],
@ -187,20 +164,26 @@ def main():
"applicationVersion": "4.0.0", "applicationVersion": "4.0.0",
} }
(LIBRARY / "metadata.json").write_text(json.dumps(meta)) (LIBRARY / "metadata.json").write_text(json.dumps(meta))
print("Library metadata written.") (LIBRARY / "actions.json").write_text("[]")
(LIBRARY / "tags.json").write_text("[]")
(LIBRARY / "saved-filters.json").write_text("[]")
(LIBRARY / "mtime.json").write_text("{}") # Will be written at end
# Import each file print("Library root files written.", flush=True)
mtime_map = {}
errors = [] errors = []
for i, (filepath, category, course) in enumerate(files, 1): for i, (filepath, category, course) in enumerate(files, 1):
try: try:
ext = filepath.suffix.lower().lstrip(".") ext = filepath.suffix.lower().lstrip(".")
stem = filepath.stem stem = filepath.stem # MUST match filename exactly for Eagle
display_name = stem # Must match filename stem exactly for Eagle to find the file
item_id = gen_id() item_id = gen_id()
item_dir = IMAGES / f"{item_id}.info" item_dir = IMAGES / f"{item_id}.info"
item_dir.mkdir(parents=True, exist_ok=True) item_dir.mkdir(parents=True, exist_ok=True)
# Hardlink file into .info dir to save space # Hardlink the file into .info dir
dest_file = item_dir / filepath.name dest_file = item_dir / filepath.name
try: try:
os.link(filepath, dest_file) os.link(filepath, dest_file)
@ -208,35 +191,27 @@ def main():
import shutil import shutil
shutil.copy2(filepath, dest_file) shutil.copy2(filepath, dest_file)
# Get metadata # Generate thumbnail
thumb_path = item_dir / f"{stem}_thumbnail.png"
generate_thumbnail(filepath, thumb_path)
# Get file stats
stat = filepath.stat() stat = filepath.stat()
size = stat.st_size size = stat.st_size
btime_ms = int(stat.st_ctime * 1000) btime_ms = int(stat.st_ctime * 1000)
mtime_ms = int(stat.st_mtime * 1000) mtime_ms = int(stat.st_mtime * 1000)
mod_ms = now_ms()
width, height, duration = 0, 0, None # Folder assignment
if ext in VIDEO_EXTS:
width, height, duration = get_video_info(filepath)
elif ext in IMAGE_EXTS:
width, height, _ = get_video_info(filepath)
# Generate thumbnail
thumb_name = f"{stem}_thumbnail.png"
thumb_path = item_dir / thumb_name
generate_thumbnail(filepath, thumb_path, duration)
has_thumb = thumb_path.exists()
# Determine folder assignment
folder_id = None folder_id = None
if category and course: if category and course:
folder_id = folder_ids.get((category, course)) folder_id = folder_ids.get((category, course))
if not folder_id and category: if not folder_id and category:
folder_id = folder_ids.get((category, None)) folder_id = folder_ids.get((category, None))
# Write metadata.json
img_meta = { img_meta = {
"id": item_id, "id": item_id,
"name": display_name, "name": stem, # Exact filename stem — Eagle matches this to the file
"size": size, "size": size,
"btime": btime_ms, "btime": btime_ms,
"mtime": mtime_ms, "mtime": mtime_ms,
@ -246,17 +221,16 @@ def main():
"isDeleted": False, "isDeleted": False,
"url": "", "url": "",
"annotation": "", "annotation": "",
"modificationTime": now_ms(), "modificationTime": mod_ms,
"width": width, "width": 0,
"height": height, "height": 0,
"resolutionWidth": width, "resolutionWidth": 0,
"resolutionHeight": height, "resolutionHeight": 0,
"lastModified": now_ms(), "lastModified": mod_ms,
} }
if duration is not None:
img_meta["duration"] = duration
(item_dir / "metadata.json").write_text(json.dumps(img_meta)) (item_dir / "metadata.json").write_text(json.dumps(img_meta))
mtime_map[item_id] = mod_ms
if i % 50 == 0 or i == len(files): if i % 50 == 0 or i == len(files):
print(f"Progress: {i}/{len(files)}", flush=True) print(f"Progress: {i}/{len(files)}", flush=True)
@ -265,12 +239,16 @@ def main():
errors.append(f"{filepath}: {e}") errors.append(f"{filepath}: {e}")
print(f"ERROR: {filepath}: {e}", file=sys.stderr) print(f"ERROR: {filepath}: {e}", file=sys.stderr)
print(f"\n=== Import complete ===") # Write final mtime.json with all item IDs
print(f"Items imported: {i}") mtime_map["all"] = len(mtime_map)
print(f"Errors: {len(errors)}") (LIBRARY / "mtime.json").write_text(json.dumps(mtime_map))
print(f"\n=== Import complete ===", flush=True)
print(f"Items imported: {len(mtime_map) - 1}", flush=True)
print(f"Errors: {len(errors)}", flush=True)
if errors: if errors:
for e in errors[:10]: for e in errors[:10]:
print(f" {e}") print(f" {e}", flush=True)
if __name__ == "__main__": if __name__ == "__main__":
main() main()