eagle-viewer/eagle_api_import.py

#!/usr/bin/env python3
"""
eagle_api_import.py
Run this on your Mac while Eagle is open with the Masterclass library active.
Uses Eagle's local REST API to properly import all Masterclass files.

Usage:
  python3 eagle_api_import.py

Requirements:
- Eagle must be open
- Masterclass library must be the active library in Eagle
- NAS must be mounted at /Volumes/steve/
"""

import os
import json
import time
import urllib.request
import urllib.error
from pathlib import Path
from collections import defaultdict

EAGLE_API = "http://localhost:41595"
SOURCE = Path("/Volumes/steve/Courses/masterclass")

SUPPORTED_EXTS = {
    "mp4", "mov", "m4v", "avi", "mkv", "webm",
    "mp3", "m4a", "aac", "wav", "flac",
    "jpg", "jpeg", "png", "gif", "webp",
    "pdf",
}

def api(method, endpoint, data=None):
    url = f"{EAGLE_API}{endpoint}"
    body = json.dumps(data).encode() if data else None
    req = urllib.request.Request(
        url, data=body,
        headers={"Content-Type": "application/json"},
        method=method
    )
    try:
        with urllib.request.urlopen(req, timeout=10) as resp:
            return json.loads(resp.read())
    except Exception as e:
        return {"status": "error", "message": str(e)}

def get(endpoint):
    return api("GET", endpoint)

def post(endpoint, data):
    return api("POST", endpoint, data)

def collect_files():
    files = []
    for root, dirs, filenames in os.walk(SOURCE):
        dirs[:] = sorted([d for d in dirs if d != "@eaDir"])
        root_path = Path(root)
        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 create_folders():
    """Create folder hierarchy in Eagle and return {(category, course): folder_id}."""
    # First get existing folders
    lib_info = get("/api/library/info")
    existing = {}
    if lib_info.get("status") == "success":
        def extract(folders, parent=None):
            for f in folders:
                existing[f["name"]] = f["id"]
                extract(f.get("children", []), f["id"])
        extract(lib_info.get("data", {}).get("library", {}).get("folders", []))
    
    print(f"Existing folders: {len(existing)}")
    
    folder_ids = {}
    
    # Get all categories and courses from files
    tree = defaultdict(set)
    for _, category, course in collect_files():
        if category:
            tree[category].add(course or "")

    for category in sorted(tree):
        # Create category folder if needed
        if category not in existing:
            resp = post("/api/folder/create", {"folderName": category})
            if resp.get("status") == "success":
                cat_id = resp["data"]["id"]
                existing[category] = cat_id
                print(f"  Created folder: {category}")
                time.sleep(0.1)
            else:
                print(f"  ERROR creating {category}: {resp}")
                continue
        else:
            cat_id = existing[category]
        
        folder_ids[(category, None)] = cat_id

        for course in sorted(tree[category]):
            if not course:
                continue
            if course not in existing:
                resp = post("/api/folder/create", {
                    "folderName": course,
                    "parent": cat_id
                })
                if resp.get("status") == "success":
                    course_id = resp["data"]["id"]
                    existing[course] = course_id
                    print(f"    Created subfolder: {category}/{course}")
                    time.sleep(0.1)
                else:
                    print(f"    ERROR creating {category}/{course}: {resp}")
                    continue
            else:
                course_id = existing[course]
            
            folder_ids[(category, course)] = course_id

    return folder_ids

def main():
    # Check Eagle is running
    info = get("/api/application/info")
    if info.get("status") != "success":
        print("ERROR: Eagle API not reachable. Make sure Eagle is open.")
        return

    print(f"Eagle version: {info['data']['version']}")

    # Check current library
    lib_info = get("/api/library/info")
    if lib_info.get("status") == "success":
        lib_name = lib_info.get("data", {}).get("library", {}).get("name", "unknown")
        lib_path = lib_info.get("data", {}).get("library", {}).get("path", "")
        print(f"Current library: {lib_name}")
        print(f"Library path: {lib_path}")
        if "masterclass" not in lib_name.lower() and "masterclass" not in lib_path.lower():
            confirm = input("WARNING: Active library doesn't look like Masterclass. Continue? (y/n): ")
            if confirm.lower() != "y":
                return

    print("\nCreating folder structure...")
    folder_ids = create_folders()
    print(f"Folders ready: {len(folder_ids)}")

    print("\nCollecting files...")
    files = collect_files()
    print(f"Files to import: {len(files)}")
    input("Press Enter to start import (or Ctrl+C to cancel)...")

    done = 0
    errors = []
    BATCH_SIZE = 20

    # Import in batches using addFromPaths
    batch = []
    for i, (filepath, category, course) in enumerate(files, 1):
        stem = filepath.stem
        folder_id = folder_ids.get((category, course)) or folder_ids.get((category, None))

        item = {
            "path": str(filepath),
            "name": stem,
            "tags": [],
            "annotation": "",
        }
        if folder_id:
            item["folderId"] = folder_id

        batch.append(item)

        if len(batch) >= BATCH_SIZE or i == len(files):
            resp = post("/api/item/addFromPaths", {"items": batch})
            if resp.get("status") == "success":
                done += len(batch)
            else:
                # Fall back to individual adds
                for item in batch:
                    r = post("/api/item/addFromPath", item)
                    if r.get("status") == "success":
                        done += 1
                    else:
                        errors.append(f"{item['path']}: {r.get('message', r)}")
            batch = []

            if done % 50 == 0 or i == len(files):
                print(f"Progress: {done}/{len(files)} done, {len(errors)} errors")
            time.sleep(0.05)  # Gentle rate limiting

    print(f"\n=== Import complete ===")
    print(f"Imported: {done}")
    print(f"Errors: {len(errors)}")
    if errors:
        for e in errors[:10]:
            print(f"  {e}")

if __name__ == "__main__":
    main()