#!/usr/bin/env python3 """ import_masterclass.py Creates a Masterclass.library in Eagle format from /volume1/steve/Courses/masterclass/ Run on the Synology NAS. """ import os import json import time import subprocess import sys import hashlib from pathlib import Path SOURCE = Path("/volume1/steve/Courses/masterclass") LIBRARY = Path("/volume1/steve/Eagle_Libraries/Masterclass.library") IMAGES = LIBRARY / "images" SUPPORTED_EXTS = { "mp4", "mov", "m4v", "avi", "mkv", "webm", # video "mp3", "m4a", "aac", "wav", "flac", # audio "jpg", "jpeg", "png", "gif", "webp", # image "pdf", # docs } VIDEO_EXTS = {"mp4", "mov", "m4v", "avi", "mkv", "webm"} IMAGE_EXTS = {"jpg", "jpeg", "png", "gif", "webp"} # ID counter _id_counter = 0 _id_prefix = f"MC{int(time.time()) % 10000000:07d}" def gen_id(): global _id_counter _id_counter += 1 return f"{_id_prefix}{_id_counter:05d}" def now_ms(): return int(time.time() * 1000) def get_video_info(filepath): """Returns (width, height, duration) via ffprobe.""" 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(".") try: if ext in VIDEO_EXTS: seek = "00:00:10" if duration and duration > 0: t = max(0, duration * 0.1) seek = f"{int(t // 3600):02d}:{int((t % 3600) // 60):02d}:{t % 60:05.2f}" subprocess.run( ["ffmpeg", "-v", "quiet", "-ss", seek, "-i", str(filepath), "-vframes", "1", "-vf", "scale=320:-1", str(thumb_path)], timeout=60, capture_output=True ) elif ext in IMAGE_EXTS: subprocess.run( ["ffmpeg", "-v", "quiet", "-i", str(filepath), "-vf", "scale=320:-1", str(thumb_path)], timeout=30, capture_output=True ) except Exception: pass 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(): """Walk source dir, return list of (filepath, category, course) tuples.""" files = [] for root, dirs, filenames in os.walk(SOURCE): dirs[:] = [d for d in sorted(dirs) if d != "@eaDir"] root_path = Path(root) # Determine category and course from path depth rel = root_path.relative_to(SOURCE) parts = rel.parts category = parts[0] if len(parts) >= 1 else None course = parts[1] if len(parts) >= 2 else None if category is None: continue for fname in sorted(filenames): ext = fname.rsplit(".", 1)[-1].lower() if "." in fname else "" if ext not in SUPPORTED_EXTS: continue files.append((root_path / fname, category, course)) return files def build_folder_tree(files): """Build Eagle folder hierarchy from category/course structure.""" # Collect unique category -> set of courses from collections import defaultdict, OrderedDict tree = defaultdict(set) for _, category, course in files: if category: tree[category].add(course or "") folder_ids = {} # (category, course) -> id folders_json = [] for category in sorted(tree): cat_id = gen_id() folder_ids[(category, None)] = cat_id children = [] for course in sorted(tree[category]): if not course: continue course_id = gen_id() folder_ids[(category, course)] = course_id children.append({ "id": course_id, "name": course.replace("-", " ").replace("_", " ").title(), "description": "", "children": [], "modificationTime": now_ms(), "tags": [], "password": "", "passwordTips": "", }) folders_json.append({ "id": cat_id, "name": category.replace("-", " ").replace("_", " ").title(), "description": "", "children": children, "modificationTime": now_ms(), "tags": [], "password": "", "passwordTips": "", }) return folders_json, folder_ids def main(): print(f"Source: {SOURCE}") print(f"Library: {LIBRARY}") if not SOURCE.exists(): print("ERROR: Source directory not found.", file=sys.stderr) sys.exit(1) # Collect files print("Scanning files...") files = collect_files() print(f"Found {len(files)} files to import.") # Build folder tree folders_json, folder_ids = build_folder_tree(files) # Create library structure IMAGES.mkdir(parents=True, exist_ok=True) # Write library metadata.json meta = { "folders": folders_json, "smartFolders": [], "quickAccess": [], "tagsGroups": [], "modificationTime": now_ms(), "applicationVersion": "4.0.0", } (LIBRARY / "metadata.json").write_text(json.dumps(meta)) print("Library metadata written.") # Import each file errors = [] for i, (filepath, category, course) in enumerate(files, 1): try: ext = filepath.suffix.lower().lstrip(".") stem = filepath.stem display_name = clean_name(stem) item_id = gen_id() item_dir = IMAGES / f"{item_id}.info" item_dir.mkdir(parents=True, exist_ok=True) # Hardlink file into .info dir to save space dest_file = item_dir / filepath.name try: os.link(filepath, dest_file) except OSError: import shutil shutil.copy2(filepath, dest_file) # Get metadata stat = filepath.stat() size = stat.st_size btime_ms = int(stat.st_ctime * 1000) mtime_ms = int(stat.st_mtime * 1000) width, height, duration = 0, 0, None 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 if category and course: folder_id = folder_ids.get((category, course)) if not folder_id and category: folder_id = folder_ids.get((category, None)) # Write metadata.json img_meta = { "id": item_id, "name": display_name, "size": size, "btime": btime_ms, "mtime": mtime_ms, "ext": ext, "tags": [], "folders": [folder_id] if folder_id else [], "isDeleted": False, "url": "", "annotation": "", "modificationTime": now_ms(), "width": width, "height": height, "resolutionWidth": width, "resolutionHeight": height, "lastModified": now_ms(), } if duration is not None: img_meta["duration"] = duration (item_dir / "metadata.json").write_text(json.dumps(img_meta)) if i % 50 == 0 or i == len(files): print(f"Progress: {i}/{len(files)}", flush=True) except Exception as e: errors.append(f"{filepath}: {e}") print(f"ERROR: {filepath}: {e}", file=sys.stderr) print(f"\n=== Import complete ===") print(f"Items imported: {i}") print(f"Errors: {len(errors)}") if errors: for e in errors[:10]: print(f" {e}") if __name__ == "__main__": main()