AddonsManager.py

#!/usr/bin/env python3
"""
ESO AddOn Manager - CLI
Bridges any addon directory into the Game Pass ESO managed-addon system
using NTFS directory junctions, requiring no game file modifications.
"""

import argparse
import ctypes
import ctypes.wintypes
import json
import os
import re
import subprocess
import sys
import time
import uuid
from pathlib import Path

# ---------------------------------------------------------------------------
# Paths
# ---------------------------------------------------------------------------

LOCAL_APPDATA   = Path(os.environ["LOCALAPPDATA"])
USER_PROFILE    = Path(os.environ["USERPROFILE"])

ESO_LOCAL_BASE  = LOCAL_APPDATA / "Elder Scrolls Online"
DEFAULT_CHANNEL = "xb1cert"


def _get_documents_folder() -> Path:
    """Return the user's real Documents folder via SHGetFolderPathW."""
    CSIDL_PERSONAL     = 5
    SHGFP_TYPE_CURRENT = 0
    buf = ctypes.create_unicode_buffer(ctypes.wintypes.MAX_PATH)
    ctypes.windll.shell32.SHGetFolderPathW(
        None, CSIDL_PERSONAL, None, SHGFP_TYPE_CURRENT, buf
    )
    return Path(buf.value)


DEFAULT_SOURCE  = str(_get_documents_folder() / "Elder Scrolls Online" / DEFAULT_CHANNEL / "AddOns")

REROUTER_TAG    = "__addon_rerouter__"   # written into our content_installation_info.json

# ---------------------------------------------------------------------------
# Admin check / elevation
# ---------------------------------------------------------------------------

def _is_admin() -> bool:
    try:
        return bool(ctypes.windll.shell32.IsUserAnAdmin())
    except Exception:
        return False


def _require_admin():
    if _is_admin():
        return
    print("Administrator privileges are required to create directory junctions.")
    print("Re-launching elevated …")
    args = " ".join(f'"{a}"' for a in sys.argv)
    ctypes.windll.shell32.ShellExecuteW(None, "runas", sys.executable, args, None, 1)
    sys.exit(0)

# ---------------------------------------------------------------------------
# NTFS junction helpers
# ---------------------------------------------------------------------------

def _is_reparse_point(path: Path) -> bool:
    """Return True if *path* is an NTFS reparse point (junction or symlink)."""
    INVALID = 0xFFFFFFFF
    FILE_ATTRIBUTE_REPARSE_POINT = 0x400
    attrs = ctypes.windll.kernel32.GetFileAttributesW(str(path))
    if attrs == INVALID:
        return False
    return bool(attrs & FILE_ATTRIBUTE_REPARSE_POINT)


def _create_junction(link: Path, target: Path) -> bool:
    r = subprocess.run(
        ["cmd", "/c", "mklink", "/J", str(link), str(target)],
        capture_output=True, text=True,
    )
    return r.returncode == 0


def _remove_junction(link: Path) -> bool:
    """Remove a junction without touching the target directory."""
    r = subprocess.run(
        ["cmd", "/c", "rmdir", str(link)],
        capture_output=True, text=True,
    )
    return r.returncode == 0


def _safe_remove_uuid_dir(uuid_dir: Path, addon_name: str, managed_root: Path):
    """
    Remove a rerouter-created UUID directory safely.
    Never uses shutil.rmtree - it follows NTFS junctions and would
    delete the source addon files.  os.rmdir() removes only the junction
    point itself, leaving the target untouched.
    """
    # Refuse to operate outside managed_root
    try:
        uuid_dir.resolve().relative_to(managed_root.resolve())
    except ValueError:
        return

    junction = uuid_dir / addon_name
    if _is_reparse_point(junction):
        try:
            os.rmdir(str(junction))
        except OSError:
            return  # couldn't remove junction - abort rather than risk anything
    elif junction.exists():
        return  # real directory we didn't create - leave it alone

    # Remove only the specific file we wrote; then the (now-empty) dir
    try:
        (uuid_dir / "content_installation_info.json").unlink()
    except OSError:
        pass
    try:
        uuid_dir.rmdir()
    except OSError:
        pass

# ---------------------------------------------------------------------------
# UUID / mapping helpers
# ---------------------------------------------------------------------------

# We derive per-addon UUIDs deterministically from the addon name so that
# running the tool twice gives identical results and the mapping survives
# re-installation of the tool itself.
_UUID_NAMESPACE = uuid.UUID("f4ad7000-e500-4000-9000-000000000001")


def _addon_uuid(addon_name: str) -> str:
    return str(uuid.uuid5(_UUID_NAMESPACE, addon_name.lower()))


def _map_path(channel: str) -> Path:
    return ESO_LOCAL_BASE / channel / "CachedData" / "AddonRouterMap.json"


def _load_map(channel: str) -> dict:
    p = _map_path(channel)
    if p.exists():
        try:
            with open(p) as f:
                return json.load(f)
        except Exception:
            pass
    return {"version": 1, "managed": {}}


def _save_map(channel: str, data: dict):
    p = _map_path(channel)
    p.parent.mkdir(parents=True, exist_ok=True)
    with open(p, "w") as f:
        json.dump(data, f, indent=2)

# ---------------------------------------------------------------------------
# Addon scanning
# ---------------------------------------------------------------------------

def _scan_source(source: Path) -> dict:
    """Return {addon_name: {'path': Path, 'addon_file': Path}} for each addon folder."""
    result = {}
    if not source.is_dir():
        return result
    for entry in source.iterdir():
        if not entry.is_dir():
            continue
        addon_files = list(entry.glob("*.addon"))
        if addon_files:
            result[entry.name] = {"path": entry, "addon_file": addon_files[0]}
    return result


def _read_addon_version(addon_file: Path) -> str:
    """Extract ## AddOnVersion or ## Version from a .addon manifest."""
    try:
        for line in addon_file.read_text(encoding="utf-8", errors="ignore").splitlines():
            m = re.match(r"##\s*AddOnVersion:\s*(.+)", line)
            if m:
                return m.group(1).strip()
            m = re.match(r"##\s*Version:\s*(.+)", line)
            if m:
                return m.group(1).strip()
    except Exception:
        pass
    return "custom"

# ---------------------------------------------------------------------------
# Core: create / update one managed-addon entry
# ---------------------------------------------------------------------------

def _upsert_entry(
    addon_name: str,
    addon_info: dict,
    addon_uuid: str,
    managed_root: Path,
    dry_run: bool,
) -> str:
    """
    Ensure AddOnsManaged/{uuid}/{AddonName} is a junction pointing at the
    source addon folder.  Returns 'added', 'updated', 'skipped', or 'failed'.
    """
    uuid_dir      = managed_root / addon_uuid
    junction_path = uuid_dir / addon_name

    # Determine what already exists
    if junction_path.exists() or _is_reparse_point(junction_path):
        if _is_reparse_point(junction_path):
            action = "updated"
        else:
            # Real directory we didn't create - leave it alone
            return "skipped"
    else:
        action = "added"

    if dry_run:
        return action

    # Materialise UUID directory
    uuid_dir.mkdir(parents=True, exist_ok=True)

    # Write content_installation_info.json
    now     = str(int(time.time()))
    version = _read_addon_version(addon_info["addon_file"])
    info    = {
        "contentInstallationInfoVersion": 1,
        "contentId":                      addon_uuid,
        "installationTime":               now,
        "releaseVersion":                 version,
        "releaseCreationTime":            now,
        "ignoreUpdates":                  True,
        REROUTER_TAG:                     True,
    }
    with open(uuid_dir / "content_installation_info.json", "w") as f:
        json.dump(info, f)

    # (Re)create junction
    if _is_reparse_point(junction_path):
        _remove_junction(junction_path)
    elif junction_path.exists():
        pass  # won't reach here due to earlier check

    ok = _create_junction(junction_path, addon_info["path"])
    return action if ok else "failed"

# ---------------------------------------------------------------------------
# Commands
# ---------------------------------------------------------------------------

def cmd_sync(source: Path, channel: str, dry_run: bool, remove_stale: bool):
    managed_root = ESO_LOCAL_BASE / channel / "CachedData" / "AddOnsManaged"

    print(f"Source  : {source}")
    print(f"Target  : {managed_root}")
    if dry_run:
        print("(dry run - no changes will be made)")
    print()

    if not source.is_dir():
        print(f"Source directory does not exist: {source}")
        print("Create it and place addon folders inside, then run sync again.")
        return

    managed_root.mkdir(parents=True, exist_ok=True)
    map_data     = _load_map(channel)
    source_addons = _scan_source(source)

    print(f"Addons found in source: {len(source_addons)}")
    print()

    counts = {"added": 0, "updated": 0, "skipped": 0, "failed": 0}

    for name, info in sorted(source_addons.items()):
        a_uuid  = map_data["managed"].get(name) or _addon_uuid(name)
        map_data["managed"][name] = a_uuid
        result  = _upsert_entry(name, info, a_uuid, managed_root, dry_run)
        counts[result] += 1
        tag = result.upper()[:3]
        print(f"  [{tag}] {name}")

    # Optionally remove junctions for addons no longer in source.
    # Guard: if source scan returned nothing, skip stale removal entirely -
    # an empty scan most likely means the source path is wrong, and treating
    # all tracked addons as stale would remove all junctions unnecessarily.
    if remove_stale and source_addons:
        stale_names = set(map_data["managed"].keys()) - set(source_addons.keys())
        for name in sorted(stale_names):
            a_uuid    = map_data["managed"][name]
            uuid_dir  = managed_root / a_uuid
            info_file = uuid_dir / "content_installation_info.json"
            # Only remove entries that we created
            try:
                with open(info_file) as f:
                    meta = json.load(f)
                if not meta.get(REROUTER_TAG):
                    continue
            except Exception:
                continue

            print(f"  [REM] {name} (not in source)")
            if not dry_run:
                _safe_remove_uuid_dir(uuid_dir, name, managed_root)
                del map_data["managed"][name]

    if not dry_run:
        _save_map(channel, map_data)

    print()
    print(
        f"Complete.  Added: {counts['added']}  "
        f"Updated: {counts['updated']}  "
        f"Skipped: {counts['skipped']}  "
        f"Failed: {counts['failed']}"
    )


def cmd_status(channel: str):
    managed_root = ESO_LOCAL_BASE / channel / "CachedData" / "AddOnsManaged"
    plain_link   = ESO_LOCAL_BASE / channel / "AddOns"
    map_data     = _load_map(channel)

    print(f"Channel   : {channel}")
    print(f"Data root : {ESO_LOCAL_BASE / channel}")
    print()

    # Plain AddOns junction status
    if plain_link.exists():
        if _is_reparse_point(plain_link):
            print(f"Plain AddOns\\ : junction present -> {plain_link}")
        else:
            print(f"Plain AddOns\\ : real directory (not a junction)")
    else:
        print(f"Plain AddOns\\ : not present")
    print()

    # Managed entries we own
    owned = map_data.get("managed", {})
    print(f"Rerouted addons ({len(owned)}):")
    if not owned:
        print("  (none)")
    for name, a_uuid in sorted(owned.items()):
        junction = managed_root / a_uuid / name
        if _is_reparse_point(junction):
            state = "OK"
        elif junction.is_dir():
            state = "DIR (not junction?)"
        else:
            state = "MISSING"
        print(f"  {state:20s} {name}")


def cmd_clean(channel: str, dry_run: bool):
    """Remove all junction entries created by this tool."""
    managed_root = ESO_LOCAL_BASE / channel / "CachedData" / "AddOnsManaged"
    map_data     = _load_map(channel)
    removed      = 0

    for name, a_uuid in list(map_data.get("managed", {}).items()):
        uuid_dir  = managed_root / a_uuid
        info_file = uuid_dir / "content_installation_info.json"
        try:
            with open(info_file) as f:
                meta = json.load(f)
            if not meta.get(REROUTER_TAG):
                continue
        except Exception:
            continue

        print(f"  [REMOVE] {name}")
        if not dry_run:
            _safe_remove_uuid_dir(uuid_dir, name, managed_root)
            del map_data["managed"][name]
            removed += 1

    if not dry_run:
        _save_map(channel, map_data)

    print()
    print(f"Removed {removed} rerouted addon(s).")

# ---------------------------------------------------------------------------
# Entry point
# ---------------------------------------------------------------------------

def main():
    p = argparse.ArgumentParser(
        description="ESO Game Pass Addons Rerouter",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
commands:
  sync          Link addons from source dir into the Game Pass addon system
  status        Show what the rerouter has set up
  clean         Remove all rerouter-created junctions
        """,
    )
    p.add_argument(
        "--channel",
        default=DEFAULT_CHANNEL,
        help=f"ESO channel folder name (default: {DEFAULT_CHANNEL})",
    )
    sub = p.add_subparsers(dest="command", metavar="command")

    # sync
    sp = sub.add_parser("sync", help="Sync addons source -> Game Pass managed system")
    sp.add_argument(
        "--source",
        default=str(DEFAULT_SOURCE),
        help=f"Addons directory to read from (default: {DEFAULT_SOURCE})",
    )
    sp.add_argument("--dry-run", action="store_true", help="Preview without making changes")
    sp.add_argument("--remove-stale", action="store_true",
                    help="Remove junctions for addons no longer in source")

    # status
    sub.add_parser("status", help="Show rerouter status")

    # clean
    cp = sub.add_parser("clean", help="Remove all rerouter-created junctions")
    cp.add_argument("--dry-run", action="store_true", help="Preview without making changes")

    args = p.parse_args()

    if args.command is None:
        p.print_help()
        return

    channel = args.channel

    if args.command == "sync":
        _require_admin()
        cmd_sync(
            source=Path(args.source),
            channel=channel,
            dry_run=args.dry_run,
            remove_stale=args.remove_stale,
        )

    elif args.command == "status":
        cmd_status(channel)

    elif args.command == "clean":
        _require_admin()
        cmd_clean(channel, dry_run=args.dry_run)


if __name__ == "__main__":
    main()