feat: tfiles — native GTK4 file manager

Trash via Gio.File.trash(), confirmed permanent delete, zip-slip-guarded
extraction, two-phase batch rename, freedesktop thumbnail cache, async
volume/clipboard handling.

Known issues (from review, to fix next): directory listing runs on the
main thread; undo of a move can overwrite a file re-created at the origin.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
2026-07-02 21:22:08 +02:00
parent 369b4583ef
commit ccc2d4e15d
19 changed files with 4782 additions and 0 deletions
+10
View File
@@ -0,0 +1,10 @@
"""TANINUX Files — nativer Dateimanager im Apple-Look von TANINUX.
Schwesternapp zu TANINUX Music (`tmusic`): teilt sich dieselbe libadwaita-freie
UI-Naht (``gui/ui/native`` + ``style_apple.css``) und Akzent-/Hell-Dunkel-Logik
(``gui/style``). Aufbau analog zur Music-Shell:
places.py Sidebar-Orte (XDG-Verzeichnisse + eingehängte Laufwerke)
window.py Shell (Sidebar · Toolbar/Breadcrumb · Datei-Raster)
app.py Einstiegspunkt `tfiles`
"""
+4
View File
@@ -0,0 +1,4 @@
from taninux.files.app import main
if __name__ == "__main__":
raise SystemExit(main())
+53
View File
@@ -0,0 +1,53 @@
"""TANINUX Files — Einstiegspunkt (`tfiles`).
Nativer Dateimanager im Apple-Look von TANINUX. Lädt dasselbe Apple-Stylesheet
und die Akzent-/Hell-Dunkel-Logik wie TANINUX Music, öffnet dann die Datei-Shell.
"""
from __future__ import annotations
import logging
import sys
import gi
gi.require_version("Gtk", "4.0")
from gi.repository import Gio, Gtk # noqa: E402
from taninux.gui import APP_ID, style
from taninux.gui.ui import native
logger = logging.getLogger(__name__)
class FilesApp(Gtk.Application):
def __init__(self) -> None:
super().__init__(
application_id=f"{APP_ID}.Files",
flags=Gio.ApplicationFlags.DEFAULT_FLAGS,
)
self._window: Gtk.Window | None = None
def do_activate(self) -> None:
if self._window is not None:
self._window.present()
return
native.ensure_css()
try:
style.apply_persisted() # Akzentfarbe + Hell/Dunkel wie der Rest von TANINUX
except Exception: # noqa: BLE001
logger.debug("style.apply_persisted() übersprungen", exc_info=True)
from taninux.files.window import FilesWindow
self._window = FilesWindow(self)
self._window.present()
def main(argv: list[str] | None = None) -> int:
app = FilesApp()
return app.run(argv if argv is not None else sys.argv)
if __name__ == "__main__":
raise SystemExit(main())
+292
View File
@@ -0,0 +1,292 @@
"""Archiv-Modul: Komprimieren und Entpacken von Dateien.
Selbst-enthaltenes Modul fuer den GTK4-Dateimanager. Verwendet ausschliesslich
die Python-Standardbibliothek (zipfile, tarfile, threading) sowie GLib, um
Callbacks an die Hauptschleife zu uebergeben. Die eigentliche Arbeit laeuft in
einem Hintergrund-Thread, damit die Oberflaeche nicht blockiert.
"""
from __future__ import annotations
import os
import tarfile
import threading
import zipfile
from pathlib import Path
from typing import Callable, Iterable, List, Tuple
import gi
gi.require_version("Gtk", "4.0")
from gi.repository import GLib # noqa: E402
# Supported create formats, in menu order: (id, human_label)
FORMATS: List[Tuple[str, str]] = [
("zip", "Zip"),
("tar.gz", "Tar.gz"),
("tar.xz", "Tar.xz"),
]
# Map create-format id -> (file extension, tarfile mode or None for zip)
_CREATE_EXT = {
"zip": ".zip",
"tar.gz": ".tar.gz",
"tar.xz": ".tar.xz",
}
_TAR_MODE = {
"tar.gz": "w:gz",
"tar.xz": "w:xz",
}
# Known double / compound extensions (longest first matters).
_ARCHIVE_EXTS = (
".tar.gz",
".tar.bz2",
".tar.xz",
".tgz",
".tbz2",
".txz",
".tar",
".zip",
)
_ARCHIVE_CONTENT_TYPES = {
"application/zip",
"application/x-zip",
"application/x-zip-compressed",
"application/x-tar",
"application/gzip",
"application/x-gzip",
"application/x-xz",
"application/x-bzip2",
"application/x-bzip",
"application/x-compressed-tar",
"application/x-gtar",
"application/x-bzip-compressed-tar",
"application/x-xz-compressed-tar",
"application/x-lzma-compressed-tar",
}
def is_archive(path: Path, content_type: str = "") -> bool:
"""True if `path` looks like a supported, extractable archive.
Cheap check, no IO: matches by file extension or by content_type.
"""
name = path.name.lower()
for ext in _ARCHIVE_EXTS:
if name.endswith(ext):
return True
if content_type and content_type.lower() in _ARCHIVE_CONTENT_TYPES:
return True
return False
def _strip_archive_ext(name: str) -> str:
"""Return the archive stem, stripping known (double) extensions."""
lower = name.lower()
for ext in _ARCHIVE_EXTS:
if lower.endswith(ext):
return name[: -len(ext)]
# Fall back to a single-suffix strip.
return Path(name).stem
def _collision_free(base_dir: Path, stem: str, ext: str) -> Path:
"""Return a path in `base_dir` that does not yet exist.
`ext` is "" for directories or a full extension like ".tar.gz" for files.
Appends " 2", " 3", ... before the extension on collision.
"""
candidate = base_dir / f"{stem}{ext}"
if not candidate.exists():
return candidate
n = 2
while True:
candidate = base_dir / f"{stem} {n}{ext}"
if not candidate.exists():
return candidate
n += 1
def _run_in_thread(target) -> None:
threading.Thread(target=target, daemon=True).start()
def _finish(on_done: Callable[["Path | None", str], None], result, message: str) -> None:
"""Marshal the callback onto the GLib main loop. Runs once."""
def _cb():
on_done(result, message)
return False
GLib.idle_add(_cb)
def compress(
paths: Iterable[Path],
dest_dir: Path,
fmt: str,
on_done: Callable[["Path | None", str], None],
) -> None:
"""Create one archive in `dest_dir` containing all `paths`.
Files and directories are stored with their basenames as top-level entries.
Runs in a background thread; never raises.
"""
items = [Path(p) for p in paths]
dest_dir = Path(dest_dir)
def work():
try:
if fmt not in _CREATE_EXT:
_finish(on_done, None, f"Unbekanntes Format: {fmt}")
return
if not items:
_finish(on_done, None, "Keine Dateien angegeben.")
return
ext = _CREATE_EXT[fmt]
if len(items) == 1:
stem = items[0].name
else:
stem = "Archive"
archive_path = _collision_free(dest_dir, stem, ext)
if fmt == "zip":
_make_zip(items, archive_path)
else:
_make_tar(items, archive_path, _TAR_MODE[fmt])
_finish(on_done, archive_path, "")
except Exception as exc: # noqa: BLE001
_finish(on_done, None, str(exc))
_run_in_thread(work)
def _make_zip(items: List[Path], archive_path: Path) -> None:
with zipfile.ZipFile(archive_path, "w", zipfile.ZIP_DEFLATED) as zf:
for item in items:
arcbase = item.name
if item.is_dir():
# Ensure an entry for the directory itself, then walk it.
zf.write(item, arcbase)
for root, dirs, files in os.walk(item):
root_p = Path(root)
rel = root_p.relative_to(item)
for d in dirs:
dpath = root_p / d
arcname = os.path.join(arcbase, rel.as_posix(), d) if rel.as_posix() != "." else os.path.join(arcbase, d)
zf.write(dpath, arcname)
for f in files:
fpath = root_p / f
arcname = os.path.join(arcbase, rel.as_posix(), f) if rel.as_posix() != "." else os.path.join(arcbase, f)
zf.write(fpath, arcname)
else:
zf.write(item, arcbase)
def _make_tar(items: List[Path], archive_path: Path, mode: str) -> None:
with tarfile.open(archive_path, mode) as tf:
for item in items:
tf.add(item, arcname=item.name)
def extract(
archive: Path,
dest_dir: Path,
on_done: Callable[["Path | None", str], None],
) -> None:
"""Extract `archive` into a new collision-free subfolder of `dest_dir`.
The subfolder is named after the archive's stem. Members whose resolved path
escapes the target dir (zip-slip / path traversal), absolute members, and
symlinks pointing outside are skipped. Runs in a background thread; never
raises.
"""
archive = Path(archive)
dest_dir = Path(dest_dir)
def work():
try:
stem = _strip_archive_ext(archive.name)
if not stem:
stem = archive.name
extract_dir = _collision_free(dest_dir, stem, "")
extract_dir.mkdir(parents=True, exist_ok=False)
if zipfile.is_zipfile(archive):
_extract_zip(archive, extract_dir)
elif tarfile.is_tarfile(archive):
_extract_tar(archive, extract_dir)
else:
# Last resort: let tarfile try to open it.
try:
_extract_tar(archive, extract_dir)
except Exception:
_finish(on_done, None, "Unbekanntes oder beschaedigtes Archiv.")
return
_finish(on_done, extract_dir, "")
except Exception as exc: # noqa: BLE001
_finish(on_done, None, str(exc))
_run_in_thread(work)
def _is_safe(extract_dir: Path, member_name: str) -> "Path | None":
"""Return a resolved, safe target path inside extract_dir, or None."""
if not member_name:
return None
if os.path.isabs(member_name) or member_name.startswith(("/", "\\")):
return None
base = extract_dir.resolve()
target = (extract_dir / member_name).resolve()
try:
if not target.is_relative_to(base):
return None
except AttributeError: # Python < 3.9 fallback
if os.path.commonpath([str(base), str(target)]) != str(base):
return None
return target
def _extract_zip(archive: Path, extract_dir: Path) -> None:
with zipfile.ZipFile(archive) as zf:
for info in zf.infolist():
target = _is_safe(extract_dir, info.filename)
if target is None:
continue
if info.is_dir():
target.mkdir(parents=True, exist_ok=True)
continue
target.parent.mkdir(parents=True, exist_ok=True)
with zf.open(info) as src, open(target, "wb") as dst:
while True:
chunk = src.read(1024 * 64)
if not chunk:
break
dst.write(chunk)
def _extract_tar(archive: Path, extract_dir: Path) -> None:
base = extract_dir.resolve()
with tarfile.open(archive) as tf:
for member in tf.getmembers():
target = _is_safe(extract_dir, member.name)
if target is None:
continue
# Reject symlinks/hardlinks pointing outside the target dir.
if member.issym() or member.islnk():
link_target = (target.parent / member.linkname).resolve()
try:
safe = link_target.is_relative_to(base)
except AttributeError:
safe = os.path.commonpath([str(base), str(link_target)]) == str(base)
if not safe:
continue
try:
tf.extract(member, extract_dir, filter="data")
except TypeError:
# Python < 3.12 has no `filter` parameter.
tf.extract(member, extract_dir)
+199
View File
@@ -0,0 +1,199 @@
"""Stapelweise Umbenennung (Batch-Rename) Engine.
Reine Logik zum Berechnen neuer Dateinamen (für eine Live-Vorschau) und zum
Anwenden der Umbenennung im Hintergrund-Thread. Der Dialog wird anderswo gebaut.
"""
from dataclasses import dataclass, field
from pathlib import Path
from typing import Callable, List, Tuple
import re
import os
import threading
import gi
gi.require_version("Gtk", "4.0")
from gi.repository import GLib
@dataclass
class RenameRule:
find: str = "" # text (or regex if use_regex) to replace in the NAME STEM
replace: str = ""
use_regex: bool = False
prefix: str = "" # prepended to the stem
suffix: str = "" # appended to the stem (before the extension)
numbering: bool = False # append a sequence number to the stem
start: int = 1
digits: int = 1 # zero-pad the number to this many digits
case: str = "keep" # "keep" | "lower" | "upper" | "title"
def _split_stem_ext(path: Path) -> Tuple[str, str]:
"""Return (stem, ext) for a path.
For a directory (or a name without a dot) treat the whole name as the stem
and ext="". For a file use the last suffix as ext.
"""
name = path.name
try:
is_dir = path.is_dir()
except OSError:
is_dir = False
if is_dir:
return name, ""
ext = path.suffix
if not ext:
return name, ""
return name[: -len(ext)], ext
def _apply_case(stem: str, case: str) -> str:
if case == "lower":
return stem.lower()
if case == "upper":
return stem.upper()
if case == "title":
return stem.title()
return stem
def _transform_stem(stem: str, rule: RenameRule, index: int) -> str:
# 1) find/replace
if rule.find:
if rule.use_regex:
try:
stem = re.sub(rule.find, rule.replace, stem)
except re.error:
pass # leave stem unchanged on regex error
else:
stem = stem.replace(rule.find, rule.replace)
# 2) case
stem = _apply_case(stem, rule.case)
# 3) prefix + stem + suffix
stem = f"{rule.prefix}{stem}{rule.suffix}"
# 4) numbering
if rule.numbering:
n = rule.start + index
digits = rule.digits if rule.digits and rule.digits > 0 else 1
stem = f"{stem}{n:0{digits}d}"
return stem
def preview(paths: List[Path], rule: RenameRule) -> List[Tuple[Path, str]]:
"""Compute the new basename for each path (see module/API docs)."""
result: List[Tuple[Path, str]] = []
used: set = set()
for index, path in enumerate(paths):
stem, ext = _split_stem_ext(path)
new_stem = _transform_stem(stem, rule, index)
new_name = f"{new_stem}{ext}"
# Ensure uniqueness within the batch (case-sensitive).
if new_name in used:
counter = 2
candidate = f"{new_stem}-{counter}{ext}"
while candidate in used:
counter += 1
candidate = f"{new_stem}-{counter}{ext}"
new_name = candidate
used.add(new_name)
result.append((path, new_name))
return result
def _do_apply(paths: List[Path], rule: RenameRule,
on_done: Callable[[List[Tuple[Path, Path]], List[Tuple[Path, str]]], None]) -> None:
renamed: List[Tuple[Path, Path]] = []
errors: List[Tuple[Path, str]] = []
try:
plan = preview(paths, rule)
except Exception as exc: # never raise
GLib.idle_add(on_done, renamed, [(p, str(exc)) for p in paths])
return
# Build (old_path, final_path) list, skipping no-ops.
targets: List[Tuple[Path, Path]] = []
final_names_by_dir: dict = {}
for old_path, new_name in plan:
final_path = old_path.parent / new_name
final_names_by_dir.setdefault(str(old_path.parent), set()).add(new_name)
if final_path == old_path:
continue
targets.append((old_path, final_path))
# Phase 1: move each source to a unique temp name in the same dir.
# This avoids intermediate collisions (e.g. swaps / shifted numbering).
temp_map: List[Tuple[Path, Path, Path]] = [] # (old, temp, final)
for old_path, final_path in targets:
parent = old_path.parent
batch_names = final_names_by_dir.get(str(parent), set())
# If the final target already exists OUTSIDE the batch, record an error.
try:
target_exists = final_path.exists()
except OSError:
target_exists = False
if target_exists and final_path.name not in batch_names:
errors.append((old_path, f"Ziel existiert bereits: {final_path.name}"))
continue
# Also guard: target exists, is in batch, but is not itself a source
# being moved away -> still a real collision.
if target_exists and final_path != old_path:
source_set = {o for o, _ in targets}
if final_path not in source_set:
errors.append((old_path, f"Ziel existiert bereits: {final_path.name}"))
continue
# Create a unique temp name in the same dir.
temp_path = None
attempt = 0
while True:
candidate = parent / f".{old_path.name}.tanin-rename-{os.getpid()}-{attempt}.tmp"
if not candidate.exists():
temp_path = candidate
break
attempt += 1
try:
os.rename(old_path, temp_path)
except OSError as exc:
errors.append((old_path, str(exc)))
continue
temp_map.append((old_path, temp_path, final_path))
# Phase 2: move temp -> final.
for old_path, temp_path, final_path in temp_map:
try:
if final_path.exists():
# Collision appeared (something outside our control); roll back this one.
os.rename(temp_path, old_path)
errors.append((old_path, f"Ziel existiert bereits: {final_path.name}"))
continue
os.rename(temp_path, final_path)
renamed.append((old_path, final_path))
except OSError as exc:
# Try to roll back to the original name.
try:
os.rename(temp_path, old_path)
except OSError:
pass
errors.append((old_path, str(exc)))
GLib.idle_add(on_done, renamed, errors)
def apply(paths: List[Path], rule: RenameRule,
on_done: Callable[[List[Tuple[Path, Path]], List[Tuple[Path, str]]], None]) -> None:
"""Apply the rename in a background thread; call on_done via GLib.idle_add."""
thread = threading.Thread(
target=_do_apply, args=(paths, rule, on_done), daemon=True
)
thread.start()
+207
View File
@@ -0,0 +1,207 @@
"""System-Zwischenablage für Dateien (Copy/Cut/Paste).
Interoperiert mit GNOME Files (Nautilus) und anderen Anwendungen über die
De-facto-Standardformate:
- ``x-special/gnome-copied-files``: erste Zeile ``copy`` oder ``cut``, danach
je eine ``file://``-URI pro Zeile (ohne abschließenden Zeilenumbruch).
- ``text/uri-list``: die URIs durch ``\\r\\n`` getrennt (für generische Apps).
"""
from pathlib import Path
from typing import Callable, List
import gi
gi.require_version("Gdk", "4.0")
gi.require_version("Gtk", "4.0")
from gi.repository import Gdk, Gio, GLib, Gtk # noqa: E402
# MIME-Typen, die wir anbieten und lesen.
_MIME_GNOME = "x-special/gnome-copied-files"
_MIME_URILIST = "text/uri-list"
def _paths_to_uris(paths: List[Path]) -> List[str]:
"""Wandelt Pfade in ``file://``-URIs um (überspringt unkonvertierbare)."""
uris: List[str] = []
for p in paths:
uri = Gio.File.new_for_path(str(p)).get_uri()
if uri:
uris.append(uri)
return uris
def _uri_to_path(uri: str) -> Path | None:
"""Wandelt eine ``file://``-URI zurück in einen Pfad (oder ``None``)."""
path = Gio.File.new_for_uri(uri).get_path()
return Path(path) if path else None
class FileClipboard:
"""Kapselt eine ``Gdk.Clipboard`` für Datei-Copy/Cut/Paste."""
def __init__(self, clipboard: Gdk.Clipboard) -> None:
# Die vom Aufrufer übergebene Zwischenablage merken
# (z. B. ``display.get_clipboard()``).
self._clipboard = clipboard
# ------------------------------------------------------------------
# Schreiben
# ------------------------------------------------------------------
def _set(self, operation: str, paths: List[Path]) -> None:
"""Legt die Pfade als ``operation`` (copy/cut) in beiden Formaten ab."""
uris = _paths_to_uris(paths)
# x-special/gnome-copied-files: Op-Zeile + URIs, durch \n getrennt,
# ohne abschließenden Zeilenumbruch.
gnome_payload = "\n".join([operation, *uris]).encode("utf-8")
# text/uri-list: URIs durch \r\n getrennt.
urilist_payload = "\r\n".join(uris).encode("utf-8")
prov_gnome = Gdk.ContentProvider.new_for_bytes(
_MIME_GNOME, GLib.Bytes.new(gnome_payload)
)
prov_urilist = Gdk.ContentProvider.new_for_bytes(
_MIME_URILIST, GLib.Bytes.new(urilist_payload)
)
provider = Gdk.ContentProvider.new_union([prov_gnome, prov_urilist])
self._clipboard.set_content(provider)
def copy(self, paths: List[Path]) -> None:
"""Legt die Pfade als COPY-Operation in die Zwischenablage."""
self._set("copy", paths)
def cut(self, paths: List[Path]) -> None:
"""Legt die Pfade als CUT/MOVE-Operation in die Zwischenablage."""
self._set("cut", paths)
# ------------------------------------------------------------------
# Lesen
# ------------------------------------------------------------------
def has_files(self) -> bool:
"""True, wenn die Zwischenablage eines unserer Datei-Formate anbietet."""
try:
formats = self._clipboard.get_formats()
return formats.contain_mime_type(_MIME_GNOME) or formats.contain_mime_type(
_MIME_URILIST
)
except Exception:
return False
def read(self, callback: Callable[[List[Path], bool], None]) -> None:
"""Liest die Zwischenablage asynchron und ruft ``callback(paths, is_cut)``.
Bevorzugt ``x-special/gnome-copied-files`` (liefert Operation + Pfade),
fällt auf ``text/uri-list`` zurück (Operation = copy). Liefert nichts
Brauchbares vor, wird ``callback([], False)`` aufgerufen. Wirft nie.
"""
def deliver(paths: List[Path], is_cut: bool) -> None:
try:
callback(paths, is_cut)
except Exception:
# Fehler im Nutzer-Callback nicht weiterreichen.
pass
def on_read(clipboard: Gdk.Clipboard, result: Gio.AsyncResult) -> None:
try:
stream, mime = clipboard.read_finish(result)
except GLib.Error:
# Nichts Brauchbares in der Zwischenablage.
deliver([], False)
return
except Exception:
deliver([], False)
return
if stream is None:
deliver([], False)
return
# Den Stream asynchron auslesen, damit der Mainloop nicht
# blockiert (synchrones Lesen der eigenen Auswahl kann unter
# Wayland deadlocken).
self._read_all_async(
stream, lambda data: deliver(*self._parse(data, mime))
)
try:
self._clipboard.read_async(
[_MIME_GNOME, _MIME_URILIST],
GLib.PRIORITY_DEFAULT,
None,
on_read,
)
except Exception:
# Asynchrone Anforderung über den Mainloop nachreichen.
GLib.idle_add(lambda: (deliver([], False), False)[1])
@staticmethod
def _read_all_async(
stream: Gio.InputStream, done: Callable[[bytes], None]
) -> None:
"""Liest den gesamten Stream asynchron, ruft ``done(bytes)``."""
chunks: List[bytes] = []
def on_chunk(s: Gio.InputStream, result: Gio.AsyncResult) -> None:
try:
buf = s.read_bytes_finish(result)
except Exception:
try:
s.close(None)
except Exception:
pass
done(b"".join(chunks))
return
size = buf.get_size() if buf is not None else 0
if size == 0:
# Ende des Streams.
try:
s.close(None)
except Exception:
pass
done(b"".join(chunks))
return
chunks.append(buf.get_data() or b"")
s.read_bytes_async(65536, GLib.PRIORITY_DEFAULT, None, on_chunk)
stream.read_bytes_async(65536, GLib.PRIORITY_DEFAULT, None, on_chunk)
@staticmethod
def _parse(data: bytes, mime: str) -> tuple[List[Path], bool]:
"""Parst die Rohdaten je nach ausgewähltem MIME-Typ."""
text = data.decode("utf-8", errors="replace")
is_cut = False
uri_lines: List[str]
if mime == _MIME_GNOME:
# Erste nicht-leere Zeile ist die Operation.
lines = text.split("\n")
op = ""
rest_start = 0
for i, line in enumerate(lines):
if line.strip():
op = line.strip().lower()
rest_start = i + 1
break
is_cut = op == "cut"
uri_lines = lines[rest_start:]
else:
# text/uri-list: durch \r\n (oder \n) getrennt, evtl. #-Kommentare.
uri_lines = text.replace("\r\n", "\n").split("\n")
paths: List[Path] = []
for line in uri_lines:
line = line.strip()
if not line or line.startswith("#"):
continue
p = _uri_to_path(line)
if p is not None:
paths.append(p)
return paths, is_cut
+167
View File
@@ -0,0 +1,167 @@
"""Favoriten-Verwaltung für die Dateimanager-Seitenleiste.
Speichert benutzererstellte Lesezeichen ("Favoriten") als JSON und
benachrichtigt registrierte Beobachter, sobald sich die Liste ändert,
damit die Seitenleiste neu aufgebaut werden kann.
"""
from __future__ import annotations
import json
from pathlib import Path
from typing import Callable
import gi
gi.require_version("Gtk", "4.0")
from gi.repository import GLib # noqa: E402
_VERSION = 1
def _config_path() -> Path:
"""Pfad zur JSON-Datei mit den gespeicherten Lesezeichen."""
return Path(GLib.get_user_config_dir()) / "taninux" / "files-bookmarks.json"
def _norm(path: Path) -> Path:
"""Pfad auf eine vergleichbare, absolute Form bringen.
Verwendet expanduser().resolve(); bei kaputten Pfaden wird auf eine
einfache absolute Form zurückgegriffen, ohne eine Ausnahme auszulösen.
"""
try:
return Path(path).expanduser().resolve()
except (OSError, RuntimeError, ValueError):
try:
return Path(path).expanduser().absolute()
except (OSError, RuntimeError, ValueError):
return Path(path)
class Favorites:
"""Persistente Sammlung von Seitenleisten-Lesezeichen."""
def __init__(self) -> None:
self._paths: list[Path] = []
self._callbacks: list[Callable[[], None]] = []
self._load()
# ------------------------------------------------------------------
# Persistenz
# ------------------------------------------------------------------
def _load(self) -> None:
"""Lesezeichen von der Platte laden; bei Fehlern leer starten."""
path = _config_path()
try:
raw = path.read_text(encoding="utf-8")
data = json.loads(raw)
except (OSError, ValueError):
self._paths = []
return
if not isinstance(data, dict):
self._paths = []
return
entries = data.get("paths")
if not isinstance(entries, list):
self._paths = []
return
result: list[Path] = []
seen: set[Path] = set()
for entry in entries:
if not isinstance(entry, str):
continue
p = _norm(Path(entry))
if p in seen:
continue
seen.add(p)
result.append(p)
self._paths = result
def _save(self) -> None:
"""Aktuellen Zustand als JSON schreiben (Verzeichnis bei Bedarf anlegen)."""
path = _config_path()
try:
path.parent.mkdir(parents=True, exist_ok=True)
except OSError:
return
payload = {
"version": _VERSION,
"paths": [str(p) for p in self._paths],
}
try:
text = json.dumps(payload, indent=2, ensure_ascii=False)
tmp = path.with_suffix(path.suffix + ".tmp")
try:
tmp.write_text(text, encoding="utf-8")
tmp.replace(path)
except OSError:
# Fallback: einfacher direkter Schreibvorgang.
path.write_text(text, encoding="utf-8")
except OSError:
return
# ------------------------------------------------------------------
# Beobachter
# ------------------------------------------------------------------
def connect(self, callback: Callable[[], None]) -> None:
"""Beobachter registrieren, der nach jedem add/remove aufgerufen wird."""
self._callbacks.append(callback)
def _notify(self) -> None:
for cb in list(self._callbacks):
try:
cb()
except Exception:
# Ein fehlerhafter Beobachter darf die anderen nicht brechen.
pass
# ------------------------------------------------------------------
# Öffentliche API
# ------------------------------------------------------------------
def list(self) -> list[Path]:
"""Aktuelle Lesezeichen in Reihenfolge; nur existierende Verzeichnisse."""
result: list[Path] = []
for p in self._paths:
try:
if p.is_dir():
result.append(p)
except OSError:
continue
return result
def contains(self, path: Path) -> bool:
target = _norm(path)
return any(p == target for p in self._paths)
def add(self, path: Path) -> bool:
"""Lesezeichen anhängen, falls noch nicht vorhanden und ein Verzeichnis."""
target = _norm(path)
try:
if not target.is_dir():
return False
except OSError:
return False
if any(p == target for p in self._paths):
return False
self._paths.append(target)
self._save()
self._notify()
return True
def remove(self, path: Path) -> bool:
"""Lesezeichen entfernen, falls vorhanden."""
target = _norm(path)
for i, p in enumerate(self._paths):
if p == target:
del self._paths[i]
self._save()
self._notify()
return True
return False
+215
View File
@@ -0,0 +1,215 @@
"""Dateioperationen für den Dateimanager.
Kopieren/Verschieben laufen in einem Hintergrund-Thread; der Abschluss wird
über GLib.idle_add zurück auf die GLib-Hauptschleife marshallt, damit UI-Code
sicher ausgeführt werden kann. Papierkorb, Umbenennen und Ordner-Erstellung
sind synchron.
"""
from __future__ import annotations
import shutil
import threading
from pathlib import Path
from typing import Callable, Iterable
import gi
gi.require_version("Gtk", "4.0")
from gi.repository import Gio, GLib # noqa: E402
def unique_target(dest_dir: Path, name: str) -> Path:
"""Liefert einen kollisionsfreien Pfad innerhalb von ``dest_dir``.
Existiert ``name`` bereits, wird vor der Erweiterung ein Suffix eingefügt:
``file.txt`` -> ``file copy.txt`` -> ``file copy 2.txt`` -> ...
Für Verzeichnisse / Namen ohne Erweiterung wird `` copy``, `` copy 2`` usw.
angehängt. Gibt niemals einen bereits existierenden Pfad zurück.
"""
dest_dir = Path(dest_dir)
candidate = dest_dir / name
if not candidate.exists():
return candidate
p = Path(name)
suffix = p.suffix
stem = p.name[: len(p.name) - len(suffix)] if suffix else p.name
n = 1
while True:
if n == 1:
new_stem = f"{stem} copy"
else:
new_stem = f"{stem} copy {n}"
candidate = dest_dir / f"{new_stem}{suffix}"
if not candidate.exists():
return candidate
n += 1
def _is_into_subtree(src: Path, dest_dir: Path) -> bool:
"""True, wenn ``dest_dir`` gleich ``src`` ist oder innerhalb von ``src`` liegt."""
src_r = src.resolve()
dest_r = dest_dir.resolve()
if src_r == dest_r:
return True
return src_r in dest_r.parents
def _do_copy(
sources: list[Path],
dest_dir: Path,
move: bool,
on_done: Callable[[list[Path], list[tuple[Path, str]]], None] | None,
) -> None:
succeeded: list[Path] = []
errors: list[tuple[Path, str]] = []
for src in sources:
try:
if not src.exists():
errors.append((src, "Quelle existiert nicht"))
continue
# No-op move: src already lives in dest_dir.
if move and src.parent.resolve() == dest_dir.resolve():
continue
if src.is_dir() and _is_into_subtree(src, dest_dir):
raise ValueError(
"Verzeichnis kann nicht in sich selbst kopiert werden"
)
target = unique_target(dest_dir, src.name)
if move:
shutil.move(str(src), str(target))
elif src.is_dir():
shutil.copytree(str(src), str(target))
else:
shutil.copy2(str(src), str(target))
succeeded.append(target)
except Exception as exc: # noqa: BLE001 - record, never crash
errors.append((src, str(exc)))
if on_done is not None:
GLib.idle_add(lambda: (on_done(succeeded, errors), False)[1])
def copy_paths(
sources: Iterable[Path],
dest_dir: Path,
*,
move: bool,
on_done: Callable[[list[Path], list[tuple[Path, str]]], None] | None = None,
) -> None:
"""Kopiert/verschiebt ``sources`` nach ``dest_dir`` im Hintergrund-Thread.
Für jede Quelle wird ein kollisionssicheres Ziel via
:func:`unique_target` bestimmt. ``on_done(succeeded, errors)`` wird nach
Abschluss über ``GLib.idle_add`` auf der Hauptschleife aufgerufen.
"""
srcs = [Path(s) for s in sources]
dest = Path(dest_dir)
thread = threading.Thread(
target=_do_copy,
args=(srcs, dest, move, on_done),
daemon=True,
)
thread.start()
def trash_paths(paths: Iterable[Path]) -> list[tuple[Path, str]]:
"""Verschiebt jeden Pfad synchron in den Papierkorb.
Gibt eine Liste von ``(path, message)`` für Fehlschläge zurück
(leere Liste = alles ok).
"""
errors: list[tuple[Path, str]] = []
for p in paths:
p = Path(p)
try:
Gio.File.new_for_path(str(p)).trash(None)
except Exception as exc: # noqa: BLE001
errors.append((p, str(exc)))
return errors
def delete_paths(paths: Iterable[Path]) -> list[tuple[Path, str]]:
"""Löscht jeden Pfad endgültig (rekursiv für Ordner).
Gibt eine Liste von ``(path, message)`` für Fehlschläge zurück
(leere Liste = alles ok).
"""
errors: list[tuple[Path, str]] = []
for p in paths:
p = Path(p)
try:
if p.is_dir() and not p.is_symlink():
shutil.rmtree(str(p))
else:
p.unlink()
except Exception as exc: # noqa: BLE001
errors.append((p, str(exc)))
return errors
def rename_path(path: Path, new_name: str) -> tuple[bool, str]:
"""Benennt ``path`` innerhalb desselben Elternverzeichnisses um.
Leere Namen oder Namen mit ``/`` werden abgelehnt.
Gibt ``(True, "")`` bei Erfolg, sonst ``(False, message)`` zurück.
"""
path = Path(path)
name = (new_name or "").strip()
if not name:
return (False, "Leerer Name")
if "/" in name:
return (False, 'Name darf kein "/" enthalten')
try:
target = path.parent / name
if target.exists():
return (False, "Ziel existiert bereits")
path.rename(target)
return (True, "")
except Exception as exc: # noqa: BLE001
return (False, str(exc))
def make_dir(parent: Path, name: str) -> tuple[bool, str]:
"""Erstellt ``parent/name`` (ohne Eltern, darf nicht existieren).
Gibt ``(True, "")`` / ``(False, message)`` zurück.
"""
parent = Path(parent)
nm = (name or "").strip()
if not nm:
return (False, "Leerer Name")
if "/" in nm:
return (False, 'Name darf kein "/" enthalten')
try:
(parent / nm).mkdir(parents=False, exist_ok=False)
return (True, "")
except Exception as exc: # noqa: BLE001
return (False, str(exc))
def make_file(parent: Path, name: str) -> tuple[bool, str]:
"""Erstellt eine leere Datei ``parent/name`` (darf nicht existieren).
Gibt ``(True, "")`` / ``(False, message)`` zurück.
"""
parent = Path(parent)
nm = (name or "").strip()
if not nm:
return (False, "Leerer Name")
if "/" in nm:
return (False, 'Name darf kein "/" enthalten')
try:
(parent / nm).touch(exist_ok=False)
return (True, "")
except Exception as exc: # noqa: BLE001
return (False, str(exc))
+128
View File
@@ -0,0 +1,128 @@
"""Datenmodell für den Dateimanager.
Stellt :class:`FileItem`-Objekte bereit, an die eine ``Gtk.GridView``- bzw.
``Gtk.ColumnView``-Factory binden kann, sowie ``list_directory`` zum
Auflisten eines Verzeichnisses.
"""
import gi
gi.require_version("Gtk", "4.0")
from pathlib import Path
from gi.repository import Gio, GLib, GObject
# Gio-Abfrageattribute, die beim Enumerieren angefordert werden.
ATTRS = ",".join(
[
"standard::name",
"standard::display-name",
"standard::icon",
"standard::type",
"standard::is-hidden",
"standard::is-backup",
"standard::content-type",
"standard::size",
"time::modified",
]
)
_UNITS = ("bytes", "KB", "MB", "GB", "TB")
class FileItem(GObject.Object):
"""Ein einzelner Eintrag (Datei oder Ordner) im Dateimanager."""
__gtype_name__ = "TaninFileItem"
name = GObject.Property(type=str, default="")
path = GObject.Property(type=str, default="")
is_dir = GObject.Property(type=bool, default=False)
size = GObject.Property(type=GObject.TYPE_INT64, default=0)
mtime = GObject.Property(type=GObject.TYPE_INT64, default=0)
content_type = GObject.Property(type=str, default="")
def __init__(self, **kwargs):
super().__init__(**kwargs)
# Kein GObject.Property: Gio.Icon lässt sich bequemer direkt setzen.
self.icon = None
@classmethod
def from_info(cls, info: Gio.FileInfo, parent: Path) -> "FileItem":
name = info.get_name()
display_name = info.get_display_name() or name
is_dir = info.get_file_type() == Gio.FileType.DIRECTORY
dt = info.get_modification_date_time()
mtime = dt.to_unix() if dt is not None else 0
item = cls(
name=display_name,
path=str(parent / name),
is_dir=is_dir,
size=0 if is_dir else info.get_size(),
mtime=mtime,
content_type=info.get_content_type() or "",
)
item.icon = info.get_icon()
return item
def as_path(self) -> Path:
return Path(self.path)
def size_text(self) -> str:
if self.is_dir:
return ""
value = float(self.size)
unit_index = 0
while value >= 1024 and unit_index < len(_UNITS) - 1:
value /= 1024
unit_index += 1
if unit_index == 0:
return f"{int(value)} {_UNITS[0]}"
if value >= 10:
return f"{value:.0f} {_UNITS[unit_index]}"
return f"{value:.1f} {_UNITS[unit_index]}"
def mtime_text(self) -> str:
if self.mtime <= 0:
return ""
dt = GLib.DateTime.new_from_unix_local(self.mtime)
if dt is None:
return ""
return dt.format("%d %b %Y %H:%M")
def list_directory(path: Path, show_hidden: bool) -> list[FileItem]:
"""Listet die Kinder von ``path`` auf.
Versteckte/Backup-Einträge werden ohne ``show_hidden`` übersprungen.
Sortierung: Verzeichnisse zuerst, danach alphabetisch (case-insensitive).
Bei nicht lesbarem Verzeichnis wird ``[]`` zurückgegeben.
"""
gfile = Gio.File.new_for_path(str(path))
try:
enumerator = gfile.enumerate_children(
ATTRS, Gio.FileQueryInfoFlags.NONE, None
)
except GLib.Error:
return []
items: list[FileItem] = []
try:
while True:
try:
info = enumerator.next_file(None)
except GLib.Error:
break
if info is None:
break
if not show_hidden and (info.get_is_hidden() or info.get_is_backup()):
continue
items.append(FileItem.from_info(info, path))
finally:
enumerator.close(None)
items.sort(key=lambda i: (not i.is_dir, i.name.casefold()))
return items
+74
View File
@@ -0,0 +1,74 @@
"""Sidebar-Orte: XDG-Standardverzeichnisse + eingehängte Laufwerke.
Liefert reine Datenstrukturen (Icon, Label, Pfad) — die Window-Schicht baut
daraus die Sidebar-Zeilen. So bleibt die Ortelogik testbar und GTK-arm.
"""
from __future__ import annotations
import logging
from dataclasses import dataclass
from pathlib import Path
from typing import List
import gi
gi.require_version("Gtk", "4.0")
from gi.repository import Gio, GLib # noqa: E402
logger = logging.getLogger(__name__)
@dataclass(frozen=True)
class Place:
icon: str
label: str
path: Path
# (XDG-Enum, Icon-Name, Anzeigename) — fehlende/auf $HOME zeigende werden gefiltert.
_SPECIALS = [
(GLib.UserDirectory.DIRECTORY_DESKTOP, "user-desktop-symbolic", "Desktop"),
(GLib.UserDirectory.DIRECTORY_DOCUMENTS, "folder-documents-symbolic", "Documents"),
(GLib.UserDirectory.DIRECTORY_DOWNLOAD, "folder-download-symbolic", "Downloads"),
(GLib.UserDirectory.DIRECTORY_MUSIC, "folder-music-symbolic", "Music"),
(GLib.UserDirectory.DIRECTORY_PICTURES, "folder-pictures-symbolic", "Pictures"),
(GLib.UserDirectory.DIRECTORY_VIDEOS, "folder-videos-symbolic", "Videos"),
]
def user_places() -> List[Place]:
"""Home + vorhandene XDG-Sonderverzeichnisse (ohne Dubletten zu Home)."""
home = Path.home()
out = [Place("user-home-symbolic", "Home", home)]
for directory, icon, label in _SPECIALS:
raw = GLib.get_user_special_dir(directory)
if not raw:
continue
path = Path(raw)
if path == home or not path.is_dir():
continue
out.append(Place(icon, label, path))
return out
def mounted_places() -> List[Place]:
"""Eingehängte, vom Nutzer entfernbare/sichtbare Laufwerke (USB, Platten…)."""
out: List[Place] = []
try:
monitor = Gio.VolumeMonitor.get()
except Exception: # noqa: BLE001
logger.debug("VolumeMonitor nicht verfügbar", exc_info=True)
return out
for mount in monitor.get_mounts():
try:
if mount.is_shadowed():
continue
root = mount.get_root()
path = root.get_path() if root is not None else None
if not path:
continue
out.append(Place("drive-removable-media-symbolic", mount.get_name(), Path(path)))
except Exception: # noqa: BLE001
logger.debug("Mount übersprungen", exc_info=True)
return out
+328
View File
@@ -0,0 +1,328 @@
"""Quick-Look-Vorschaufläche für den TANINUX-Dateimanager.
Zeigt eine einzelne Datei je nach Typ groß an (Bild / Text / PDF / generisch).
Schweres Laden geschieht in Hintergrund-Threads; ein Token-Guard sorgt dafür,
dass ein langsamer Ladevorgang für einen alten Pfad niemals einen neueren
Request überschreibt.
"""
from __future__ import annotations
import threading
from pathlib import Path
import gi
gi.require_version("Gtk", "4.0")
gi.require_version("Gdk", "4.0")
gi.require_version("GdkPixbuf", "2.0")
from gi.repository import Gdk, GdkPixbuf, Gio, GLib, Gtk
try:
gi.require_version("Poppler", "0.18")
from gi.repository import Poppler
except (ValueError, ImportError):
Poppler = None
try:
import cairo
except ImportError:
cairo = None
# Allowlist textueller Content-Types (zusätzlich zu allem unter "text/").
_TEXTUAL_TYPES = {
"application/json",
"application/xml",
"application/javascript",
"application/x-shellscript",
"application/x-yaml",
"application/toml",
"application/x-desktop",
"application/x-python",
"application/x-perl",
"application/x-ruby",
}
# Dateiendungen, die wir als Text behandeln, auch wenn der Content-Type generisch ist.
_TEXTUAL_SUFFIXES = {
".py", ".js", ".ts", ".rs", ".c", ".h", ".cpp", ".go", ".sh", ".md",
".txt", ".json", ".toml", ".yaml", ".yml", ".xml", ".html", ".css",
".ini", ".conf", ".log",
}
_MAX_TEXT_BYTES = 100_000
_IMG_MAX = 1600
_PDF_TARGET = 1200
def _is_textual(content_type: str, path: Path) -> bool:
"""True, wenn der Inhalt sinnvoll als Text vorschaubar ist."""
ct = (content_type or "").lower()
if ct.startswith("text/"):
return True
if ct in _TEXTUAL_TYPES:
return True
if "+xml" in ct or "+json" in ct:
return True
try:
if path.suffix.lower() in _TEXTUAL_SUFFIXES:
return True
except Exception:
pass
return False
def _humanize_size(num: int) -> str:
"""Größe in 1024er-Schritten (B/KB/MB/GB/TB)."""
size = float(num)
for unit in ("B", "KB", "MB", "GB", "TB"):
if abs(size) < 1024.0 or unit == "TB":
if unit == "B":
return f"{int(size)} {unit}"
return f"{size:.1f} {unit}"
size /= 1024.0
return f"{size:.1f} TB"
class PreviewPane(Gtk.Box):
"""Quick-Look-Vorschaufläche: zeigt eine Datei je nach Typ groß an."""
def __init__(self) -> None:
super().__init__(orientation=Gtk.Orientation.VERTICAL)
self.add_css_class("tn-preview")
self.set_hexpand(True)
self.set_vexpand(True)
self._token = 0
self._stack = Gtk.Stack()
self._stack.set_hexpand(True)
self._stack.set_vexpand(True)
self.append(self._stack)
# --- "image" ---------------------------------------------------------
self._picture = Gtk.Picture()
self._picture.set_content_fit(Gtk.ContentFit.CONTAIN)
self._picture.set_hexpand(True)
self._picture.set_vexpand(True)
self._stack.add_named(self._picture, "image")
# --- "text" ----------------------------------------------------------
self._scroller = Gtk.ScrolledWindow()
self._scroller.set_hexpand(True)
self._scroller.set_vexpand(True)
self._scroller.set_policy(Gtk.PolicyType.AUTOMATIC, Gtk.PolicyType.AUTOMATIC)
self._textview = Gtk.TextView()
self._textview.set_editable(False)
self._textview.set_monospace(True)
self._textview.set_cursor_visible(False)
self._textview.set_wrap_mode(Gtk.WrapMode.WORD_CHAR)
self._scroller.set_child(self._textview)
self._stack.add_named(self._scroller, "text")
# --- "generic" -------------------------------------------------------
generic = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=12)
generic.set_hexpand(True)
generic.set_vexpand(True)
generic.set_halign(Gtk.Align.CENTER)
generic.set_valign(Gtk.Align.CENTER)
self._gen_icon = Gtk.Image()
self._gen_icon.set_pixel_size(128)
generic.append(self._gen_icon)
self._gen_name = Gtk.Label()
self._gen_name.add_css_class("tn-preview-name")
self._gen_name.set_wrap(True)
self._gen_name.set_justify(Gtk.Justification.CENTER)
generic.append(self._gen_name)
self._gen_type = Gtk.Label()
self._gen_type.add_css_class("tn-preview-meta")
generic.append(self._gen_type)
self._gen_size = Gtk.Label()
self._gen_size.add_css_class("tn-preview-meta")
generic.append(self._gen_size)
self._stack.add_named(generic, "generic")
self._stack.set_visible_child_name("generic")
# ------------------------------------------------------------------ public
def show_path(self, path: Path, content_type: str) -> None:
"""Aktualisiert die Vorschau auf `path`. Niemals werfend."""
try:
path = Path(path)
self._token += 1
token = self._token
content_type = content_type or ""
if content_type.startswith("image/"):
self._load_image_async(path, token)
elif content_type == "application/pdf" and Poppler is not None and cairo is not None:
self._load_pdf_async(path, token)
elif _is_textual(content_type, path):
self._load_text_async(path, token)
else:
self._show_generic(path, content_type)
except Exception:
# Defensive: niemals nach außen werfen.
try:
self._show_generic(Path(path), content_type or "")
except Exception:
pass
def clear(self) -> None:
"""Zurück in einen neutralen Leerzustand; In-Flight-Loads verwerfen."""
self._token += 1
self._gen_icon.clear()
self._gen_name.set_text("")
self._gen_type.set_text("")
self._gen_size.set_text("")
self._stack.set_visible_child_name("generic")
# ------------------------------------------------------------------- image
def _load_image_async(self, path: Path, token: int) -> None:
def work() -> None:
try:
pixbuf = GdkPixbuf.Pixbuf.new_from_file_at_scale(
str(path), _IMG_MAX, _IMG_MAX, True
)
except Exception:
GLib.idle_add(self._apply_generic_fallback, path, token)
return
GLib.idle_add(self._apply_pixbuf, pixbuf, token)
threading.Thread(target=work, daemon=True).start()
def _apply_pixbuf(self, pixbuf: GdkPixbuf.Pixbuf, token: int) -> bool:
if token != self._token:
return False
try:
texture = Gdk.Texture.new_for_pixbuf(pixbuf)
self._picture.set_paintable(texture)
self._stack.set_visible_child_name("image")
except Exception:
pass
return False
# -------------------------------------------------------------------- text
def _load_text_async(self, path: Path, token: int) -> None:
def work() -> None:
try:
with open(path, "rb") as fh:
raw = fh.read(_MAX_TEXT_BYTES)
text = raw.decode("utf-8", errors="replace")
except Exception:
GLib.idle_add(self._apply_generic_fallback, path, token)
return
GLib.idle_add(self._apply_text, text, token)
threading.Thread(target=work, daemon=True).start()
def _apply_text(self, text: str, token: int) -> bool:
if token != self._token:
return False
try:
self._textview.get_buffer().set_text(text)
self._stack.set_visible_child_name("text")
except Exception:
pass
return False
# --------------------------------------------------------------------- pdf
def _load_pdf_async(self, path: Path, token: int) -> None:
def work() -> None:
try:
uri = Gio.File.new_for_path(str(path)).get_uri()
doc = Poppler.Document.new_from_file(uri, None)
page = doc.get_page(0)
if page is None:
raise ValueError("no page 0")
p_w, p_h = page.get_size()
if p_w <= 0 or p_h <= 0:
raise ValueError("bad page size")
scale = _PDF_TARGET / max(p_w, p_h)
w = max(1, int(p_w * scale))
h = max(1, int(p_h * scale))
surface = cairo.ImageSurface(cairo.FORMAT_ARGB32, w, h)
ctx = cairo.Context(surface)
# weißer Hintergrund (PDF-Seiten sind oft transparent)
ctx.set_source_rgb(1, 1, 1)
ctx.paint()
ctx.scale(scale, scale)
page.render(ctx)
surface.flush()
data = bytes(surface.get_data())
stride = surface.get_stride()
except Exception:
GLib.idle_add(self._apply_generic_fallback, path, token)
return
GLib.idle_add(self._apply_pdf, data, w, h, stride, token)
threading.Thread(target=work, daemon=True).start()
def _apply_pdf(self, data: bytes, w: int, h: int, stride: int, token: int) -> bool:
if token != self._token:
return False
try:
gbytes = GLib.Bytes.new(data)
texture = Gdk.MemoryTexture.new(
w, h, Gdk.MemoryFormat.B8G8R8A8_PREMULTIPLIED, gbytes, stride
)
self._picture.set_paintable(texture)
self._stack.set_visible_child_name("image")
except Exception:
pass
return False
# ----------------------------------------------------------------- generic
def _apply_generic_fallback(self, path: Path, token: int) -> bool:
if token != self._token:
return False
self._show_generic(path, "")
return False
def _show_generic(self, path: Path, content_type: str) -> None:
# Icon
try:
gfile = Gio.File.new_for_path(str(path))
info = gfile.query_info(
"standard::icon", Gio.FileQueryInfoFlags.NONE, None
)
icon = info.get_icon()
if icon is not None:
self._gen_icon.set_from_gicon(icon)
else:
self._gen_icon.set_from_icon_name("text-x-generic")
except Exception:
self._gen_icon.set_from_icon_name("text-x-generic")
# Name
try:
self._gen_name.set_text(path.name)
except Exception:
self._gen_name.set_text("")
# Typbeschreibung
desc = ""
try:
if content_type:
desc = Gio.content_type_get_description(content_type) or ""
except Exception:
desc = ""
self._gen_type.set_text(desc)
# Größe
size_text = ""
try:
size_text = _humanize_size(path.stat().st_size)
except Exception:
size_text = ""
self._gen_size.set_text(size_text)
self._stack.set_visible_child_name("generic")
+324
View File
@@ -0,0 +1,324 @@
"""Eigenschaften-Dialog für eine Datei oder einen Ordner.
Zeigt Typ, Größe, Ort, Änderungsdatum und Rechte in einem macOS-artigen
Karten-Layout (System-Settings-Optik). Nutzt das native, libadwaita-freie
Widget-Vokabular aus ``taninux.gui.ui.native`` (``Group`` / ``Row`` / die
``.tn-*``-CSS-Klassen) für visuelle Konsistenz mit der restlichen App.
Öffentliche API::
present_properties(parent: Gtk.Widget, path: Path) -> None
Die Ordnergröße wird rekursiv in einem Hintergrund-Thread berechnet (os.walk),
damit der Dialog nie blockiert; das Ergebnis wird per ``GLib.idle_add`` live in
die Zeile geschrieben.
"""
from __future__ import annotations
import os
import stat as _stat
import threading
from datetime import datetime
from pathlib import Path
import gi
gi.require_version("Gtk", "4.0")
from gi.repository import Gio, GLib, Gtk # noqa: E402
from taninux.gui.ui import native # noqa: E402
# pwd/grp gibt es nur auf POSIX — defensiv importieren, damit der Modul-Import
# auf anderen Plattformen nicht scheitert.
try: # pragma: no cover - plattformabhängig
import pwd as _pwd
except Exception: # noqa: BLE001
_pwd = None
try: # pragma: no cover - plattformabhängig
import grp as _grp
except Exception: # noqa: BLE001
_grp = None
_DASH = "" # Platzhalter, wenn ein Wert nicht ermittelbar ist.
# --------------------------------------------------------------------------- #
# Hilfsfunktionen
# --------------------------------------------------------------------------- #
def _humanize(size: int) -> str:
"""Bytes in eine lesbare Größe (1024er-Schritte, bytes/KB/MB/GB/TB)."""
value = float(size)
for unit in ("bytes", "KB", "MB", "GB", "TB"):
if value < 1024.0 or unit == "TB":
if unit == "bytes":
return f"{int(value)} {unit}"
return f"{value:.1f} {unit}"
value /= 1024.0
return f"{int(size)} bytes" # unerreichbar, aber robust
def _size_label(size: int) -> str:
"""„1.2 MB (1,234,567 bytes)" — humanisiert plus exakte Byte-Zahl."""
if size < 1024:
return _humanize(size)
return f"{_humanize(size)} ({size:,} bytes)"
def _icon_for(gfile: Gio.File) -> "Gio.Icon | None":
"""Gio.Icon aus ``standard::icon`` abfragen (oder None bei Fehler)."""
try:
info = gfile.query_info(
"standard::icon", Gio.FileQueryInfoFlags.NONE, None
)
return info.get_icon()
except Exception: # noqa: BLE001
return None
def _content_type(path: Path, is_dir: bool) -> tuple[str, str]:
"""(content_type, beschreibung) für den Pfad ermitteln."""
if is_dir:
ctype = "inode/directory"
else:
guess = Gio.content_type_guess(str(path), None)
ctype = guess[0] if guess and guess[0] else "application/octet-stream"
try:
desc = Gio.content_type_get_description(ctype)
except Exception: # noqa: BLE001
desc = ctype
return ctype, desc or ctype
def _owner_group(st: os.stat_result) -> str:
"""„user · group" aus uid/gid (Namen, wenn auflösbar)."""
user = str(st.st_uid)
group = str(st.st_gid)
if _pwd is not None:
try:
user = _pwd.getpwuid(st.st_uid).pw_name
except Exception: # noqa: BLE001
pass
if _grp is not None:
try:
group = _grp.getgrgid(st.st_gid).gr_name
except Exception: # noqa: BLE001
pass
return f"{user} · {group}"
def _format_time(epoch: float) -> str:
"""Lokalisiertes Datum/Uhrzeit aus einem Unix-Timestamp."""
try:
return datetime.fromtimestamp(epoch).strftime("%c")
except Exception: # noqa: BLE001
return _DASH
# --------------------------------------------------------------------------- #
# Wert-Zeile (Titel links, Wert rechts) — auf Basis der nativen Row
# --------------------------------------------------------------------------- #
def _value_row(title: str, value: str) -> tuple[native.Row, Gtk.Label]:
"""Eine Zeile mit fixem Titel und rechtsbündigem Wert-Label.
Gibt (Row, Wert-Label) zurück, damit der Aufrufer den Wert später (z. B.
nach der Ordnergröße-Berechnung) aktualisieren kann.
"""
row = native.Row(title=title)
row.set_activatable(False)
val = Gtk.Label(label=value, xalign=1.0)
val.add_css_class("tn-subtitle")
val.set_selectable(True)
val.set_wrap(True)
val.set_max_width_chars(34)
val.set_justify(Gtk.Justification.RIGHT)
row.add_suffix(val)
return row, val
# --------------------------------------------------------------------------- #
# Öffentliche API
# --------------------------------------------------------------------------- #
def present_properties(parent: Gtk.Widget, path: Path) -> None:
"""Baut ein modales Eigenschaften-Fenster für ``path`` und zeigt es.
Transient für das Toplevel von ``parent``; übernimmt dessen Dunkel-Zustand.
Robust gegen fehlende Dateien / Rechtefehler (zeigt " statt zu crashen).
"""
native.ensure_css()
path = Path(path)
# -- stat (defensiv) ----------------------------------------------------
try:
st = path.lstat()
except Exception: # noqa: BLE001
st = None
is_dir = bool(st is not None and _stat.S_ISDIR(st.st_mode))
if st is None: # Fallback, falls lstat scheiterte
is_dir = path.is_dir()
gfile = Gio.File.new_for_path(str(path))
ctype, type_desc = _content_type(path, is_dir)
# -- Fenster ------------------------------------------------------------
win = Gtk.Window(modal=True, resizable=True)
win.set_default_size(380, 520)
win.set_title(f"{path.name or path} — Eigenschaften")
# Toplevel ermitteln: transient + Dunkel-Zustand übernehmen.
root = parent.get_root() if hasattr(parent, "get_root") else None
if isinstance(root, Gtk.Window):
win.set_transient_for(root)
dark = bool(isinstance(root, Gtk.Widget) and root.has_css_class("tn-dark"))
native.set_dark(win, dark)
# -- Scrollender Inhalt -------------------------------------------------
scroller = Gtk.ScrolledWindow(hexpand=True, vexpand=True)
scroller.set_policy(Gtk.PolicyType.NEVER, Gtk.PolicyType.AUTOMATIC)
scroller.add_css_class("tn-page")
win.set_child(scroller)
column = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=24)
column.set_margin_top(24)
column.set_margin_bottom(24)
column.set_margin_start(20)
column.set_margin_end(20)
scroller.set_child(column)
# -- Kopf: großes Icon + Name (fett) + Typbeschreibung ------------------
header = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=8)
header.set_halign(Gtk.Align.CENTER)
icon = _icon_for(gfile)
img = (
Gtk.Image.new_from_gicon(icon)
if icon is not None
else Gtk.Image.new_from_icon_name(
"folder" if is_dir else "text-x-generic"
)
)
img.set_pixel_size(64)
img.set_halign(Gtk.Align.CENTER)
header.append(img)
name_lbl = Gtk.Label(label=path.name or str(path))
name_lbl.add_css_class("title-3")
name_lbl.set_wrap(True)
name_lbl.set_justify(Gtk.Justification.CENTER)
name_lbl.set_max_width_chars(36)
header.append(name_lbl)
type_hdr = Gtk.Label(label=type_desc)
type_hdr.add_css_class("tn-subtitle")
type_hdr.set_wrap(True)
type_hdr.set_justify(Gtk.Justification.CENTER)
header.append(type_hdr)
column.append(header)
# -- Detail-Karte -------------------------------------------------------
group = native.Group()
# Typ
type_row, _ = _value_row("Type", type_desc)
group.add(type_row)
# Größe — Datei sofort, Ordner per Hintergrund-Thread.
if is_dir:
size_row, size_val = _value_row("Size", "Calculating…")
else:
size_bytes = st.st_size if st is not None else 0
size_row, size_val = _value_row(
"Size", _size_label(size_bytes) if st is not None else _DASH
)
group.add(size_row)
# Ort (Elternverzeichnis)
parent_dir = str(path.parent) if str(path.parent) else _DASH
loc_row, _ = _value_row("Location", parent_dir)
group.add(loc_row)
# Geändert
mod_row, _ = _value_row(
"Modified", _format_time(st.st_mtime) if st is not None else _DASH
)
group.add(mod_row)
# Zugegriffen (optional)
acc_row, _ = _value_row(
"Accessed", _format_time(st.st_atime) if st is not None else _DASH
)
group.add(acc_row)
# Rechte: drwxr-xr-x · user · group
if st is not None:
try:
modestr = _stat.filemode(st.st_mode)
except Exception: # noqa: BLE001
modestr = _DASH
perms = f"{modestr}\n{_owner_group(st)}"
else:
perms = _DASH
perm_row, _ = _value_row("Permissions", perms)
group.add(perm_row)
column.append(group)
# -- Schließen-Button ---------------------------------------------------
actions = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL)
actions.set_halign(Gtk.Align.CENTER)
close_btn = Gtk.Button(label="Close")
close_btn.add_css_class("pill")
close_btn.connect("clicked", lambda _b: win.close())
actions.append(close_btn)
column.append(actions)
# -- Ordnergröße im Hintergrund -----------------------------------------
if is_dir:
_spawn_dir_size(win, size_val, path)
win.present()
def _spawn_dir_size(win: Gtk.Window, label: Gtk.Label, path: Path) -> None:
"""Berechnet die rekursive Ordnergröße in einem Thread und füllt ``label``.
Aktualisiert live (Zwischenstände alle paar tausend Einträge) und schreibt
am Ende X MB (N items)". Alle UI-Updates laufen über ``GLib.idle_add``.
"""
cancelled = {"flag": False}
win.connect("destroy", lambda *_: cancelled.__setitem__("flag", True))
def _push(total: int, items: int, done: bool) -> bool:
if cancelled["flag"]:
return False
if done:
label.set_label(f"{_humanize(total)} ({items:,} items)")
else:
label.set_label(f"Calculating… {_humanize(total)}")
return False # einmalig
def _work() -> None:
total = 0
items = 0
last_push = 0.0
try:
for dirpath, dirnames, filenames in os.walk(str(path)):
if cancelled["flag"]:
return
for name in dirnames + filenames:
items += 1
try:
total += os.lstat(os.path.join(dirpath, name)).st_size
except OSError:
pass
# Gelegentlicher Zwischenstand (nicht zu oft → wenig Overhead).
now = GLib.get_monotonic_time()
if now - last_push > 200_000: # ~200 ms
last_push = now
GLib.idle_add(_push, total, items, False)
except Exception: # noqa: BLE001
pass
GLib.idle_add(_push, total, items, True)
threading.Thread(target=_work, daemon=True).start()
+109
View File
@@ -0,0 +1,109 @@
"""Einstellungen fuer den TANINUX-Dateimanager.
Speichert ein paar Praeferenzen sitzungsuebergreifend als JSON unter
``$XDG_CONFIG_HOME/taninux/files-settings.json``. Robust gegen fehlende
Datei, kaputtes JSON und falsche Typen -- es wird niemals eine Ausnahme
ausgeloest, im Zweifel gelten die Defaults.
"""
from __future__ import annotations
import json
import os
from pathlib import Path
from typing import Any
import gi
gi.require_version("Gtk", "4.0")
from gi.repository import GLib # noqa: E402
DEFAULTS = {
"view_mode": "grid", # "grid" | "list"
"sort_key": "name", # name|size|mtime|type
"sort_descending": False,
"folders_first": True,
"show_hidden": False,
"width": 1120,
"height": 740,
}
_CONFIG_FILENAME = "files-settings.json"
def _config_path() -> Path:
return Path(GLib.get_user_config_dir()) / "taninux" / _CONFIG_FILENAME
def _coerce(key: str, value: Any) -> Any:
"""Light type-coercion. Return the coerced value, or DEFAULTS[key] on
a clearly wrong type."""
default = DEFAULTS[key]
# Booleans must be checked before int (bool is a subclass of int).
if isinstance(default, bool):
if isinstance(value, bool):
return value
return default
if isinstance(default, int):
try:
if isinstance(value, bool):
return default
return int(value)
except (TypeError, ValueError):
return default
if isinstance(default, str):
if isinstance(value, str):
return value
return default
return value
class Settings:
def __init__(self) -> None:
self._values: dict[str, Any] = dict(DEFAULTS)
self._path = _config_path()
self._load()
def _load(self) -> None:
try:
with open(self._path, "r", encoding="utf-8") as fh:
data = json.load(fh)
except (OSError, ValueError):
return
if not isinstance(data, dict):
return
for key, raw in data.items():
if key not in DEFAULTS:
continue
self._values[key] = _coerce(key, raw)
def get(self, key: str) -> Any:
return self._values.get(key, DEFAULTS.get(key))
def set(self, key: str, value: Any) -> None:
if key not in DEFAULTS:
return
self._values[key] = _coerce(key, value)
def save(self) -> None:
try:
self._path.parent.mkdir(parents=True, exist_ok=True)
tmp = self._path.with_suffix(self._path.suffix + ".tmp")
with open(tmp, "w", encoding="utf-8") as fh:
json.dump(self._values, fh, indent=2)
fh.flush()
os.fsync(fh.fileno())
os.replace(tmp, self._path)
except OSError:
return
def update(self, **kwargs: Any) -> None:
for key, value in kwargs.items():
self.set(key, value)
self.save()
+82
View File
@@ -0,0 +1,82 @@
"""Sortierung von Dateieinträgen.
Dieses Modul stellt einen :class:`Gtk.Sorter` bereit, der eine Liste von
``FileItem``-Objekten (siehe :mod:`taninux.files.model`) ordnet. Die Sortierung
ist rein und seiteneffektfrei: ein :class:`SortSpec` beschreibt den Schlüssel,
die Richtung und ob Ordner zuerst erscheinen.
"""
from dataclasses import dataclass
import gi
gi.require_version("Gtk", "4.0")
from gi.repository import Gtk
# sort keys
KEY_NAME = "name"
KEY_SIZE = "size"
KEY_MTIME = "mtime"
KEY_TYPE = "type"
KEYS = (KEY_NAME, KEY_SIZE, KEY_MTIME, KEY_TYPE) # iteration order for a dropdown
LABELS = {KEY_NAME: "Name", KEY_SIZE: "Size", KEY_MTIME: "Modified", KEY_TYPE: "Type"}
@dataclass
class SortSpec:
key: str = KEY_NAME # one of KEYS
descending: bool = False
folders_first: bool = True
def _cmp(a, b) -> int:
"""Vergleicht zwei vergleichbare Werte und liefert -1/0/1."""
if a < b:
return -1
if a > b:
return 1
return 0
def _name_key(item) -> str:
return (item.name or "").casefold()
def make_sorter(spec: SortSpec) -> Gtk.Sorter:
"""Erzeugt einen :class:`Gtk.CustomSorter` gemäß ``spec``."""
key = spec.key
descending = spec.descending
folders_first = spec.folders_first
def compare(a, b, *args) -> int:
# 1) Ordner vor Dateien -- gewinnt immer, unabhängig von `descending`.
if folders_first and a.is_dir != b.is_dir:
return -1 if a.is_dir else 1
# 2) Vergleich nach Schlüssel.
if key == KEY_SIZE:
result = _cmp(a.size, b.size)
elif key == KEY_MTIME:
result = _cmp(a.mtime, b.mtime)
elif key == KEY_TYPE:
result = _cmp((a.content_type or "").casefold(),
(b.content_type or "").casefold())
if result == 0:
# Gleichstand beim Typ -> nach Name (immer aufsteigend).
result = _cmp(_name_key(a), _name_key(b))
else: # KEY_NAME (Standard)
result = _cmp(_name_key(a), _name_key(b))
# Nur den Schlüsselvergleich invertieren.
if descending:
result = -result
if result != 0:
return result
# 3) Finaler Tie-Break: Name aufsteigend, nie von `descending` betroffen.
return _cmp(_name_key(a), _name_key(b))
return Gtk.CustomSorter.new(compare, None)
+262
View File
@@ -0,0 +1,262 @@
"""Asynchroner Thumbnail-Generator fuer den Dateimanager.
Laedt und skaliert Bilder in einem Hintergrund-Thread und liefert eine
``Gdk.Texture`` auf der GLib-Hauptschleife zurueck. Die Texture-Erzeugung
und jeglicher GTK-Zugriff finden ausschliesslich im Hauptthread statt; im
Worker-Thread wird lediglich der ``GdkPixbuf.Pixbuf`` erzeugt.
Zusaetzlich wird ein persistenter On-Disk-Cache nach der freedesktop
Thumbnail-Spezifikation gefuehrt (``$XDG_CACHE_HOME/thumbnails/large``),
damit Thumbnails Neustarts ueberleben und mit anderen Dateimanagern
interoperabel sind. Saemtliche Disk-IO laeuft im Worker-Thread.
"""
from __future__ import annotations
import hashlib
import os
import tempfile
import threading
from collections import OrderedDict
from pathlib import Path
from typing import Callable
import gi
gi.require_version("Gdk", "4.0")
gi.require_version("GdkPixbuf", "2.0")
gi.require_version("Gtk", "4.0")
from gi.repository import Gdk, GdkPixbuf, Gio, GLib # noqa: E402
# Maximale Anzahl gecachter Textures, bevor die aeltesten verworfen werden.
_CACHE_CAP = 512
class Thumbnailer:
"""Erzeugt und cached Bild-Thumbnails als ``Gdk.Texture``."""
def __init__(self, size: int = 256, max_bytes: int = 32 * 1024 * 1024) -> None:
# size = maximale Breite/Hoehe des Thumbnails (Seitenverhaeltnis bleibt erhalten).
# max_bytes = Bilddateien groesser als dieser Wert werden uebersprungen.
self.size = int(size)
self.max_bytes = int(max_bytes)
# Cache: (str(path), mtime_int) -> Gdk.Texture, aeltester Eintrag zuerst.
self._cache: "OrderedDict[tuple[str, int], Gdk.Texture]" = OrderedDict()
# Schluessel, fuer die gerade ein Worker laeuft (Doppel-Requests vermeiden).
self._inflight: set[tuple[str, int]] = set()
self._lock = threading.Lock()
# On-Disk-Cache-Verzeichnis (freedesktop Thumbnail-Spec, 256px-Bucket).
self._disk_dir = Path(GLib.get_user_cache_dir()) / "thumbnails" / "large"
# -- Hilfsfunktionen ----------------------------------------------------
def _mtime_int(self, path: Path) -> "int | None":
try:
return int(path.stat().st_mtime)
except OSError:
return None
def _cache_get(self, key: tuple[str, int]) -> "Gdk.Texture | None":
with self._lock:
tex = self._cache.get(key)
if tex is not None:
# Als zuletzt benutzt markieren.
self._cache.move_to_end(key)
return tex
def _cache_put(self, key: tuple[str, int], tex: "Gdk.Texture") -> None:
with self._lock:
self._cache[key] = tex
self._cache.move_to_end(key)
while len(self._cache) > _CACHE_CAP:
self._cache.popitem(last=False)
# -- On-Disk-Cache (freedesktop) ---------------------------------------
def _file_uri(self, path: Path) -> str:
# Kanonische file://-URI fuer den gegebenen Pfad.
return Gio.File.new_for_path(str(path)).get_uri()
def _disk_path(self, uri: str) -> Path:
# md5(uri).hexdigest() + ".png" gemaess Spezifikation.
digest = hashlib.md5(uri.encode("utf-8")).hexdigest()
return self._disk_dir / f"{digest}.png"
def _disk_load(self, path: Path, mtime: int) -> "GdkPixbuf.Pixbuf | None":
# Laedt das On-Disk-Thumbnail nur, wenn Thumb::MTime mit der Quelle
# uebereinstimmt. Laeuft im Worker-Thread. Wirft niemals.
try:
uri = self._file_uri(path)
disk = self._disk_path(uri)
if not disk.exists():
return None
pixbuf = GdkPixbuf.Pixbuf.new_from_file(str(disk))
stored = pixbuf.get_option("tEXt::Thumb::MTime")
if stored is None:
stored = pixbuf.get_option("Thumb::MTime")
if stored is None or str(stored) != str(int(mtime)):
return None
return pixbuf
except Exception:
return None
def _disk_save(self, path: Path, mtime: int, pixbuf: "GdkPixbuf.Pixbuf") -> None:
# Schreibt das Thumbnail mit eingebetteten Thumb::URI/Thumb::MTime-Keys
# atomar in den Cache. Laeuft im Worker-Thread. Wirft niemals.
try:
uri = self._file_uri(path)
disk = self._disk_path(uri)
try:
self._disk_dir.mkdir(mode=0o700, parents=True, exist_ok=True)
except OSError:
return
fd, tmp_name = tempfile.mkstemp(suffix=".png", dir=str(self._disk_dir))
os.close(fd)
try:
pixbuf.savev(
tmp_name,
"png",
["tEXt::Thumb::URI", "tEXt::Thumb::MTime"],
[uri, str(int(mtime))],
)
os.replace(tmp_name, str(disk))
except Exception:
try:
os.unlink(tmp_name)
except OSError:
pass
except Exception:
pass
# -- Oeffentliche API ---------------------------------------------------
def can_thumbnail(self, path: Path, content_type: str) -> bool:
# Nur fuer Raster-Bild-Content-Types, deren Dateigroesse <= max_bytes ist.
if not content_type or not content_type.startswith("image/"):
return False
try:
return path.stat().st_size <= self.max_bytes
except OSError:
return False
def get_cached(self, path: Path) -> "Gdk.Texture | None":
# Liefert eine gecachte Texture fuer (path, mtime), sonst None.
# SYNCHRON, nur Speicher -- keinerlei Disk-IO.
mtime = self._mtime_int(path)
if mtime is None:
return None
return self._cache_get((str(path), mtime))
def request(
self,
path: Path,
content_type: str,
callback: Callable[["Gdk.Texture | None"], None],
) -> None:
# Cache-Treffer -> sofort (synchron) zuruecksenden.
cached = self.get_cached(path)
if cached is not None:
try:
callback(cached)
except Exception:
pass
return
if not self.can_thumbnail(path, content_type):
try:
callback(None)
except Exception:
pass
return
mtime = self._mtime_int(path)
if mtime is None:
try:
callback(None)
except Exception:
pass
return
key = (str(path), mtime)
# Bereits in Arbeit -> nicht erneut starten, aber Callback bedienen.
with self._lock:
if key in self._inflight:
start = False
else:
self._inflight.add(key)
start = True
if not start:
# Ein anderer Request laeuft schon; Cache wird gefuellt, hier nur
# bestmoeglich nachreichen.
GLib.idle_add(self._deliver_later, key, callback)
return
thread = threading.Thread(
target=self._worker,
args=(path, mtime, key, callback),
daemon=True,
)
thread.start()
# -- Interna ------------------------------------------------------------
def _worker(
self,
path: Path,
mtime: int,
key: tuple[str, int],
callback: Callable[["Gdk.Texture | None"], None],
) -> None:
pixbuf = None
from_disk = False
try:
# Zuerst den On-Disk-Cache pruefen (Thumb::MTime muss passen).
pixbuf = self._disk_load(path, mtime)
if pixbuf is not None:
from_disk = True
else:
pixbuf = GdkPixbuf.Pixbuf.new_from_file_at_scale(
str(path), self.size, self.size, True
)
except Exception:
pixbuf = None
# Frisch generierte Thumbnails persistent ablegen (best-effort).
if pixbuf is not None and not from_disk:
self._disk_save(path, mtime, pixbuf)
# Finisher im Hauptthread: Texture erzeugen, cachen, Callback aufrufen.
GLib.idle_add(self._finish, key, pixbuf, callback)
def _finish(
self,
key: tuple[str, int],
pixbuf: "GdkPixbuf.Pixbuf | None",
callback: Callable[["Gdk.Texture | None"], None],
) -> bool:
tex = None
try:
if pixbuf is not None:
tex = Gdk.Texture.new_for_pixbuf(pixbuf)
self._cache_put(key, tex)
except Exception:
tex = None
finally:
with self._lock:
self._inflight.discard(key)
try:
callback(tex)
except Exception:
pass
return False # nur einmal ausfuehren
def _deliver_later(
self,
key: tuple[str, int],
callback: Callable[["Gdk.Texture | None"], None],
) -> bool:
try:
callback(self._cache_get(key))
except Exception:
pass
return False
+183
View File
@@ -0,0 +1,183 @@
"""Trash-Ansicht für den Dateimanager.
Listet Einträge aus dem Papierkorb (``trash:///``) auf und erlaubt das
Wiederherstellen an den ursprünglichen Ort, das endgültige Löschen einzelner
Einträge sowie das vollständige Leeren des Papierkorbs. Alle schreibenden
Operationen laufen in einem Hintergrund-Thread und melden ihr Ergebnis über
``GLib.idle_add`` zurück an den GTK-Mainloop.
"""
from __future__ import annotations
import threading
from dataclasses import dataclass
from typing import Callable, Iterable, List
import gi
gi.require_version("Gtk", "4.0")
from gi.repository import Gio, GLib # noqa: E402
_ATTRS = (
"standard::name,standard::display-name,standard::icon,standard::type,"
"standard::size,standard::content-type,trash::orig-path,trash::deletion-date"
)
@dataclass
class TrashEntry:
name: str
icon: object
size: int
is_dir: bool
content_type: str
orig_path: str
deletion_date: str
gfile: object
def _idle(on_done: Callable[[int, List[str]], None], count: int, errors: List[str]) -> None:
GLib.idle_add(lambda: (on_done(count, errors), False)[1])
def list_trash() -> List[TrashEntry]:
"""Enumeriert den Papierkorb. Wirft nie -> bei Fehler leere Liste."""
entries: List[TrashEntry] = []
try:
tdir = Gio.File.new_for_uri("trash:///")
en = tdir.enumerate_children(_ATTRS, Gio.FileQueryInfoFlags.NONE, None)
except GLib.Error:
return entries
while True:
try:
info = en.next_file(None)
except GLib.Error:
break
if info is None:
break
try:
name = info.get_name()
display = info.get_display_name() or name
icon = info.get_icon()
try:
size = info.get_size()
except Exception:
size = 0
is_dir = info.get_file_type() == Gio.FileType.DIRECTORY
content_type = info.get_content_type() or ""
orig = info.get_attribute_byte_string("trash::orig-path") or ""
ddate = info.get_attribute_string("trash::deletion-date") or ""
gfile = tdir.get_child(name)
entries.append(
TrashEntry(
name=display,
icon=icon,
size=size or 0,
is_dir=is_dir,
content_type=content_type,
orig_path=orig,
deletion_date=ddate,
gfile=gfile,
)
)
except Exception:
continue
try:
en.close(None)
except GLib.Error:
pass
return entries
def _ensure_parent(orig_path: str) -> None:
"""Legt das Elternverzeichnis von *orig_path* an (ignoriert "existiert")."""
try:
parent = Gio.File.new_for_path(orig_path).get_parent()
if parent is None:
return
try:
parent.make_directory_with_parents(None)
except GLib.Error as exc:
if not exc.matches(Gio.io_error_quark(), Gio.IOErrorEnum.EXISTS):
raise
except GLib.Error:
pass
def restore(entries: Iterable[TrashEntry],
on_done: Callable[[int, List[str]], None]) -> None:
"""Stellt Einträge an ihren Ursprungsort wieder her (Hintergrund-Thread)."""
items = list(entries)
def work() -> None:
restored = 0
errors: List[str] = []
for e in items:
try:
orig = e.orig_path
if not orig:
errors.append(f"{e.name}: unbekannter Ursprungsort")
continue
_ensure_parent(orig)
dest = Gio.File.new_for_path(orig)
e.gfile.move(
dest,
Gio.FileCopyFlags.NOFOLLOW_SYMLINKS | Gio.FileCopyFlags.ALL_METADATA,
None,
None,
None,
)
restored += 1
except GLib.Error as exc:
errors.append(f"{e.name}: {exc.message}")
except Exception as exc: # pragma: no cover - defensive
errors.append(f"{e.name}: {exc}")
_idle(on_done, restored, errors)
threading.Thread(target=work, daemon=True).start()
def delete(entries: Iterable[TrashEntry],
on_done: Callable[[int, List[str]], None]) -> None:
"""Löscht Einträge endgültig aus dem Papierkorb (Hintergrund-Thread)."""
items = list(entries)
def work() -> None:
deleted = 0
errors: List[str] = []
for e in items:
try:
e.gfile.delete(None)
deleted += 1
except GLib.Error as exc:
errors.append(f"{e.name}: {exc.message}")
except Exception as exc: # pragma: no cover - defensive
errors.append(f"{e.name}: {exc}")
_idle(on_done, deleted, errors)
threading.Thread(target=work, daemon=True).start()
def empty(on_done: Callable[[int, List[str]], None]) -> None:
"""Leert den gesamten Papierkorb (Hintergrund-Thread)."""
def work() -> None:
deleted = 0
errors: List[str] = []
try:
items = list_trash()
except Exception:
items = []
for e in items:
try:
e.gfile.delete(None)
deleted += 1
except GLib.Error as exc:
errors.append(f"{e.name}: {exc.message}")
except Exception as exc: # pragma: no cover - defensive
errors.append(f"{e.name}: {exc}")
_idle(on_done, deleted, errors)
threading.Thread(target=work, daemon=True).start()
+137
View File
@@ -0,0 +1,137 @@
"""Undo-Stack fuer Dateioperationen.
Selbststaendiges Modul: zeichnet ausgefuehrte Operationen auf und kehrt sie
auf Ctrl+Z um. Verschieben wird im Hintergrund-Thread rueckgaengig gemacht;
das Ergebnis wird ueber GLib.idle_add im Hauptloop zurueckgemeldet.
"""
from __future__ import annotations
import os
import shutil
import threading
from dataclasses import dataclass, field
from pathlib import Path
from typing import Callable, List, Tuple
import gi
gi.require_version("Gtk", "4.0")
from gi.repository import Gio, GLib # noqa: E402
@dataclass
class _Op:
kind: str # "move" | "copy" | "rename" | "create"
label: str
pairs: List[Tuple[Path, Path]] = field(default_factory=list) # move/rename
paths: List[Path] = field(default_factory=list) # copy/create
def _trash_or_delete(path: Path) -> None:
"""Send a path to trash; fall back to permanent delete. May raise."""
try:
Gio.File.new_for_path(str(path)).trash(None)
return
except Exception:
pass
# Fallback: permanent delete.
if path.is_dir() and not path.is_symlink():
shutil.rmtree(str(path))
else:
path.unlink()
class UndoStack:
def __init__(self) -> None:
self._stack: List[_Op] = []
# -- recording (called AFTER an operation succeeds) --
def record_move(self, pairs: List[Tuple[Path, Path]]) -> None:
if not pairs:
return
self._stack.append(
_Op(kind="move", label="Undo Move", pairs=[(Path(o), Path(n)) for o, n in pairs])
)
def record_copy(self, created: List[Path]) -> None:
if not created:
return
self._stack.append(
_Op(kind="copy", label="Undo Copy", paths=[Path(p) for p in created])
)
def record_rename(self, old_path: Path, new_path: Path) -> None:
self._stack.append(
_Op(kind="rename", label="Undo Rename", pairs=[(Path(old_path), Path(new_path))])
)
def record_create(self, path: Path) -> None:
self._stack.append(
_Op(kind="create", label="Undo Create", paths=[Path(path)])
)
# -- query --
def can_undo(self) -> bool:
return bool(self._stack)
def describe_next(self) -> str:
return self._stack[-1].label if self._stack else ""
# -- perform --
def undo(self, on_done: Callable[[str], None]) -> None:
if not self._stack:
try:
on_done("")
except Exception:
pass
return
op = self._stack.pop()
def _finish(message: str) -> None:
def _cb() -> bool:
try:
on_done(message)
except Exception:
pass
return False
GLib.idle_add(_cb)
def _work() -> None:
errors = 0
try:
if op.kind == "move":
for original, new in op.pairs:
try:
if not new.exists():
continue
shutil.move(str(new), str(original))
except Exception:
errors += 1
elif op.kind == "rename":
for old_path, new_path in op.pairs:
try:
if not new_path.exists():
continue
os.rename(str(new_path), str(old_path))
except Exception:
errors += 1
elif op.kind in ("copy", "create"):
for p in op.paths:
try:
if not (p.exists() or p.is_symlink()):
continue
_trash_or_delete(p)
except Exception:
errors += 1
except Exception:
errors += 1
if errors:
_finish(f"Undo failed for {errors} item(s)")
else:
_finish("")
threading.Thread(target=_work, daemon=True).start()
+135
View File
@@ -0,0 +1,135 @@
"""Eingehängte Laufwerke für die Sidebar — reich aufgelöst und auswerfbar.
Anders als ``places.py`` behält dieses Modul das ``Gio.Mount``-Handle, damit
die UI ein Laufwerk später aushängen oder auswerfen kann.
"""
from __future__ import annotations
import logging
from dataclasses import dataclass
from pathlib import Path
from typing import Callable, List
import gi
gi.require_version("Gtk", "4.0")
from gi.repository import Gio, GLib # noqa: E402
logger = logging.getLogger(__name__)
@dataclass
class MountedVolume:
name: str
path: Path
icon: str # icon name, e.g. "drive-removable-media-symbolic" / "drive-harddisk-symbolic"
can_eject: bool # True if the mount can be ejected or unmounted
mount: object # the Gio.Mount handle (kept for unmount/eject)
def _is_removable(mount) -> bool:
"""Best-effort: ist das zugrundeliegende Laufwerk entfernbar? Alles None-fest."""
try:
vol = mount.get_volume()
if vol is None:
return False
drive = vol.get_drive()
if drive is None:
return False
if drive.is_removable():
return True
if drive.is_media_removable():
return True
except Exception: # noqa: BLE001
logger.debug("Removable-Erkennung fehlgeschlagen", exc_info=True)
return False
def list_volumes() -> List[MountedVolume]:
"""Eingehängte Laufwerke mit lokalem Pfad — reich aufgelöst. Wirft nie."""
out: List[MountedVolume] = []
try:
monitor = Gio.VolumeMonitor.get()
except Exception: # noqa: BLE001
logger.debug("VolumeMonitor nicht verfügbar", exc_info=True)
return out
try:
mounts = monitor.get_mounts()
except Exception: # noqa: BLE001
logger.debug("get_mounts() fehlgeschlagen", exc_info=True)
return out
for mount in mounts:
try:
if mount.is_shadowed():
continue
root = mount.get_root()
path = root.get_path() if root is not None else None
if not path:
continue
can_eject = bool(mount.can_eject() or mount.can_unmount())
icon = (
"drive-removable-media-symbolic"
if _is_removable(mount)
else "drive-harddisk-symbolic"
)
out.append(
MountedVolume(
name=mount.get_name(),
path=Path(path),
icon=icon,
can_eject=can_eject,
mount=mount,
)
)
except Exception: # noqa: BLE001
logger.debug("Mount übersprungen", exc_info=True)
return out
def eject_or_unmount(mount, on_done: Callable[[bool, str], None]) -> None:
"""Auswerfen (falls möglich), sonst aushängen — asynchron. Wirft nie.
``on_done(ok, message)`` wird in der GLib-Hauptschleife aufgerufen; bei
Erfolg ist ``message`` "".
"""
def _eject_finish(_mount, result) -> None:
try:
mount.eject_with_operation_finish(result)
on_done(True, "")
except GLib.Error as err:
on_done(False, err.message)
except Exception as err: # noqa: BLE001
on_done(False, str(err))
def _unmount_finish(_mount, result) -> None:
try:
mount.unmount_with_operation_finish(result)
on_done(True, "")
except GLib.Error as err:
on_done(False, err.message)
except Exception as err: # noqa: BLE001
on_done(False, str(err))
try:
if mount.can_eject():
mount.eject_with_operation(
Gio.MountUnmountFlags.NONE,
Gio.MountOperation(),
None,
_eject_finish,
)
elif mount.can_unmount():
mount.unmount_with_operation(
Gio.MountUnmountFlags.NONE,
Gio.MountOperation(),
None,
_unmount_finish,
)
else:
on_done(False, "Cannot eject")
except Exception as err: # noqa: BLE001
logger.debug("eject_or_unmount fehlgeschlagen", exc_info=True)
on_done(False, str(err))
File diff suppressed because it is too large Load Diff