import os import subprocess import threading import time import uuid import json 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 read_metadata(path): """Read common ID3/FFmpeg metadata from an MP3.""" try: result = subprocess.run( ["ffprobe", "-v", "error", "-show_entries", "format_tags=date,artist,title,album", "-of", "json", str(path)], capture_output=True, text=True, timeout=10, ) if result.returncode != 0: return {} data = json.loads(result.stdout) tags = data.get("format", {}).get("tags", {}) or {} return {str(k).lower(): str(v) for k, v in tags.items() if v is not None} except Exception: return {} def metadata_datetime(path): tags = read_metadata(path) value = tags.get("date", "") if value: try: dt = datetime.fromisoformat(value.replace("Z", "+00:00")) return dt.astimezone(ZoneInfo("Europe/Berlin")).strftime("%d.%m.%Y %H:%M:%S") except ValueError: pass return format_datetime(path.stat().st_mtime) def metadata_form_value(path): tags = read_metadata(path) value = tags.get("date", "") if value: try: dt = datetime.fromisoformat(value.replace("Z", "+00:00")) return dt.astimezone(ZoneInfo("Europe/Berlin")).strftime("%Y-%m-%dT%H:%M") except ValueError: pass return datetime.fromtimestamp(path.stat().st_mtime, ZoneInfo("Europe/Berlin")).strftime("%Y-%m-%dT%H:%M") def create_finished(filename, start, end, metadata=None, 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 metadata = metadata or {} 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", "-metadata", f"date={metadata.get('date', '')}", "-metadata", f"artist={metadata.get('artist', '')}", "-metadata", f"title={metadata.get('title', '')}", "-metadata", f"album={metadata.get('album', '')}", "-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, metadata): 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, metadata=metadata, 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.

Metadaten der fertigen Aufnahme

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 %}
DateinameGrößeDauer
{{ item.filename }} {{ item.size }} {{ item.duration }} ✂ Bearbeiten
{% else %}
Keine unbearbeiteten Aufnahmen vorhanden.
{% endif %}

Bearbeitete Aufnahmen

{% if finished %} {% for item in finished %} {% endfor %}
ZeitpunktRednerBeschreibungVeranstaltungDauer
{{ item.datetime }} {{ item.artist or "–" }} {{ item.title or "–" }} {{ item.album or "–" }} {{ 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() tags = read_metadata(path) finished.append({ "filename": path.name, "datetime": tags.get("date", "") or format_datetime(stat.st_mtime), "artist": tags.get("artist", ""), "title": tags.get("title", ""), "album": tags.get("album", ""), "size": format_size(stat.st_size), "duration": format_duration(get_duration(path)), "sort_date": tags.get("date", "") or datetime.fromtimestamp(stat.st_mtime, ZoneInfo("Europe/Berlin")).isoformat(), }) except FileNotFoundError: pass recordings.sort(key=lambda x: x["mtime"], reverse=True) finished.sort(key=lambda x: x["sort_date"], 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, metadata=read_metadata(path), metadata_datetime=metadata_form_value(path), 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") metadata = data.get("metadata") or {} 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()} metadata = { "date": str(metadata.get("date", "")).strip(), "artist": str(metadata.get("artist", "")).strip(), "title": str(metadata.get("title", "")).strip(), "album": str(metadata.get("album", "")).strip(), } if not metadata["date"]: return jsonify({"success": False, "error": "Zeitpunkt fehlt"}), 400 try: dt = datetime.fromisoformat(metadata["date"]) if dt.tzinfo is None: dt = dt.replace(tzinfo=ZoneInfo("Europe/Berlin")) metadata["date"] = dt.isoformat() except ValueError: return jsonify({"success": False, "error": "Ungültiger Zeitpunkt"}), 400 threading.Thread(target=run_edit_job, args=(job_id, filename, start_f, end_f, metadata), 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)