pgl_drive_mapper.py

import ctypes
import os
import queue
import shutil
import string
import subprocess
import sys
import threading
import tkinter as tk
from pathlib import Path
from tkinter import messagebox, scrolledtext, ttk

APP_NAME = "PGL Drive Mapper"
DEFAULT_REMOTE = "pgl_e2"
DEFAULT_DRIVE = "X:"
DEFAULT_CACHE_DIR = r"C:\rclone-cache"
DEFAULT_BUCKET = "superpgl"
DEFAULT_ENDPOINT = "https://YOUR-ENDPOINT-HERE"


class App(tk.Tk):
    def __init__(self):
        super().__init__()
        self.title(APP_NAME)
        self.geometry("1080x820")
        self.minsize(980, 760)
        self.configure(bg="#f5f7fb")
        self.protocol("WM_DELETE_WINDOW", self.on_close)

        self.log_queue = queue.Queue()
        self.mount_process = None
        self.worker = None

        self._build_vars()
        self._configure_style()
        self._build_ui()
        self.after(150, self._drain_log_queue)
        self.refresh_drive_letters()
        self.refresh_preflight()

    def _build_vars(self):
        self.access_key = tk.StringVar()
        self.secret_key = tk.StringVar()
        self.endpoint = tk.StringVar(value=DEFAULT_ENDPOINT)
        self.bucket = tk.StringVar(value=DEFAULT_BUCKET)
        self.drive = tk.StringVar(value=DEFAULT_DRIVE)
        self.remote = tk.StringVar(value=DEFAULT_REMOTE)
        self.cache_dir = tk.StringVar(value=DEFAULT_CACHE_DIR)
        self.status = tk.StringVar(value="Ready")
        self.rclone_status = tk.StringVar(value="Checking...")
        self.winfsp_status = tk.StringVar(value="Checking...")
        self.admin_status = tk.StringVar(value="Checking...")
        self.mount_target = tk.StringVar(value="Not mounted")

    def _configure_style(self):
        style = ttk.Style(self)
        try:
            style.theme_use("clam")
        except tk.TclError:
            pass

        self.option_add("*Font", "{Segoe UI} 10")
        style.configure("App.TFrame", background="#f5f7fb")
        style.configure("Card.TFrame", background="#ffffff", relief="flat")
        style.configure("Title.TLabel", background="#f5f7fb", foreground="#111827", font=("Segoe UI Semibold", 24))
        style.configure("Sub.TLabel", background="#f5f7fb", foreground="#6b7280", font=("Segoe UI", 10))
        style.configure("CardTitle.TLabel", background="#ffffff", foreground="#111827", font=("Segoe UI Semibold", 11))
        style.configure("CardText.TLabel", background="#ffffff", foreground="#374151", font=("Segoe UI", 10))
        style.configure("Primary.TButton", font=("Segoe UI Semibold", 10))
        style.configure("Secondary.TButton", font=("Segoe UI", 10))
        style.configure("Modern.TEntry", fieldbackground="#f9fafb", padding=8)
        style.configure("Modern.TCombobox", fieldbackground="#f9fafb", padding=6)
        style.configure("Treeview", rowheight=26, font=("Segoe UI", 10))
        style.configure("Treeview.Heading", font=("Segoe UI Semibold", 10))
        style.map("Treeview", background=[("selected", "#dbeafe")], foreground=[("selected", "#111827")])

    def _build_ui(self):
        root = ttk.Frame(self, style="App.TFrame", padding=18)
        root.pack(fill="both", expand=True)
        root.columnconfigure(0, weight=1)
        root.rowconfigure(3, weight=1)
        root.rowconfigure(4, weight=1)

        header = ttk.Frame(root, style="App.TFrame")
        header.grid(row=0, column=0, sticky="ew", pady=(0, 14))
        header.columnconfigure(0, weight=1)
        ttk.Label(header, text=APP_NAME, style="Title.TLabel").grid(row=0, column=0, sticky="w")
        ttk.Label(
            header,
            text="Create an IDrive e2 remote, preview the bucket, then mount or disconnect cleanly.",
            style="Sub.TLabel",
        ).grid(row=1, column=0, sticky="w", pady=(2, 0))
        ttk.Label(header, textvariable=self.status, style="Sub.TLabel").grid(row=0, column=1, rowspan=2, sticky="e")

        top_cards = ttk.Frame(root, style="App.TFrame")
        top_cards.grid(row=1, column=0, sticky="ew", pady=(0, 14))
        for i in range(3):
            top_cards.columnconfigure(i, weight=1)
        self._build_status_card(top_cards, 0, "rclone", self.rclone_status)
        self._build_status_card(top_cards, 1, "WinFsp", self.winfsp_status)
        self._build_status_card(top_cards, 2, "Session", self.admin_status)

        content = ttk.Frame(root, style="App.TFrame")
        content.grid(row=2, column=0, sticky="nsew", pady=(0, 14))
        content.columnconfigure(0, weight=3)
        content.columnconfigure(1, weight=2)

        form_card = ttk.Frame(content, style="Card.TFrame", padding=16)
        form_card.grid(row=0, column=0, sticky="nsew", padx=(0, 10))
        form_card.columnconfigure(1, weight=1)
        ttk.Label(form_card, text="Connection", style="CardTitle.TLabel").grid(row=0, column=0, columnspan=2, sticky="w", pady=(0, 10))

        self._modern_entry(form_card, "Access Key", self.access_key, 1)
        self._modern_entry(form_card, "Secret Key", self.secret_key, 2, show="*")
        self._modern_entry(form_card, "Endpoint", self.endpoint, 3)
        self._modern_entry(form_card, "Bucket", self.bucket, 4)
        self._modern_entry(form_card, "Remote Name", self.remote, 5)
        self._modern_entry(form_card, "Cache Directory", self.cache_dir, 6)
        self._drive_dropdown(form_card, 7)

        actions = ttk.Frame(form_card, style="Card.TFrame")
        actions.grid(row=8, column=0, columnspan=2, sticky="ew", pady=(14, 0))
        for i in range(4):
            actions.columnconfigure(i, weight=1)
        ttk.Button(actions, text="Refresh", style="Secondary.TButton", command=self.refresh_all).grid(row=0, column=0, sticky="ew", padx=(0, 8))
        ttk.Button(actions, text="Create Remote", style="Secondary.TButton", command=self.create_remote).grid(row=0, column=1, sticky="ew", padx=(0, 8))
        ttk.Button(actions, text="Test & Preview", style="Primary.TButton", command=self.test_and_preview).grid(row=0, column=2, sticky="ew", padx=(0, 8))
        ttk.Button(actions, text="Mount", style="Primary.TButton", command=self.mount).grid(row=0, column=3, sticky="ew")

        side_card = ttk.Frame(content, style="Card.TFrame", padding=16)
        side_card.grid(row=0, column=1, sticky="nsew")
        side_card.columnconfigure(0, weight=1)
        ttk.Label(side_card, text="Mount Control", style="CardTitle.TLabel").grid(row=0, column=0, sticky="w")
        ttk.Label(side_card, textvariable=self.mount_target, style="CardText.TLabel", wraplength=300).grid(row=1, column=0, sticky="w", pady=(8, 14))
        ttk.Button(side_card, text="Disconnect", style="Secondary.TButton", command=self.disconnect).grid(row=2, column=0, sticky="ew", pady=(0, 8))
        #ttk.Button(side_card, text="Open Mounted Drive", style="Secondary.TButton", command=self.open_mounted_drive).grid(row=3, column=0, sticky="ew", pady=(0, 8))
        ttk.Button(side_card, text="Open Install Guide", style="Secondary.TButton", command=self.open_install_instructions).grid(row=4, column=0, sticky="ew", pady=(0, 8))
        ttk.Button(side_card, text="Open rclone Config", style="Secondary.TButton", command=self.open_rclone_config_folder).grid(row=5, column=0, sticky="ew")

        preview_card = ttk.Frame(root, style="Card.TFrame", padding=16)
        preview_card.grid(row=3, column=0, sticky="nsew", pady=(0, 14))
        preview_card.columnconfigure(0, weight=1)
        preview_card.rowconfigure(1, weight=1)
        ttk.Label(preview_card, text="Bucket Preview", style="CardTitle.TLabel").grid(row=0, column=0, sticky="w", pady=(0, 10))
        self.tree = ttk.Treeview(preview_card, columns=("type", "path"), show="tree headings")
        self.tree.heading("#0", text="Name")
        self.tree.heading("type", text="Type")
        self.tree.heading("path", text="Path")
        self.tree.column("#0", width=260)
        self.tree.column("type", width=120, anchor="center")
        self.tree.column("path", width=580)
        self.tree.grid(row=1, column=0, sticky="nsew")
        tree_scroll = ttk.Scrollbar(preview_card, orient="vertical", command=self.tree.yview)
        tree_scroll.grid(row=1, column=1, sticky="ns")
        self.tree.configure(yscrollcommand=tree_scroll.set)

        log_card = ttk.Frame(root, style="Card.TFrame", padding=16)
        log_card.grid(row=4, column=0, sticky="nsew")
        log_card.columnconfigure(0, weight=1)
        log_card.rowconfigure(1, weight=1)
        ttk.Label(log_card, text="Log", style="CardTitle.TLabel").grid(row=0, column=0, sticky="w", pady=(0, 10))
        self.log_box = scrolledtext.ScrolledText(
            log_card,
            height=12,
            bg="#0f172a",
            fg="#e5e7eb",
            insertbackground="#e5e7eb",
            relief="flat",
            font=("Consolas", 10),
            wrap="word",
            padx=10,
            pady=10,
        )
        self.log_box.grid(row=1, column=0, sticky="nsew")

    def _build_status_card(self, parent, col, title, variable):
        card = ttk.Frame(parent, style="Card.TFrame", padding=14)
        card.grid(row=0, column=col, sticky="ew", padx=(0 if col == 0 else 8, 0))
        ttk.Label(card, text=title, style="CardTitle.TLabel").pack(anchor="w")
        ttk.Label(card, textvariable=variable, style="CardText.TLabel", wraplength=260).pack(anchor="w", pady=(6, 0))

    def _modern_entry(self, parent, label, variable, row, show=None):
        ttk.Label(parent, text=label, style="CardText.TLabel").grid(row=row, column=0, sticky="w", padx=(0, 12), pady=6)
        entry = ttk.Entry(parent, textvariable=variable, style="Modern.TEntry", show=show)
        entry.grid(row=row, column=1, sticky="ew", pady=6)

    def _drive_dropdown(self, parent, row):
        ttk.Label(parent, text="Drive Letter", style="CardText.TLabel").grid(row=row, column=0, sticky="w", padx=(0, 12), pady=6)
        wrap = ttk.Frame(parent, style="Card.TFrame")
        wrap.grid(row=row, column=1, sticky="ew", pady=6)
        wrap.columnconfigure(0, weight=1)
        self.drive_combo = ttk.Combobox(wrap, textvariable=self.drive, state="readonly", style="Modern.TCombobox")
        self.drive_combo.grid(row=0, column=0, sticky="ew")
        ttk.Button(wrap, text="↻", width=3, command=self.refresh_drive_letters).grid(row=0, column=1, padx=(8, 0))

    def log(self, msg):
        self.log_queue.put(msg)

    def _drain_log_queue(self):
        while not self.log_queue.empty():
            self.log_box.insert("end", self.log_queue.get() + "\n")
            self.log_box.see("end")
        self.after(150, self._drain_log_queue)

    def set_status(self, text):
        self.status.set(text)
        self.log(text)

    def is_admin(self):
        try:
            return bool(ctypes.windll.shell32.IsUserAnAdmin())
        except Exception:
            return False

    def refresh_preflight(self):
        self.rclone_status.set(shutil.which("rclone") or "rclone not found in PATH")
        self.winfsp_status.set("Installed" if self.find_winfsp() else "WinFsp not found")
        self.admin_status.set("Administrator session" if self.is_admin() else "Normal user session")

    def refresh_all(self):
        self.refresh_preflight()
        self.refresh_drive_letters()
        self.clear_preview()
        self.set_status("Refreshed environment checks.")

    def refresh_drive_letters(self):
        letters = self.get_available_drive_choices()
        self.drive_combo["values"] = letters
        current = self.drive.get().upper().strip() or DEFAULT_DRIVE
        if current not in letters:
            self.drive.set(letters[0] if letters else DEFAULT_DRIVE)

    def get_available_drive_choices(self):
        used = set()
        for letter in string.ascii_uppercase:
            if os.path.exists(f"{letter}:\\"):
                used.add(letter)
        all_letters = list(string.ascii_uppercase)
        display = []
        preferred = []
        for letter in all_letters:
            label = f"{letter}: ({'Mapped/Used' if letter in used else 'Available'})"
            if letter not in used:
                preferred.append(label)
            else:
                display.append(label)
        choices = preferred + display
        current_drive = self.drive.get().upper().replace(":", "")
        if current_drive and current_drive not in used:
            formatted = f"{current_drive}: (Available)"
            if formatted not in choices:
                choices.insert(0, formatted)
        return choices

    def parse_drive_value(self):
        raw = self.drive.get().strip()
        if not raw:
            return DEFAULT_DRIVE
        return raw.split(":")[0].upper() + ":"

    def ensure_requirements(self):
        if not shutil.which("rclone"):
            messagebox.showerror(APP_NAME, "rclone was not found in PATH.")
            return False
        if not self.find_winfsp():
            messagebox.showerror(APP_NAME, "WinFsp was not found.")
            return False
        return True

    def validate_inputs(self):
        required = {
            "Access Key": self.access_key.get().strip(),
            "Secret Key": self.secret_key.get().strip(),
            "Endpoint": self.endpoint.get().strip(),
            "Bucket": self.bucket.get().strip(),
            "Remote Name": self.remote.get().strip(),
            "Cache Directory": self.cache_dir.get().strip(),
        }
        missing = [name for name, value in required.items() if not value]
        if missing:
            messagebox.showerror(APP_NAME, f"Missing required fields: {', '.join(missing)}")
            return False
        drive = self.parse_drive_value()
        if len(drive) != 2 or drive[1] != ":":
            messagebox.showerror(APP_NAME, "Choose a valid drive letter.")
            return False
        return True

    def create_remote(self):
        if not self.ensure_requirements() or not self.validate_inputs():
            return
        self.run_async(self._create_remote_task)

    def _create_remote_task(self):
        self.set_status("Creating or updating rclone remote...")
        cmd = [
            "rclone", "config", "create",
            self.remote.get().strip(),
            "s3",
            "provider", "IDrive",
            "access_key_id", self.access_key.get().strip(),
            "secret_access_key", self.secret_key.get().strip(),
            "endpoint", self.endpoint.get().strip(),
        ]
        self.run(cmd, redact={self.access_key.get().strip(), self.secret_key.get().strip()})
        self.set_status("Remote creation finished.")

    def test_and_preview(self):
        if not self.ensure_requirements() or not self.validate_inputs():
            return
        self.run_async(self._test_and_preview_task)

    def _test_and_preview_task(self):
        self.set_status("Testing connection and building preview...")
        remote_root = f"{self.remote.get().strip()}:"
        target = self.get_target()
        ok_root, root_output = self.run_capture(["rclone", "lsd", remote_root])
        if ok_root:
            self.log(root_output)
        ok, output = self.run_capture(["rclone", "lsf", target, "--max-depth", "3", "--dirs-only", "-R"])
        self.after(0, self.clear_preview)
        if ok:
            self.after(0, lambda: self.populate_preview(output))
            self.set_status("Preview ready.")
        else:
            self.log(output)
            self.set_status("Preview failed. Check bucket, endpoint, or credentials.")

    def clear_preview(self):
        for item in self.tree.get_children():
            self.tree.delete(item)

    def populate_preview(self, output):
        self.clear_preview()
        root_label = self.bucket.get().strip() or "bucket"
        root_id = self.tree.insert("", "end", text=root_label, values=("Bucket", root_label), open=True)
        nodes = {"": root_id}
        lines = [line.strip().strip("/") for line in output.splitlines() if line.strip()]
        if not lines:
            self.tree.insert(root_id, "end", text="(No folders found at this depth)", values=("Info", ""))
            return
        for path in lines:
            parts = path.split("/")
            cumulative = []
            parent_key = ""
            for part in parts:
                cumulative.append(part)
                key = "/".join(cumulative)
                if key not in nodes:
                    parent_id = nodes[parent_key]
                    nodes[key] = self.tree.insert(parent_id, "end", text=part, values=("Folder", key), open=False)
                parent_key = key

    def mount(self):
        if self.mount_process and self.mount_process.poll() is None:
            messagebox.showinfo(APP_NAME, "A mount is already active.")
            return
        if not self.ensure_requirements() or not self.validate_inputs():
            return
        self.run_async(self._mount_task)

    def _mount_task(self):
        target = self.get_target()
        drive = self.parse_drive_value()
        cache_dir = self.cache_dir.get().strip()
        Path(cache_dir).mkdir(parents=True, exist_ok=True)

        cmd = [
            "rclone", "mount",
            target,
            drive,
            "--vfs-cache-mode", "writes",
            "--cache-dir", cache_dir,
            "--links",
        ]

        self.set_status(f"Mounting {target} to {drive}...")
        self.log("$ " + subprocess.list2cmdline(cmd))
        try:
            self.mount_process = subprocess.Popen(
                cmd,
                stdout=subprocess.PIPE,
                stderr=subprocess.STDOUT,
                text=True,
                bufsize=1,
            )
            self.mount_target.set(f"Mounted target: {target} → {drive}")
            self.after(0, self.refresh_drive_letters)
            threading.Thread(target=self._read_mount_output, daemon=True).start()
        except Exception as exc:
            self.log(f"Failed to start mount: {exc}")
            self.set_status("Mount failed to start.")

    def _read_mount_output(self):
        if not self.mount_process or not self.mount_process.stdout:
            return
        for line in self.mount_process.stdout:
            self.log(line.rstrip())
        rc = self.mount_process.wait()
        self.mount_target.set("Not mounted")
        self.after(0, self.refresh_drive_letters)
        self.set_status(f"Mount process exited with code {rc}.")

    def disconnect(self):
        if not self.mount_process or self.mount_process.poll() is not None:
            self.set_status("No active mount process to disconnect.")
            return
        self.set_status("Disconnecting mount...")
        try:
            self.mount_process.terminate()
            self.mount_process.wait(timeout=5)
        except subprocess.TimeoutExpired:
            self.mount_process.kill()
        except Exception as exc:
            self.log(f"Disconnect error: {exc}")
        finally:
            self.mount_target.set("Not mounted")
            self.after(0, self.refresh_drive_letters)
            self.set_status("Disconnected.")

    def open_mounted_drive(self):
        drive = self.parse_drive_value()
        if os.path.exists(drive + "\\"):
            os.startfile(drive + "\\")
        else:
            messagebox.showinfo(APP_NAME, f"{drive} is not currently visible.")

    def on_close(self):
        if self.mount_process and self.mount_process.poll() is None:
            if not messagebox.askyesno(APP_NAME, "Disconnect the mounted drive and close the app?"):
                return
            self.disconnect()
        self.destroy()

    def get_target(self):
        return f"{self.remote.get().strip()}:{self.bucket.get().strip()}"

    def find_winfsp(self):
        candidates = [
            Path(r"C:\Program Files\WinFsp"),
            Path(r"C:\Program Files (x86)\WinFsp"),
        ]
        return any(path.exists() for path in candidates)

    def open_install_instructions(self):
        import webbrowser
        webbrowser.open("https://wiki.projectgamelab.org/en/documentation/map-idrive-E2-to-windows")

    def open_rclone_config_folder(self):
        if not shutil.which("rclone"):
            messagebox.showerror(APP_NAME, "rclone is not in PATH.")
            return
        ok, output = self.run_capture(["rclone", "config", "file"])
        if not ok:
            messagebox.showerror(APP_NAME, "Could not locate rclone config.")
            return
        path = output.strip()
        if path:
            os.startfile(str(Path(path).parent))

    def run_async(self, func):
        if self.worker and self.worker.is_alive():
            messagebox.showinfo(APP_NAME, "Another task is already running.")
            return
        self.worker = threading.Thread(target=func, daemon=True)
        self.worker.start()

    def run(self, cmd, redact=None):
        redact = redact or set()
        logged = []
        for part in cmd:
            masked = part
            for secret in redact:
                if secret:
                    masked = masked.replace(secret, "***REDACTED***")
            logged.append(masked)
        self.log("$ " + subprocess.list2cmdline(logged))
        result = subprocess.run(cmd, capture_output=True, text=True)
        if result.stdout:
            self.log(result.stdout.strip())
        if result.stderr:
            self.log(result.stderr.strip())
        self.log(f"Exit code: {result.returncode}")
        return result.returncode == 0

    def run_capture(self, cmd):
        self.log("$ " + subprocess.list2cmdline(cmd))
        result = subprocess.run(cmd, capture_output=True, text=True)
        output = (result.stdout or "") + ("\n" + result.stderr if result.stderr else "")
        return result.returncode == 0, output.strip()


if __name__ == "__main__":
    if sys.platform != "win32":
        print("This app is intended for Windows.")
    app = App()
    app.mainloop()