update_yandex_sdk.py

"""Coordinate package imports with the active Unity Editor instance."""
 
from __future__ import annotations
 
import json
import os
import re
import shlex
import shutil
import subprocess
import sys
import time
import uuid
from contextlib import contextmanager, nullcontext
from pathlib import Path
from typing import Iterator
 
 
ASSETS_DIRECTORY = "Assets"
TARGET_DIRECTORY = "YandexMobileAds"
PRESERVED_DIRECTORY = "AdRevenue"
PACKAGE_EXTENSION = ".unitypackage"
YANDEX_MOBILE_ADS_PACKAGE_PREFIX = "yandex-mobileads-"
MEDIATION_ADAPTER_PACKAGE_PREFIX = "mobileads-"
XDG_CONFIG_HOME_ENVIRONMENT_VARIABLE = "XDG_CONFIG_HOME"
XDG_DOWNLOAD_DIR_ENVIRONMENT_VARIABLE = "XDG_DOWNLOAD_DIR"
XDG_USER_DIRECTORIES_FILENAME = "user-dirs.dirs"
BRIDGE_DIRECTORY = "YandexSdkUpdater"
REQUEST_FILENAME = "request.json"
RESULT_FILENAME = "result.json"
RECOVERY_FILENAME = "recovery.json"
LOCK_FILENAME = "session.lock"
ACTIVE_SESSION_FILENAME = "active-session.json"
STARTUP_ATTEMPT_ENVIRONMENT_VARIABLE = "YANDEX_SDK_UPDATER_STARTUP_ATTEMPT"
MANUAL_IMPORT_COMPLETION_FIELD = "ManualImportCompletionConfirmed"
CORE_PACKAGE_PATTERN = re.compile(
    rf"^{re.escape(YANDEX_MOBILE_ADS_PACKAGE_PREFIX)}"
    r"(lite|mediation)-(\d+(?:\.\d+)+)\.unitypackage$",
    re.IGNORECASE,
)
CANCEL_COMMANDS = frozenset({"q", "quit", "n", "no", "cancel"})
POLL_INTERVAL_SECONDS = 0.35
EDITOR_RESPONSE_TIMEOUT_SECONDS = 60
EDITOR_BUSY_RETRY_SECONDS = 60
EDITOR_NOT_READY_ERROR = "The editor reported an error: The editor is not ready to process the request."
EDITOR_WAIT_WARNING_SECONDS = 30
EDITOR_STATUS_INTERVAL_SECONDS = 10
UNITY_BACKGROUND_RESPONSE_GRACE_SECONDS = 3.0
UNITY_FOCUS_DELAY_SECONDS = 0.65
UNITY_FOCUS_VERIFY_TIMEOUT_SECONDS = 1.0
UNITY_FOCUS_VERIFY_INTERVAL_SECONDS = 0.05
WINDOW_RESTORE_COMMAND = 9
UNITY_FOCUS_MAX_WINDOWS = 16
ESCAPE_KEY = "\x1b"
 
 
class UpdateError(Exception):
    """Represent an operation that cannot proceed safely."""
 
 
class UserCancelled(Exception):
    """Represent an explicit request to cancel the operation."""
 
 
class EscapeCancelled(UserCancelled):
    """Represent a request to exit immediately using Escape."""
 
 
class EditorOperationError(UpdateError):
    """Represent a definitively failed editor operation."""
 
 
class EditorOperationCancelled(UserCancelled):
    """Represent an operation definitively cancelled by the editor."""
 
 
class EditorOperationInterrupted(UpdateError):
    """Represent an editor import whose completion must be verified manually."""
 
 
def get_downloads_directory() -> Path:
    """Resolve the configured downloads directory on Linux, with a home-folder fallback."""
    fallback = Path.home() / "Downloads"
    if not sys.platform.startswith("linux"):
        return fallback
 
    home = os.environ.get("HOME") or str(Path.home())
 
    def expand_path(value: str) -> Path:
        value = re.sub(r"\$\{HOME\}|\$HOME(?=/|$)", lambda _: home, value)
        return Path(os.path.expandvars(value)).expanduser()
 
    configured_directory = os.environ.get(XDG_DOWNLOAD_DIR_ENVIRONMENT_VARIABLE)
    if configured_directory:
        path = expand_path(configured_directory)
        if path.is_absolute():
            return path
 
    configured_home = os.environ.get(XDG_CONFIG_HOME_ENVIRONMENT_VARIABLE)
    config_home = expand_path(configured_home) if configured_home else Path(home) / ".config"
    if not config_home.is_absolute():
        config_home = Path(home) / ".config"
 
    user_directories_file = config_home / XDG_USER_DIRECTORIES_FILENAME
    try:
        lines = user_directories_file.read_text(encoding="utf-8").splitlines()
    except OSError:
        return fallback
 
    for line in lines:
        match = re.match(r"^\s*XDG_DOWNLOAD_DIR\s*=\s*(.*?)\s*$", line)
        if not match:
            continue
 
        try:
            values = shlex.split(match.group(1), comments=True, posix=True)
        except ValueError:
            continue
 
        if len(values) != 1:
            continue
 
        path = expand_path(values[0])
        if path.is_absolute():
            return path
 
    return fallback
 
 
def find_assets_directory(script_directory: Path) -> Path:
    """Find the Unity assets root enclosing the script.
 
    Args:
        script_directory: Directory containing the running script.
 
    Returns:
        The nearest matching assets directory.
 
    Raises:
        UpdateError: The script is outside an assets directory.
    """
    for directory in (script_directory, *script_directory.parents):
        if directory.name.casefold() == ASSETS_DIRECTORY.casefold():
            return directory
 
    raise UpdateError(
        f"The script is not inside an {ASSETS_DIRECTORY} directory. "
        "Move it into the Unity project."
    )
 
 
def validate_script_directory(script_directory: Path) -> None:
    """Enforce the intended script location.
 
    Args:
        script_directory: Directory containing the running script.
 
    Raises:
        UpdateError: The script is outside the required directory.
    """
    if script_directory.name.casefold() != TARGET_DIRECTORY.casefold():
        raise UpdateError(
            f"Run this script from a {TARGET_DIRECTORY} directory within {ASSETS_DIRECTORY}."
        )
 
 
def version_key(version: str) -> tuple[int, ...]:
    """Convert a dotted version into a sortable numeric tuple.
 
    Args:
        version: Dotted version number.
 
    Returns:
        Parsed version components.
    """
    return tuple(int(part) for part in version.split("."))
 
 
def discover_packages(downloads_directory: Path) -> tuple[str, list[Path], list[str]]:
    """Identify the newest compatible set of downloaded packages.
 
    Args:
        downloads_directory: Folder containing package archives.
 
    Returns:
        Selected version, ordered package paths, and ignored core versions.
 
    Raises:
        UpdateError: No suitable core packages are available.
    """
    if not downloads_directory.is_dir():
        raise UpdateError(f"The downloads directory was not found: {downloads_directory}")
 
    packages = sorted(
        (
            path
            for path in downloads_directory.iterdir()
            if path.is_file() and path.suffix.casefold() == PACKAGE_EXTENSION
        ),
        key=lambda path: path.name.casefold(),
    )
 
    core_packages: dict[tuple[str, str], Path] = {}
    for package in packages:
        match = CORE_PACKAGE_PATTERN.fullmatch(package.name)
        if match:
            kind, version = match.groups()
            core_packages[(kind.casefold(), version)] = package
 
    if not core_packages:
        raise UpdateError(
            "No core SDK packages were found. At least one matching "
            "lite or mediation package is required."
        )
 
    versions = {version for _, version in core_packages}
    version = max(versions, key=version_key)
    lite = core_packages.get(("lite", version))
    mediation = core_packages.get(("mediation", version))
    mediation_suffix = f"-mediation-{version}{PACKAGE_EXTENSION}".casefold()
    mediation_adapters = sorted(
        (
            package
            for package in packages
            if package.name.casefold().startswith(
                (YANDEX_MOBILE_ADS_PACKAGE_PREFIX, MEDIATION_ADAPTER_PACKAGE_PREFIX)
            )
            and package.name.casefold().endswith(mediation_suffix)
            and package != mediation
        ),
        key=lambda path: path.name.casefold(),
    )
    ordered = [lite] if lite is not None else []
    ordered.extend(mediation_adapters)
    if mediation is not None:
        ordered.append(mediation)
 
    ignored = sorted(versions - {version}, key=version_key)
    return version, ordered, ignored
 
 
def metadata_path(asset_path: Path) -> Path:
    """Get the companion metadata path for an asset.
 
    Args:
        asset_path: Asset directory or file path.
 
    Returns:
        The expected companion metadata file path.
    """
    return asset_path.with_name(f"{asset_path.name}.meta")
 
 
def path_exists(path: Path) -> bool:
    """Check whether a path is occupied, including dangling links.
 
    Args:
        path: Path to inspect.
 
    Returns:
        Whether the path is present or is a symbolic link.
    """
    return path.exists() or path.is_symlink()
 
 
def check_escape() -> None:
    """Cancel immediately when Escape is pressed in a Windows console.
 
    Raises:
        EscapeCancelled: Escape was pressed.
    """
    if sys.platform != "win32":
        return
 
    import msvcrt
 
    while msvcrt.kbhit():
        key = msvcrt.getwch()
        if key == ESCAPE_KEY:
            raise EscapeCancelled("The operation was cancelled by the user (Esc).")
        if key in ("\x00", "\xe0"):
            msvcrt.getwch()
 
 
def read_console_enter_pressed() -> bool:
    """Read a pending Enter key without blocking the editor response wait.
 
    Returns:
        Whether Enter was pressed in the interactive console.
 
    Raises:
        EscapeCancelled: Escape was pressed on Windows.
    """
    if not sys.stdin.isatty():
        return False
 
    if sys.platform == "win32":
        import msvcrt
 
        enter_pressed = False
        while msvcrt.kbhit():
            key = msvcrt.getwch()
            if key == ESCAPE_KEY:
                print()
                raise EscapeCancelled("The operation was cancelled by the user (Esc).")
            if key in ("\x00", "\xe0"):
                msvcrt.getwch()
            elif key in ("\r", "\n"):
                print()
                enter_pressed = True
        return enter_pressed
 
    import select
 
    readable, _, _ = select.select([sys.stdin], [], [], 0)
    if not readable:
        return False
 
    entered_text = sys.stdin.readline()
    return bool(entered_text)
 
 
def read_console_input(message: str) -> str:
    """Read a console response while allowing Escape without Enter.
 
    Args:
        message: Prompt to display.
 
    Returns:
        Entered text.
 
    Raises:
        EscapeCancelled: Escape was pressed.
    """
    if sys.platform != "win32":
        return input(message)
 
    import msvcrt
 
    print(message, end="", flush=True)
    characters = []
    while True:
        key = msvcrt.getwch()
        if key == ESCAPE_KEY:
            print()
            raise EscapeCancelled("The operation was cancelled by the user (Esc).")
        if key in ("\r", "\n"):
            print()
            return "".join(characters)
        if key == "\b":
            if characters:
                characters.pop()
                print("\b \b", end="", flush=True)
        elif key in ("\x00", "\xe0"):
            msvcrt.getwch()
        elif key.isprintable():
            characters.append(key)
            print(key, end="", flush=True)
 
 
def ask_to_continue(message: str) -> None:
    """Wait for confirmation, allowing cancellation by command or Escape.
 
    Args:
        message: Prompt to display.
 
    Raises:
        UserCancelled: The user entered a cancellation command.
    """
    if read_console_input(message).strip().casefold() in CANCEL_COMMANDS:
        raise UserCancelled("The operation was cancelled by the user.")
 
 
def atomic_write_json(path: Path, content: dict[str, object]) -> None:
    """Replace a small JSON document without exposing partial writes.
 
    Args:
        path: Destination path.
        content: Data to serialize.
    """
    temporary = path.with_name(f"{path.name}.{uuid.uuid4().hex}.tmp")
    try:
        temporary.write_text(json.dumps(content, ensure_ascii=True), encoding="utf-8")
        os.replace(temporary, path)
    finally:
        try:
            temporary.unlink()
        except FileNotFoundError:
            pass
 
 
def read_json(path: Path) -> dict[str, object] | None:
    """Read a state document, tolerating transient file replacement.
 
    Args:
        path: JSON document path.
 
    Returns:
        Parsed document, or None when unavailable or incomplete.
    """
    try:
        content = json.loads(path.read_text(encoding="utf-8"))
        return content if isinstance(content, dict) else None
    except (FileNotFoundError, UnicodeDecodeError, json.JSONDecodeError):
        return None
 
 
@contextmanager
def exclusive_session(lock_path: Path) -> Iterator[None]:
    """Prevent multiple processes from controlling one project simultaneously.
 
    Args:
        lock_path: Path for an operating-system-managed lock.
 
    Yields:
        Control after an exclusive lock is acquired.
 
    Raises:
        UpdateError: Another process already owns the lock.
    """
    with lock_path.open("a+b") as lock_file:
        if sys.platform == "win32":
            import msvcrt
 
            lock_file.seek(0)
            if lock_file.read(1) == b"":
                lock_file.seek(0)
                lock_file.write(b"0")
                lock_file.flush()
            lock_file.seek(0)
            try:
                msvcrt.locking(lock_file.fileno(), msvcrt.LK_NBLCK, 1)
            except OSError as exc:
                raise UpdateError("Another updater process is already running.") from exc
            try:
                yield
            finally:
                lock_file.seek(0)
                msvcrt.locking(lock_file.fileno(), msvcrt.LK_UNLCK, 1)
        else:
            import fcntl
 
            try:
                fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
            except OSError as exc:
                raise UpdateError("Another updater process is already running.") from exc
            try:
                yield
            finally:
                fcntl.flock(lock_file.fileno(), fcntl.LOCK_UN)
 
 
@contextmanager
def focus_unity_editor(project_name: str) -> Iterator[bool]:
    """Focus the matching Unity Editor until the operation is resolved.
 
    Args:
        project_name: Current Unity project directory name.
 
    Yields:
        Whether the editor was successfully focused.
    """
    if sys.platform == "win32":
        with focus_unity_editor_windows(project_name) as focused:
            yield focused
    elif sys.platform == "darwin":
        with activate_unity_windows_macos(project_name) as focused:
            yield focused
    elif sys.platform.startswith("linux"):
        with activate_unity_windows_linux(project_name) as focused:
            yield focused
    else:
        yield False
 
 
@contextmanager
def focus_unity_editor_windows(project_name: str) -> Iterator[bool]:
    """Temporarily activate the editor window of the current project on Windows.
 
    Args:
        project_name: Current Unity project directory name.
 
    Yields:
        Whether the matching editor was activated.
    """
    import ctypes
    from ctypes import wintypes
 
    user32 = ctypes.WinDLL("user32", use_last_error=True)
    kernel32 = ctypes.WinDLL("kernel32", use_last_error=True)
    user32.GetForegroundWindow.restype = wintypes.HWND
    user32.GetWindowThreadProcessId.argtypes = (wintypes.HWND, ctypes.POINTER(wintypes.DWORD))
    user32.GetWindowThreadProcessId.restype = wintypes.DWORD
    user32.IsWindowVisible.argtypes = (wintypes.HWND,)
    user32.IsIconic.argtypes = (wintypes.HWND,)
    user32.GetWindowTextLengthW.argtypes = (wintypes.HWND,)
    user32.GetWindowTextW.argtypes = (wintypes.HWND, wintypes.LPWSTR, ctypes.c_int)
    user32.SetForegroundWindow.argtypes = (wintypes.HWND,)
    user32.SetForegroundWindow.restype = wintypes.BOOL
    user32.IsWindow.argtypes = (wintypes.HWND,)
    user32.IsWindow.restype = wintypes.BOOL
    user32.ShowWindowAsync.argtypes = (wintypes.HWND, ctypes.c_int)
    user32.ShowWindowAsync.restype = wintypes.BOOL
    kernel32.OpenProcess.argtypes = (wintypes.DWORD, wintypes.BOOL, wintypes.DWORD)
    kernel32.OpenProcess.restype = wintypes.HANDLE
    kernel32.QueryFullProcessImageNameW.argtypes = (
        wintypes.HANDLE, wintypes.DWORD, wintypes.LPWSTR, ctypes.POINTER(wintypes.DWORD)
    )
    kernel32.CloseHandle.argtypes = (wintypes.HANDLE,)
 
    def focus_window(handle: int) -> bool:
        """Activate a window while observing Windows foreground restrictions.
 
        Args:
            handle: Native window handle.
 
        Returns:
            Whether the requested window actually became the foreground window.
        """
        if not user32.IsWindow(handle):
            return False
        if user32.IsIconic(handle):
            user32.ShowWindowAsync(handle, WINDOW_RESTORE_COMMAND)
        if user32.GetForegroundWindow() == handle:
            return True
        # Keep input queues separate so an unresponsive target cannot block activation.
        user32.SetForegroundWindow(handle)
        return wait_for_foreground(handle)
 
    def wait_for_foreground(handle: int) -> bool:
        """Wait briefly for asynchronous window activation to finish.
 
        Args:
            handle: Expected foreground window handle.
 
        Returns:
            Whether the expected window became active before the deadline.
        """
        deadline = time.monotonic() + UNITY_FOCUS_VERIFY_TIMEOUT_SECONDS
        while user32.GetForegroundWindow() != handle:
            check_escape()
            if time.monotonic() >= deadline:
                return False
            time.sleep(UNITY_FOCUS_VERIFY_INTERVAL_SECONDS)
        return True
 
    editors = []
    seen_processes = set()
    callback_type = ctypes.WINFUNCTYPE(wintypes.BOOL, wintypes.HWND, wintypes.LPARAM)
 
    def collect_window(handle: int, parameter: int) -> bool:
        """Collect a visible top-level window for each Unity Editor process.
 
        Args:
            handle: Native window handle.
            parameter: Unused window enumeration argument.
 
        Returns:
            Whether enumeration should continue.
        """
        if not user32.IsWindowVisible(handle) or user32.IsIconic(handle):
            return True
        title_length = user32.GetWindowTextLengthW(handle)
        if not title_length:
            return True
        process_id = wintypes.DWORD()
        user32.GetWindowThreadProcessId(handle, ctypes.byref(process_id))
        if not process_id.value or process_id.value in seen_processes:
            return True
        process = kernel32.OpenProcess(0x1000, False, process_id.value)
        if not process:
            return True
        try:
            executable = ctypes.create_unicode_buffer(4096)
            length = wintypes.DWORD(len(executable))
            if not kernel32.QueryFullProcessImageNameW(process, 0, executable, ctypes.byref(length)):
                return True
            if executable.value.rsplit("\\", 1)[-1].casefold() != "unity.exe":
                return True
            title = ctypes.create_unicode_buffer(title_length + 1)
            user32.GetWindowTextW(handle, title, len(title))
            seen_processes.add(process_id.value)
            editors.append((handle, process_id.value, title.value.casefold()))
            return len(editors) < UNITY_FOCUS_MAX_WINDOWS
        finally:
            kernel32.CloseHandle(process)
 
    callback = callback_type(collect_window)
    user32.EnumWindows.argtypes = (callback_type, wintypes.LPARAM)
    user32.EnumWindows(callback, 0)
 
    matches = [editor for editor in editors if project_name.casefold() in editor[2]]
    if len(matches) == 1:
        selected = matches[0]
    elif len(editors) == 1:
        selected = editors[0]
    else:
        selected = None
    original = user32.GetForegroundWindow()
    focused = selected is not None and focus_window(selected[0])
    try:
        yield focused
    finally:
        if focused and original and original != selected[0]:
            foreground = user32.GetForegroundWindow()
            active_process_id = wintypes.DWORD()
            user32.GetWindowThreadProcessId(foreground, ctypes.byref(active_process_id))
            if active_process_id.value == selected[1]:
                focus_window(original)
 
 
@contextmanager
def activate_unity_windows_macos(project_name: str) -> Iterator[bool]:
    """Temporarily activate this project's Unity Editor on macOS.
 
    Args:
        project_name: Current Unity project directory name.
 
    Yields:
        Whether System Events activated the matching Unity Editor process.
    """
    script = f"""
    on run argv
        set projectName to item 1 of argv
        tell application "System Events"
            set previousProcess to first application process whose frontmost is true
            set previousProcessId to unix id of previousProcess
            set editors to every application process whose name is "Unity"
            set matchingEditors to {{}}
            repeat with editorProcess in editors
                try
                    set processMatches to false
                    repeat with editorWindow in every window of editorProcess
                        ignoring case
                            if (name of editorWindow as text) contains projectName then
                                set processMatches to true
                            end if
                        end ignoring
                    end repeat
                    if processMatches then set end of matchingEditors to editorProcess
                end try
            end repeat
 
            if (count of matchingEditors) is 1 then
                set selectedEditor to item 1 of matchingEditors
            else if (count of editors) is 1 then
                set selectedEditor to item 1 of editors
            else
                return (previousProcessId as string) & "|0|0"
            end if
 
            set activatedCount to 0
            set activatedProcessId to 0
            try
                set frontmost of selectedEditor to true
                delay {UNITY_FOCUS_DELAY_SECONDS}
                set activatedProcessId to unix id of selectedEditor
                set activatedCount to 1
            end try
            return (previousProcessId as string) & "|" & (activatedProcessId as string) & "|" & (activatedCount as string)
        end tell
    end run
    """
    previous_process_id = None
    activated_process_id = None
    focused = False
    try:
        result = subprocess.run(
            ["osascript", "-e", script, project_name],
            capture_output=True,
            text=True,
            timeout=20,
            check=True,
        )
        previous_id, activated_id, activated_count = result.stdout.strip().split("|")
        previous_process_id = int(previous_id)
        activated_process_id = int(activated_id)
        focused = int(activated_count) > 0
    except (OSError, ValueError, subprocess.CalledProcessError, subprocess.TimeoutExpired):
        pass
 
    try:
        yield focused
    finally:
        if focused and previous_process_id and activated_process_id:
            restore_script = f"""
                tell application "System Events"
                    set activeProcess to first application process whose frontmost is true
                    if (unix id of activeProcess) is {activated_process_id} then
                        set previousProcess to first application process whose unix id is {previous_process_id}
                        set frontmost of previousProcess to true
                    end if
                end tell
            """
            try:
                subprocess.run(
                    ["osascript", "-e", restore_script],
                    capture_output=True,
                    text=True,
                    timeout=20,
                    check=True,
                )
            except (OSError, subprocess.CalledProcessError, subprocess.TimeoutExpired):
                pass
 
 
def run_xdotool(*arguments: str) -> str | None:
    """Run one X11 window operation without exposing command errors.
 
    Args:
        *arguments: Arguments passed to the xdotool executable.
 
    Returns:
        Captured output, or None if the operation could not complete.
    """
    try:
        result = subprocess.run(
            ["xdotool", *arguments], capture_output=True, text=True, timeout=4, check=True
        )
        return result.stdout.strip()
    except (OSError, subprocess.CalledProcessError, subprocess.TimeoutExpired):
        return None
 
 
@contextmanager
def activate_unity_windows_linux(project_name: str) -> Iterator[bool]:
    """Temporarily focus this project's Unity Editor on X11.
 
    Args:
        project_name: Current Unity project directory name.
 
    Yields:
        Whether the matching Unity Editor window was successfully activated.
    """
    if not os.environ.get("DISPLAY") or not shutil.which("xdotool"):
        yield False
        return
 
    candidates = run_xdotool("search", "--onlyvisible", "--class", "Unity")
    original = run_xdotool("getactivewindow")
    if not candidates or not original:
        yield False
        return
 
    editors = []
    seen_processes = set()
    for handle in candidates.splitlines():
        check_escape()
        process_id = run_xdotool("getwindowpid", handle)
        if not process_id or process_id in seen_processes:
            continue
        try:
            executable = (Path("/proc") / process_id / "exe").resolve(strict=True)
        except (OSError, RuntimeError):
            continue
        if executable.name.casefold() != "unity":
            continue
        seen_processes.add(process_id)
        title = run_xdotool("getwindowname", handle) or ""
        editors.append((handle, process_id, title.casefold()))
        if len(seen_processes) >= UNITY_FOCUS_MAX_WINDOWS:
            break
 
    matches = [editor for editor in editors if project_name.casefold() in editor[2]]
    if len(matches) == 1:
        selected = matches[0]
    elif len(editors) == 1:
        selected = editors[0]
    else:
        selected = None
    if selected is None:
        yield False
        return
 
    activated = run_xdotool("windowactivate", "--sync", selected[0]) is not None
    try:
        if activated:
            time.sleep(UNITY_FOCUS_DELAY_SECONDS)
        yield activated
    finally:
        if activated and run_xdotool("getactivewindow") == selected[0]:
            run_xdotool("windowactivate", original)
 
 
class EditorBridge:
    """Exchange commands and results with the active Unity Editor."""
 
    def __init__(self, directory: Path, project_name: str) -> None:
        """Initialize communication paths.
 
        Args:
            directory: Shared project-local state directory.
            project_name: Name of the Unity project to activate.
        """
        self.directory = directory
        self.project_name = project_name
        self.request_path = directory / REQUEST_FILENAME
        self.result_path = directory / RESULT_FILENAME
 
    def execute(self, action: str, **arguments: str) -> bool:
        """Send an editor action and retry only a confirmed transient readiness rejection.
 
        Args:
            action: Operation identifier recognized by the editor.
            **arguments: Operation-specific command fields.
 
        Returns:
            Whether the operation completed, including editor readiness after a Lite reload.
 
        Raises:
            UpdateError: The operation fails or cannot be acknowledged.
            UserCancelled: The operation is cancelled in Unity or the console.
        """
        deadline = time.monotonic() + EDITOR_BUSY_RETRY_SECONDS
        waiting_printed = False
        while True:
            try:
                return self._execute_once(action, **arguments)
            except EditorOperationError as exc:
                if str(exc) != EDITOR_NOT_READY_ERROR or time.monotonic() >= deadline:
                    raise
                if not waiting_printed:
                    print("Unity is refreshing assets. Waiting to retry the command...", flush=True)
                    waiting_printed = True
                check_escape()
                time.sleep(POLL_INTERVAL_SECONDS)
 
    def _execute_once(self, action: str, **arguments: str) -> bool:
        """Send a uniquely identified command and await the editor response.
 
        Args:
            action: Operation identifier recognized by the editor.
            **arguments: Operation-specific command fields.
 
        Returns:
            Whether the operation completed, including editor readiness after a Lite reload.
 
        Raises:
            UpdateError: The operation fails or cannot be acknowledged.
            UserCancelled: The operation is cancelled in Unity or the console.
        """
        command_id = uuid.uuid4().hex
        atomic_write_json(self.request_path, {
            "ID": command_id,
            "Action": action,
            "Cancelled": False,
            **arguments,
        })
 
        try:
            self._wait_for_editor(command_id, action)
            return True
        except EditorOperationInterrupted:
            print("\nThe Unity Editor reloaded during the import.")
            package = CORE_PACKAGE_PATTERN.fullmatch(Path(arguments.get("Path", "")).name)
            if action == "import" and package is not None and package.group(1).casefold() == "lite":
                self.execute("ping")
                return True
 
            ask_to_continue(
                "Finish or cancel the import in Unity, then press Enter ONLY "
                "if it completed successfully; Q cancels automatic restoration: "
            )
            return True
        except (UserCancelled, KeyboardInterrupt, UpdateError):
            self._cancel_pending_request(command_id)
            raise
 
    def _cancel_pending_request(self, command_id: str) -> None:
        """Mark an unclaimed request as cancelled so Unity cannot run it later.
 
        Args:
            command_id: Identifier of the request to cancel.
        """
        request = read_json(self.request_path)
        if request is None or request.get("ID") != command_id:
            return
 
        result = read_json(self.result_path)
        if result is not None and result.get("ID") == command_id and result.get("State") in {
            "processing",
            "completed",
            "failed",
            "cancelled",
            "interrupted",
        }:
            return
 
        request["Cancelled"] = True
        atomic_write_json(self.request_path, request)
 
    def _confirm_manual_import_completion(self, command_id: str) -> None:
        """Signal Unity that the user dismissed the empty import dialog.
 
        Args:
            command_id: Identifier of the active import request.
 
        Raises:
            UpdateError: The active import request can no longer be confirmed.
        """
        request = read_json(self.request_path)
        if request is None or request.get("ID") != command_id or request.get("Action") != "import":
            raise UpdateError("The active import request could not be confirmed manually.")
 
        request[MANUAL_IMPORT_COMPLETION_FIELD] = True
        atomic_write_json(self.request_path, request)
 
    def _wait_for_editor(self, command_id: str, action: str) -> None:
        """Wait for Unity to acknowledge and finish the request.
 
        Args:
            command_id: Identifier of the pending request.
            action: Operation requested from Unity.
 
        Raises:
            UpdateError: Unity fails or does not acknowledge the request.
            UserCancelled: The user cancels while waiting.
        """
        ping_response_received = False
        if action == "ping":
            print(
                "If Unity does not respond, select the Unity Editor window and wait for it to process "
                "the request.\nThe updater will continue automatically once Unity responds.",
                flush=True,
            )
            grace_end = time.monotonic() + UNITY_BACKGROUND_RESPONSE_GRACE_SECONDS
            while time.monotonic() < grace_end:
                check_escape()
                initial_result = read_json(self.result_path)
                if initial_result is not None and initial_result.get("ID") == command_id:
                    if initial_result.get("State") == "completed":
                        print("Unity Editor is ready.", flush=True)
                        return
                    ping_response_received = True
                    break
                time.sleep(min(0.1, max(0.0, grace_end - time.monotonic())))
 
        started = time.monotonic()
        acknowledged = False
        manual_confirmation_requested = False
        manual_confirmation_sent = False
        warning_printed = False
        last_status = started
 
        focus_needed = action == "import" or (action == "ping" and not ping_response_received)
        focus = focus_unity_editor(self.project_name) if focus_needed else nullcontext(False)
        with focus as focused:
            if focus_needed and not focused:
                if action == "ping":
                    print(
                        "The matching Unity Editor window could not be activated automatically. "
                        "Select it manually.",
                        flush=True,
                    )
                elif action == "import":
                    print(
                        "The Unity Editor window could not be activated automatically. "
                        "Select it manually to handle the package import.",
                        flush=True,
                    )
            if action == "import":
                print(
                    "If Unity shows 'Nothing to import', close that dialog and press Enter here to continue.",
                    flush=True,
                )
            while True:
                if action == "import":
                    enter_pressed = read_console_enter_pressed()
                else:
                    check_escape()
                    enter_pressed = False
 
                result = read_json(self.result_path)
                if result is not None and result.get("ID") == command_id:
                    state = result.get("State", "")
                    detail = result.get("Message", "")
                    if state == "processing":
                        acknowledged = True
                    elif state == "completed":
                        if action == "ping":
                            print("Unity Editor is ready.", flush=True)
                        return
                    elif state == "cancelled":
                        raise EditorOperationCancelled(f"The editor cancelled the operation. {detail}".strip())
                    elif state == "failed":
                        raise EditorOperationError(f"The editor reported an error: {detail}")
                    elif state == "interrupted":
                        if action != "import":
                            raise UpdateError("The editor reloaded during an asset operation. Review the project.")
                        raise EditorOperationInterrupted()
 
                if enter_pressed and acknowledged:
                    manual_confirmation_requested = True
 
                if manual_confirmation_requested and acknowledged and not manual_confirmation_sent:
                    self._confirm_manual_import_completion(command_id)
                    manual_confirmation_requested = False
                    manual_confirmation_sent = True
                    print("Manual import completion sent to Unity. Waiting for confirmation...", flush=True)
 
                now = time.monotonic()
                elapsed = now - started
                if action == "ping" and now - last_status >= EDITOR_STATUS_INTERVAL_SECONDS:
                    print(f"Still waiting for Unity Editor... {int(elapsed)}s / {EDITOR_RESPONSE_TIMEOUT_SECONDS}s", flush=True)
                    last_status = now
                if not acknowledged and elapsed > EDITOR_RESPONSE_TIMEOUT_SECONDS:
                    raise UpdateError(
                        "The Unity Editor did not respond to the request. "
                        "Check that the project is open, scripts have compiled, "
                        "and the editor bridge has compiled successfully."
                    )
                if acknowledged and elapsed > EDITOR_WAIT_WARNING_SECONDS and not warning_printed:
                    print("\nWaiting for Unity to finish the operation. Press Ctrl+C to stop waiting.")
                    warning_printed = True
                time.sleep(POLL_INTERVAL_SECONDS)
 
 
def verify_recovery(bridge: EditorBridge, local_asset: Path, root_asset: Path) -> None:
    """Handle an unfinished relocation from an earlier process.
 
    Args:
        bridge: Editor communication client.
        local_asset: Original asset directory.
        root_asset: Temporary asset directory.
 
    Raises:
        UpdateError: Recovery paths conflict or restoration fails.
        UserCancelled: Recovery is declined.
    """
    recovery_path = bridge.directory / RECOVERY_FILENAME
    journal = read_json(recovery_path)
    if journal is None:
        if path_exists(recovery_path):
            raise UpdateError(f"The recovery journal is unreadable: {recovery_path}")
        return
 
    if journal.get("source") != str(local_asset) or journal.get("destination") != str(root_asset):
        raise UpdateError(f"The recovery journal references unexpected paths: {recovery_path}")
 
    if local_asset.is_dir() and not path_exists(root_asset) and not path_exists(metadata_path(root_asset)):
        request = read_json(bridge.request_path)
        if request is not None and request.get("Action") == "move":
            result = read_json(bridge.result_path)
            if result is None or result.get("ID") != request.get("ID") or result.get("State") == "processing":
                raise UpdateError(
                    "A previous asset move has not been confirmed by the editor. "
                    "Wait for Unity to finish processing before retrying."
                )
        recovery_path.unlink()
        return
 
    if (
        not root_asset.is_dir()
        or root_asset.is_symlink()
        or not metadata_path(root_asset).is_file()
        or metadata_path(root_asset).is_symlink()
        or path_exists(local_asset)
        or path_exists(metadata_path(local_asset))
    ):
        raise UpdateError(
            "A previous relocation was interrupted and the paths are not safe to restore. "
            f"Review both locations manually:\n  {local_asset}\n  {root_asset}"
        )
 
    print(f"\nAn unfinished asset relocation was found: {root_asset}")
    ask_to_continue(
        "Ensure ALL earlier imports have finished or been cancelled in Unity. "
        "Press Enter to restore before updating; Q leaves it untouched: "
    )
    bridge.execute("move", Source=unity_asset_path(root_asset), Destination=unity_asset_path(local_asset))
    recovery_path.unlink()
    print("Previous asset relocation was restored.")
 
 
def unity_asset_path(path: Path) -> str:
    """Convert a filesystem path below the assets root into a Unity asset path.
 
    Args:
        path: Directory inside the current project's assets folder.
 
    Returns:
        Project-relative asset path using forward slashes.
    """
    assets = find_assets_directory(path)
    return f"{ASSETS_DIRECTORY}/{path.relative_to(assets).as_posix()}"
 
 
def run_update(bridge: EditorBridge, script_directory: Path, assets_directory: Path) -> None:
    """Run the update sequence while preserving movable project assets.
 
    Args:
        bridge: Editor integration client.
        script_directory: Directory containing the updater.
        assets_directory: Project assets root.
 
    Raises:
        UpdateError: A prerequisite fails or the editor reports an error.
        UserCancelled: The user or editor cancels the operation.
    """
    print("Searching for downloaded SDK packages...", flush=True)
    downloads_directory = get_downloads_directory()
    version, packages, ignored_versions = discover_packages(downloads_directory)
    local_asset = script_directory / PRESERVED_DIRECTORY
    root_asset = assets_directory / PRESERVED_DIRECTORY
 
    print("Waiting for Unity Editor to respond...", flush=True)
    bridge.execute("ping")
    print("Checking previous update state...", flush=True)
    verify_recovery(bridge, local_asset, root_asset)
 
    if path_exists(local_asset) and (not local_asset.is_dir() or local_asset.is_symlink()):
        raise UpdateError(f"The preserved asset path is not a regular directory: {local_asset}")
    if path_exists(root_asset) and (not root_asset.is_dir() or root_asset.is_symlink()):
        raise UpdateError(f"The root asset path is not a regular directory: {root_asset}")
 
    should_move = local_asset.is_dir()
    if should_move:
        if path_exists(root_asset) or path_exists(metadata_path(root_asset)):
            raise UpdateError(
                "Both preserved directory locations are occupied. "
                "No files were modified."
            )
        if not metadata_path(local_asset).is_file() or metadata_path(local_asset).is_symlink():
            raise UpdateError(f"The metadata file is missing or invalid: {metadata_path(local_asset)}")
 
    print(f"\nProject: {assets_directory.parent}")
    print(f"Downloads directory: {downloads_directory}")
    print(f"Selected version: {version}")
    if ignored_versions:
        print(f"Other core versions skipped: {', '.join(ignored_versions)}")
    print("\nImport order:")
    for index, package in enumerate(packages, start=1):
        print(f"  {index}. {package.name}")
 
    if should_move:
        print(f"\n{PRESERVED_DIRECTORY}: temporarily move to {ASSETS_DIRECTORY} using the editor.")
    elif root_asset.is_dir():
        print(f"\n{PRESERVED_DIRECTORY}: already in {ASSETS_DIRECTORY}; no relocation needed.")
    else:
        print(f"\n{PRESERVED_DIRECTORY}: absent; no relocation needed.")
 
    ask_to_continue("\nPress Enter to start, or Q to cancel: ")
 
    recovery_path = bridge.directory / RECOVERY_FILENAME
    moved = False
    restore_allowed = True
    try:
        if should_move:
            atomic_write_json(recovery_path, {
                "source": str(local_asset),
                "destination": str(root_asset),
            })
            bridge.execute("move", Source=unity_asset_path(local_asset), Destination=unity_asset_path(root_asset))
            moved = True
            print("\nPreserved directory temporarily moved by Unity Editor.")
 
        for index, package in enumerate(packages, start=1):
            print(f"\n[{index}/{len(packages)}] Importing: {package.name}")
            restore_allowed = False
            try:
                bridge.execute("import", Path=str(package.resolve()))
            except (EditorOperationError, EditorOperationCancelled):
                restore_allowed = True
                raise
            else:
                restore_allowed = True
                print(f"Completed: {package.name}")
    finally:
        if path_exists(recovery_path):
            if moved and restore_allowed:
                print("\nRestoring preserved directory using Unity Editor...")
                try:
                    bridge.execute(
                        "move",
                        Source=unity_asset_path(root_asset),
                        Destination=unity_asset_path(local_asset),
                    )
                except (UpdateError, UserCancelled, KeyboardInterrupt) as exc:
                    print(f"WARNING: Automatic asset restoration failed: {exc}")
                    print(f"Recovery information: {recovery_path}")
                    raise
                else:
                    recovery_path.unlink()
                    print("Preserved directory restored successfully.")
            else:
                print("\nWARNING: The asset location could not be confirmed automatically.")
                print(f"Check both locations: {local_asset} and {root_asset}")
                print(f"Recovery information: {recovery_path}")
 
    print("\nAll selected packages have been imported successfully.")
 
 
def claim_controller_startup_attempt() -> None:
    """Atomically claim a pending launch token, or stop after launcher cancellation."""
    attempt_path_value = os.environ.pop(STARTUP_ATTEMPT_ENVIRONMENT_VARIABLE, None)
    if not attempt_path_value:
        return
 
    attempt_path = Path(attempt_path_value)
    started_path = attempt_path.with_suffix(".started")
    try:
        os.rename(attempt_path, started_path)
    except OSError as exc:
        if not attempt_path.exists():
            raise UpdateError("The updater startup attempt was cancelled before the controller started.") from exc
        raise UpdateError(f"Could not claim the updater startup attempt: {exc}") from exc
 
 
@contextmanager
def active_update_session(directory: Path) -> Iterator[None]:
    """Publish the controller process while SDK imports and restoration are active.
 
    Args:
        directory: Shared editor bridge state directory.
 
    Yields:
        Control while editor-side SDK patching is suspended.
    """
    marker = directory / ACTIVE_SESSION_FILENAME
    atomic_write_json(marker, {"ProcessId": os.getpid()})
    try:
        yield
    finally:
        try:
            marker.unlink()
        except FileNotFoundError:
            pass
 
 
def main() -> int:
    """Validate the installation and coordinate an import session.
 
    Returns:
        Zero after a successful update or explicit cancellation.
 
    Raises:
        UpdateError: Validation, editor communication, or recovery fails.
        UserCancelled: An explicit cancellation occurs.
    """
    claim_controller_startup_attempt()
 
    script_directory = Path(__file__).resolve().parent
    assets_directory = find_assets_directory(script_directory)
    validate_script_directory(script_directory)
 
    project_directory = assets_directory.parent
    if not (project_directory / "ProjectSettings").is_dir():
        raise UpdateError(f"The Unity project settings directory is missing: {project_directory}")
 
    bridge_directory = project_directory / "Library" / BRIDGE_DIRECTORY
    bridge_directory.mkdir(parents=True, exist_ok=True)
    with exclusive_session(bridge_directory / LOCK_FILENAME), active_update_session(bridge_directory):
        run_update(EditorBridge(bridge_directory, project_directory.name), script_directory, assets_directory)
    return 0
 
 
if __name__ == "__main__":
    print("Yandex Mobile Ads SDK Updater", flush=True)
    if sys.platform == "win32":
        print("Press Esc in the updater window to cancel or close it.\n", flush=True)
    else:
        print("Press Ctrl+C in the updater window to cancel.\n", flush=True)
    exit_code = 1
    escape_requested = False
    try:
        exit_code = main()
    except EscapeCancelled as exc:
        print(f"\n{exc}")
        exit_code = 0
        escape_requested = True
    except UserCancelled as exc:
        print(f"\n{exc}")
        exit_code = 0
    except (UpdateError, OSError, EOFError, ValueError) as exc:
        print(f"\nERROR: {exc}")
    except KeyboardInterrupt:
        print("\nOperation interrupted. Check the recovery information above before retrying.")
    finally:
        if not escape_requested:
            try:
                read_console_input("\nPress Enter to close...")
            except (EOFError, KeyboardInterrupt, EscapeCancelled):
                pass
 
    sys.exit(exit_code)