Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
660bb70c5d | ||
|
|
e59932ab75 |
+15
-1
@@ -8,6 +8,19 @@ Die Version steht an genau einer Stelle: `src/netmover/__init__.py`.
|
||||
|
||||
## [Unreleased]
|
||||
|
||||
## [0.7.0] – 2026-10-12
|
||||
|
||||
### Hinzugefügt
|
||||
- **Archiv je Regel:** zusätzliche Kopie in ein Archiv (Anbindung SMB/NFS/SFTP oder
|
||||
lokaler Pfad), Ablage „wie in der Quelle“ oder „nach Datum“ (`Jahr/Monat/…`).
|
||||
Reihenfolge: erst Archiv, dann Ziel, **erst dann wird die Quelle gelöscht**.
|
||||
Scheitert das Ziel nach dem Archiv, verhindert eine Buchführung
|
||||
(`archived_files`) eine zweite Archivkopie. Im Archiv wird nie überschrieben.
|
||||
Nicht erreichbares Archiv → Regel wartet. Platzprüfung auch fürs Archiv,
|
||||
Probelauf zeigt das Archivziel. Migration 0004.
|
||||
- Export/Import enthalten die Archiv-Einstellungen; eine als Archiv benutzte
|
||||
Anbindung lässt sich nicht löschen.
|
||||
|
||||
## [0.6.0] – 2026-10-11
|
||||
|
||||
### Geändert
|
||||
@@ -246,7 +259,8 @@ Erste Version.
|
||||
- Löschen von Zugangsdaten, die noch von einem Mount benutzt werden, trennt den
|
||||
Mount nicht mehr stillschweigend ab, sondern wird von der Datenbank abgelehnt.
|
||||
|
||||
[Unreleased]: https://git.wilde-server.de/florian.wilde/netmover/compare/v0.6.0...HEAD
|
||||
[Unreleased]: https://git.wilde-server.de/florian.wilde/netmover/compare/v0.7.0...HEAD
|
||||
[0.7.0]: https://git.wilde-server.de/florian.wilde/netmover/compare/v0.6.0...v0.7.0
|
||||
[0.6.0]: https://git.wilde-server.de/florian.wilde/netmover/compare/v0.5.0...v0.6.0
|
||||
[0.5.0]: https://git.wilde-server.de/florian.wilde/netmover/compare/v0.4.0...v0.5.0
|
||||
[0.4.0]: https://git.wilde-server.de/florian.wilde/netmover/compare/v0.3.0...v0.4.0
|
||||
|
||||
@@ -20,10 +20,32 @@ Zeiten in Ortszeit (Europe/Berlin).
|
||||
| 2026-10-11 | 00:31 | 00:50 | 0:19 | – | Pausieren, Wartungsmodus, E-Mail (0.4.0) |
|
||||
| 2026-10-11 | 00:54 | 01:16 | 0:22 | – | SFTP-Quelle, Zugangsdaten, Anmeldung (0.5.0) |
|
||||
| 2026-10-11 | 08:43 | 08:55 | 0:12 | – | Anmeldung je Netz (0.6.0) |
|
||||
| **Summe** | | | **3:29** | | |
|
||||
| 2026-10-12 | 00:14 | 00:34 | 0:20 | – | Archiv je Regel (0.7.0) |
|
||||
| **Summe** | | | **3:49** | | |
|
||||
|
||||
---
|
||||
|
||||
## 2026-10-12 · 00:14–00:34 · Archiv je Regel (0.7.0)
|
||||
|
||||
**Erledigt**
|
||||
- Archivfelder an der Regel, Tabelle `archived_files` (Migration 0004).
|
||||
- Engine: Archiv zuerst, dann Ziel, dann Quelle löschen; Archiv immer „umbenennen“.
|
||||
- Runner: Ziel- und Archivaufbau über eine gemeinsame Fabrik (`_target`).
|
||||
- 14 Tests (350 gesamt); Mutationsprobe 4/4 erkannt (Buchführung, Archiv
|
||||
übersprungen, Kollisionsregel, Platzprüfung).
|
||||
|
||||
**Entscheidungen**
|
||||
- Archiv *vor* Ziel: Paperless holt Dateien sofort ab – andersherum würde ein
|
||||
gescheitertes Archiv beim Neuversuch zur doppelten Lieferung führen.
|
||||
- Archiv überschreibt nie (unabhängig von der Kollisionsregel des Ziels).
|
||||
- Ablage „nach Datum“ nach Verarbeitungsdatum (nicht Datei-mtime) – eindeutig
|
||||
nachvollziehbar, wann eine Datei durchgelaufen ist.
|
||||
|
||||
**Stolperstein**
|
||||
- Downgrade von 0004 scheiterte an den CHECK-Regeln der neuen Auswahlfelder
|
||||
(verweisen nach dem Entfernen der Spalten ins Leere) – nur dank der Migrations-
|
||||
tests mit Daten aufgefallen; Downgrade entfernt sie jetzt zuerst.
|
||||
|
||||
## 2026-10-11 · 08:43–08:55 · Anmeldung je Netz (0.6.0)
|
||||
|
||||
**Anlass** – Wunsch: Anmeldung abschalten, und zwar konfigurierbar, aus welchen Netzen
|
||||
|
||||
@@ -28,6 +28,7 @@ Legende: ✅ erledigt · 🔨 in Arbeit · 📋 geplant · 💡 Idee (noch nicht
|
||||
|
||||
| Version | Thema |
|
||||
|---|---|
|
||||
| 0.7.0 | Archiv je Regel: Kopie ins Ziel und ins Archiv, Quelle erst danach löschen |
|
||||
| 0.6.0 | Anmeldung je Netz: ohne Passwort / mit Passwort / abgewiesen, Schutz vor Aussperren |
|
||||
| 0.5.0 | SFTP als Quelle; gespeicherte Zugangsdaten für SMB/SFTP-Passwörter (Seite „Zugangsdaten“); Anmeldung abschaltbar (vertrauenswürdige Netze / ganz) |
|
||||
| 0.4.0 | Automatisch pausieren/fortsetzen bei nicht erreichbaren Anbindungen; Wartungsmodus; E-Mail-Benachrichtigung; eine Meldung pro ausgefallener Anbindung |
|
||||
|
||||
+1
-1
@@ -4,7 +4,7 @@
|
||||
services:
|
||||
netmover:
|
||||
build: .
|
||||
image: netmover:0.6.0
|
||||
image: netmover:0.7.0
|
||||
container_name: netmover
|
||||
restart: unless-stopped
|
||||
cap_add:
|
||||
|
||||
+31
-2
@@ -8,7 +8,7 @@ Diese Anleitung erklärt die Bedienung. Installation, Updates und technische Det
|
||||
stehen im [README](../README.md).
|
||||
|
||||
**Inhalt:** [Begriffe](#begriffe) · [Erste Anmeldung](#erste-anmeldung) ·
|
||||
[Zugangsdaten](#zugangsdaten) · [Anbindungen](#anbindungen) · [Regeln](#regeln) ·
|
||||
[Zugangsdaten](#zugangsdaten) · [Anbindungen](#anbindungen) · [Regeln](#regeln) · [Archiv](#archiv) ·
|
||||
[Probelauf](#probelauf) · [Status](#status) · [Wartungsmodus](#wartungsmodus) · [Protokoll](#protokoll) ·
|
||||
[Einstellungen](#einstellungen) · [Anmeldung](#anmeldung) · [Uptime Kuma](#uptime-kuma) · [Beispiele](#beispiele) · [Häufige Fragen](#fragen)
|
||||
|
||||
@@ -160,6 +160,34 @@ Gespeicherte Zugangsdaten lassen sich nur löschen, wenn sie keine Anbindung meh
|
||||
| **Aktion** | **verschieben** (Quelle wird nach erfolgreicher Übertragung gelöscht) oder **kopieren** (Quelle bleibt; jede Dateiversion wird nur einmal kopiert). |
|
||||
| **Wenn Datei existiert** | **umbenennen** (`scan.pdf` → `scan-1.pdf`, `scan-2.pdf` …), **überspringen** oder **überschreiben**. |
|
||||
|
||||
### Archiv {#archiv}
|
||||
|
||||
Optional bekommt jede Datei zusätzlich eine **Kopie ins Archiv** – z. B. Paperless als
|
||||
Ziel und ein Archivordner auf dem NAS. **Die Quelle wird erst gelöscht, wenn die Datei
|
||||
im Ziel UND im Archiv angekommen ist.**
|
||||
|
||||
| Feld | Bedeutung |
|
||||
|---|---|
|
||||
| **Archiv** | kein Archiv / eine Anbindung (SMB, NFS, SFTP) / ein lokaler Pfad |
|
||||
| **Archivordner** | Anbindung: relativer Unterordner (wird angelegt). Lokal: absoluter Pfad, muss existieren. |
|
||||
| **Ablage** | **wie in der Quelle** oder **nach Datum** – `2026/10/scan001.pdf` (Monat der Verarbeitung) |
|
||||
|
||||
So läuft es pro Datei:
|
||||
|
||||
1. **Zuerst ins Archiv** – mit denselben Sicherungen wie beim Ziel.
|
||||
2. **Dann ins Ziel.**
|
||||
3. **Dann erst** wird die Quelle gelöscht (bei „verschieben“).
|
||||
|
||||
Das Archiv kommt zuerst, weil ein Ziel wie Paperless Dateien sofort abholt: Scheitert
|
||||
das Archiv, wurde noch nichts geliefert, und die Datei bleibt in der Quelle. Scheitert
|
||||
danach das Ziel, merkt sich NetMover, dass die Datei schon archiviert ist – der nächste
|
||||
Versuch legt keine zweite Archivkopie an.
|
||||
|
||||
Im Archiv wird **nie überschrieben**: Gleiche Namen werden immer umbenannt
|
||||
(`scan-1.pdf`), egal was die Regel fürs Ziel vorsieht. Ist das Archiv nicht erreichbar,
|
||||
wartet die Regel wie bei Quelle und Ziel. Im Kopiermodus bekommen Ziel und Archiv je
|
||||
eine Kopie, die Quelle bleibt.
|
||||
|
||||
### Sicherheit und Grenzen {#regel-sicherheit}
|
||||
|
||||
| Feld | Bedeutung | Empfehlung |
|
||||
@@ -378,7 +406,8 @@ deaktiviert, Zugangsdaten müssen neu eingegeben werden.
|
||||
3. Anbindung **Paperless** (SFTP, Pfad `/opt/paperless_data/consume`, Schlüssel
|
||||
wählen, Benutzer = der Paperless-Benutzer), testen.
|
||||
4. Regel **Scans → Paperless**: Quelle NAS, Filter `*.pdf`, Ziel Paperless,
|
||||
verschieben, umbenennen, 30 s. **Probelauf**, dann aktivieren.
|
||||
verschieben, umbenennen, 30 s. Optional **Archiv**: Anbindung NAS, Ordner
|
||||
`Archiv/Scans`, Ablage nach Datum. **Probelauf**, dann aktivieren.
|
||||
|
||||
### Jede Person in einen eigenen Ordner {#beispiel-unterordner}
|
||||
|
||||
|
||||
@@ -1,3 +1,3 @@
|
||||
"""NetMover – NFS/SMB-Mounts verwalten und Dateien zeitgesteuert verschieben."""
|
||||
|
||||
__version__ = "0.6.0"
|
||||
__version__ = "0.7.0"
|
||||
|
||||
+52
-1
@@ -18,6 +18,8 @@ from sqlalchemy.orm import Session
|
||||
|
||||
from netmover.crypto import SecretBox, generate_ssh_keypair
|
||||
from netmover.models import (
|
||||
ArchiveKind,
|
||||
ArchiveLayout,
|
||||
CollisionPolicy,
|
||||
Credential,
|
||||
CredentialKind,
|
||||
@@ -424,7 +426,11 @@ def delete_mount(s: Session, mounts: MountManager, mount_id: int) -> None:
|
||||
if m is None:
|
||||
return
|
||||
used = s.scalars(
|
||||
select(Rule.name).where((Rule.src_mount_id == mount_id) | (Rule.dst_mount_id == mount_id))
|
||||
select(Rule.name).where(
|
||||
(Rule.src_mount_id == mount_id)
|
||||
| (Rule.dst_mount_id == mount_id)
|
||||
| (Rule.archive_mount_id == mount_id)
|
||||
)
|
||||
).all()
|
||||
if used:
|
||||
raise AdminError(f"Wird noch von Regeln benutzt: {', '.join(used)}")
|
||||
@@ -465,6 +471,10 @@ class RuleInput:
|
||||
dry_run: bool = False
|
||||
quarantine_path: str = ""
|
||||
quarantine_after: int = 3
|
||||
archive_kind: ArchiveKind = ArchiveKind.NONE
|
||||
archive_mount_id: int | None = None
|
||||
archive_path: str = ""
|
||||
archive_layout: ArchiveLayout = ArchiveLayout.FLAT
|
||||
extra: dict[str, str] = field(default_factory=dict)
|
||||
|
||||
|
||||
@@ -492,6 +502,10 @@ def rule_input_from_model(r: Rule) -> RuleInput:
|
||||
dry_run=r.dry_run,
|
||||
quarantine_path=r.quarantine_path or "",
|
||||
quarantine_after=r.quarantine_after,
|
||||
archive_kind=r.archive_kind,
|
||||
archive_mount_id=r.archive_mount_id,
|
||||
archive_path=r.archive_path,
|
||||
archive_layout=r.archive_layout,
|
||||
)
|
||||
|
||||
|
||||
@@ -585,6 +599,39 @@ def parse_rule_form(s: Session, form: Mapping[str, str]) -> tuple[RuleInput, Err
|
||||
d.quarantine_path = safe_relpath("quarantine_path", d.quarantine_path)
|
||||
except ValidationError as exc:
|
||||
errors["quarantine_path"] = str(exc)
|
||||
|
||||
# Archiv
|
||||
try:
|
||||
d.archive_kind = ArchiveKind(form.get("archive_kind") or "none")
|
||||
d.archive_layout = ArchiveLayout(form.get("archive_layout") or "flat")
|
||||
except ValueError:
|
||||
errors["archive_kind"] = "Ungültige Auswahl."
|
||||
d.archive_path = (form.get("archive_path") or "").strip()
|
||||
if d.archive_kind is ArchiveKind.MOUNT:
|
||||
d.archive_mount_id = _int(form, "archive_mount_id", errors)
|
||||
if d.archive_mount_id is None or s.get(Mount, d.archive_mount_id) is None:
|
||||
errors["archive_mount_id"] = "Archiv-Anbindung wählen."
|
||||
try:
|
||||
d.archive_path = safe_relpath("archive_path", d.archive_path)
|
||||
except ValidationError as exc:
|
||||
errors["archive_path"] = str(exc)
|
||||
if (
|
||||
"archive_path" not in errors
|
||||
and d.archive_mount_id == d.dst_mount_id
|
||||
and d.dst_kind == "mount"
|
||||
and d.archive_path == d.dst_path
|
||||
):
|
||||
errors["archive_path"] = "Archiv und Ziel sind derselbe Ordner."
|
||||
elif d.archive_kind is ArchiveKind.LOCAL:
|
||||
d.archive_mount_id = None
|
||||
try:
|
||||
d.archive_path = safe_abspath("archive_path", d.archive_path)
|
||||
except ValidationError as exc:
|
||||
errors["archive_path"] = str(exc)
|
||||
if d.dst_kind == "local" and d.archive_path == d.dst_path:
|
||||
errors["archive_path"] = "Archiv und Ziel sind derselbe Ordner."
|
||||
else:
|
||||
d.archive_mount_id, d.archive_path = None, ""
|
||||
return d, errors
|
||||
|
||||
|
||||
@@ -617,6 +664,10 @@ def save_rule(s: Session, d: RuleInput, rule_id: int | None = None) -> tuple[Rul
|
||||
"on_collision",
|
||||
"dry_run",
|
||||
"quarantine_after",
|
||||
"archive_kind",
|
||||
"archive_mount_id",
|
||||
"archive_path",
|
||||
"archive_layout",
|
||||
):
|
||||
setattr(r, f, getattr(d, f))
|
||||
r.set_owner = d.set_owner or None
|
||||
|
||||
@@ -0,0 +1,103 @@
|
||||
"""archive
|
||||
|
||||
Revision ID: 0004
|
||||
Revises: 0003
|
||||
Create Date: 2026-10-12
|
||||
"""
|
||||
|
||||
from collections.abc import Sequence
|
||||
|
||||
import sqlalchemy as sa
|
||||
from alembic import op
|
||||
|
||||
import netmover.models
|
||||
|
||||
revision: str = "0004"
|
||||
down_revision: str | None = "0003"
|
||||
branch_labels: str | Sequence[str] | None = None
|
||||
depends_on: str | Sequence[str] | None = None
|
||||
|
||||
|
||||
def upgrade() -> None:
|
||||
op.create_table(
|
||||
"archived_files",
|
||||
sa.Column("id", sa.Integer(), nullable=False),
|
||||
sa.Column("rule_id", sa.Integer(), nullable=False),
|
||||
sa.Column("src_relpath", sa.String(length=2048), nullable=False),
|
||||
sa.Column("size", sa.BigInteger(), nullable=False),
|
||||
sa.Column("mtime_ns", sa.BigInteger(), nullable=False),
|
||||
sa.Column("archive_path", sa.String(length=2048), nullable=False),
|
||||
sa.Column("archived_at", netmover.models.UTCDateTime(), nullable=False),
|
||||
sa.ForeignKeyConstraint(
|
||||
["rule_id"],
|
||||
["rules.id"],
|
||||
name=op.f("fk_archived_files_rule_id_rules"),
|
||||
ondelete="CASCADE",
|
||||
),
|
||||
sa.PrimaryKeyConstraint("id", name=op.f("pk_archived_files")),
|
||||
sa.UniqueConstraint(
|
||||
"rule_id",
|
||||
"src_relpath",
|
||||
"size",
|
||||
"mtime_ns",
|
||||
name=op.f("uq_archived_files_rule_id_src_relpath_size_mtime_ns"),
|
||||
),
|
||||
)
|
||||
with op.batch_alter_table("rules", schema=None) as batch_op:
|
||||
batch_op.add_column(
|
||||
sa.Column(
|
||||
"archive_kind",
|
||||
sa.Enum(
|
||||
"none",
|
||||
"mount",
|
||||
"local",
|
||||
name="archive_kind",
|
||||
native_enum=False,
|
||||
create_constraint=True,
|
||||
length=24,
|
||||
),
|
||||
server_default="none",
|
||||
nullable=False,
|
||||
)
|
||||
)
|
||||
batch_op.add_column(sa.Column("archive_mount_id", sa.Integer(), nullable=True))
|
||||
batch_op.add_column(
|
||||
sa.Column("archive_path", sa.String(length=1024), server_default="", nullable=False)
|
||||
)
|
||||
batch_op.add_column(
|
||||
sa.Column(
|
||||
"archive_layout",
|
||||
sa.Enum(
|
||||
"flat",
|
||||
"date",
|
||||
name="archive_layout",
|
||||
native_enum=False,
|
||||
create_constraint=True,
|
||||
length=24,
|
||||
),
|
||||
server_default="flat",
|
||||
nullable=False,
|
||||
)
|
||||
)
|
||||
batch_op.create_foreign_key(
|
||||
batch_op.f("fk_rules_archive_mount_id_mounts"),
|
||||
"mounts",
|
||||
["archive_mount_id"],
|
||||
["id"],
|
||||
ondelete="RESTRICT",
|
||||
)
|
||||
|
||||
|
||||
def downgrade() -> None:
|
||||
with op.batch_alter_table("rules", schema=None) as batch_op:
|
||||
batch_op.drop_constraint(batch_op.f("fk_rules_archive_mount_id_mounts"), type_="foreignkey")
|
||||
# CHECK-Regeln der Auswahlfelder zuerst entfernen – sonst verweisen sie beim
|
||||
# Neuaufbau der Tabelle auf Spalten, die es nicht mehr gibt.
|
||||
batch_op.drop_constraint(batch_op.f("ck_rules_archive_kind"), type_="check")
|
||||
batch_op.drop_constraint(batch_op.f("ck_rules_archive_layout"), type_="check")
|
||||
batch_op.drop_column("archive_layout")
|
||||
batch_op.drop_column("archive_path")
|
||||
batch_op.drop_column("archive_mount_id")
|
||||
batch_op.drop_column("archive_kind")
|
||||
|
||||
op.drop_table("archived_files")
|
||||
@@ -10,6 +10,7 @@ Abweichungen bzw. Ergänzungen gegenüber dem Mindestumfang aus der Spezifikatio
|
||||
* ``credentials.managed`` + ``mounts.username`` (Migration 0002) –
|
||||
Schlüsselverwaltung: gespeicherte SSH-Schlüssel, die mehrere Anbindungen mit
|
||||
jeweils eigenem Benutzernamen nutzen.
|
||||
* ``rules.archive_*`` + Tabelle ``archived_files`` (Migration 0004) – Archivkopie.
|
||||
* ``rules.waiting_since`` / ``rules.waiting_reason`` (Migration 0003) – Regel wartet
|
||||
auf eine nicht erreichbare Anbindung (automatisch pausiert).
|
||||
* ``mounts.state_since`` – für „hängt seit …“ / „gestört seit …“ auf der Statusseite.
|
||||
@@ -138,6 +139,17 @@ class CollisionPolicy(enum.StrEnum):
|
||||
OVERWRITE = "overwrite"
|
||||
|
||||
|
||||
class ArchiveKind(enum.StrEnum):
|
||||
NONE = "none"
|
||||
MOUNT = "mount"
|
||||
LOCAL = "local"
|
||||
|
||||
|
||||
class ArchiveLayout(enum.StrEnum):
|
||||
FLAT = "flat" # wie in der Quelle
|
||||
DATE = "date" # Jahr/Monat/…, nach Tag der Verarbeitung
|
||||
|
||||
|
||||
class RunStatus(enum.StrEnum):
|
||||
RUNNING = "running"
|
||||
OK = "ok" # keine Dateifehler
|
||||
@@ -259,6 +271,18 @@ class Rule(Base):
|
||||
quarantine_path: Mapped[str | None] = mapped_column(String(1024))
|
||||
quarantine_after: Mapped[int] = mapped_column(Integer, default=3)
|
||||
|
||||
# Archiv (Migration 0004): zusätzliche Kopie; Quelle erst löschen, wenn Ziel UND Archiv da
|
||||
archive_kind: Mapped[ArchiveKind] = mapped_column(
|
||||
_enum(ArchiveKind, "archive_kind"), default=ArchiveKind.NONE, server_default="none"
|
||||
)
|
||||
archive_mount_id: Mapped[int | None] = mapped_column(
|
||||
ForeignKey("mounts.id", ondelete="RESTRICT")
|
||||
)
|
||||
archive_path: Mapped[str] = mapped_column(String(1024), default="", server_default="")
|
||||
archive_layout: Mapped[ArchiveLayout] = mapped_column(
|
||||
_enum(ArchiveLayout, "archive_layout"), default=ArchiveLayout.FLAT, server_default="flat"
|
||||
)
|
||||
|
||||
last_run_at: Mapped[datetime | None] = mapped_column(UTCDateTime)
|
||||
last_success_at: Mapped[datetime | None] = mapped_column(UTCDateTime)
|
||||
consecutive_failures: Mapped[int] = mapped_column(Integer, default=0)
|
||||
@@ -269,6 +293,7 @@ class Rule(Base):
|
||||
|
||||
src_mount: Mapped[Mount] = relationship(foreign_keys=[src_mount_id])
|
||||
dst_mount: Mapped[Mount | None] = relationship(foreign_keys=[dst_mount_id])
|
||||
archive_mount: Mapped[Mount | None] = relationship(foreign_keys=[archive_mount_id])
|
||||
runs: Mapped[list[Run]] = relationship(
|
||||
back_populates="rule", cascade="all, delete-orphan", passive_deletes=True
|
||||
)
|
||||
@@ -330,6 +355,22 @@ class CopiedFile(Base):
|
||||
copied_at: Mapped[datetime] = mapped_column(UTCDateTime, default=utcnow)
|
||||
|
||||
|
||||
class ArchivedFile(Base):
|
||||
"""Welche Dateiversion schon im Archiv liegt – damit ein Neuversuch nach einem
|
||||
gescheiterten Ziel keine zweite Archivkopie anlegt."""
|
||||
|
||||
__tablename__ = "archived_files"
|
||||
__table_args__ = (UniqueConstraint("rule_id", "src_relpath", "size", "mtime_ns"),)
|
||||
|
||||
id: Mapped[int] = mapped_column(Integer, primary_key=True)
|
||||
rule_id: Mapped[int] = mapped_column(ForeignKey("rules.id", ondelete="CASCADE"))
|
||||
src_relpath: Mapped[str] = mapped_column(String(2048))
|
||||
size: Mapped[int] = mapped_column(BigInteger)
|
||||
mtime_ns: Mapped[int] = mapped_column(BigInteger)
|
||||
archive_path: Mapped[str] = mapped_column(String(2048))
|
||||
archived_at: Mapped[datetime] = mapped_column(UTCDateTime, default=utcnow)
|
||||
|
||||
|
||||
class FileFailure(Base):
|
||||
"""Fehlversuche pro Datei, Grundlage für die Quarantäne."""
|
||||
|
||||
|
||||
@@ -19,6 +19,7 @@ import posixpath
|
||||
import time
|
||||
from collections.abc import Callable
|
||||
from dataclasses import dataclass, field
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
@@ -72,6 +73,7 @@ class RuleSnapshot:
|
||||
set_mode: int | None = None
|
||||
quarantine_after: int = 3
|
||||
dst_root_squash: bool | None = None # nur NFS-Ziele: Ergebnis des Schreibtests
|
||||
archive_layout: str = "flat" # "flat" | "date" (Jahr/Monat/…)
|
||||
|
||||
|
||||
@dataclass
|
||||
@@ -159,6 +161,7 @@ class Engine:
|
||||
dry_run: bool = False,
|
||||
quarantine_rel: str | None = None,
|
||||
quarantine_root: Path | None = None,
|
||||
archive: Destination | None = None,
|
||||
) -> RunResult:
|
||||
"""``source``: eine :class:`Source` oder (Kurzform) ein lokales Verzeichnis.
|
||||
Quarantäne relativ zur Wurzel der Anbindung (``quarantine_rel``) oder – bei
|
||||
@@ -168,7 +171,7 @@ class Engine:
|
||||
if quarantine_root is not None and isinstance(src, LocalSource):
|
||||
quarantine_rel = str(quarantine_root.relative_to(src.root))
|
||||
try:
|
||||
self._execute(rule, src, dest, token, result, dry_run, quarantine_rel)
|
||||
self._execute(rule, src, dest, token, result, dry_run, quarantine_rel, archive)
|
||||
except Cancelled:
|
||||
result.status = RunStatus.ERROR
|
||||
result.message = "Lauf abgebrochen"
|
||||
@@ -195,10 +198,13 @@ class Engine:
|
||||
result: RunResult,
|
||||
dry_run: bool,
|
||||
quarantine_rel: str | None,
|
||||
archive: Destination | None = None,
|
||||
) -> None:
|
||||
src.check_root()
|
||||
if not dry_run:
|
||||
dest.prepare()
|
||||
if archive is not None:
|
||||
archive.prepare()
|
||||
token.check()
|
||||
|
||||
exclude = frozenset({quarantine_rel}) if quarantine_rel else frozenset()
|
||||
@@ -212,8 +218,12 @@ class Engine:
|
||||
# Kopier-Buchführung (Fallstrick 11). Gilt auch für „verschieben“, falls
|
||||
# dort einmal die Quelle nicht gelöscht werden konnte.
|
||||
done = self.ledger.copied_keys(rule.id)
|
||||
archived = self.ledger.archived_keys(rule.id) if archive is not None else set()
|
||||
if not dry_run:
|
||||
self.ledger.prune_copied(rule.id, {c.relpath for c in candidates})
|
||||
present = {c.relpath for c in candidates}
|
||||
self.ledger.prune_copied(rule.id, present)
|
||||
if archive is not None:
|
||||
self.ledger.prune_archived(rule.id, present)
|
||||
fresh = [c for c in candidates if (c.relpath, c.state.size, c.state.mtime_ns) not in done]
|
||||
unchanged = len(candidates) - len(fresh)
|
||||
|
||||
@@ -271,7 +281,11 @@ class Engine:
|
||||
failures = self.ledger.failures(rule.id)
|
||||
for c in stable:
|
||||
token.check()
|
||||
result.add(self._process(rule, c, src, dest, token, dry_run, quarantine_rel, failures))
|
||||
result.add(
|
||||
self._process(
|
||||
rule, c, src, dest, token, dry_run, quarantine_rel, failures, archive, archived
|
||||
)
|
||||
)
|
||||
|
||||
if unchanged:
|
||||
result.message = f"{unchanged} bereits übertragen"
|
||||
@@ -309,8 +323,14 @@ class Engine:
|
||||
dry_run: bool,
|
||||
quarantine_rel: str | None,
|
||||
failures: dict[str, FailureInfo],
|
||||
archive: Destination | None = None,
|
||||
archived: set[tuple[str, int, int]] | None = None,
|
||||
) -> ItemResult:
|
||||
size = c.state.size
|
||||
arch_rel = self._archive_rel(c.relpath, rule.archive_layout)
|
||||
need_archive = archive is not None and (
|
||||
(c.relpath, size, c.state.mtime_ns) not in (archived or set())
|
||||
)
|
||||
failure = failures.get(c.relpath)
|
||||
|
||||
# Quarantäne für wiederholt scheiternde Dateien
|
||||
@@ -349,9 +369,14 @@ class Engine:
|
||||
else None
|
||||
)
|
||||
verb = "kopieren" if rule.action is RuleAction.COPY else "verschieben"
|
||||
return ItemResult(
|
||||
c.relpath, ItemStatus.PLANNED, dest.describe(planned), size, msg or f"würde {verb}"
|
||||
)
|
||||
msg = msg or f"würde {verb}"
|
||||
if archive is not None:
|
||||
if need_archive:
|
||||
arch_planned = archive.plan(arch_rel, CollisionPolicy.RENAME)
|
||||
msg += f"; Archiv: {archive.describe(arch_planned or arch_rel)}"
|
||||
else:
|
||||
msg += "; bereits archiviert"
|
||||
return ItemResult(c.relpath, ItemStatus.PLANNED, dest.describe(planned), size, msg)
|
||||
|
||||
try:
|
||||
free = dest.free_bytes()
|
||||
@@ -367,10 +392,51 @@ class Engine:
|
||||
f"benötigt {_mb(int(size * SPACE_RESERVE))})",
|
||||
)
|
||||
|
||||
if need_archive and archive is not None:
|
||||
try:
|
||||
afree = archive.free_bytes()
|
||||
except OSError:
|
||||
afree = None
|
||||
if afree is not None and afree < size * SPACE_RESERVE:
|
||||
return ItemResult(
|
||||
c.relpath,
|
||||
ItemStatus.FAILED,
|
||||
archive.describe(arch_rel),
|
||||
size,
|
||||
f"zu wenig Platz im Archiv (frei {_mb(afree)}, "
|
||||
f"benötigt {_mb(int(size * SPACE_RESERVE))})",
|
||||
)
|
||||
|
||||
def verify() -> None:
|
||||
if src.state(c.relpath) != c.state:
|
||||
raise SourceChanged("Quelle hat sich während der Übertragung geändert")
|
||||
|
||||
# Archiv ZUERST: scheitert es, ist noch nichts ans Ziel (z. B. Paperless)
|
||||
# geliefert. Scheitert danach das Ziel, verhindert die Buchführung eine
|
||||
# zweite Archivkopie beim nächsten Versuch.
|
||||
archive_note: str | None = None
|
||||
if need_archive and archive is not None:
|
||||
try:
|
||||
meta, _ = self._meta(rule, src, c.relpath, dest)
|
||||
ares = archive.commit(
|
||||
_SourceFile(src, c.relpath),
|
||||
size,
|
||||
arch_rel,
|
||||
CollisionPolicy.RENAME,
|
||||
MetaOptions(meta.atime_ns, meta.mtime_ns, meta.mode, None),
|
||||
verify,
|
||||
token,
|
||||
)
|
||||
except CollisionSkip: # kann bei RENAME nicht auftreten
|
||||
return ItemResult(c.relpath, ItemStatus.SKIPPED, None, size, "Archiv belegt")
|
||||
except SourceChanged as exc:
|
||||
return ItemResult(c.relpath, ItemStatus.SKIPPED, None, size, str(exc))
|
||||
except (TransferError, OSError) as exc:
|
||||
return self._fail(rule, c, exc, dry_run, prefix="Archiv: ")
|
||||
archive_path = archive.describe(ares.final_rel)
|
||||
self.ledger.record_archive(rule.id, c.relpath, size, c.state.mtime_ns, archive_path)
|
||||
archive_note = f"archiviert: {archive_path}"
|
||||
|
||||
try:
|
||||
meta, notes = self._meta(rule, src, c.relpath, dest)
|
||||
res = None
|
||||
@@ -401,6 +467,8 @@ class Engine:
|
||||
return self._fail(rule, c, exc, dry_run)
|
||||
|
||||
notes += res.notes
|
||||
if archive_note:
|
||||
notes.insert(0, archive_note)
|
||||
target = dest.describe(res.final_rel)
|
||||
if c.relpath in failures:
|
||||
self.ledger.clear_failure(rule.id, c.relpath)
|
||||
@@ -425,10 +493,12 @@ class Engine:
|
||||
)
|
||||
return ItemResult(c.relpath, ItemStatus.MOVED, target, size, "; ".join(notes) or None)
|
||||
|
||||
def _fail(self, rule: RuleSnapshot, c: Candidate, exc: Exception, dry_run: bool) -> ItemResult:
|
||||
def _fail(
|
||||
self, rule: RuleSnapshot, c: Candidate, exc: Exception, dry_run: bool, prefix: str = ""
|
||||
) -> ItemResult:
|
||||
if isinstance(exc, OSError) and exc.errno in MOUNT_ERRNOS:
|
||||
raise RunAborted(f"Anbindung gestört: {_describe_os_error(exc)}") from exc
|
||||
msg = _describe_os_error(exc) if isinstance(exc, OSError) else str(exc)
|
||||
raise RunAborted(f"{prefix}Anbindung gestört: {_describe_os_error(exc)}") from exc
|
||||
msg = prefix + (_describe_os_error(exc) if isinstance(exc, OSError) else str(exc))
|
||||
if not dry_run:
|
||||
attempts = self.ledger.record_failure(
|
||||
rule.id, c.relpath, c.state.size, c.state.mtime_ns, msg
|
||||
@@ -437,6 +507,12 @@ class Engine:
|
||||
event(log, "file.failed", level=30, rule=rule.name, file=c.relpath, error=msg)
|
||||
return ItemResult(c.relpath, ItemStatus.FAILED, None, c.state.size, msg)
|
||||
|
||||
def _archive_rel(self, relpath: str, layout: str) -> str:
|
||||
if layout == "date":
|
||||
t = datetime.fromtimestamp(self._clock())
|
||||
return f"{t:%Y}/{t:%m}/{relpath}"
|
||||
return relpath
|
||||
|
||||
def _quarantine(
|
||||
self, rule: RuleSnapshot, c: Candidate, src: Source, q_rel: str, attempts: int
|
||||
) -> ItemResult:
|
||||
|
||||
@@ -14,7 +14,7 @@ from sqlalchemy import delete, select
|
||||
from sqlalchemy.orm import Session, sessionmaker
|
||||
|
||||
from netmover.db import session_scope
|
||||
from netmover.models import CopiedFile, FileFailure, utcnow
|
||||
from netmover.models import ArchivedFile, CopiedFile, FileFailure, utcnow
|
||||
|
||||
CopyKey = tuple[str, int, int] # (relpath, size, mtime_ns)
|
||||
|
||||
@@ -72,6 +72,53 @@ class Ledger:
|
||||
s.execute(delete(CopiedFile).where(CopiedFile.id.in_(stale)))
|
||||
return len(stale)
|
||||
|
||||
# -- Archiv ------------------------------------------------------------------------
|
||||
|
||||
def archived_keys(self, rule_id: int) -> set[CopyKey]:
|
||||
with session_scope(self.session_factory) as s:
|
||||
rows = s.execute(
|
||||
select(ArchivedFile.src_relpath, ArchivedFile.size, ArchivedFile.mtime_ns).where(
|
||||
ArchivedFile.rule_id == rule_id
|
||||
)
|
||||
).all()
|
||||
return {(r[0], r[1], r[2]) for r in rows}
|
||||
|
||||
def record_archive(
|
||||
self, rule_id: int, relpath: str, size: int, mtime_ns: int, archive_path: str
|
||||
) -> None:
|
||||
with session_scope(self.session_factory) as s:
|
||||
exists = s.scalar(
|
||||
select(ArchivedFile.id).where(
|
||||
ArchivedFile.rule_id == rule_id,
|
||||
ArchivedFile.src_relpath == relpath,
|
||||
ArchivedFile.size == size,
|
||||
ArchivedFile.mtime_ns == mtime_ns,
|
||||
)
|
||||
)
|
||||
if exists is None:
|
||||
s.add(
|
||||
ArchivedFile(
|
||||
rule_id=rule_id,
|
||||
src_relpath=relpath,
|
||||
size=size,
|
||||
mtime_ns=mtime_ns,
|
||||
archive_path=archive_path,
|
||||
)
|
||||
)
|
||||
|
||||
def prune_archived(self, rule_id: int, present: set[str]) -> int:
|
||||
"""Wie :meth:`prune_copied`: nur nach einem vollständigen Scan aufrufen."""
|
||||
with session_scope(self.session_factory) as s:
|
||||
rows = s.execute(
|
||||
select(ArchivedFile.id, ArchivedFile.src_relpath).where(
|
||||
ArchivedFile.rule_id == rule_id
|
||||
)
|
||||
).all()
|
||||
stale = [r[0] for r in rows if r[1] not in present]
|
||||
if stale:
|
||||
s.execute(delete(ArchivedFile).where(ArchivedFile.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()
|
||||
|
||||
@@ -56,6 +56,11 @@ class RulePlan:
|
||||
quarantine_path: str | None
|
||||
op_timeout: int
|
||||
dry_run: bool
|
||||
archive_kind: str = "none"
|
||||
archive_mount_id: int | None = None
|
||||
archive_type: MountType | None = None
|
||||
archive_name: str | None = None
|
||||
archive_path: str = ""
|
||||
|
||||
|
||||
def parse_owner(value: str | None) -> tuple[int, int] | None:
|
||||
@@ -92,6 +97,7 @@ def load_plan(session: Session, rule_id: int) -> RulePlan:
|
||||
set_mode=parse_mode(r.set_mode),
|
||||
quarantine_after=r.quarantine_after,
|
||||
dst_root_squash=dst.root_squash if dst is not None else None,
|
||||
archive_layout=r.archive_layout.value,
|
||||
)
|
||||
return RulePlan(
|
||||
snapshot=snap,
|
||||
@@ -109,6 +115,11 @@ def load_plan(session: Session, rule_id: int) -> RulePlan:
|
||||
else None,
|
||||
op_timeout=r.op_timeout_seconds,
|
||||
dry_run=r.dry_run,
|
||||
archive_kind=r.archive_kind.value,
|
||||
archive_mount_id=r.archive_mount_id if r.archive_kind.value == "mount" else None,
|
||||
archive_type=r.archive_mount.type if r.archive_mount is not None else None,
|
||||
archive_name=r.archive_mount.name if r.archive_mount is not None else None,
|
||||
archive_path=r.archive_path,
|
||||
)
|
||||
|
||||
|
||||
@@ -176,24 +187,52 @@ class RuleRunner:
|
||||
if plan.dst_mount_id is not None
|
||||
else None
|
||||
)
|
||||
arch_mp = (
|
||||
self.mounts.check_alive(plan.archive_mount_id)
|
||||
if plan.archive_mount_id is not None
|
||||
else None
|
||||
)
|
||||
except MountUnavailable as exc:
|
||||
return RunResult(RunStatus.ERROR, dry, message=str(exc), unavailable=True)
|
||||
|
||||
try:
|
||||
make_source = self._source_factory(plan, src_mp)
|
||||
src_root = src_mp / plan.src_subpath if src_mp and plan.src_subpath else src_mp
|
||||
make_dest = self._dest_factory(plan, dst_mp, src_root)
|
||||
make_dest = self._target(
|
||||
"dst_path",
|
||||
plan.dst_mount_id,
|
||||
plan.dst_type,
|
||||
plan.dst_name,
|
||||
plan.dst_path,
|
||||
dst_mp,
|
||||
src_root,
|
||||
)
|
||||
make_archive = (
|
||||
self._target(
|
||||
"archive_path",
|
||||
plan.archive_mount_id,
|
||||
plan.archive_type,
|
||||
plan.archive_name,
|
||||
plan.archive_path,
|
||||
arch_mp,
|
||||
src_root,
|
||||
)
|
||||
if plan.archive_kind != "none"
|
||||
else None
|
||||
)
|
||||
except (ValidationError, TransferError, SftpError) as exc:
|
||||
return RunResult(RunStatus.ERROR, dry, message=str(exc))
|
||||
|
||||
def work(token: CancelToken) -> RunResult:
|
||||
source = make_source()
|
||||
opened: list[Source | Destination] = []
|
||||
try:
|
||||
source = make_source()
|
||||
opened.append(source)
|
||||
dest = make_dest()
|
||||
except BaseException:
|
||||
source.close()
|
||||
raise
|
||||
try:
|
||||
opened.append(dest)
|
||||
archive = make_archive() if make_archive is not None else None
|
||||
if archive is not None:
|
||||
opened.append(archive)
|
||||
return self.engine.execute(
|
||||
plan.snapshot,
|
||||
source,
|
||||
@@ -201,10 +240,11 @@ class RuleRunner:
|
||||
token,
|
||||
dry_run=dry,
|
||||
quarantine_rel=plan.quarantine_path,
|
||||
archive=archive,
|
||||
)
|
||||
finally:
|
||||
dest.close()
|
||||
source.close()
|
||||
for item in reversed(opened):
|
||||
item.close()
|
||||
|
||||
def finish() -> None:
|
||||
self._started.pop(rule_id, None)
|
||||
@@ -251,41 +291,50 @@ class RuleRunner:
|
||||
|
||||
return make_local
|
||||
|
||||
def _dest_factory(
|
||||
self, plan: RulePlan, dst_mp: Path | None, src_root: Path | None
|
||||
def _target(
|
||||
self,
|
||||
field: str,
|
||||
mount_id: int | None,
|
||||
mount_type: MountType | None,
|
||||
mount_name: str | None,
|
||||
path: str,
|
||||
mp: Path | None,
|
||||
src_root: Path | None,
|
||||
) -> Callable[[], Destination]:
|
||||
if plan.dst_mount_id is None:
|
||||
root = Path(safe_abspath("dst_path", plan.dst_path))
|
||||
"""Ziel oder Archiv: lokaler Pfad (``mount_id`` None), SFTP oder Kernel-Mount."""
|
||||
what = "Archiv" if field == "archive_path" else "Ziel"
|
||||
if mount_id is None:
|
||||
root = Path(safe_abspath(field, path))
|
||||
if self._under_mount_base(root):
|
||||
base = self.mounts.mount_base
|
||||
raise ValidationError(
|
||||
"dst_path",
|
||||
f"Lokales Ziel liegt unter {base} – bitte die Anbindung als Ziel wählen, "
|
||||
field,
|
||||
f"Lokales {what} liegt unter {base} – bitte die Anbindung als {what} wählen, "
|
||||
"damit geprüft wird, ob sie eingehängt ist.",
|
||||
)
|
||||
kind = "local"
|
||||
local_root = root
|
||||
create = False
|
||||
elif plan.dst_type is MountType.SFTP:
|
||||
scfg = self.mounts.sftp_config_for(plan.dst_mount_id)
|
||||
sub = safe_relpath("dst_path", plan.dst_path)
|
||||
elif mount_type is MountType.SFTP:
|
||||
scfg = self.mounts.sftp_config_for(mount_id)
|
||||
sub = safe_relpath(field, path)
|
||||
remote = posixpath.join(scfg.base_path, sub) if sub else scfg.base_path
|
||||
label = plan.dst_name or "sftp"
|
||||
label = mount_name or "sftp"
|
||||
|
||||
def make_sftp() -> Destination:
|
||||
return SftpDestination(connect(scfg), remote, label=label)
|
||||
|
||||
return make_sftp
|
||||
else:
|
||||
if dst_mp is None:
|
||||
raise TransferError("Ziel-Anbindung ist nicht eingehängt")
|
||||
sub = safe_relpath("dst_path", plan.dst_path)
|
||||
local_root = dst_mp / sub if sub else dst_mp
|
||||
kind = "cifs" if plan.dst_type is MountType.CIFS else "nfs"
|
||||
if mp is None:
|
||||
raise TransferError(f"{what}-Anbindung ist nicht eingehängt")
|
||||
sub = safe_relpath(field, path)
|
||||
local_root = mp / sub if sub else mp
|
||||
kind = "cifs" if mount_type is MountType.CIFS else "nfs"
|
||||
create = True
|
||||
|
||||
if src_root is not None and local_root.resolve() == src_root.resolve():
|
||||
raise ValidationError("dst_path", "Quelle und Ziel sind derselbe Ordner")
|
||||
raise ValidationError(field, f"Quelle und {what} sind derselbe Ordner")
|
||||
|
||||
def make_local() -> Destination:
|
||||
return LocalDestination(local_root, kind, create_root=create) # type: ignore[arg-type]
|
||||
|
||||
@@ -16,6 +16,8 @@ from sqlalchemy.orm import Session
|
||||
|
||||
from netmover import __version__
|
||||
from netmover.models import (
|
||||
ArchiveKind,
|
||||
ArchiveLayout,
|
||||
CollisionPolicy,
|
||||
Mount,
|
||||
MountState,
|
||||
@@ -72,6 +74,9 @@ RULE_FIELDS = (
|
||||
"dry_run",
|
||||
"quarantine_path",
|
||||
"quarantine_after",
|
||||
"archive_kind",
|
||||
"archive_path",
|
||||
"archive_layout",
|
||||
)
|
||||
SETTING_FIELDS = (
|
||||
"retention_days",
|
||||
@@ -99,6 +104,7 @@ def export_config(s: Session, settings: AppSettings) -> dict[str, Any]:
|
||||
**{f: _plain(getattr(r, f)) for f in RULE_FIELDS},
|
||||
"src_mount": names[r.src_mount_id],
|
||||
"dst_mount": names.get(r.dst_mount_id) if r.dst_mount_id else None,
|
||||
"archive_mount": names.get(r.archive_mount_id) if r.archive_mount_id else None,
|
||||
}
|
||||
for r in s.scalars(select(Rule).order_by(Rule.name))
|
||||
],
|
||||
@@ -165,13 +171,16 @@ def import_config(s: Session, mounts: MountManager, data: dict[str, Any]) -> Imp
|
||||
try:
|
||||
src = ids[raw["src_mount"]]
|
||||
dst = ids[raw["dst_mount"]] if raw.get("dst_mount") else None
|
||||
r = Rule(src_mount_id=src, dst_mount_id=dst)
|
||||
arch = ids[raw["archive_mount"]] if raw.get("archive_mount") else None
|
||||
r = Rule(src_mount_id=src, dst_mount_id=dst, archive_mount_id=arch)
|
||||
for f in RULE_FIELDS:
|
||||
if f in raw:
|
||||
setattr(r, f, raw[f])
|
||||
r.pattern_type = PatternType(r.pattern_type)
|
||||
r.action = RuleAction(r.action)
|
||||
r.on_collision = CollisionPolicy(r.on_collision)
|
||||
r.archive_kind = ArchiveKind(r.archive_kind or "none")
|
||||
r.archive_layout = ArchiveLayout(r.archive_layout or "flat")
|
||||
r.src_subpath = safe_relpath("src_subpath", r.src_subpath or "")
|
||||
if int(r.interval_seconds) < 10:
|
||||
raise ValueError("Intervall < 10 s")
|
||||
|
||||
+12
-1
@@ -422,13 +422,24 @@ class RuleView:
|
||||
target: str
|
||||
|
||||
|
||||
def _archive_target(r: Rule) -> str | None:
|
||||
if r.archive_kind.value == "none":
|
||||
return None
|
||||
if r.archive_kind.value == "local" or r.archive_mount is None:
|
||||
place = r.archive_path
|
||||
else:
|
||||
place = f"{r.archive_mount.name}:/{r.archive_path}"
|
||||
return place + (" (nach Datum)" if r.archive_layout.value == "date" else "")
|
||||
|
||||
|
||||
def _rule_target(r: Rule) -> str:
|
||||
src = f"{r.src_mount.name}:/{r.src_subpath}" if r.src_subpath else f"{r.src_mount.name}:/"
|
||||
if r.dst_mount is None:
|
||||
dst = r.dst_path
|
||||
else:
|
||||
dst = f"{r.dst_mount.name}:/{r.dst_path}" if r.dst_path else f"{r.dst_mount.name}:/"
|
||||
return f"{src} → {dst}"
|
||||
archive = _archive_target(r)
|
||||
return f"{src} → {dst}" + (f" + Archiv {archive}" if archive else "")
|
||||
|
||||
|
||||
async def _status_context(request: Request) -> dict[str, Any]:
|
||||
|
||||
@@ -30,6 +30,13 @@
|
||||
{{ select('on_collision', 'Wenn Datei existiert', [('rename', 'umbenennen (scan-1.pdf)'), ('skip', 'überspringen'), ('overwrite', 'überschreiben')], d.on_collision.value, errors=errors) }}
|
||||
</div></fieldset>
|
||||
|
||||
<fieldset><legend>Archiv <a class="legend-help" href="/help#archiv" title="Hilfe">?</a></legend><div class="fields">
|
||||
{{ select('archive_kind', 'Archiv', [('none', 'kein Archiv'), ('mount', 'Anbindung'), ('local', 'Lokaler Pfad')], d.archive_kind.value, errors=errors, hint='Zusätzliche Kopie. Die Quelle wird erst gelöscht, wenn Ziel UND Archiv sie haben.') }}
|
||||
<div data-show-for="archive_kind:mount">{{ select('archive_mount_id', 'Archiv-Anbindung', dst_options, d.archive_mount_id or '', errors=errors) }}</div>
|
||||
<div data-show-for="archive_kind:mount local">{{ field('archive_path', 'Archivordner', d.archive_path, errors=errors, hint='Anbindung: relativer Unterordner (wird angelegt). Lokal: absoluter Pfad, muss existieren.') }}</div>
|
||||
<div data-show-for="archive_kind:mount local">{{ select('archive_layout', 'Ablage', [('flat', 'wie in der Quelle'), ('date', 'nach Datum (Jahr/Monat)')], d.archive_layout.value, errors=errors, hint='Im Archiv wird nie überschrieben – gleiche Namen werden umbenannt.') }}</div>
|
||||
</div></fieldset>
|
||||
|
||||
<fieldset><legend>Sicherheit und Grenzen <a class="legend-help" href="/help#regel-sicherheit" title="Hilfe">?</a></legend><div class="fields">
|
||||
{{ field('stability_seconds', 'Stabilitätsfenster (s)', d.stability_seconds, type='number', errors=errors, attrs='min="0"', hint='Größe und Zeit müssen so lange gleich bleiben') }}
|
||||
{{ field('min_age_seconds', 'Mindestalter (s)', d.min_age_seconds, type='number', errors=errors, attrs='min="0"') }}
|
||||
|
||||
@@ -0,0 +1,351 @@
|
||||
"""0.7.0: Archiv – Kopie ans Ziel UND ins Archiv, Quelle erst danach löschen."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
from collections.abc import Iterator
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.orm import Session, sessionmaker
|
||||
|
||||
from netmover.db import session_scope
|
||||
from netmover.models import (
|
||||
ArchivedFile,
|
||||
ArchiveKind,
|
||||
CollisionPolicy,
|
||||
ItemStatus,
|
||||
Rule,
|
||||
RuleAction,
|
||||
RunStatus,
|
||||
)
|
||||
from netmover.mounting.fake import FakeMountBackend
|
||||
from netmover.mounting.manager import MountManager
|
||||
from netmover.mover.destinations import LocalDestination, TransferError
|
||||
from netmover.mover.engine import Engine, RuleSnapshot, RunResult
|
||||
from netmover.mover.ledger import Ledger
|
||||
from netmover.mover.runner import RuleRunner
|
||||
from netmover.timeouts import CancelToken
|
||||
from tests.conftest import MountFactory
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def rule_id(make_mount: MountFactory, session_factory: sessionmaker[Session]) -> int:
|
||||
mid = make_mount()
|
||||
with session_scope(session_factory) as s:
|
||||
r = Rule(name="r", src_mount_id=mid, dst_path="/x")
|
||||
s.add(r)
|
||||
s.flush()
|
||||
return r.id
|
||||
|
||||
|
||||
FIXED = datetime(2026, 10, 12, 9, 30).timestamp()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def engine(session_factory: sessionmaker[Session]) -> Engine:
|
||||
return Engine(Ledger(session_factory), sleep=lambda _: None, clock=lambda: FIXED)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def dirs(tmp_path: Path) -> tuple[Path, Path, Path]:
|
||||
src, dst, arch = tmp_path / "src", tmp_path / "dst", tmp_path / "archiv"
|
||||
for d in (src, dst, arch):
|
||||
d.mkdir()
|
||||
return src, dst, arch
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def cross_device(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""tmp_path liegt auf einem Gerät – realistischer Weg über Kopie erzwingen."""
|
||||
monkeypatch.setattr(LocalDestination, "move_same_device", lambda *a, **k: None)
|
||||
|
||||
|
||||
def snap(rule_id: int, **kw: object) -> RuleSnapshot:
|
||||
return RuleSnapshot(id=rule_id, name="r", stability_seconds=0, **kw) # type: ignore[arg-type]
|
||||
|
||||
|
||||
def run(
|
||||
engine: Engine, rule: RuleSnapshot, src: Path, dst: Path, arch: Path | None, **kw: object
|
||||
) -> RunResult:
|
||||
archive = LocalDestination(arch) if arch is not None else None
|
||||
return engine.execute(rule, src, LocalDestination(dst), CancelToken(), archive=archive, **kw) # type: ignore[arg-type]
|
||||
|
||||
|
||||
def write(p: Path, data: bytes = b"inhalt") -> Path:
|
||||
p.parent.mkdir(parents=True, exist_ok=True)
|
||||
p.write_bytes(data)
|
||||
return p
|
||||
|
||||
|
||||
def names(d: Path) -> list[str]:
|
||||
return sorted(str(p.relative_to(d)) for p in d.rglob("*") if p.is_file())
|
||||
|
||||
|
||||
def test_move_delivers_to_target_and_archive_then_deletes_source(
|
||||
engine: Engine, rule_id: int, dirs: tuple[Path, Path, Path]
|
||||
) -> None:
|
||||
src, dst, arch = dirs
|
||||
data = os.urandom(200_000)
|
||||
write(src / "scan001.pdf", data)
|
||||
result = run(engine, snap(rule_id), src, dst, arch)
|
||||
assert result.status is RunStatus.OK
|
||||
item = result.items[0]
|
||||
assert item.status is ItemStatus.MOVED and item.message and "archiviert:" in item.message
|
||||
assert (dst / "scan001.pdf").read_bytes() == data == (arch / "scan001.pdf").read_bytes()
|
||||
assert not (src / "scan001.pdf").exists()
|
||||
assert names(dst) == ["scan001.pdf"] and names(arch) == ["scan001.pdf"] # keine .part-Reste
|
||||
|
||||
|
||||
def test_date_layout(engine: Engine, rule_id: int, dirs: tuple[Path, Path, Path]) -> None:
|
||||
src, dst, arch = dirs
|
||||
write(src / "max" / "a.pdf")
|
||||
run(engine, snap(rule_id, recursive=True, archive_layout="date"), src, dst, arch)
|
||||
assert names(arch) == ["2026/10/max/a.pdf"] and names(dst) == ["max/a.pdf"]
|
||||
|
||||
|
||||
def test_archive_failure_delivers_nothing_and_keeps_source(
|
||||
engine: Engine, rule_id: int, dirs: tuple[Path, Path, Path], monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
src, dst, arch = dirs
|
||||
write(src / "a.pdf")
|
||||
real = LocalDestination._link_noreplace
|
||||
|
||||
def fail_in_archive(self: LocalDestination, tmp: str, final: str) -> None:
|
||||
if self.root == arch:
|
||||
raise TransferError("Archiv voll")
|
||||
real(self, tmp, final)
|
||||
|
||||
monkeypatch.setattr(LocalDestination, "_link_noreplace", fail_in_archive)
|
||||
result = run(engine, snap(rule_id), src, dst, arch)
|
||||
item = result.items[0]
|
||||
assert item.status is ItemStatus.FAILED and item.message == "Archiv: Archiv voll (Versuch 1)"
|
||||
assert (src / "a.pdf").exists() and names(dst) == [] and names(arch) == []
|
||||
|
||||
|
||||
def test_target_failure_after_archive_never_archives_twice(
|
||||
engine: Engine,
|
||||
rule_id: int,
|
||||
dirs: tuple[Path, Path, Path],
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
session_factory: sessionmaker[Session],
|
||||
) -> None:
|
||||
src, dst, arch = dirs
|
||||
write(src / "a.pdf")
|
||||
real = LocalDestination._link_noreplace
|
||||
|
||||
def fail_in_target(self: LocalDestination, tmp: str, final: str) -> None:
|
||||
if self.root == dst:
|
||||
raise TransferError("Ziel gesperrt")
|
||||
real(self, tmp, final)
|
||||
|
||||
monkeypatch.setattr(LocalDestination, "_link_noreplace", fail_in_target)
|
||||
first = run(engine, snap(rule_id), src, dst, arch)
|
||||
assert first.items[0].status is ItemStatus.FAILED
|
||||
assert (src / "a.pdf").exists() and names(arch) == ["a.pdf"] and names(dst) == []
|
||||
with session_factory() as s:
|
||||
assert len(s.scalars(select(ArchivedFile)).all()) == 1
|
||||
|
||||
monkeypatch.setattr(LocalDestination, "_link_noreplace", real) # Ziel wieder frei
|
||||
second = run(engine, snap(rule_id), src, dst, arch)
|
||||
assert second.items[0].status is ItemStatus.MOVED
|
||||
assert names(arch) == ["a.pdf"] # keine zweite Archivkopie (a-1.pdf)
|
||||
assert names(dst) == ["a.pdf"] and not (src / "a.pdf").exists()
|
||||
|
||||
|
||||
def test_archive_never_overwrites(
|
||||
engine: Engine, rule_id: int, dirs: tuple[Path, Path, Path]
|
||||
) -> None:
|
||||
src, dst, arch = dirs
|
||||
write(arch / "scan.pdf", b"alt im Archiv")
|
||||
write(src / "scan.pdf", b"neu")
|
||||
run(engine, snap(rule_id, on_collision=CollisionPolicy.OVERWRITE), src, dst, arch)
|
||||
assert (arch / "scan.pdf").read_bytes() == b"alt im Archiv"
|
||||
assert (arch / "scan-1.pdf").read_bytes() == b"neu"
|
||||
|
||||
|
||||
def test_no_space_in_archive_writes_nothing(
|
||||
engine: Engine, rule_id: int, dirs: tuple[Path, Path, Path], monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
src, dst, arch = dirs
|
||||
write(src / "a.pdf", b"x" * 1000)
|
||||
real = LocalDestination.free_bytes
|
||||
monkeypatch.setattr(
|
||||
LocalDestination, "free_bytes", lambda self: 10 if self.root == arch else real(self)
|
||||
)
|
||||
result = run(engine, snap(rule_id), src, dst, arch)
|
||||
assert "zu wenig Platz im Archiv" in (result.items[0].message or "")
|
||||
assert names(dst) == [] and names(arch) == [] and (src / "a.pdf").exists()
|
||||
|
||||
|
||||
def test_source_changed_during_archive_copy(
|
||||
engine: Engine, rule_id: int, dirs: tuple[Path, Path, Path], monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
src, dst, arch = dirs
|
||||
f = write(src / "a.pdf", b"original")
|
||||
real = LocalDestination._apply_meta
|
||||
|
||||
def change(self: LocalDestination, rel: str, meta: object) -> list[str]:
|
||||
if self.root == arch:
|
||||
f.write_bytes(b"geaendert!!")
|
||||
return real(self, rel, meta) # type: ignore[arg-type]
|
||||
|
||||
monkeypatch.setattr(LocalDestination, "_apply_meta", change)
|
||||
result = run(engine, snap(rule_id), src, dst, arch)
|
||||
assert result.items[0].status is ItemStatus.SKIPPED
|
||||
assert names(dst) == [] and names(arch) == [] and f.exists()
|
||||
|
||||
|
||||
def test_copy_mode_with_archive(
|
||||
engine: Engine, rule_id: int, dirs: tuple[Path, Path, Path]
|
||||
) -> None:
|
||||
src, dst, arch = dirs
|
||||
f = write(src / "a.pdf", b"v1")
|
||||
rule = snap(rule_id, action=RuleAction.COPY)
|
||||
for _ in range(3):
|
||||
run(engine, rule, src, dst, arch)
|
||||
assert f.exists() and names(dst) == ["a.pdf"] and names(arch) == ["a.pdf"]
|
||||
f.write_bytes(b"version 2")
|
||||
run(engine, rule, src, dst, arch)
|
||||
assert names(dst) == ["a-1.pdf", "a.pdf"] and names(arch) == ["a-1.pdf", "a.pdf"]
|
||||
|
||||
|
||||
def test_dry_run_shows_archive_and_writes_nothing(
|
||||
engine: Engine, rule_id: int, dirs: tuple[Path, Path, Path]
|
||||
) -> None:
|
||||
src, dst, arch = dirs
|
||||
write(src / "a.pdf")
|
||||
result = run(engine, snap(rule_id, archive_layout="date"), src, dst, arch, dry_run=True)
|
||||
assert result.items[0].status is ItemStatus.PLANNED
|
||||
assert f"Archiv: {arch / '2026/10/a.pdf'}" in (result.items[0].message or "")
|
||||
assert names(dst) == [] and names(arch) == [] and (src / "a.pdf").exists()
|
||||
|
||||
|
||||
def test_without_archive_unchanged(
|
||||
engine: Engine, rule_id: int, dirs: tuple[Path, Path, Path]
|
||||
) -> None:
|
||||
src, dst, arch = dirs
|
||||
write(src / "a.pdf")
|
||||
result = run(engine, snap(rule_id), src, dst, None)
|
||||
assert result.items[0].message is None and names(arch) == []
|
||||
|
||||
|
||||
# -- Regellauf ----------------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def setup(
|
||||
manager: MountManager,
|
||||
fake_backend: FakeMountBackend,
|
||||
make_mount: MountFactory,
|
||||
session_factory: sessionmaker[Session],
|
||||
tmp_path: Path,
|
||||
) -> Iterator[tuple[int, Path, Path, Path]]:
|
||||
scans = fake_backend.export("nas", "scans")
|
||||
arch_remote = fake_backend.export("nas", "archiv")
|
||||
src_id = make_mount("scans", share="scans")
|
||||
arch_id = make_mount("archiv", share="archiv")
|
||||
manager.mount(src_id)
|
||||
manager.mount(arch_id)
|
||||
consume = tmp_path / "consume"
|
||||
consume.mkdir()
|
||||
with session_scope(session_factory) as s:
|
||||
r = Rule(
|
||||
name="Scans → Paperless + Archiv",
|
||||
src_mount_id=src_id,
|
||||
dst_path=str(consume),
|
||||
stability_seconds=0,
|
||||
archive_kind=ArchiveKind.MOUNT,
|
||||
archive_mount_id=arch_id,
|
||||
archive_path="scans",
|
||||
)
|
||||
s.add(r)
|
||||
s.flush()
|
||||
rid = r.id
|
||||
yield rid, scans, consume, arch_remote
|
||||
|
||||
|
||||
def test_runner_with_archive_mount(
|
||||
setup: tuple[int, Path, Path, Path],
|
||||
manager: MountManager,
|
||||
session_factory: sessionmaker[Session],
|
||||
) -> None:
|
||||
rid, scans, consume, arch = setup
|
||||
write(scans / "doc.pdf", b"%PDF")
|
||||
result = RuleRunner(session_factory, manager, sleep=lambda _: None).run(rid)
|
||||
assert result.status is RunStatus.OK, result.message
|
||||
assert (
|
||||
(consume / "doc.pdf").read_bytes() == b"%PDF" == (arch / "scans" / "doc.pdf").read_bytes()
|
||||
)
|
||||
assert not (scans / "doc.pdf").exists()
|
||||
|
||||
|
||||
def test_unavailable_archive_makes_rule_wait(
|
||||
setup: tuple[int, Path, Path, Path],
|
||||
manager: MountManager,
|
||||
fake_backend: FakeMountBackend,
|
||||
session_factory: sessionmaker[Session],
|
||||
) -> None:
|
||||
rid, scans, consume, _arch = setup
|
||||
write(scans / "doc.pdf")
|
||||
fake_backend.vanish(manager.mountpoint_for("archiv"))
|
||||
result = RuleRunner(session_factory, manager, sleep=lambda _: None).run(rid)
|
||||
assert result.unavailable and result.status is RunStatus.ERROR
|
||||
assert (scans / "doc.pdf").exists() and list(consume.iterdir()) == []
|
||||
|
||||
|
||||
def test_mount_used_as_archive_cannot_be_deleted(
|
||||
setup: tuple[int, Path, Path, Path],
|
||||
session_factory: sessionmaker[Session],
|
||||
manager: MountManager,
|
||||
) -> None:
|
||||
from netmover.admin import AdminError, delete_mount
|
||||
from netmover.models import Mount
|
||||
|
||||
with session_scope(session_factory) as s:
|
||||
arch_id = s.scalar(select(Mount.id).where(Mount.name == "archiv"))
|
||||
assert arch_id is not None
|
||||
with pytest.raises(AdminError, match="Wird noch von Regeln benutzt"):
|
||||
delete_mount(s, manager, arch_id)
|
||||
|
||||
|
||||
def test_rule_form_archive_validation(
|
||||
session_factory: sessionmaker[Session], make_mount: MountFactory
|
||||
) -> None:
|
||||
from netmover.admin import parse_rule_form, save_rule
|
||||
|
||||
mid = make_mount()
|
||||
base = {
|
||||
"name": "x",
|
||||
"src_mount_id": str(mid),
|
||||
"dst_kind": "local",
|
||||
"dst_path": "/srv/consume",
|
||||
"interval_seconds": "30",
|
||||
"stability_seconds": "5",
|
||||
"op_timeout_seconds": "120",
|
||||
}
|
||||
with session_scope(session_factory) as s:
|
||||
_, err = parse_rule_form(
|
||||
s, {**base, "archive_kind": "local", "archive_path": "/srv/consume"}
|
||||
)
|
||||
assert "derselbe Ordner" in err["archive_path"]
|
||||
_, err = parse_rule_form(s, {**base, "archive_kind": "mount", "archive_path": "a"})
|
||||
assert "archive_mount_id" in err
|
||||
_, err = parse_rule_form(s, {**base, "archive_kind": "local", "archive_path": "relativ"})
|
||||
assert "absoluten Pfad" in err["archive_path"]
|
||||
d, err = parse_rule_form(
|
||||
s,
|
||||
{
|
||||
**base,
|
||||
"archive_kind": "mount",
|
||||
"archive_mount_id": str(mid),
|
||||
"archive_path": "archiv/scans",
|
||||
"archive_layout": "date",
|
||||
},
|
||||
)
|
||||
assert err == {}
|
||||
r, _ = save_rule(s, d)
|
||||
assert r is not None and r.archive_kind is ArchiveKind.MOUNT
|
||||
assert r.archive_path == "archiv/scans" and r.archive_layout.value == "date"
|
||||
Reference in New Issue
Block a user