From b252b83eb680fcd546605915207504529ef3fd02 Mon Sep 17 00:00:00 2001 From: bernd Date: Tue, 29 Sep 2026 07:43:00 +0200 Subject: [PATCH] =?UTF-8?q?editor.py=20hinzugef=C3=BCgt?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- editor.py | 568 ++++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 568 insertions(+) create mode 100644 editor.py diff --git a/editor.py b/editor.py new file mode 100644 index 0000000..4b9dfdd --- /dev/null +++ b/editor.py @@ -0,0 +1,568 @@ +import os +import subprocess +from datetime import datetime +from pathlib import Path +from zoneinfo import ZoneInfo + +from flask import Flask, abort, jsonify, render_template_string, request, send_from_directory, url_for + + +# --------------------------------------------------------------------------- +# Configuration +# --------------------------------------------------------------------------- + +RECORDINGS_DIR = Path(os.environ.get("RECORDINGS_DIR", "/recordings")) +FINISHED_DIR = Path(os.environ.get("FINISHED_DIR", "/finished")) +WAVEFORMS_DIR = Path(os.environ.get("WAVEFORMS_DIR", "/waveforms")) +WEB_PORT = int(os.environ.get("WEB_PORT", "8080")) + + +# --------------------------------------------------------------------------- +# Application +# --------------------------------------------------------------------------- + +app = Flask(__name__) + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +def format_datetime(timestamp): + return datetime.fromtimestamp(timestamp, ZoneInfo("Europe/Berlin")).strftime( + "%d.%m.%Y %H:%M:%S" + ) + + +def format_size(size): + if size < 1024: + return f"{size} B" + if size < 1024 ** 2: + return f"{size / 1024:.1f} KB" + if size < 1024 ** 3: + return f"{size / 1024 ** 2:.1f} MB" + return f"{size / 1024 ** 3:.2f} GB" + + +def format_duration(seconds): + if seconds is None: + return "–" + seconds = int(seconds) + hours = seconds // 3600 + minutes = (seconds % 3600) // 60 + seconds = seconds % 60 + if hours: + return f"{hours}:{minutes:02d}:{seconds:02d}" + return f"{minutes}:{seconds:02d}" + + +def safe_filename(filename): + if not filename or Path(filename).name != filename: + return None + if not filename.lower().endswith(".mp3"): + return None + return filename + + +def source_path(filename): + filename = safe_filename(filename) + if filename is None: + return None + path = RECORDINGS_DIR / filename + return path if path.is_file() else None + + +def finished_path(filename): + filename = safe_filename(filename) + if filename is None: + return None + return FINISHED_DIR / filename + + +def get_duration(path): + try: + result = subprocess.run( + [ + "ffprobe", "-v", "error", + "-show_entries", "format=duration", + "-of", "default=noprint_wrappers=1:nokey=1", + str(path), + ], + capture_output=True, text=True, timeout=10, + ) + if result.returncode != 0: + return None + return float(result.stdout.strip()) + except (OSError, ValueError, subprocess.SubprocessError): + return None + + +def waveform_path(filename): + return WAVEFORMS_DIR / f"{filename}.png" + + +def create_waveform(filename): + source = source_path(filename) + if source is None: + return None + + output = waveform_path(filename) + WAVEFORMS_DIR.mkdir(parents=True, exist_ok=True) + + try: + if output.is_file() and output.stat().st_mtime >= source.stat().st_mtime: + return output + except FileNotFoundError: + pass + + temp = output.with_suffix(".tmp.png") + try: + result = subprocess.run( + [ + "ffmpeg", "-hide_banner", "-loglevel", "error", "-y", + "-i", str(source), + "-filter_complex", + "aformat=channel_layouts=mono,showwavespic=s=2400x240:split_channels=0", + "-frames:v", "1", + str(temp), + ], + capture_output=True, text=True, timeout=120, + ) + if result.returncode != 0 or not temp.is_file(): + print(f"[waveform] failed for {filename}: {result.stderr.strip()}", flush=True) + return None + temp.replace(output) + return output + except Exception as exc: + print(f"[waveform] failed for {filename}: {exc}", flush=True) + try: + temp.unlink(missing_ok=True) + except OSError: + pass + return None + + +def create_finished(filename, start, end): + source = source_path(filename) + if source is None: + raise ValueError("Aufnahme nicht gefunden") + + try: + start = float(start) + end = float(end) + except (TypeError, ValueError): + raise ValueError("Ungültige Zeitangabe") + + duration = get_duration(source) + if duration is None: + raise ValueError("Aufnahmedauer konnte nicht ermittelt werden") + + if start < 0 or end > duration or end <= start: + raise ValueError("Ungültiger Auswahlbereich") + + if end - start < 0.5: + raise ValueError("Der ausgewählte Bereich ist zu kurz") + + FINISHED_DIR.mkdir(parents=True, exist_ok=True) + destination = finished_path(filename) + temp = FINISHED_DIR / f".{filename}.tmp.mp3" + + result = subprocess.run( + [ + "ffmpeg", "-hide_banner", "-loglevel", "warning", "-y", + "-i", str(source), + "-ss", f"{start:.3f}", + "-t", f"{end - start:.3f}", + "-c:a", "libmp3lame", + "-b:a", "128k", + str(temp), + ], + capture_output=True, text=True, timeout=1800, + ) + + if result.returncode != 0 or not temp.is_file(): + try: + temp.unlink(missing_ok=True) + except OSError: + pass + print(f"[edit] ffmpeg failed for {filename}: {result.stderr.strip()}", flush=True) + raise RuntimeError("Schneiden fehlgeschlagen") + + temp.replace(destination) + return destination, get_duration(destination) + + +# --------------------------------------------------------------------------- +# HTML +# --------------------------------------------------------------------------- + +EDITOR_HTML = """ + + + + + +Event Audio – Aufnahme bearbeiten + + + +
+← Zurück zu den Aufnahmen +

Aufnahme bearbeiten

+
{{ filename }} · Originaldauer {{ duration }}
+ +
+
+ Wellenform +
+
+
+
+
+ 00:00:00 + {{ duration }} +
+ + + +
+ + + + + 00:00:00 +
+ +
+
Anfang00:00:00
+
Ende00:00:00
+
Neue Dauer00:00:00
+
+ +
+ + Original herunterladen +
+
Die erste Version der Aufnahme wird als Original auf dem Server behalten. Weitere Schnitte sind möglich und können jederzeit auf das Original zurückgesetzt werden.
+
+
+
+ + + +""" + + +INDEX_HTML = """ + + + + + +Event Audio – Editor + + + +
+

Event Audio – Editor

+
Fertige Aufnahmen aus {{ recordings_dir }}
+ +
+

Aufnahmen bearbeiten

+{% if recordings %} + + + +{% for item in recordings %} + + + + + + + +{% endfor %} + +
AufnahmeDatumGrößeDauer
{{ item.filename }}{{ item.datetime }}{{ item.size }}{{ item.duration }}✂ Bearbeiten
+{% else %}
Keine fertigen Aufnahmen vorhanden.
{% endif %} +
+ +
+

Bearbeitete Aufnahmen

+{% if finished %} + + + +{% for item in finished %} + + + + + + + +{% endfor %} + +
DateiDatumGrößeDauer
{{ item.filename }}{{ item.datetime }}{{ item.size }}{{ item.duration }}▶ Abspielen / Download
+{% else %}
Noch keine bearbeiteten Aufnahmen vorhanden.
{% endif %} +
+
+ + +""" + + +# --------------------------------------------------------------------------- +# Routes +# --------------------------------------------------------------------------- + +@app.route("/") +def index(): + RECORDINGS_DIR.mkdir(parents=True, exist_ok=True) + FINISHED_DIR.mkdir(parents=True, exist_ok=True) + + recordings = [] + for path in RECORDINGS_DIR.glob("*.mp3"): + try: + stat = path.stat() + recordings.append({ + "filename": path.name, + "datetime": format_datetime(stat.st_mtime), + "size": format_size(stat.st_size), + "duration": format_duration(get_duration(path)), + "mtime": stat.st_mtime, + }) + except FileNotFoundError: + pass + + finished = [] + for path in FINISHED_DIR.glob("*.mp3"): + try: + stat = path.stat() + finished.append({ + "filename": path.name, + "datetime": format_datetime(stat.st_mtime), + "size": format_size(stat.st_size), + "duration": format_duration(get_duration(path)), + "mtime": stat.st_mtime, + }) + except FileNotFoundError: + pass + + recordings.sort(key=lambda x: x["mtime"], reverse=True) + finished.sort(key=lambda x: x["mtime"], reverse=True) + + return render_template_string( + INDEX_HTML, + recordings=recordings, + finished=finished, + recordings_dir=str(RECORDINGS_DIR), + ) + + +@app.route("/edit/") +def edit_recording(filename): + path = source_path(filename) + if path is None: + abort(404) + + duration = get_duration(path) + if duration is None: + abort(500, description="Aufnahmedauer konnte nicht ermittelt werden") + + waveform = create_waveform(filename) + if waveform is None: + abort(500, description="Wellenform konnte nicht erzeugt werden") + + return render_template_string( + EDITOR_HTML, + filename=filename, + duration=format_duration(duration), + duration_seconds=duration, + waveform_url=url_for("waveform", filename=filename), + audio_url=url_for("recording", filename=filename), + ) + + +@app.route("/waveforms/") +def waveform(filename): + if source_path(filename) is None: + abort(404) + output = create_waveform(filename) + if output is None: + abort(404) + return send_from_directory(WAVEFORMS_DIR, output.name, mimetype="image/png") + + +@app.route("/recordings/") +def recording(filename): + path = source_path(filename) + if path is None: + abort(404) + return send_from_directory(RECORDINGS_DIR, path.name, mimetype="audio/mpeg") + + +@app.route("/finished/") +def finished_file(filename): + path = finished_path(filename) + if path is None or not path.is_file(): + abort(404) + return send_from_directory(FINISHED_DIR, path.name, mimetype="audio/mpeg", as_attachment=False) + + +@app.route("/api/edit", methods=["POST"]) +def api_edit(): + data = request.get_json(silent=True) or {} + filename = data.get("filename") + + try: + destination, duration = create_finished(filename, data.get("start"), data.get("end")) + return jsonify({ + "success": True, + "filename": destination.name, + "duration": duration, + "duration_formatted": format_duration(duration), + }) + except ValueError as exc: + return jsonify({"success": False, "error": str(exc)}), 400 + except Exception as exc: + print(f"[edit] failed for {filename}: {exc}", flush=True) + return jsonify({"success": False, "error": "Schneiden fehlgeschlagen"}), 500 + + +if __name__ == "__main__": + RECORDINGS_DIR.mkdir(parents=True, exist_ok=True) + FINISHED_DIR.mkdir(parents=True, exist_ok=True) + app.run(host="0.0.0.0", port=WEB_PORT)