runtime.py

"""Desktop paths, ArcGIS Python discovery, and an isolated read-only worker."""
import json
import os
import subprocess
import sys
import tempfile
import threading
import time
from contextlib import ExitStack
from pathlib import Path

WORKER_IDLE_TIMEOUT = 300


def finish_process(process):
    """Release process/file handles before removing its portable scratch folder."""
    if process.poll() is None:
        process.kill()
    process.wait(timeout=10)


def cleanup_scan(session):
    # Windows can briefly retain an event-file handle after process termination.
    for attempt in range(30):
        try:
            session.cleanup()
            return
        except (PermissionError, NotADirectoryError):
            # Python 3.11's TemporaryDirectory can turn a transient Windows
            # sharing violation on a file into NotADirectoryError while retrying.
            if attempt == 29:
                raise
            time.sleep(0.2)


def assets():
    return Path(getattr(sys, "_MEIPASS", Path(__file__).parent))


def interface_html(directory):
    directory = Path(directory)
    html = (directory / "index.html").read_text(encoding="utf-8").replace('<script src="demo.js"></script>', '')
    for tag, name, opening, closing in (
            ('<link rel="stylesheet" href="app.css">', "app.css", '<style>', '</style>'),
            ('<script src="app.js"></script>', "app.js", '<script>', '</script>')):
        if tag in html:
            html = html.replace(tag, opening + (directory / name).read_text(encoding="utf-8") + closing)
    return html


def app_dir():
    return Path(sys.executable).resolve().parent if getattr(sys, "frozen", False) else Path(__file__).resolve().parent


def data_dir():
    # A copied app folder carries its preferences with it. Never fall back to AppData.
    path = app_dir() / "Data"
    try:
        path.mkdir(parents=True, exist_ok=True)
    except OSError as exc:
        raise RuntimeError("Extract the whole app folder to a location you can write to, such as Documents or a USB drive.") from exc
    return path


def clear_browser_cache(directory):
    """Remove only this app's generated cache folders, never saved lists or links."""
    import re
    import shutil
    directory = Path(directory).resolve()
    for path in directory.glob("webview-*"):
        if (re.fullmatch(r"webview-[a-z0-9_]{8}", path.name) and path.is_dir()
                and not path.is_symlink() and path.resolve() == path.absolute()):
            try:
                shutil.rmtree(path)
            except OSError:
                pass  # An exiting browser or antivirus may still hold a file.


def close_browser_session(session):
    # WebView2 may release its files slightly after the window is gone.
    for attempt in range(30):
        session.cleanup()
        if not Path(session.name).exists():
            return
        time.sleep(0.2)


class AppInstance:
    def __init__(self, directory):
        self.handle = None
        if os.name != "nt":
            return
        import ctypes
        import hashlib
        from ctypes import wintypes
        self.kernel = ctypes.WinDLL("kernel32", use_last_error=True)
        self.kernel.CreateMutexW.argtypes = [ctypes.c_void_p, wintypes.BOOL, wintypes.LPCWSTR]
        self.kernel.CreateMutexW.restype = wintypes.HANDLE
        self.kernel.CloseHandle.argtypes = [wintypes.HANDLE]
        key = hashlib.sha256(os.path.normcase(str(Path(directory).resolve())).encode("utf-8")).hexdigest()
        self.handle = self.kernel.CreateMutexW(None, False, "Local\\TSMIS.ProjectInspector." + key)
        if not self.handle:
            raise ctypes.WinError(ctypes.get_last_error())
        if ctypes.get_last_error() == 183:  # ERROR_ALREADY_EXISTS
            self.close()
            raise RuntimeError("This app is already open from this folder. Use its existing window.")

    def close(self):
        if self.handle:
            self.kernel.CloseHandle(self.handle)
            self.handle = None


def prepare_desktop():
    if os.name != "nt":
        return
    import ctypes
    ctypes.windll.shell32.SetCurrentProcessExplicitAppUserModelID("TSMIS.ProjectInspector")
    if not getattr(sys, "frozen", False):
        return
    import hashlib
    from bundle_info import ASSEMBLIES
    # Explorer can propagate the ZIP's Internet zone to managed DLLs. Trust only
    # the three shipped assemblies whose hashes are compiled into this executable.
    for relative, expected in ASSEMBLIES.items():
        path = assets() / relative
        if not path.is_file() or hashlib.sha256(path.read_bytes()).hexdigest() != expected:
            raise RuntimeError("An app file is missing or damaged: " + path.name + ". Extract the complete ZIP again. If IT removed the file, ask IT to review the app.")
        try:
            Path(str(path) + ":Zone.Identifier").unlink(missing_ok=True)
        except OSError as exc:
            raise RuntimeError("Windows has blocked a downloaded app file. Right-click the original ZIP, choose Properties > Unblock, then extract it again.") from exc
    # Ask Explorer to refresh this executable's cached icon after an update.
    notify = ctypes.windll.shell32.SHChangeNotify
    notify.argtypes = [ctypes.c_long, ctypes.c_uint, ctypes.c_wchar_p, ctypes.c_void_p]
    notify.restype = None
    notify(0x2000, 0x0005, str(Path(sys.executable).resolve()), None)  # UPDATEITEM, PATHW


def check_webview_runtime():
    if os.name != "nt":
        return
    import ctypes
    if getattr(sys, "frozen", False):
        library = assets() / "webview"
    else:
        import importlib.util
        library = Path(importlib.util.find_spec("webview").origin).parent
    loader_path = library / "lib" / "runtimes" / "win-x64" / "native" / "WebView2Loader.dll"
    if not loader_path.is_file():
        raise RuntimeError("An app support file is missing: WebView2Loader.dll. Extract the complete ZIP again. If IT removed the file, ask IT to review the app.")
    loader = ctypes.WinDLL(str(loader_path))
    probe = loader.GetAvailableCoreWebView2BrowserVersionString
    probe.argtypes = [ctypes.c_wchar_p, ctypes.POINTER(ctypes.c_void_p)]
    probe.restype = ctypes.c_long
    version = ctypes.c_void_p()
    try:
        if probe(None, ctypes.byref(version)) < 0 or not version.value:
            raise RuntimeError("Microsoft Edge WebView2 Runtime was not found. It normally comes with Windows 11 and is already on most Windows 10 PCs. Ask IT to install or repair Microsoft Edge WebView2 Runtime.")
    finally:
        if version.value:
            ctypes.windll.ole32.CoTaskMemFree.argtypes = [ctypes.c_void_p]
            ctypes.windll.ole32.CoTaskMemFree(version)


def default_folder():
    documents = Path.home() / "Documents"
    if os.name == "nt":
        try:
            import winreg
            with winreg.OpenKey(winreg.HKEY_CURRENT_USER, r"Software\Microsoft\Windows\CurrentVersion\Explorer\User Shell Folders") as key:
                documents = Path(os.path.expandvars(winreg.QueryValueEx(key, "Personal")[0]))
        except OSError:
            pass
    return str(documents / "ArcGIS")


def find_arcgis_python():
    roots = [Path(os.environ.get("ProgramFiles", r"C:\Program Files")) / "ArcGIS" / "Pro",
             Path(os.environ.get("LOCALAPPDATA", str(Path.home()))) / "Programs" / "ArcGIS" / "Pro"]
    if os.name == "nt":
        import winreg
        for hive in (winreg.HKEY_CURRENT_USER, winreg.HKEY_LOCAL_MACHINE):
            try:
                with winreg.OpenKey(hive, r"SOFTWARE\ESRI\ArcGISPro", 0, winreg.KEY_READ | winreg.KEY_WOW64_64KEY) as key:
                    roots.insert(0, Path(winreg.QueryValueEx(key, "InstallDir")[0]))
            except OSError:
                pass
    for root in roots:
        candidate = root / "bin" / "Python" / "envs" / "arcgispro-py3" / "python.exe"
        if candidate.is_file():
            return str(candidate)
    return ""


def worker_environment(python):
    env = {k: v for k, v in os.environ.items() if not k.startswith(("PYTHON", "_PYI"))}
    prefix = Path(python).parent
    paths = [prefix, prefix / "Library" / "bin", prefix / "Scripts"]
    # Direct python.exe avoids requiring a shell on restricted work PCs.
    # The environment's ArcGISPro.pth supplies ArcPy; PATH supplies native DLLs.
    for p in prefix.parents:
        if p.name.lower() == "pro" and (p / "bin").is_dir():
            paths.append(p / "bin")
            break
    inherited = env.get("PATH", "").split(os.pathsep)
    if getattr(sys, "frozen", False):
        inherited = [p for p in inherited if p and not Path(p).is_relative_to(assets())]
    env["PATH"] = os.pathsep.join(str(p) for p in paths) + os.pathsep + os.pathsep.join(inherited)
    env["CONDA_PREFIX"] = str(prefix)
    env["PYTHONUTF8"] = "1"
    env["PYTHONIOENCODING"] = "utf-8"
    env["PYTHONDONTWRITEBYTECODE"] = "1"
    return env



_spawn_lock = threading.Lock()


def start_external(command, **kwargs):
    # Reset PyInstaller's inherited DLL directory so ArcGIS loads its own DLLs.
    with _spawn_lock:
        if os.name == "nt" and getattr(sys, "frozen", False):
            import ctypes
            kernel = ctypes.windll.kernel32
            old = ctypes.create_unicode_buffer(32768)
            kernel.GetDllDirectoryW(len(old), old)
            kernel.SetDllDirectoryW(None)
            try:
                return subprocess.Popen(command, **kwargs)
            finally:
                kernel.SetDllDirectoryW(old.value or None)
        return subprocess.Popen(command, **kwargs)


def balanced_groups(projects, count):
    """Put expensive projects first and balance their measured read times."""
    groups, totals = [[] for _ in range(count)], [0.0] * count
    for project in sorted(projects, key=lambda p: p.get("timings", {}).get("total_seconds", 0), reverse=True):
        index = min(range(count), key=lambda i: (totals[i], len(groups[i])))
        groups[index].append(project["path"])
        totals[index] += project.get("timings", {}).get("total_seconds", 0)
    return [group for group in groups if group]


def run_readers(owner, python, requests, children, progress):
    """Run up to two isolated readers, retaining partial results even on failure."""
    if not 1 <= len(requests) <= 2:
        raise ValueError("A scan needs one or two readers.")
    try:
        for request in requests:
            with owner.lock:
                if owner.cancelled.is_set():
                    break
                child = ScanRunner()
                children.append(child)
                owner._children = children[:]
                child.start(python, request)
        while children:
            snapshots = [child.snapshot() for child in children]
            progress(children, snapshots)
            if owner.cancelled.is_set() or any(state["error"] for state in snapshots):
                for child in children:
                    child.stop()
                break
            if not any(state["running"] for state in snapshots):
                break
            owner.cancelled.wait(0.12)
    finally:
        try:
            for child in children:
                child.close()
            progress(children, [child.snapshot() for child in children])
        finally:
            with owner.lock:
                owner._children = []


class ScanRunner:
    def __init__(self, on_complete=None):
        self.lock = threading.RLock()
        self.cancelled = threading.Event()
        self.process = None
        self._children = []
        self._thread = None
        self._on_complete = on_complete
        self.state = {"running": False, "message": "Ready to scan", "total": 0, "completed": 0,
                      "error": "", "result": None, "last_refreshed": None, "save_error": ""}

    def restore(self, record=None):
        with self.lock:
            if self.state["running"]:
                raise ValueError("Finish or stop the refresh before switching folders.")
            result = record["result"] if record else None
            count = len(result["projects"]) if result else 0
            self.state.update(result=result, error="", save_error="", completed=count, total=count,
                              message="Saved list" if result else "Ready", test_progress=None,
                              last_refreshed=record["refreshed_at"] if record else None)

    def snapshot(self):
        with self.lock:
            # Poll summaries; layer detail is fetched only when a project is selected.
            state = {k: v for k, v in self.state.items() if k != "result"}
            result = self.state["result"]
            state["projects"] = [{k: v for k, v in p.items() if k not in {"rows", "connection_metadata", "timings"}}
                                  for p in result["projects"]] if result else []
            state["warnings"] = result.get("warnings", []) if result else []
            state["diagnostic_scan"] = bool(result and result.get("diagnostic_scan"))
            state["has_result"] = result is not None
            state["complete"] = bool(result and result.get("complete"))
            state["root"] = result.get("root", "") if result else ""
            performance = result.get("performance", {}) if result else {}
            state["test_summary"] = performance.get("summary", "")
            state["test_status"] = performance.get("status", "")
            return state

    def start(self, python, request, readers=1):
        with self.lock:
            if self.state["running"]:
                raise ValueError("A scan is already running.")
            if readers not in (1, 2) or (readers == 2 and request.get("diagnostics")):
                raise ValueError("Use one reader for a diagnostic baseline, or up to two for Refresh.")
            if not (request.get("_timing_pass") or request.get("_probe_only") or "_project_paths" in request) and not Path(request["root"]).is_dir():
                raise ValueError("Choose an existing folder containing your ArcGIS Pro projects.")
            if not python or not Path(python).is_file():
                raise ValueError("ArcGIS Pro was not found on this PC. On the work PC, open Settings to choose ArcGIS Pro's python.exe if it isn't detected.")
            if not request.get("match", "").strip():
                raise ValueError("Enter the TSMIS identifier in Settings (usually tsmis).")
            costs = {}
            for project in (self.state["result"] or {}).get("projects", []):
                timings = project.get("timings")
                seconds = timings.get("total_seconds") if isinstance(timings, dict) else None
                # Saved durations are scheduling hints, never a reason to make
                # a list unusable. Ignore damaged or unreasonable values.
                if isinstance(seconds, (int, float)) and 0 < seconds <= 86400:
                    costs[os.path.normcase(project["path"])] = seconds
            self.cancelled.clear()
            self.state.update(running=True, message="Finding projects…", total=0, completed=0, error="", save_error="",
                              test_progress=None,
                              result={**request, "diagnostic_scan": bool(request.get("diagnostics")),
                                      "python_executable": python, "projects": [], "warnings": [], "complete": False})
            if request.get("diagnostics") and not request.get("_timing_pass"):
                self.state["result"]["performance"] = {"schema": 1, "status": "running", "passes": [], "summary": "Initial diagnostic scan…"}
            self._thread = threading.Thread(target=self._run, args=(python, request, readers, costs), daemon=True)
            self._thread.start()

    def stop(self):
        self.cancelled.set()
        with self.lock:
            for child in self._children:
                child.stop()

    def close(self):
        self.stop()
        if self._thread:
            self._thread.join(timeout=12)
        process = self.process
        if process and process.poll() is None:
            process.kill()
            process.wait(timeout=5)
        if self._thread and self._thread.is_alive():
            self._thread.join(timeout=5)
        for child in self._children:
            child.close()

    def _event(self, event):
        with self.lock:
            result = self.state["result"]
            kind = event.get("type")
            if kind == "begin":
                result.update(event["metadata"])
            elif kind == "discovered":
                self.state["total"] = event["total"]
                result["warnings"] = event["warnings"]
                result["projects_found"] = event["total"]
            elif kind == "progress":
                self.state["message"] = event["message"]
                if result.get("diagnostics"):
                    result["last_progress"] = event["message"]
            elif kind == "project":
                result["projects"].append(event["project"])
                self.state["completed"] = len(result["projects"])
            elif kind == "runtime":
                result["arcgis_version"] = event["arcgis_version"]
            elif kind == "timings":
                result.setdefault("timings", {}).update(event["timings"])
            elif kind == "file_probe":
                result["file_probe"] = event["files"]
                self.state["completed"] = len(event["files"])
            elif kind == "read_stage":
                result["last_read"] = {k: v for k, v in event.items() if k != "type"}
                self.state["message"] = event["stage"] + " · " + Path(event["project"]).name
            elif kind == "done":
                result.update(complete=event.get("complete", True), arcgis_version=event["arcgis_version"])
                if not result["complete"]:
                    self.state["error"] = "Some folders could not be read. Your previous saved list is unchanged."
                self.state["message"] = "Scan complete" if self.state["total"] else "No .aprx projects found in this folder"
            elif kind == "fatal":
                self.state["error"] = event["message"]
                result["error"] = event["message"]

    def _run_parallel(self, python, request, costs):
        def discovering(children, snapshots):
            if snapshots:
                with self.lock:
                    self.state["total"] = snapshots[0]["total"]
                    self.state["result"]["warnings"] = snapshots[0]["warnings"]

        probes = []
        run_readers(self, python, [dict(request, _probe_only=True)], probes, discovering)
        if self.cancelled.is_set():
            return
        probe = probes[0].state
        found = probe["result"] or {}
        if probe["error"] or not found.get("complete") or "file_probe" not in found:
            raise RuntimeError(probe["error"] or "The project folder could not be fully checked. Previous saved list kept.")
        files = sorted(found["file_probe"], key=str.casefold)
        with self.lock:
            result = self.state["result"]
            result.update({key: found[key] for key in ("app_version", "scan_time", "python_version") if key in found})
            result.update(projects_found=len(files), reader_count=min(2, len(files)))
            if not files:
                result.update(complete=True, arcgis_version="Not needed; no projects found")
                self.state["message"] = "No .aprx projects found in this folder"
                return
        known = [costs[os.path.normcase(path)] for path in files if os.path.normcase(path) in costs]
        default_cost = sum(known) / len(known) if known else 1.0
        groups = balanced_groups([{"path": path, "timings": {"total_seconds": costs.get(os.path.normcase(path), default_cost)}}
                                  for path in files], 2)
        requests = [dict(request, _project_paths=group) for group in groups]
        received, offsets = set(), [0] * len(groups)

        def reading(children, snapshots):
            # Append once in completion order, so an open detail view keeps the
            # same project index when the other reader finishes a project.
            for index, child in enumerate(children):
                with child.lock:
                    projects = list((child.state["result"] or {}).get("projects", []))
                for project in projects[offsets[index]:]:
                    path = project["path"]
                    if path not in groups[index] or path in received:
                        raise RuntimeError("A reader returned duplicate or unexpected projects. Previous saved list kept.")
                    received.add(path)
                    self._event({"type": "project", "project": project})
                offsets[index] = len(projects)
            messages = [state["message"] for state in snapshots if state["running"]]
            with self.lock:
                self.state["message"] = " / ".join(messages) if messages else "Finishing scan…"

        children = []
        run_readers(self, python, requests, children, reading)
        if self.cancelled.is_set():
            return
        for child in children:
            if child.state["error"] or not (child.state["result"] or {}).get("complete"):
                raise RuntimeError(child.state["error"] or "A project reader did not finish. Previous saved list kept.")
        if received != set(files):
            raise RuntimeError("Some project results are missing. Previous saved list kept.")
        with self.lock:
            self.state["result"].update(complete=True, arcgis_version=children[0].state["result"].get("arcgis_version", "Unknown"))
            self.state["message"] = "Scan complete"

    def _run(self, python, request, readers=1, costs=None):
        process = None
        launched = None
        try:
            if readers == 2:
                launched = time.monotonic()
                self._run_parallel(python, request, costs or {})
                return
            with ExitStack() as cleanup:
                session = tempfile.TemporaryDirectory(prefix="scan-", dir=data_dir())
                cleanup.callback(cleanup_scan, session)
                folder = Path(session.name)
                request_path, event_path = folder / "request.json", folder / "events.jsonl"
                request_path.write_text(json.dumps(request), encoding="utf-8")
                event_path.touch()
                script = assets() / "worker" / "worker.py" if getattr(sys, "frozen", False) else assets() / "worker.py"
                launched = time.monotonic()
                process = start_external([python, "-B", "-u", str(script), "--request", str(request_path), "--events", str(event_path)],
                                           cwd=script.parent, env=worker_environment(python), stdin=subprocess.DEVNULL,
                                           stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
                                           creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0))
                self.process = process
                cleanup.callback(finish_process, process)
                started = last_event = time.monotonic()
                with event_path.open(encoding="utf-8") as stream:
                    while True:
                        if self.cancelled.is_set():
                            process.terminate()
                            with self.lock:
                                self.state["message"] = "Scan stopped. Completed projects are available."
                                self.state["result"]["cancelled"] = True
                            break
                        position = stream.tell()
                        line = stream.readline()
                        if line and line.endswith("\n"):
                            if request.get("diagnostics"):
                                with self.lock:
                                    self.state["result"].setdefault("timings", {}).setdefault("first_event_seconds", time.monotonic() - launched)
                            self._event(json.loads(line))
                            last_event = time.monotonic()
                            continue
                        stream.seek(position)
                        if process.poll() is not None:
                            # A final read closes the race between EOF and worker exit.
                            for tail in stream:
                                if tail.strip():
                                    self._event(json.loads(tail))
                            break
                        if time.monotonic() - last_event > WORKER_IDLE_TIMEOUT:
                            with self.lock:
                                self.state["error"] = "ArcGIS did not respond for five minutes. Completed projects are available; check the last project and retry."
                            process.terminate()
                            raise RuntimeError(self.state["error"])
                        if time.monotonic() - started > 25 and self.state["completed"] == 0:
                            with self.lock:
                                self.state["message"] = "Waiting for ArcGIS Pro… You can stop the scan if it cannot sign in or read a project."
                        time.sleep(0.12)
                process.wait(timeout=10)
                with self.lock:
                    if request.get("diagnostics"):
                        self.state["result"].setdefault("timings", {})["wall_seconds"] = time.monotonic() - launched
                    if (process.returncode != 0 or not self.state["result"]["complete"]) and not self.state["error"] and not self.cancelled.is_set():
                        self.state["error"] = "ArcGIS Python stopped before finishing. Open ArcGIS Pro and sign in, then verify its Python environment in Settings."
                if (request.get("diagnostics") and not request.get("_timing_pass")
                        and not self.state["error"] and not self.cancelled.is_set()):
                    from diagnostics import run_tests
                    self.process = None
                    run_tests(self, python, request, folder)
        except Exception as exc:
            with self.lock:
                self.state["error"] = self.state["error"] or (str(exc) if isinstance(exc, RuntimeError) else "Could not run the ArcGIS reader (" + type(exc).__name__ + "). Check the Python path and folder permissions.")
        finally:
            for child in self._children:
                child.close()
            self._children = []
            if process and process.poll() is None:
                process.kill()
                process.wait()
            self.process = None
            with self.lock:
                if (request.get("diagnostics") or readers == 2) and launched is not None:
                    self.state["result"].setdefault("timings", {}).setdefault("wall_seconds", time.monotonic() - launched)
                self.state["result"]["error"] = self.state["error"]
                self.state["result"]["projects_read"] = self.state["completed"]
                if self.state["error"] or self.cancelled.is_set():
                    self.state["result"]["complete"] = False
                    if self.cancelled.is_set():
                        self.state["result"]["cancelled"] = True
                        self.state["message"] = "Scan stopped. Completed projects are available."
                    performance = self.state["result"].get("performance")
                    if performance:
                        performance["status"] = "cancelled" if self.cancelled.is_set() else "incomplete"
                        performance["summary"] = "Test stopped. Save the diagnostic ZIP for partial timings." if self.cancelled.is_set() else "Test incomplete. Save the diagnostic ZIP for details."
                if self.state["result"]["complete"] and self._on_complete:
                    try:
                        self.state["last_refreshed"] = self._on_complete(self.state["result"])
                    except Exception:
                        self.state["save_error"] = "This list could not be saved. Your previous saved list is unchanged. Check that the app folder is writable."
                self.state["running"] = False