SaraFem/PortalService.py

#!/usr/bin/env python3
"""org.freedesktop.impl.portal.FileChooser backend, so SaraFem shows up as
the actual system file picker (e.g. a browser's "choose an image to upload"
dialog), not just an app you open manually.

Registered via Packaging/SaraFem.portal + Packaging/org.freedesktop.impl.portal.desktop.SaraFem.service
and preferred for FileChooser specifically via
~/.config/xdg-desktop-portal/cosmic-portals.conf — only that one interface is
overridden, so a bug here can't affect any other portal (Screenshot,
Settings, etc.), and the override is one file to delete to revert.

A sibling module (not under Packaging/) so PyInstaller's static import
analysis bundles it — and its own dbus_next import — automatically, the same
way it already bundles Core.py/Browser.py/etc, with no special build flags.
Runnable two ways: `python Main.py --portal-service` (dev venv or a packaged
build) invokes Run() directly; Packaging/org.freedesktop.impl.portal.desktop.SaraFem.service's
Exec= does exactly that via Packaging/Launch.sh.

Each request here spawns the picker UI (Picker.py, via Main.py --pick-*) and
waits for it to exit, reading back the JSON result file it writes — the
D-Bus plumbing lives entirely in this file, not in the GUI.
"""

import asyncio
import json
import sys
import tempfile
import urllib.parse
from pathlib import Path
from typing import Optional

from dbus_next import BusType, Variant
from dbus_next.aio import MessageBus
from dbus_next.service import ServiceInterface, method

BusName = "org.freedesktop.impl.portal.desktop.SaraFem"
ObjectPath = "/org/freedesktop/portal/desktop"
RepoRoot = Path(__file__).resolve().parent
Launcher = RepoRoot / "Packaging" / "Launch.sh"


def _PickerCommand() -> list[str]:
    """A packaged (PyInstaller-frozen) install has no venv and no separate
    Launch.sh — just this one executable, which re-invokes itself with
    --pick-* flags for the picker UI (see Main.py's --portal-service
    branch). Running from source still goes through Launch.sh."""
    if getattr(sys, "frozen", False):
        return [sys.executable]
    return [str(Launcher)]


def _PathToUri(EntryPath: Path) -> str:
    return "file://" + urllib.parse.quote(str(EntryPath))


def _DecodeNulPath(RawBytes: bytes) -> Optional[Path]:
    Text = bytes(RawBytes).rstrip(b"\x00").decode(errors="replace")
    return Path(Text) if Text else None


class _RequestObject(ServiceInterface):
    """Minimal org.freedesktop.impl.portal.Request: lets the calling app
    (via the portal frontend) cancel a pending pick — e.g. it closed its own
    upload dialog while SaraFem's picker was still open — by killing our
    picker subprocess instead of leaving it orphaned."""

    def __init__(self, Bus: MessageBus, HandlePath: str):
        super().__init__("org.freedesktop.impl.portal.Request")
        self.Bus = Bus
        self.HandlePath = HandlePath
        self.Process: Optional[asyncio.subprocess.Process] = None
        self.Cancelled = asyncio.Event()
        Bus.export(HandlePath, self)

    @method()
    def Close(self) -> None:
        self.Cancelled.set()
        if self.Process is not None and self.Process.returncode is None:
            self.Process.terminate()

    def Unexport(self) -> None:
        try:
            self.Bus.unexport(self.HandlePath, self)
        except Exception:
            pass


class FileChooserPortal(ServiceInterface):
    def __init__(self, Bus: MessageBus):
        super().__init__("org.freedesktop.impl.portal.FileChooser")
        self.Bus = Bus

    async def _RunPicker(
        self, HandlePath: str, Title: str, ExtraArgs: list[str], StartPath: Optional[Path]
    ) -> tuple[int, dict]:
        Request = _RequestObject(self.Bus, HandlePath)
        ResultFile = tempfile.NamedTemporaryFile(prefix="sarafem_pick_", suffix=".json", delete=False)
        ResultPath = Path(ResultFile.name)
        ResultFile.close()

        Args = [*_PickerCommand(), *ExtraArgs, "--title", Title, "--result-file", str(ResultPath)]
        if StartPath is not None:
            Args.append(str(StartPath))

        try:
            Process = await asyncio.create_subprocess_exec(*Args)
            Request.Process = Process
            await Process.wait()
        finally:
            Request.Unexport()

        if Request.Cancelled.is_set():
            ResultPath.unlink(missing_ok=True)
            return 2, {}

        try:
            Data = json.loads(ResultPath.read_text())
        except (OSError, ValueError):
            ResultPath.unlink(missing_ok=True)
            return 2, {}
        ResultPath.unlink(missing_ok=True)

        if Data.get("cancelled"):
            return 1, {}

        Uris = [_PathToUri(Path(P)) for P in Data.get("paths", [])]
        if not Uris:
            return 1, {}
        return 0, {"uris": Variant("as", Uris)}

    def _FilterArgs(self, Options: dict) -> list[str]:
        FiltersVariant = Options.get("filters")
        if FiltersVariant is None:
            return []
        Args = []
        for Label, Patterns in FiltersVariant.value:
            # Pattern type 0 = glob, 1 = mimetype (see the FileChooser portal
            # spec) — only globs translate directly to Picker.py's filters;
            # mimetype-only filter groups are skipped (files stay unfiltered
            # for that group rather than guessing a glob from a MIME type).
            Globs = [Pattern for (Kind, Pattern) in Patterns if Kind == 0]
            if Globs:
                Args += ["--filter", f"{Label}:{','.join(Globs)}"]
        return Args

    @method()
    async def OpenFile(
        self, handle: "o", app_id: "s", parent_window: "s", title: "s", options: "a{sv}"
    ) -> "ua{sv}":
        Multiple = options.get("multiple", Variant("b", False)).value
        Directory = options.get("directory", Variant("b", False)).value
        CurrentFolder = options.get("current_folder")
        StartPath = _DecodeNulPath(CurrentFolder.value) if CurrentFolder is not None else None

        ExtraArgs = ["--pick-folder"] if Directory else ["--pick-open"]
        if Multiple:
            ExtraArgs.append("--multiple")
        ExtraArgs += self._FilterArgs(options)

        Response, Results = await self._RunPicker(handle, title or "Open File", ExtraArgs, StartPath)
        return [Response, Results]

    @method()
    async def SaveFile(
        self, handle: "o", app_id: "s", parent_window: "s", title: "s", options: "a{sv}"
    ) -> "ua{sv}":
        CurrentName = options.get("current_name")
        SuggestedName = CurrentName.value if CurrentName is not None else ""
        CurrentFolder = options.get("current_folder")
        StartPath = _DecodeNulPath(CurrentFolder.value) if CurrentFolder is not None else None

        ExtraArgs = ["--pick-save"]
        if SuggestedName:
            ExtraArgs += ["--suggested-name", SuggestedName]

        Response, Results = await self._RunPicker(handle, title or "Save File", ExtraArgs, StartPath)
        return [Response, Results]

    @method()
    async def SaveFiles(
        self, handle: "o", app_id: "s", parent_window: "s", title: "s", options: "a{sv}"
    ) -> "ua{sv}":
        # SaraFem's picker only does a single Save destination — still a
        # reasonable fallback for a rarely-used multi-file-save call.
        return await self.SaveFile(handle, app_id, parent_window, title, options)


async def Run() -> None:
    Bus = await MessageBus(bus_type=BusType.SESSION).connect()
    Bus.export(ObjectPath, FileChooserPortal(Bus))
    await Bus.request_name(BusName)
    await Bus.wait_for_disconnect()


if __name__ == "__main__":
    try:
        asyncio.run(Run())
    except KeyboardInterrupt:
        sys.exit(0)