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 write_metadata(path, metadata):
"""Rewrite only the MP3 metadata without re-encoding the audio."""
temp = path.with_name(f".{path.name}.metadata.mp3")
values = {
"date": str(metadata.get("date", "")).strip(),
"artist": str(metadata.get("artist", "")).strip(),
"title": str(metadata.get("title", "")).strip(),
"album": str(metadata.get("album", "")).strip(),
}
cmd = [
"ffmpeg", "-hide_banner", "-loglevel", "error", "-y",
"-i", str(path),
"-map", "0",
"-c", "copy",
"-map_metadata", "-1",
]
for key, value in values.items():
cmd += ["-metadata", f"{key}={value}"]
cmd.append(str(temp))
try:
result = subprocess.run(cmd, capture_output=True, text=True, timeout=120)
if result.returncode != 0 or not temp.is_file() or temp.stat().st_size == 0:
temp.unlink(missing_ok=True)
raise ValueError("ID3-Tags konnten nicht gespeichert werden")
temp.replace(path)
except subprocess.TimeoutExpired:
temp.unlink(missing_ok=True)
raise ValueError("Speichern der ID3-Tags hat zu lange gedauert")
except OSError as exc:
temp.unlink(missing_ok=True)
raise ValueError(f"Speichern der ID3-Tags fehlgeschlagen: {exc}")
def update_finished_file(old_filename, new_filename, metadata):
"""Update metadata and optionally rename a finished file in one operation."""
old_path = finished_path(old_filename)
new_filename = clean_new_filename(new_filename)
if old_path is None or not old_path.is_file():
raise ValueError("Datei nicht gefunden")
if new_filename is None:
raise ValueError("Ungültiger Dateiname")
new_path = FINISHED_DIR / new_filename
if new_path != old_path and new_path.exists():
raise ValueError("Eine Datei mit diesem Namen existiert bereits")
date = str(metadata.get("date", "")).strip()
if not date:
raise ValueError("Zeitpunkt fehlt")
try:
dt = datetime.fromisoformat(date)
if dt.tzinfo is None:
dt = dt.replace(tzinfo=ZoneInfo("Europe/Berlin"))
date = dt.isoformat()
except ValueError:
raise ValueError("Ungültiger Zeitpunkt")
values = {
"date": date,
"artist": str(metadata.get("artist", "")).strip(),
"title": str(metadata.get("title", "")).strip(),
"album": str(metadata.get("album", "")).strip(),
}
# Build the complete new MP3 first. The original remains untouched until
# FFmpeg has successfully written the new file.
temp = FINISHED_DIR / f".{new_path.name}.update.tmp.mp3"
temp.unlink(missing_ok=True)
cmd = [
"ffmpeg", "-hide_banner", "-loglevel", "error", "-y",
"-i", str(old_path),
"-map", "0",
"-c", "copy",
"-map_metadata", "-1",
]
for key, value in values.items():
cmd += ["-metadata", f"{key}={value}"]
cmd.append(str(temp))
try:
result = subprocess.run(cmd, capture_output=True, text=True, timeout=120)
if result.returncode != 0 or not temp.is_file() or temp.stat().st_size == 0:
raise ValueError("ID3-Tags konnten nicht gespeichert werden")
temp.replace(new_path)
if new_path != old_path:
old_path.unlink()
return new_path
except subprocess.TimeoutExpired:
temp.unlink(missing_ok=True)
raise ValueError("Speichern der Datei hat zu lange gedauert")
except OSError as exc:
temp.unlink(missing_ok=True)
raise ValueError(f"Datei konnte nicht gespeichert werden: {exc}")
except ValueError:
temp.unlink(missing_ok=True)
raise
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 }}
00:00:00
{{ duration }}
Zoom: 100 %
00:00:00
Anfang00:00:00
Ende00:00:00
Neue Dauer00:00:00
Beim Erstellen einer fertigen Aufnahme bleibt das Original unverändert.
"""
INDEX_HTML = """
Event Audio – Editor
Event Audio – Editor
Fertige Aufnahmen aus {{ recordings_dir }}
Aufnahmen bearbeiten
{% if recordings %}
| Dateiname | Größe | Dauer | |
{% for item in recordings %}
| {{ item.filename }} |
{{ item.size }} |
{{ item.duration }} |
✂ Bearbeiten
|
{% endfor %}
{% else %}
Keine unbearbeiteten Aufnahmen vorhanden.
{% endif %}
Bearbeitete Aufnahmen
{% if finished %}
| Zeitpunkt | Redner | Beschreibung | Veranstaltung | Dauer | |
{% for item in finished %}
| {{ item.datetime }} |
{{ item.artist or "–" }} |
{{ item.title or "–" }} |
{{ item.album or "–" }} |
{{ item.duration }} |
▶ Abspielen / Download
|
{% endfor %}
{% 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": metadata_datetime(path),
"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 == "update_finished":
if kind != "finished":
return jsonify({"success": False, "error": "Metadaten können nur bei bearbeiteten Aufnahmen geändert werden"}), 400
metadata = data.get("metadata") or {}
path = update_finished_file(filename, data.get("new_filename"), metadata)
return jsonify({"success": True, "filename": path.name})
if action == "update_metadata":
if kind != "finished":
return jsonify({"success": False, "error": "Metadaten können nur bei bearbeiteten Aufnahmen geändert werden"}), 400
path = finished_path(filename)
if path is None or not path.is_file():
raise ValueError("Datei nicht gefunden")
metadata = data.get("metadata") or {}
date = str(metadata.get("date", "")).strip()
if not date:
raise ValueError("Zeitpunkt fehlt")
try:
dt = datetime.fromisoformat(date)
if dt.tzinfo is None:
dt = dt.replace(tzinfo=ZoneInfo("Europe/Berlin"))
date = dt.isoformat()
except ValueError:
raise ValueError("Ungültiger Zeitpunkt")
write_metadata(path, {
"date": date,
"artist": str(metadata.get("artist", "")).strip(),
"title": str(metadata.get("title", "")).strip(),
"album": str(metadata.get("album", "")).strip(),
})
return jsonify({"success": True})
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)