feat: Regeln kopieren – vorausgefüllt, zunächst deaktiviert

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

CHANGELOG.md

@@ -8,6 +8,14 @@ Die Version steht an genau einer Stelle: `src/netmover/__init__.py`.
 
 ## [Unreleased]
 
+## [0.9.0] – 2026-10-15
+
+### Hinzugefügt
+- **Regeln kopieren** (Knopf „Kopieren“): alle Einstellungen inklusive Archiv
+  übernommen, Name „… (Kopie)“. Die Kopie ist zunächst **deaktiviert**, damit sie nicht
+  mit dem Original um dieselben Dateien konkurriert; Protokoll und Buchführung
+  beginnen neu.
+
 ## [0.8.0] – 2026-10-12
 
 ### Hinzugefügt
@@ -269,7 +277,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.8.0...HEAD
+[Unreleased]: https://git.wilde-server.de/florian.wilde/netmover/compare/v0.9.0...HEAD
+[0.9.0]: https://git.wilde-server.de/florian.wilde/netmover/compare/v0.8.0...v0.9.0
 [0.8.0]: https://git.wilde-server.de/florian.wilde/netmover/compare/v0.7.0...v0.8.0
 [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

DEVLOG.md

@@ -22,10 +22,22 @@ Zeiten in Ortszeit (Europe/Berlin).
 | 2026-10-11 | 08:43 | 08:55 | 0:12  | –       | Anmeldung je Netz (0.6.0)              |
 | 2026-10-12 | 00:14 | 00:34 | 0:20  | –       | Archiv je Regel (0.7.0)                |
 | 2026-10-12 | 00:46 | 00:55 | 0:09  | –       | Anbindungen kopieren (0.8.0)           |
-| **Summe**  |       |       | **3:58** |      |                                        |
+| 2026-10-15 | 09:17 | 09:26 | 0:09  | –       | Regeln kopieren (0.9.0)                |
+| **Summe**  |       |       | **4:07** |      |                                        |
 
 ---
 
+## 2026-10-15 · 09:17–09:26 · Regeln kopieren (0.9.0)
+
+**Erledigt** – Knopf „Kopieren“, Route `/rules/{id}/copy` (GET Formular, POST
+speichern), `rule_copy_name`. 5 Tests (362 gesamt); Gegenprobe „Kopie aktiv“ schlägt
+an (2 Tests).
+
+**Entscheidung** – Kopie startet deaktiviert: Mit identischer Quelle würden Original
+und Kopie sonst sofort parallel dieselben Dateien bearbeiten (doppelte Lieferung bzw.
+Kollisionen). Protokoll und Buchführung (`copied_files`, `archived_files`) werden
+bewusst nicht mitkopiert – die Kopie ist eine neue Regel.
+
 ## 2026-10-12 · 00:46–00:55 · Anbindungen kopieren (0.8.0)
 
 **Erledigt** – Knopf „Kopieren“, Route `/mounts/{id}/copy` (GET Formular, POST

ROADMAP.md

@@ -28,6 +28,7 @@ Legende: ✅ erledigt · 🔨 in Arbeit · 📋 geplant · 💡 Idee (noch nicht
 
 | Version | Thema |
 |---|---|
+| 0.9.0 | Regeln kopieren (Kopie zunächst deaktiviert) |
 | 0.8.0 | Anbindungen kopieren inklusive Zugangsdaten |
 | 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 |

compose.yml

@@ -4,7 +4,7 @@
 services:
   netmover:
     build: .
-    image: netmover:0.8.0
+    image: netmover:0.9.0
     container_name: netmover
     restart: unless-stopped
     cap_add:

docs/ANLEITUNG.md

@@ -226,6 +226,17 @@ Synology-Verwaltungsordner (`@eaDir`, `#recycle`).
 | **Modus setzen** | z. B. `0644`; leer = wie die Quelldatei |
 | **Quarantäneordner** | Dateien, die mehrmals hintereinander scheitern, werden dorthin verschoben (relativ zur Quell-Anbindung), statt es endlos zu versuchen. Leer = keine Quarantäne. |
 
+### Regel kopieren {#regel-kopieren}
+
+Für eine ähnliche Regel – etwa für einen zweiten Quell-Unterordner – bei der Vorlage
+auf **Kopieren** klicken. Alle Einstellungen (Filter, Ziel, Archiv, Sicherheit,
+Rechte) sind übernommen, der Name lautet „… (Kopie)“.
+
+Die Kopie ist zunächst **deaktiviert**: Mit gleicher Quelle würde sie sonst sofort mit
+dem Original um dieselben Dateien konkurrieren. Also Quelle bzw. Ziel anpassen,
+speichern, **Probelauf**, dann **Aktivieren**. Protokoll und Buchführung beginnen bei
+der Kopie neu.
+
 ### Probelauf {#probelauf}
 
 **Probelauf** zeigt in einer Tabelle, welche Datei wohin ginge – **ohne etwas zu
@@ -428,7 +439,9 @@ deaktiviert, Zugangsdaten müssen neu eingegeben werden.
 ### Jede Person in einen eigenen Ordner {#beispiel-unterordner}
 
 Scannt jede Person in einen eigenen Unterordner (`Scans/max`, `Scans/erika`): pro
-Person eine Regel mit eigenem **Unterordner** in der Quelle und eigenem Ziel.
+Person eine Regel mit eigenem **Unterordner** in der Quelle und eigenem Ziel. Am
+schnellsten: die erste Regel einrichten, dann **Kopieren** und nur Unterordner und
+Ziel ändern.
 
 ### Kopie als zusätzliche Sicherung {#beispiel-kopie}
 

src/netmover/__init__.py

@@ -1,3 +1,3 @@
 """NetMover – NFS/SMB-Mounts verwalten und Dateien zeitgesteuert verschieben."""
 
-__version__ = "0.8.0"
+__version__ = "0.9.0"

src/netmover/admin.py

@@ -668,6 +668,16 @@ def parse_rule_form(s: Session, form: Mapping[str, str]) -> tuple[RuleInput, Err
     return d, errors
 
 
+def rule_copy_name(s: Session, name: str) -> str:
+    """„Scans (Kopie)“, „Scans (Kopie 2)“, … – frei in der Datenbank."""
+    base = f"{name} (Kopie)"[:100]
+    candidate, n = base, 1
+    while s.scalar(select(Rule.id).where(Rule.name == candidate)) is not None:
+        n += 1
+        candidate = f"{name} (Kopie {n})"[:100]
+    return candidate
+
+
 def save_rule(s: Session, d: RuleInput, rule_id: int | None = None) -> tuple[Rule | None, Errors]:
     dup = s.scalar(select(Rule.id).where(Rule.name == d.name))
     if dup is not None and dup != rule_id:

src/netmover/web/app.py

@@ -44,6 +44,7 @@ from netmover.admin import (
     mount_input_from_model,
     parse_mount_form,
     parse_rule_form,
+    rule_copy_name,
     rule_input_from_model,
     save_mount,
     save_rule,
@@ -902,6 +903,7 @@ def _rule_form(
     errors: dict[str, str],
     rule_id: int | None,
     status_code: int = 200,
+    copy_of: str | None = None,
 ) -> Response:
     with nm(request).session_factory() as s:
         mounts = s.scalars(select(Mount).order_by(Mount.name)).all()
@@ -915,6 +917,7 @@ def _rule_form(
         d=d,
         errors=errors,
         rule_id=rule_id,
+        copy_of=copy_of,
         src_mounts=mounts,
         src_options=[("", "– wählen –")] + [(m.id, f"{m.name} ({m.type.value})") for m in mounts],
         dst_options=[("", "– wählen –")] + [(m.id, f"{m.name} ({m.type.value})") for m in mounts],
@@ -942,25 +945,47 @@ async def rule_edit(request: Request, rule_id: int) -> Response:
     return _rule_form(request, d, {}, rule_id)
 
 
+@protected.get("/rules/{source_id}/copy", response_class=HTMLResponse)
+async def rule_copy(request: Request, source_id: int) -> Response:
+    """Neue Regel, vorausgefüllt aus einer bestehenden – zunächst deaktiviert, damit sie
+    nicht sofort mit dem Original um dieselben Dateien konkurriert."""
+    with nm(request).session_factory() as s:
+        src = s.get(Rule, source_id)
+        if src is None:
+            return redirect("/rules")
+        d = rule_input_from_model(src)
+        d.name = rule_copy_name(s, src.name)
+        d.enabled = False
+        source_name = src.name
+    return _rule_form(request, d, {}, None, copy_of=source_name)
+
+
 @protected.post("/rules/new", response_class=HTMLResponse)
 @protected.post("/rules/{rule_id}/edit", response_class=HTMLResponse)
-async def rule_save(request: Request, rule_id: int | None = None) -> Response:
+@protected.post("/rules/{source_id}/copy", response_class=HTMLResponse)
+async def rule_save(
+    request: Request, rule_id: int | None = None, source_id: int | None = None
+) -> Response:
     state = nm(request)
     form = _form_dict(await request.form())
     with session_scope(state.session_factory) as s:
+        source = s.get(Rule, source_id) if source_id is not None else None
+        copy_of = source.name if source is not None else None
         d, errors = parse_rule_form(s, form)
         r = None
         if not errors:
             try:
-                r, errors = save_rule(s, d, rule_id)
+                r, errors = save_rule(s, d, rule_id)  # Kopie: rule_id ist None → neue Regel
             except AdminError as exc:
                 errors = {"_": str(exc)}
         if errors or r is None:
             s.rollback()
-            return _rule_form(request, d, errors, rule_id, status_code=400)
+            return _rule_form(request, d, errors, rule_id, status_code=400, copy_of=copy_of)
-        name = r.name
+        name, enabled = r.name, r.enabled
     state.scheduler.sync()
     flash(request, f"Regel „{name}“ gespeichert.")
+    if copy_of and not enabled:
+        flash(request, "Die Kopie ist deaktiviert. Probelauf machen, dann aktivieren.", "info")
     return redirect("/rules")
 
 

src/netmover/web/templates/rule_form.html

@@ -1,8 +1,11 @@
 {% extends "base.html" %}
 {% from "macros.html" import field, select, check %}
-{% block title %}{{ 'Regel bearbeiten' if rule_id else 'Neue Regel' }}{% endblock %}
+{% block title %}{{ 'Regel bearbeiten' if rule_id else ('Regel kopieren' if copy_of else 'Neue Regel') }}{% endblock %}
 {% block content %}
-<h1>{{ 'Regel bearbeiten' if rule_id else 'Neue Regel' }}</h1>
+<h1>{% if rule_id %}Regel bearbeiten{% elif copy_of %}Kopie von „{{ copy_of }}“{% else %}Neue Regel{% endif %}</h1>
+{% if copy_of %}<p class="msg info small">Alle Einstellungen stammen aus „{{ copy_of }}“. Die Kopie ist zunächst
+  <strong>deaktiviert</strong>, damit sie nicht mit dem Original um dieselben Dateien konkurriert – Quelle bzw. Ziel
+  anpassen, speichern, Probelauf machen, dann aktivieren. Protokoll und Buchführung beginnen bei der Kopie neu.</p>{% endif %}
 {% if errors.get('_') %}<p class="msg error">{{ errors['_'] }}</p>{% endif %}
 {% if not src_mounts %}<p class="msg info">Zuerst eine Anbindung als Quelle <a href="/mounts/new">anlegen</a>.</p>{% endif %}
 <form class="form" method="post" data-dynamic>

src/netmover/web/templates/rules.html

@@ -18,6 +18,7 @@
       <button class="small" hx-post="/rules/{{ r.id }}/dryrun" hx-target="#dry-{{ r.id }}" hx-swap="innerHTML">Probelauf <span class="htmx-indicator">…</span></button>
       <form class="inline" method="post" action="/rules/{{ r.id }}/toggle">{{ csrf_field(csrf) }}<button class="small">{{ 'Deaktivieren' if r.enabled else 'Aktivieren' }}</button></form>
       <a class="btn small" href="/rules/{{ r.id }}/edit">Bearbeiten</a>
+      <a class="btn small" href="/rules/{{ r.id }}/copy" title="Neue Regel mit denselben Einstellungen">Kopieren</a>
       <form class="inline" method="post" action="/rules/{{ r.id }}/delete" data-confirm="Regel „{{ r.name }}“ und ihr Protokoll löschen?">{{ csrf_field(csrf) }}<button class="small danger">Löschen</button></form>
     </div>
   </div>

tests/test_copy_rule.py

@@ -0,0 +1,168 @@
+"""0.9.0: Regeln kopieren – vorausgefüllt, zunächst deaktiviert."""
+
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Any
+
+from fastapi.testclient import TestClient
+from sqlalchemy import select
+
+from netmover.db import session_scope
+from netmover.models import (
+    ArchiveKind,
+    ArchiveLayout,
+    CopiedFile,
+    Mount,
+    MountType,
+    Rule,
+    RuleAction,
+    Run,
+)
+from tests.webutil import post
+
+
+def _setup(state: Any, tmp_path: Path) -> tuple[int, int]:
+    consume = tmp_path / "consume"
+    consume.mkdir()
+    with session_scope(state.session_factory) as s:
+        m = Mount(
+            name="NAS",
+            slug="nas",
+            type=MountType.CIFS,
+            server="nas",
+            share="scans",
+            mountpoint=str(tmp_path / "mnt" / "nas"),
+        )
+        a = Mount(
+            name="Archiv",
+            slug="archiv",
+            type=MountType.CIFS,
+            server="nas",
+            share="archiv",
+            mountpoint=str(tmp_path / "mnt" / "archiv"),
+        )
+        s.add_all([m, a])
+        s.flush()
+        r = Rule(
+            name="Scans → Paperless",
+            src_mount_id=m.id,
+            src_subpath="max",
+            dst_path=str(consume),
+            pattern="*.pdf;*.jpg",
+            recursive=True,
+            interval_seconds=45,
+            stability_seconds=7,
+            action=RuleAction.COPY,
+            set_mode="0640",
+            quarantine_path="_q",
+            archive_kind=ArchiveKind.MOUNT,
+            archive_mount_id=a.id,
+            archive_path="scans",
+            archive_layout=ArchiveLayout.DATE,
+        )
+        s.add(r)
+        s.flush()
+        s.add(Run(rule_id=r.id))
+        s.add(CopiedFile(rule_id=r.id, src_relpath="a.pdf", size=1, mtime_ns=1, dst_path="x"))
+        return r.id, m.id
+
+
+def _rule(state: Any, name: str) -> Rule:
+    with state.session_factory() as s:
+        r = s.scalars(select(Rule).where(Rule.name == name)).one()
+        s.expunge(r)
+        return r
+
+
+def _form_from_page(html: str) -> dict[str, str]:
+    """Wie ein Browser: alle Felder des Formulars mit ihren vorausgefüllten Werten."""
+    import re
+
+    form: dict[str, str] = {}
+    for m in re.finditer(r'<input[^>]*name="([^"]+)"[^>]*>', html):
+        tag, name = m.group(0), m.group(1)
+        if 'type="checkbox"' in tag:
+            if " checked" in tag:
+                form[name] = "on"
+        else:
+            v = re.search(r'value="([^"]*)"', tag)
+            form[name] = v.group(1) if v else ""
+    for m in re.finditer(r'<select[^>]*name="([^"]+)"[^>]*>(.*?)</select>', html, re.S):
+        sel = re.search(r'<option value="([^"]*)" selected>', m.group(2))
+        form[m.group(1)] = sel.group(1) if sel else ""
+    return form
+
+
+def test_copy_form_is_prefilled_and_disabled(
+    client: TestClient, state: Any, tmp_path: Path
+) -> None:
+    rid, _ = _setup(state, tmp_path)
+    assert f'href="/rules/{rid}/copy"' in client.get("/rules").text
+    html = client.get(f"/rules/{rid}/copy").text
+    assert "Kopie von „Scans → Paperless“" in html and "deaktiviert" in html
+    form = _form_from_page(html)
+    assert form["name"] == "Scans → Paperless (Kopie)"
+    assert "enabled" not in form  # Häkchen „Aktiv“ nicht gesetzt
+    assert form["pattern"] == "*.pdf;*.jpg" and form["src_subpath"] == "max"
+    assert form["archive_kind"] == "mount" and form["archive_layout"] == "date"
+
+
+def test_copy_saves_new_rule_with_all_settings(
+    client: TestClient, state: Any, tmp_path: Path
+) -> None:
+    rid, _ = _setup(state, tmp_path)
+    form = _form_from_page(client.get(f"/rules/{rid}/copy").text)
+    form["src_subpath"] = "erika"  # typische Anpassung
+    r = post(client, f"/rules/{rid}/copy", form)
+    assert r.status_code == 303, r.text
+    assert "Die Kopie ist deaktiviert" in client.get("/rules").text
+
+    orig, copy = _rule(state, "Scans → Paperless"), _rule(state, "Scans → Paperless (Kopie)")
+    assert copy.id != orig.id and copy.enabled is False and orig.enabled is True
+    assert copy.src_subpath == "erika" and orig.src_subpath == "max"
+    for f in (
+        "src_mount_id",
+        "dst_path",
+        "pattern",
+        "recursive",
+        "interval_seconds",
+        "stability_seconds",
+        "action",
+        "set_mode",
+        "quarantine_path",
+        "archive_kind",
+        "archive_mount_id",
+        "archive_path",
+        "archive_layout",
+    ):
+        assert getattr(copy, f) == getattr(orig, f), f
+    # Protokoll und Buchführung beginnen neu
+    with state.session_factory() as s:
+        assert s.scalars(select(Run).where(Run.rule_id == copy.id)).all() == []
+        assert s.scalars(select(CopiedFile).where(CopiedFile.rule_id == copy.id)).all() == []
+    assert state.scheduler.next_run(copy.id) is None  # deaktiviert → kein Termin
+
+
+def test_copy_names_count_up(client: TestClient, state: Any, tmp_path: Path) -> None:
+    rid, _ = _setup(state, tmp_path)
+    for expected in ("Scans → Paperless (Kopie)", "Scans → Paperless (Kopie 2)"):
+        form = _form_from_page(client.get(f"/rules/{rid}/copy").text)
+        assert form["name"] == expected
+        post(client, f"/rules/{rid}/copy", form)
+
+
+def test_copy_validation_error_keeps_context(
+    client: TestClient, state: Any, tmp_path: Path
+) -> None:
+    rid, _ = _setup(state, tmp_path)
+    form = _form_from_page(client.get(f"/rules/{rid}/copy").text)
+    form["name"] = "Scans → Paperless"  # schon vergeben
+    r = post(client, f"/rules/{rid}/copy", form)
+    assert r.status_code == 400 and "bereits vergeben" in r.text and "Kopie von" in r.text
+    with state.session_factory() as s:
+        assert len(s.scalars(select(Rule)).all()) == 1
+
+
+def test_copy_of_missing_rule_redirects(client: TestClient) -> None:
+    assert client.get("/rules/4711/copy").headers["location"] == "/rules"