From c3a42ce6f5cee9fa563a46d337c434e292c0fd2c Mon Sep 17 00:00:00 2001 From: Florian Wilde Date: Sat, 10 Oct 2026 19:06:14 +0200 Subject: [PATCH] =?UTF-8?q?feat:=20Scan,=20Namenskollisionen,=20Buchf?= =?UTF-8?q?=C3=BChrung=20und=20Watchdog?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Kandidatensuche mit Filtern und Ausschlüssen (versteckt, unfertig, Synology-Ordner, Symlinks), Kollisionsnamen, Ledger für copied_files und file_failures, Inaktivitäts-Watchdog mit CancelToken. Co-Authored-By: Claude Opus 5.5 --- src/netmover/mover/__init__.py | 1 + src/netmover/mover/ledger.py | 108 ++++++++++++++++++++++++++++++++ src/netmover/mover/naming.py | 22 +++++++ src/netmover/mover/scan.py | 110 +++++++++++++++++++++++++++++++++ src/netmover/timeouts.py | 74 ++++++++++++++++++++++ 5 files changed, 315 insertions(+) create mode 100644 src/netmover/mover/__init__.py create mode 100644 src/netmover/mover/ledger.py create mode 100644 src/netmover/mover/naming.py create mode 100644 src/netmover/mover/scan.py diff --git a/src/netmover/mover/__init__.py b/src/netmover/mover/__init__.py new file mode 100644 index 0000000..ae53bcb --- /dev/null +++ b/src/netmover/mover/__init__.py @@ -0,0 +1 @@ +"""Mover: Dateien finden, auf Stabilität prüfen und sicher ans Ziel bringen.""" diff --git a/src/netmover/mover/ledger.py b/src/netmover/mover/ledger.py new file mode 100644 index 0000000..09c3767 --- /dev/null +++ b/src/netmover/mover/ledger.py @@ -0,0 +1,108 @@ +"""Buchführung in der Datenbank. + +* ``copied_files`` (Fallstrick 11): Welche Dateiversion (Pfad, Größe, mtime) wurde + schon übertragen? Verhindert endloses Neu-Kopieren im Kopiermodus – und im + Verschiebemodus Duplikate, wenn die Quelle nicht gelöscht werden konnte. +* ``file_failures``: Fehlversuche pro Datei als Grundlage der Quarantäne. +""" + +from __future__ import annotations + +from dataclasses import dataclass + +from sqlalchemy import delete, select +from sqlalchemy.orm import Session, sessionmaker + +from netmover.db import session_scope +from netmover.models import CopiedFile, FileFailure, utcnow + +CopyKey = tuple[str, int, int] # (relpath, size, mtime_ns) + + +@dataclass(frozen=True) +class FailureInfo: + attempts: int + size: int + mtime_ns: int + + +class Ledger: + def __init__(self, session_factory: sessionmaker[Session]) -> None: + self.session_factory = session_factory + + def copied_keys(self, rule_id: int) -> set[CopyKey]: + with session_scope(self.session_factory) as s: + rows = s.execute( + select(CopiedFile.src_relpath, CopiedFile.size, CopiedFile.mtime_ns).where( + CopiedFile.rule_id == rule_id + ) + ).all() + return {(r[0], r[1], r[2]) for r in rows} + + def record_copy(self, rule_id: int, relpath: str, size: int, mtime_ns: int, dst: str) -> None: + with session_scope(self.session_factory) as s: + exists = s.scalar( + select(CopiedFile.id).where( + CopiedFile.rule_id == rule_id, + CopiedFile.src_relpath == relpath, + CopiedFile.size == size, + CopiedFile.mtime_ns == mtime_ns, + ) + ) + if exists is None: + s.add( + CopiedFile( + rule_id=rule_id, + src_relpath=relpath, + size=size, + mtime_ns=mtime_ns, + dst_path=dst, + ) + ) + + def prune_copied(self, rule_id: int, present: set[str]) -> int: + """Entfernt Einträge, deren Quelldatei verschwunden ist. Nur nach einem + *vollständigen* Scan aufrufen – sonst würde alles erneut kopiert.""" + with session_scope(self.session_factory) as s: + rows = s.execute( + select(CopiedFile.id, CopiedFile.src_relpath).where(CopiedFile.rule_id == rule_id) + ).all() + stale = [r[0] for r in rows if r[1] not in present] + if stale: + s.execute(delete(CopiedFile).where(CopiedFile.id.in_(stale))) + return len(stale) + + def failures(self, rule_id: int) -> dict[str, FailureInfo]: + with session_scope(self.session_factory) as s: + rows = s.scalars(select(FileFailure).where(FileFailure.rule_id == rule_id)).all() + return {r.src_relpath: FailureInfo(r.attempts, r.size, r.mtime_ns) for r in rows} + + def record_failure( + self, rule_id: int, relpath: str, size: int, mtime_ns: int, error: str + ) -> int: + with session_scope(self.session_factory) as s: + row = s.scalar( + select(FileFailure).where( + FileFailure.rule_id == rule_id, FileFailure.src_relpath == relpath + ) + ) + if row is None: + row = FileFailure( + rule_id=rule_id, src_relpath=relpath, size=size, mtime_ns=mtime_ns, attempts=0 + ) + s.add(row) + elif (row.size, row.mtime_ns) != (size, mtime_ns): + # neue Dateiversion – Zählung beginnt von vorn + row.size, row.mtime_ns, row.attempts = size, mtime_ns, 0 + row.attempts += 1 + row.last_error = error + row.last_attempt_at = utcnow() + return row.attempts + + def clear_failure(self, rule_id: int, relpath: str) -> None: + with session_scope(self.session_factory) as s: + s.execute( + delete(FileFailure).where( + FileFailure.rule_id == rule_id, FileFailure.src_relpath == relpath + ) + ) diff --git a/src/netmover/mover/naming.py b/src/netmover/mover/naming.py new file mode 100644 index 0000000..491a316 --- /dev/null +++ b/src/netmover/mover/naming.py @@ -0,0 +1,22 @@ +"""Namenskollisionen (Fallstrick 4): ``scan001.pdf`` → ``scan001-1.pdf`` → … + +Die Kandidaten werden nur erzeugt; ob ein Name frei ist, entscheidet jedes Mal +das Ziel-Dateisystem (kein Vorhalten im Speicher). +""" + +from __future__ import annotations + +import posixpath +from collections.abc import Iterator + +MAX_SUFFIX = 9999 + + +def collision_candidates(relpath: str) -> Iterator[str]: + yield relpath + directory, name = posixpath.split(relpath) + stem, dot, ext = name.rpartition(".") + if not stem: # kein Punkt bzw. nur führender Punkt + stem, dot, ext = name, "", "" + for n in range(1, MAX_SUFFIX + 1): + yield posixpath.join(directory, f"{stem}-{n}{dot}{ext}") diff --git a/src/netmover/mover/scan.py b/src/netmover/mover/scan.py new file mode 100644 index 0000000..ca57c2c --- /dev/null +++ b/src/netmover/mover/scan.py @@ -0,0 +1,110 @@ +"""Kandidaten im Quellordner finden (Fallstrick 1, erster Teil). + +Übersprungen werden immer: +* versteckte Dateien und Ordner (führender Punkt) – darunter auch unsere eigenen + ``.netmover-*.part``-Temporärdateien, +* typische Endungen unfertiger Übertragungen, +* Symlinks und alles, was keine reguläre Datei ist, +* Synology-Verwaltungsordner (``@eaDir``, ``#recycle``, ``#snapshot``). +""" + +from __future__ import annotations + +import fnmatch +import os +import re +import stat +from collections.abc import Iterator +from dataclasses import dataclass +from pathlib import Path + +from netmover.models import PatternType + +PARTIAL_SUFFIXES = (".part", ".tmp", ".crdownload", ".filepart", ".!ut") +IGNORED_DIRS = frozenset({"@eaDir", "#recycle", "#snapshot", "@tmp", "@sharebin"}) + + +@dataclass(frozen=True) +class FileState: + size: int + mtime_ns: int + + +@dataclass(frozen=True) +class Candidate: + relpath: str # POSIX, relativ zum Quellordner + path: Path + state: FileState + + +def is_ignored_name(name: str) -> bool: + return name.startswith(".") or name.lower().endswith(PARTIAL_SUFFIXES) + + +class Matcher: + """Glob: gegen den Dateinamen, ohne Groß-/Kleinschreibung, mehrere Muster mit + „;“ getrennt (``*.pdf;*.jpg``). Regex: ``re.search`` gegen den relativen Pfad.""" + + def __init__(self, pattern: str, pattern_type: PatternType) -> None: + self.pattern_type = pattern_type + if pattern_type is PatternType.REGEX: + self._regex = re.compile(pattern) + self._globs: list[str] = [] + else: + self._globs = [p.strip().lower() for p in pattern.split(";") if p.strip()] or ["*"] + + def __call__(self, relpath: str) -> bool: + if self.pattern_type is PatternType.REGEX: + return self._regex.search(relpath) is not None + name = relpath.rsplit("/", 1)[-1].lower() + return any(fnmatch.fnmatchcase(name, g) for g in self._globs) + + +def validate_pattern(pattern: str, pattern_type: PatternType) -> None: + if pattern_type is PatternType.REGEX: + re.compile(pattern) + + +def scan( + root: Path, + matcher: Matcher, + recursive: bool, + exclude: frozenset[Path] = frozenset(), +) -> Iterator[Candidate]: + """Liefert passende reguläre Dateien. ``exclude``: Verzeichnisse (realpath), die + nicht betreten werden – z. B. Quarantäne oder ein Ziel innerhalb der Quelle.""" + + def walk(directory: Path, prefix: str) -> Iterator[Candidate]: + with os.scandir(directory) as it: + entries = sorted(it, key=lambda e: e.name) + for entry in entries: + if entry.name.startswith("."): + continue + rel = f"{prefix}{entry.name}" + if entry.is_dir(follow_symlinks=False): + if ( + recursive + and entry.name not in IGNORED_DIRS + and Path(entry.path).resolve() not in exclude + ): + yield from walk(Path(entry.path), f"{rel}/") + continue + if is_ignored_name(entry.name): + continue + st = entry.stat(follow_symlinks=False) + if not stat.S_ISREG(st.st_mode): + continue + if matcher(rel): + yield Candidate(rel, Path(entry.path), FileState(st.st_size, st.st_mtime_ns)) + + yield from walk(root, "") + + +def current_state(path: Path) -> FileState | None: + try: + st = os.stat(path, follow_symlinks=False) + except FileNotFoundError: + return None + if not stat.S_ISREG(st.st_mode): + return None + return FileState(st.st_size, st.st_mtime_ns) diff --git a/src/netmover/timeouts.py b/src/netmover/timeouts.py index 17d7441..172d131 100644 --- a/src/netmover/timeouts.py +++ b/src/netmover/timeouts.py @@ -10,6 +10,7 @@ Regeln. Bewusst kein Thread-Pool – hängende Threads würden ihn auffüllen. from __future__ import annotations import threading +import time from collections.abc import Callable from netmover.logging_setup import event, get_logger @@ -56,3 +57,76 @@ def run_with_timeout[T](fn: Callable[[], T], seconds: float, what: str) -> T: if error: raise error[0] return result[0] + + +class Cancelled(Exception): + """Lauf wurde abgebrochen (Zeitlimit oder Dienstende).""" + + +class CancelToken: + """Abbruch-Flag plus Lebenszeichen für :func:`run_with_watchdog`. + + Der Worker ruft :meth:`check` regelmäßig auf (pro Datei, pro Datenblock). + Das meldet Fortschritt und wirft :class:`Cancelled`, sobald abgebrochen wurde. + """ + + def __init__(self) -> None: + self._event = threading.Event() + self.last_beat = time.monotonic() + + def beat(self) -> None: + self.last_beat = time.monotonic() + + def cancel(self) -> None: + self._event.set() + + @property + def cancelled(self) -> bool: + return self._event.is_set() + + def check(self) -> None: + if self._event.is_set(): + raise Cancelled + self.beat() + + +def run_with_watchdog[T]( + fn: Callable[[CancelToken], T], + idle_timeout: float, + what: str, + *, + on_finish: Callable[[], None] | None = None, + poll: float = 0.1, +) -> T: + """Wie :func:`run_with_timeout`, aber das Zeitlimit gilt für *Inaktivität*: + Ein großer Transfer, der stetig Fortschritt meldet, wird nicht abgebrochen, + ein hängender Dateizugriff schon. ``on_finish`` läuft im Worker-Thread, wenn + dieser wirklich endet – auch lange nach einer Zeitüberschreitung.""" + token = CancelToken() + result: list[T] = [] + error: list[BaseException] = [] + done = threading.Event() + + def target() -> None: + try: + result.append(fn(token)) + except BaseException as exc: + error.append(exc) + finally: + done.set() + if on_finish is not None: + on_finish() + + thread = threading.Thread(target=target, name=f"nm-worker:{what}", daemon=True) + thread.start() + while not done.wait(poll): + if time.monotonic() - token.last_beat > idle_timeout: + token.cancel() + global _abandoned + with _abandoned_lock: + _abandoned += 1 + event(log, "worker.timeout", level=30, what=what, idle_seconds=idle_timeout) + raise OperationTimeout(what, idle_timeout) + if error: + raise error[0] + return result[0]