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 clean_new_filename(filename): """Validate and normalize a user-supplied MP3 filename.""" if not isinstance(filename, str): return None filename = filename.strip() if not filename: return None if Path(filename).name != filename: return None if filename.lower().endswith(".mp3"): filename = filename[:-4] if not filename or filename in {".", ".."}: return None if any(ch in filename for ch in '/\\\x00'): return None return filename + ".mp3" def directory_for_kind(kind): if kind == "recording": return RECORDINGS_DIR if kind == "finished": return FINISHED_DIR return None def path_for_kind(kind, filename): filename = safe_filename(filename) directory = directory_for_kind(kind) if filename is None or directory is None: return None return directory / filename def rename_file(kind, old_filename, new_filename): old_path = path_for_kind(kind, old_filename) new_filename = clean_new_filename(new_filename) if old_path is None or new_filename is None or not old_path.is_file(): raise ValueError("Ungültiger Dateiname") new_path = old_path.with_name(new_filename) if new_path.exists(): raise ValueError("Eine Datei mit diesem Namen existiert bereits") old_waveform = waveform_path(old_path.name) if kind == "recording" else None new_waveform = waveform_path(new_path.name) if kind == "recording" else None old_path.rename(new_path) if old_waveform and old_waveform.is_file(): try: old_waveform.rename(new_waveform) except OSError: # The audio file was renamed successfully; an old waveform cache # is harmless and will be regenerated under the new name. pass return new_path def delete_file(kind, filename): path = path_for_kind(kind, filename) if path is None or not path.is_file(): raise ValueError("Datei nicht gefunden") path.unlink() if kind == "recording": try: waveform_path(path.name).unlink(missing_ok=True) except OSError: pass 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", "-fflags", "+genpts", "-ss", f"{start:.3f}", "-i", str(source), "-t", f"{end - start:.3f}", "-map", "0:a:0", "-vn", "-sn", "-dn", "-af", "loudnorm=I=-16:TP=-1.5:LRA=11", "-c:a", "libmp3lame", "-b:a", "128k", "-avoid_negative_ts", "make_zero", 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 }}
Zoom: 100 % 00:00:00
Anfang00:00:00
Ende00:00:00
Neue Dauer00:00:00
Original herunterladen
Beim Erstellen einer fertigen Aufnahme bleibt das Original unverändert. Das Original kann hier bewusst gelöscht werden; bereits erstellte fertige Aufnahmen bleiben dabei erhalten.
""" 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/file", methods=["POST"]) def api_file(): data = request.get_json(silent=True) or {} action = data.get("action") kind = data.get("kind") filename = data.get("filename") if kind not in {"recording", "finished"}: return jsonify({"success": False, "error": "Ungültiger Dateityp"}), 400 try: if action == "rename": path = rename_file(kind, filename, data.get("new_filename")) return jsonify({"success": True, "filename": path.name}) if action == "delete": delete_file(kind, filename) return jsonify({"success": True}) return jsonify({"success": False, "error": "Ungültige Aktion"}), 400 except ValueError as exc: return jsonify({"success": False, "error": str(exc)}), 400 except OSError as exc: print(f"[file] operation failed for {filename}: {exc}", flush=True) return jsonify({"success": False, "error": "Dateioperation fehlgeschlagen"}), 500 @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)