import os import subprocess import threading import time import uuid 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")) NORMALIZE = os.environ.get("NORMALIZE", "true").lower() not in ("0", "false", "no") NORMALIZE_TARGET = os.environ.get("NORMALIZE_TARGET", "-16") NORMALIZE_TRUE_PEAK = os.environ.get("NORMALIZE_TRUE_PEAK", "-1.5") NORMALIZE_LRA = os.environ.get("NORMALIZE_LRA", "11") JOBS = {} JOBS_LOCK = threading.Lock() # --------------------------------------------------------------------------- # 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, progress_callback=None): 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" selected_duration = end - start cmd = [ "ffmpeg", "-hide_banner", "-loglevel", "error", "-y", "-fflags", "+genpts", "-ss", f"{start:.3f}", "-i", str(source), "-t", f"{selected_duration:.3f}", "-map", "0:a:0", "-vn", "-sn", "-dn", *( ["-af", f"loudnorm=I={NORMALIZE_TARGET}:TP={NORMALIZE_TRUE_PEAK}:LRA={NORMALIZE_LRA}"] if NORMALIZE else [] ), "-c:a", "libmp3lame", "-b:a", "128k", "-avoid_negative_ts", "make_zero", "-progress", "pipe:1", "-nostats", str(temp), ] try: process = subprocess.Popen( cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, bufsize=1 ) stderr_lines = [] stderr_thread = threading.Thread( target=lambda: stderr_lines.extend(process.stderr.readlines()), daemon=True, ) stderr_thread.start() last_progress = -1 while True: line = process.stdout.readline() if not line: if process.poll() is not None: break time.sleep(0.05) continue line = line.strip() if line.startswith("out_time_us="): try: out_us = int(line.split("=", 1)[1]) progress = max(0.0, min(100.0, out_us / 1_000_000 / selected_duration * 100.0)) if progress_callback and (progress - last_progress >= 0.2 or progress >= 100): progress_callback(progress) last_progress = progress except (ValueError, ZeroDivisionError): pass returncode = process.wait() stderr_thread.join(timeout=2) stderr = "".join(stderr_lines).strip() if returncode != 0 or not temp.is_file(): try: temp.unlink(missing_ok=True) except OSError: pass print(f"[edit] ffmpeg failed for {filename}: {stderr}", flush=True) raise RuntimeError("Schneiden fehlgeschlagen" + (f": {stderr[-500]}" if stderr else "")) temp.replace(destination) if progress_callback: progress_callback(100.0) return destination, get_duration(destination) except Exception: try: temp.unlink(missing_ok=True) except OSError: pass raise def run_edit_job(job_id, filename, start, end): def update(**values): with JOBS_LOCK: job = JOBS.get(job_id) if job: job.update(values) update(status="running", progress=0.0) try: destination, duration = create_finished( filename, start, end, progress_callback=lambda value: update(progress=value), ) update(status="done", progress=100.0, filename=destination.name, duration=duration, finished_at=time.time()) except Exception as exc: print(f"[edit] job {job_id} failed for {filename}: {exc}", flush=True) update(status="error", error=str(exc), finished_at=time.time()) # --------------------------------------------------------------------------- # 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.
Verarbeitung läuft …0 %
""" 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") start = data.get("start") end = data.get("end") source = source_path(filename) if source is None: return jsonify({"success": False, "error": "Aufnahme nicht gefunden"}), 404 try: start_f = float(start) end_f = float(end) except (TypeError, ValueError): return jsonify({"success": False, "error": "Ungültige Zeitangabe"}), 400 if end_f <= start_f or end_f - start_f < 0.5: return jsonify({"success": False, "error": "Ungültiger Auswahlbereich"}), 400 job_id = uuid.uuid4().hex with JOBS_LOCK: JOBS[job_id] = {"status":"queued", "progress":0.0, "filename":filename, "created_at":time.time()} threading.Thread(target=run_edit_job, args=(job_id, filename, start_f, end_f), daemon=True).start() return jsonify({"success": True, "job_id": job_id}), 202 @app.route("/api/edit/status/") def api_edit_status(job_id): with JOBS_LOCK: job = JOBS.get(job_id) if not job: return jsonify({"success": False, "error": "Job nicht gefunden"}), 404 result = dict(job) return jsonify({"success": True, **result}) 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)