fix(filesystem): select files via FileManager1 DBus on Linux

open-file-location now asks the freedesktop org.freedesktop.FileManager1
ShowItems interface to reveal the file selected (GNOME Files, Dolphin,
Nemo, Caja, recent Thunar), falling back to plain xdg-open when the
interface is unavailable. Also align the Linux branch with the other
location handlers: headless sessions return clipboard mode instead of
silently failing.
This commit is contained in:
Will Miao
2026-10-10 07:15:21 +08:00
parent 406096c0e1
commit 7937aeaa04
2 changed files with 147 additions and 2 deletions
+46 -2
View File
@@ -23,6 +23,7 @@ import tempfile
import zipfile
from dataclasses import dataclass
from datetime import datetime, timezone
from urllib.parse import quote
from typing import Any, Awaitable, Callable, Dict, Mapping, Protocol, Sequence
from aiohttp import web
@@ -439,6 +440,35 @@ def _has_gui_display() -> bool:
return bool(os.environ.get("DISPLAY") or os.environ.get("WAYLAND_DISPLAY"))
def _select_in_linux_file_manager(file_path: str) -> bool:
"""Reveal ``file_path`` selected in the desktop file manager.
Uses the freedesktop ``org.freedesktop.FileManager1.ShowItems`` DBus
interface, which is supported by GNOME Files (Nautilus), KDE Dolphin,
Nemo, Caja, and recent Thunar. Returns ``False`` when the interface is
unavailable so the caller can fall back to plain ``xdg-open``.
"""
uri = "file://" + quote(file_path)
try:
result = subprocess.run(
[
"dbus-send",
"--session",
"--print-reply",
"--dest=org.freedesktop.FileManager1",
"/org/freedesktop/FileManager1",
"org.freedesktop.FileManager1.ShowItems",
f"array:string:{uri}",
"string:",
],
capture_output=True,
timeout=5,
)
except (FileNotFoundError, subprocess.SubprocessError, OSError):
return False
return result.returncode == 0
class PromptServerProtocol(Protocol):
"""Subset of PromptServer used by the handlers."""
@@ -3512,8 +3542,22 @@ class FileSystemHandler:
elif sys.platform == "darwin":
subprocess.Popen(["open", "-R", file_path])
else:
folder = os.path.dirname(file_path)
subprocess.Popen(["xdg-open", folder])
if not _has_gui_display():
# Headless/SSH session: no file manager to open, so
# hand the path to the browser for copying instead
# of reporting a success that never happened.
return web.json_response(
{
"success": True,
"message": "Headless session: path available for copying",
"path": file_path,
"mode": "clipboard",
}
)
selected = _select_in_linux_file_manager(file_path)
if not selected:
folder = os.path.dirname(file_path)
subprocess.Popen(["xdg-open", folder])
return web.json_response(
{
+101
View File
@@ -810,6 +810,107 @@ async def test_open_wildcards_location_creates_and_opens_directory(tmp_path, mon
assert calls == [["xdg-open", str(wildcards_dir)]]
@pytest.mark.asyncio
async def test_open_file_location_linux_selects_via_dbus(tmp_path, monkeypatch):
"""On a Linux desktop the handler should ask the freedesktop FileManager1
DBus service to open the folder with the file selected."""
model_file = tmp_path / "model.safetensors"
model_file.write_text("x", encoding="utf-8")
handler = FileSystemHandler(settings_service=SimpleNamespace())
monkeypatch.setenv("DISPLAY", ":0")
monkeypatch.setattr("py.routes.handlers.misc_handlers._is_docker", lambda: False)
monkeypatch.setattr("py.routes.handlers.misc_handlers._is_wsl", lambda: False)
run_calls = []
popen_calls = []
def fake_run(args, **kwargs):
run_calls.append(args)
return SimpleNamespace(returncode=0)
monkeypatch.setattr(subprocess, "run", fake_run)
monkeypatch.setattr(subprocess, "Popen", lambda *args, **kwargs: popen_calls.append(args))
request = FakeRequest(json_data={"file_path": str(model_file)})
response = await handler.open_file_location(request) # pyright: ignore[reportArgumentType]
payload = _json_payload(response)
assert response.status == 200
assert payload["success"] is True
assert len(run_calls) == 1
dbus_cmd = run_calls[0]
assert dbus_cmd[0] == "dbus-send"
assert any(
arg.startswith("array:string:file://") and "model.safetensors" in arg
for arg in dbus_cmd
)
assert popen_calls == []
@pytest.mark.asyncio
async def test_open_file_location_linux_falls_back_to_xdg_open(tmp_path, monkeypatch):
"""When the FileManager1 DBus interface is unavailable, fall back to
opening the containing folder with xdg-open."""
model_file = tmp_path / "model.safetensors"
model_file.write_text("x", encoding="utf-8")
handler = FileSystemHandler(settings_service=SimpleNamespace())
monkeypatch.setenv("DISPLAY", ":0")
monkeypatch.setattr("py.routes.handlers.misc_handlers._is_docker", lambda: False)
monkeypatch.setattr("py.routes.handlers.misc_handlers._is_wsl", lambda: False)
popen_calls = []
monkeypatch.setattr(
subprocess, "run", lambda *args, **kwargs: SimpleNamespace(returncode=1)
)
monkeypatch.setattr(
subprocess, "Popen", lambda *args, **kwargs: popen_calls.append(args[0])
)
request = FakeRequest(json_data={"file_path": str(model_file)})
response = await handler.open_file_location(request) # pyright: ignore[reportArgumentType]
payload = _json_payload(response)
assert response.status == 200
assert payload["success"] is True
assert popen_calls == [["xdg-open", str(tmp_path)]]
@pytest.mark.asyncio
async def test_open_file_location_headless_returns_clipboard_mode(tmp_path, monkeypatch):
"""Without a GUI session there is no file manager to open; the handler
must hand the path to the browser instead of failing silently."""
model_file = tmp_path / "model.safetensors"
model_file.write_text("x", encoding="utf-8")
handler = FileSystemHandler(settings_service=SimpleNamespace())
monkeypatch.delenv("DISPLAY", raising=False)
monkeypatch.delenv("WAYLAND_DISPLAY", raising=False)
monkeypatch.setattr("py.routes.handlers.misc_handlers._is_docker", lambda: False)
monkeypatch.setattr("py.routes.handlers.misc_handlers._is_wsl", lambda: False)
popen_calls = []
run_calls = []
monkeypatch.setattr(subprocess, "Popen", lambda *args, **kwargs: popen_calls.append(args))
monkeypatch.setattr(subprocess, "run", lambda *args, **kwargs: run_calls.append(args))
request = FakeRequest(json_data={"file_path": str(model_file)})
response = await handler.open_file_location(request) # pyright: ignore[reportArgumentType]
payload = _json_payload(response)
assert response.status == 200
assert payload["success"] is True
assert payload["mode"] == "clipboard"
assert payload["path"] == str(model_file)
assert popen_calls == []
assert run_calls == []
class RecordingRouter:
def __init__(self):
self.calls = []