feat: Scan, Namenskollisionen, Buchführung und Watchdog
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 <noreply@anthropic.com>
This commit is contained in:
Florian Wilde
co-authored by
Claude Opus 5.5
Claude Opus 5.5
parent
e5ed8c4d88
commit
c3a42ce6f5
@@ -0,0 +1 @@
|
||||
"""Mover: Dateien finden, auf Stabilität prüfen und sicher ans Ziel bringen."""
|
||||
@@ -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
|
||||
)
|
||||
)
|
||||
@@ -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}")
|
||||
@@ -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)
|
||||
@@ -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]
|
||||
|
||||
Reference in New Issue
Block a user