Benutzer:Flavius
Auf server3579 läuft eine automatische Pipeline, die private Video-Downloads auf der Storage Box (/mnt/storagebox/FlaviusPrivat) bereinigt und sortiert.
|
|
🧠 Entwicklung & Historie (Lessons Learned)[Bearbeiten]
Die Pipeline entstand iterativ über mehrere Zwischenschritte, bei denen typische Edge-Cases gelöst wurden:
- Symptom „Keine Dateien gefunden“: Das Ursprungsskript suchte nur lose Dateien. Die echte Struktur nutzt jedoch immer eigene Unterordner pro Film.
- Symptom „Ordner leer“ (Tiefe Strukturen): Videos lagen oft tief in Release-Unterordnern.
video-flach.pyzieht diese nun hoch. Gibt es Unklarheiten (mehrere oder keine Videos), wird der Ordner sicherheitshalber übersprungen. - Symptom „Keine Gruppenbildung“: Ordner nutzen oft das Datumsformat (
.JJ.MM.TT.). Die Erkennung trennt nun strikt zwischen echten Datumsreihen und vierstelligen Jahreszahlen wie(2010), damit unterschiedliche Filme nicht fälschlich gruppiert werden. - Fehlerschutz bei parallelem Upload: Um Datenverlust während laufender Uploads zu vermeiden, prüfen die Skripte, ob die Dateigröße über einen kurzen Zeitraum absolut stabil bleibt und die Datei seit mindestens 2 Minuten nicht mehr verändert wurde (Settle-Phase).
---
💾 Quellcode & Wiederherstellung[Bearbeiten]
Um die Skripte im Falle eines Datenverlusts wiederherzustellen, können die folgenden Blöcke ausgeklappt und kopiert werden.
1. Skripte (`/root/.local/bin/`)[Bearbeiten]
<source lang="python">
- !/usr/bin/env python3
import argparse import shutil import time from pathlib import Path
VIDEO_EXTENSIONS = {".mp4", ".mkv", ".avi", ".mov", ".webm", ".m4v", ".wmv", ".ts"} SETTLE_SEC = 120 PROBE_SEC = 15
def find_videos(folder: Path):
return [p for p in folder.rglob("*") if p.is_file() and p.suffix.lower() in VIDEO_EXTENSIONS]
def fertige_videos(videos):
snapshot = {}
for v in videos:
try:
snapshot[v] = v.stat().st_size
except OSError:
pass
if not snapshot:
return set()
time.sleep(PROBE_SEC)
fertig = set()
for v, groesse_vorher in snapshot.items():
try:
st = v.stat()
except OSError:
continue
if st.st_size == groesse_vorher and (time.time() - st.st_mtime) >= SETTLE_SEC:
fertig.add(v)
return fertig
def main():
parser = argparse.ArgumentParser(description="Filme aus Zwischenordnern eine Ebene hochziehen")
parser.add_argument("ordner", type=Path, help="Ordner mit den Namens-Unterordnern")
parser.add_argument("--apply", action="store_true", help="Tatsaechlich verschieben")
args = parser.parse_args()
if args.apply:
fertig = fertige_videos(find_videos(args.ordner))
else:
fertig = set(find_videos(args.ordner))
geaendert = False
for name_dir in sorted(args.ordner.iterdir()):
if not name_dir.is_dir():
continue
videos = find_videos(name_dir)
if len(videos) == 0 or len(videos) > 1:
continue
video = videos[0]
if video not in fertig or video.parent == name_dir:
continue
geaendert = True
inner = video.parent
if args.apply:
for f in sorted(inner.iterdir()):
ziel = name_dir / f.name
if ziel.exists():
continue
shutil.move(str(f), str(ziel))
current = inner
while current != name_dir:
eltern = current.parent
try:
current.rmdir()
except OSError:
break
current = eltern
if __name__ == "__main__":
main()
</source>
<source lang="python">
- !/usr/bin/env python3
import argparse import re import shutil import time from collections import defaultdict from pathlib import Path
VIDEO_EXTENSIONS = {".mp4", ".mkv", ".avi", ".mov", ".webm", ".m4v", ".wmv", ".ts"} SETTLE_SEC = 120 PROBE_SEC = 15
DATE_PATTERN = re.compile(r"^(.+?)\.\d{2}\.\d{2}\.\d{2}(?:\.|\s|$)") SUFFIX_PATTERN = re.compile(r"\s*[-_]?\s*[\(\[]?(?:teil|part|pt|folge)?\.?\s*(?<!\d)\d{1,3}[\)\]]?\s*$", re.IGNORECASE)
def base_name(name: str) -> str:
m = DATE_PATTERN.match(name)
if m:
return m.group(1).strip()
stripped = SUFFIX_PATTERN.sub("", name).strip()
return stripped if stripped else name
def has_video(folder: Path) -> bool:
return any(p.is_file() and p.suffix.lower() in VIDEO_EXTENSIONS for p in folder.iterdir())
def videos_in_folder(folder: Path):
return [p for p in folder.iterdir() if p.is_file() and p.suffix.lower() in VIDEO_EXTENSIONS]
def fertige_videos(videos):
snapshot = {}
for v in videos:
try: snapshot[v] = v.stat().st_size
except OSError: pass
if not snapshot: return set()
time.sleep(PROBE_SEC)
fertig = set()
for v, groesse_vorher in snapshot.items():
try: st = v.stat()
except OSError: continue
if st.st_size == groesse_vorher and (time.time() - st.st_mtime) >= SETTLE_SEC:
fertig.add(v)
return fertig
def main():
parser = argparse.ArgumentParser(description="Video-Ordner nach Basis-Namen gruppieren")
parser.add_argument("ordner", type=Path, help="Ordner mit den Film-Unterordnern")
parser.add_argument("--apply", action="store_true", help="Tatsaechlich verschieben")
args = parser.parse_args()
alle_videos = [v for d in args.ordner.iterdir() if d.is_dir() for v in videos_in_folder(d)] fertig = fertige_videos(alle_videos) if args.apply else set(alle_videos)
groups = defaultdict(list)
for d in sorted(args.ordner.iterdir()):
if d.is_dir() and has_video(d):
groups[base_name(d.name)].append(d)
for name, folders in groups.items():
if len(folders) < 2: continue
target_dir = args.ordner / name
sources = [f for f in folders if f != target_dir]
if args.apply:
target_dir.mkdir(exist_ok=True)
for src in sources:
if not all(v in fertig for v in videos_in_folder(src)): continue
for f in sorted(src.iterdir()):
ziel = target_dir / f.name
if ziel.exists(): continue
shutil.move(str(f), str(ziel))
try: src.rmdir()
except OSError: pass
if __name__ == "__main__":
main()
</source>
<source lang="bash">
- !/usr/bin/env bash
set -euo pipefail
ZIEL="/mnt/storagebox/FlaviusPrivat" MOUNT="/mnt/storagebox"
if ! mountpoint -q "$MOUNT"; then
echo "Storage Box ($MOUNT) ist nicht gemountet – breche ab." exit 0
fi
echo "=== $(date '+%F %T') : Aufraeumen gestartet ===" python3 /root/.local/bin/video-flach.py "$ZIEL" --apply python3 /root/.local/bin/video-gruppieren.py "$ZIEL" --apply echo "=== $(date '+%F %T') : Aufraeumen fertig ===" </source>
2. systemd Konfiguration (`/etc/systemd/system/`)[Bearbeiten]
<source lang="ini"> [Unit] Description=Videos flach machen und gruppieren After=network-online.target Wants=network-online.target
[Service] Type=oneshot ExecStart=/root/.local/bin/video-aufraeumen.sh </source>
<source lang="ini"> [Unit] Description=Videos regelmaessig aufraeumen (flach + gruppieren)
[Timer] OnBootSec=5min OnUnitActiveSec=30min Persistent=true
[Install] WantedBy=timers.target </source>
3. Einrichtung nach Wiederherstellung[Bearbeiten]
Um die Pipeline auf einem neuen System zu aktivieren, folgende Befehle ausführen: <source lang="bash"> chmod +x /root/.local/bin/video-flach.py chmod +x /root/.local/bin/video-gruppieren.py chmod +x /root/.local/bin/video-aufraeumen.sh
systemctl daemon-reload systemctl enable --now video-aufraeumen.timer </source>