pascal/youtube
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
youtube/schnitt/pipeline/pinolib.py
Lara c27a7ab9f5 Pino-Material aus Laras Chat und Plan-Dokument ins Repo übernehmen 
Alle 44 Tabs des Plan-Dokuments (Drehbücher, Sprechtexte DE/ES/FR,
Clip-Prompts, Adventskalender, Kanaltexte, Produktionsliste) als Markdown
unter drehbuecher/, konzept/, ideen/ und veroeffentlichung/. Dazu Projektstand
und Entscheidungen (wissen/), Schnitt-Pipeline, Shorts-Katalog,
Figuren-Referenzbilder und das Kinderkanal-Team als Subagenten
(.claude/agents/). Videos, Startbilder und Ton bleiben außerhalb des Git.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-28 21:50:01 +01:00

495 lines
18 KiB
Python
Raw Blame History

# -*- coding: utf-8 -*-
"""Gemeinsame Funktionen der Schnitt-Pipeline (Pino, Noa und Lulu).
Wird von den anderen Skripten importiert. Nicht direkt aufrufen.
Regeln:
- Es wird nur unterhalb von schnitt\\ geschrieben (ausser mit --ausserhalb).
- Vorhandene Dateien werden nie ueberschrieben (ausser mit --ueberschreiben).
- Originalclips werden nur gelesen.
"""
import csv
import json
import os
import re
import shutil
import subprocess
import sys
from pathlib import Path
SCHNITT = Path(__file__).resolve().parents[1] # ...\Pino-Kanal\schnitt
PROJEKT = SCHNITT.parent # ...\Pino-Kanal
TMP = SCHNITT / "tmp"
BERICHTE = SCHNITT / "berichte"
AUSGABE = SCHNITT / "ausgabe"
FFMPEG_STANDARD = Path(
r"C:\Users\lk-lo\AppData\Local\Microsoft\WinGet\Packages"
r"\Gyan.FFmpeg_Microsoft.Winget.Source_8wekyb3d8bbwe"
r"\ffmpeg-9.0-full_build\bin\ffmpeg.exe")
SPRACHEN = ("de", "es", "fr")
FPS = 24
LUFS_ZIEL = -14.0
TP_ZIEL = -1.5
LRA_ZIEL = 11.0
AUDIO_ENDUNGEN = (".wav", ".mp3", ".m4a", ".flac", ".ogg", ".aac")
MAX_SEGMENTE = 70
class Fehler(Exception):
"""Verstaendlicher Fehler fuer die Nutzerin (kein Traceback)."""
def init_konsole():
for s in (sys.stdout, sys.stderr):
try:
s.reconfigure(encoding="utf-8", errors="replace")
except Exception:
pass
# ----------------------------------------------------------------- Programme
def finde_programm(name):
kandidaten = []
env = os.environ.get("PINO_FFMPEG_ORDNER")
if env:
kandidaten.append(Path(env) / (name + ".exe"))
kandidaten.append(FFMPEG_STANDARD.with_name(name + ".exe"))
for k in kandidaten:
if k.exists():
return str(k)
p = shutil.which(name)
if p:
return p
raise Fehler(f"{name} nicht gefunden. Pfad in pinolib.py (FFMPEG_STANDARD) "
f"oder Umgebungsvariable PINO_FFMPEG_ORDNER setzen.")
FFMPEG = None
FFPROBE = None
def programme():
global FFMPEG, FFPROBE
if FFMPEG is None:
FFMPEG = finde_programm("ffmpeg")
FFPROBE = finde_programm("ffprobe")
return FFMPEG, FFPROBE
def lauf(args, dry=False):
"""Fuehrt ein Programm aus. Gibt (returncode, stdout, stderr) zurueck."""
if dry:
print(" [trocken] " + subprocess.list2cmdline([str(a) for a in args]))
return 0, "", ""
r = subprocess.run([str(a) for a in args], capture_output=True)
return (r.returncode, r.stdout.decode("utf-8", "replace"),
r.stderr.decode("utf-8", "replace"))
def ffmpeg_lauf(args, dry=False, info=False, was="ffmpeg"):
ff, _ = programme()
basis = [ff, "-y", "-hide_banner", "-nostdin", "-nostats",
"-loglevel", "info" if info else "error"]
rc, out, err = lauf(basis + list(args), dry=dry)
if rc != 0:
raise Fehler(f"{was} fehlgeschlagen (Code {rc}):\n{err.strip()[-1500:]}")
return out, err
_probe_cache = {}
def probe(pfad):
pfad = str(pfad)
if pfad in _probe_cache:
return _probe_cache[pfad]
_, fp = programme()
rc, out, err = lauf([fp, "-v", "error", "-show_entries",
"format=duration:stream=index,codec_type,codec_name,width,height,"
"r_frame_rate,sample_rate,channels,duration",
"-of", "json", pfad])
if rc != 0:
raise Fehler(f"ffprobe kann {pfad} nicht lesen:\n{err.strip()[-500:]}")
d = json.loads(out)
erg = {"dauer": float(d.get("format", {}).get("duration", 0) or 0),
"video": None, "audio": None}
for s in d.get("streams", []):
if s.get("codec_type") == "video" and erg["video"] is None:
erg["video"] = s
if s.get("codec_type") == "audio" and erg["audio"] is None:
erg["audio"] = s
_probe_cache[pfad] = erg
return erg
def dauer(pfad):
return probe(pfad)["dauer"]
def messe_lautheit(pfad):
"""Integrierte Lautheit (LUFS), LRA und True Peak einer Datei."""
ff, _ = programme()
rc, out, err = lauf([ff, "-hide_banner", "-nostdin", "-nostats", "-i", str(pfad),
"-vn", "-af", "ebur128=peak=true", "-f", "null", "-"])
teil = err.split("Summary:")[-1] if "Summary:" in err else err
def zahl(muster):
m = re.findall(muster, teil)
return float(m[-1]) if m else None
return {"lufs": zahl(r"I:\s+(-?[\d.]+)\s+LUFS"),
"lra": zahl(r"LRA:\s+(-?[\d.]+)\s+LU"),
"tp": zahl(r"Peak:\s+(-?[\d.]+)\s+dBFS")}
# --------------------------------------------------------------- Schreibschutz
def pruefe_ziel(pfad, ausserhalb_ok=False):
p = Path(pfad).resolve()
if not ausserhalb_ok and SCHNITT not in p.parents and p != SCHNITT:
raise Fehler(f"Ziel {p} liegt ausserhalb von {SCHNITT}. "
f"Mit --ausserhalb ausdruecklich erlauben.")
return p
def schreibschutz(pfad, ueberschreiben=False):
p = Path(pfad)
if p.exists() and not ueberschreiben:
raise Fehler(f"{p} existiert schon. Nichts wird ueberschrieben "
f"(--ueberschreiben, um es doch zu tun).")
def tmp_datei(name):
TMP.mkdir(parents=True, exist_ok=True)
return TMP / name
def aufraeumen(dateien):
"""Loescht nur eigene Temp-Dateien in schnitt\\tmp."""
for d in dateien:
try:
p = Path(d).resolve()
if TMP in p.parents and p.exists():
p.unlink()
except Exception:
pass
# -------------------------------------------------------------- Zahlen / Listen
def zahl(text, standard=None):
if text is None:
return standard
t = str(text).strip().replace(",", ".")
if t == "":
return standard
return float(t)
def liste_zahlen(text, standard):
t = (text or "").strip().replace(",", ".")
if not t:
return [standard]
return [float(x) for x in t.split("|") if x.strip() != ""]
def liste_texte(text, standard):
t = (text or "").strip().lower()
if not t:
return [standard]
return [x.strip() for x in re.split(r"[|,;]", t) if x.strip()]
def auf_laenge(werte, n, name):
if len(werte) == 1:
return werte * n
if len(werte) != n:
raise Fehler(f"{name}: {len(werte)} Werte, aber {n} Clips.")
return werte
# ---------------------------------------------------------------------- Clips
_TOKEN = re.compile(r"^([ft])(\d+)_c(\d+)$", re.I)
_RANGE = re.compile(r"^([ft]\d+_c)(\d+)-(?:\1)?(\d+)$", re.I)
def expandiere(text):
"""'f01_c08-13, f01_c20' -> ['f01_c08', ..., 'f01_c13', 'f01_c20']"""
erg = []
for tok in re.split(r"[,\s]+", (text or "").strip()):
if not tok:
continue
m = _RANGE.match(tok)
if m:
pre, a, b = m.group(1), int(m.group(2)), int(m.group(3))
if b < a:
raise Fehler(f"Bereich rueckwaerts: {tok}")
breite = max(2, len(m.group(2)))
erg += [f"{pre}{n:0{breite}d}" for n in range(a, b + 1)]
else:
erg.append(tok)
return erg
def clip_pfad(token, clips_root):
"""Findet die Clipdatei zu 'f01_c09', 't09_c03' oder 'k_05' / 'k_01_kerze3'."""
root = Path(clips_root)
m = _TOKEN.match(token)
if m:
art, nn, cc = m.group(1).lower(), int(m.group(2)), int(m.group(3))
ordner = root / (f"folge{nn:02d}" if art == "f" else "tuerchen")
namen = [f"{art}{nn:02d}_c{cc:02d}.mp4", f"{art}{nn}_c{cc:02d}.mp4",
f"{art}{nn:02d}_c{cc}.mp4"]
for n in namen:
if (ordner / n).exists():
return ordner / n
for n in namen: # Notfall: irgendwo unter clips\
treffer = list(root.rglob(n))
if treffer:
return treffer[0]
return None
if token.lower().startswith("k_"):
ordner = root / "bausteine"
genau = ordner / (token + ".mp4")
if genau.exists():
return genau
treffer = sorted(ordner.glob(token + "*.mp4")) if ordner.exists() else []
if len(treffer) == 1:
return treffer[0]
if len(treffer) > 1:
raise Fehler(f"Baustein '{token}' ist mehrdeutig: "
+ ", ".join(t.name for t in treffer))
return None
# direkter Dateiname / Pfad
p = Path(token)
if p.exists():
return p
if (root / token).exists():
return root / token
return None
def suche_clips(tokens, clips_root):
pfade, fehlt = [], []
for t in tokens:
p = clip_pfad(t, clips_root)
if p is None:
fehlt.append(t)
pfade.append(p)
if fehlt:
raise Fehler("Clips fehlen unter " + str(clips_root) + ": " + ", ".join(fehlt))
return pfade
# -------------------------------------------------------------------- Segmente
class Seg:
def __init__(self, token, pfad, in_s=0.0, out_s=None, tempo=1.0, halten_s=0.0,
uebergang_s=0.5, modus="blur", crop_x=0.5):
self.token = token
self.pfad = Path(pfad)
d = dauer(pfad)
self.in_s = max(0.0, in_s)
self.out_s = min(out_s, d) if out_s and out_s > 0 else d
if self.out_s <= self.in_s:
raise Fehler(f"{token}: out_s ({self.out_s}) <= in_s ({self.in_s})")
self.tempo = tempo
self.halten_s = max(0.0, halten_s)
self.uebergang_s = uebergang_s
self.modus = modus
self.crop_x = min(1.0, max(0.0, crop_x))
def dauer_s(self):
frames = max(1, round(((self.out_s - self.in_s) / self.tempo) * FPS))
return frames / FPS + self.halten_s
def seg_filter(i, s, ziel):
"""Filterkette fuer ein Segment. ziel = 'quer' (1280x720) oder 'hoch' (1080x1920)."""
pre = (f"[{i}:v]trim=start={s.in_s:.4f}:end={s.out_s:.4f},"
f"setpts=(PTS-STARTPTS)/{s.tempo:.4f},fps={FPS}")
hold = f",tpad=stop_mode=clone:stop_duration={s.halten_s:.4f}" if s.halten_s > 0 else ""
if ziel == "quer":
return (f"{pre},scale=1280:720:force_original_aspect_ratio=decrease:flags=lanczos,"
f"pad=1280:720:(ow-iw)/2:(oh-ih)/2:color=black,setsar=1,"
f"format=yuv420p{hold}[v{i}]")
if s.modus == "crop":
return (f"{pre},crop=w='trunc(ih*9/16/2)*2':h=ih:x='(iw-ow)*{s.crop_x:.3f}':y=0,"
f"scale=1080:1920:flags=lanczos,setsar=1,format=yuv420p{hold}[v{i}]")
# blur = Blur-Pad: scharfes 16:9-Bild in der Mitte, unscharfer Rahmen
return (f"{pre},split[a{i}][b{i}];"
f"[a{i}]scale=270:480:force_original_aspect_ratio=increase,crop=270:480,"
f"gblur=sigma=5,scale=1080:1920:flags=bicubic,eq=brightness=-0.04:saturation=0.85[bg{i}];"
f"[b{i}]scale=1080:-2:flags=lanczos[fg{i}];"
f"[bg{i}][fg{i}]overlay=(W-w)/2:(H-h)/2,setsar=1,format=yuv420p{hold}[v{i}]")
def bild_graph(segs, ziel, start=None, ende=None, extra_s=0.0, ausblenden_s=1.0):
"""Baut den Filtergraphen der Bildspur.
Gibt (graph_text, laenge_ohne_extra, gesamtlaenge) zurueck, Ausgang [vout]."""
if not segs:
raise Fehler("Keine Clips in der Bildspur.")
if len(segs) > MAX_SEGMENTE:
raise Fehler(f"{len(segs)} Segmente sind zu viel fuer einen Durchgang "
f"(max. {MAX_SEGMENTE}). Gesamtfolgen aus fertigen Tuerchen-"
f"Exporten zusammensetzen, nicht aus Einzelclips.")
teile = [seg_filter(i, s, ziel) for i, s in enumerate(segs)]
d = [s.dauer_s() for s in segs]
letzt = "v0"
akk = d[0]
for i in range(1, len(segs)):
t = segs[i - 1].uebergang_s
t = max(1.0 / FPS, min(t, 1.0, d[i] - 1.0 / FPS, akk - 1.0 / FPS))
off = akk - t
teile.append(f"[{letzt}][v{i}]xfade=transition=fade:duration={t:.4f}:offset={off:.4f}[x{i}]")
akk = akk + d[i] - t
letzt = f"x{i}"
gesamt = akk
nach = []
laenge = gesamt
if start is not None or ende is not None:
a = start or 0.0
b = min(ende if ende else gesamt, gesamt)
nach.append(f"trim=start={a:.4f}:end={b:.4f}")
nach.append("setpts=PTS-STARTPTS")
laenge = b - a
gesamtlaenge = laenge + max(0.0, extra_s)
if extra_s > 0:
nach.append(f"tpad=stop_mode=clone:stop_duration={extra_s:.4f}")
if ausblenden_s > 0:
nach.append(f"fade=t=out:st={max(0.0, gesamtlaenge - ausblenden_s):.4f}:d={ausblenden_s:.4f}")
nach.append("format=yuv420p")
teile.append(f"[{letzt}]" + ",".join(nach) + "[vout]")
return ";\n".join(teile), laenge, gesamtlaenge
def schreibe_graph(text, name):
p = tmp_datei(name)
p.write_text(text, encoding="utf-8")
return p
def kodiere_bild(clip_pfade, graph_text, ziel_datei, tmpname, dry=False):
gp = schreibe_graph(graph_text, tmpname + "_video.txt")
args = []
for p in clip_pfade:
args += ["-i", str(p)]
args += ["-/filter_complex", str(gp), "-map", "[vout]", "-an",
"-c:v", "libx264", "-preset", "medium", "-crf", "18", "-pix_fmt", "yuv420p",
"-profile:v", "high", "-r", str(FPS), "-movflags", "+faststart",
str(ziel_datei)]
ffmpeg_lauf(args, dry=dry, was="Bildspur")
return gp
# ----------------------------------------------------------------------- Audio
def audio_graph(idx_voice, idx_musik, T, ausblenden_s, musik_db=-20.0, duck=True):
a = (f"[{idx_voice}:a]aformat=sample_fmts=fltp:sample_rates=48000:channel_layouts=stereo,"
f"apad=whole_dur={T:.3f},atrim=duration={T:.3f},asetpts=PTS-STARTPTS[voice]")
g = [a]
if idx_musik is not None:
g.append(f"[{idx_musik}:a]aformat=sample_fmts=fltp:sample_rates=48000:channel_layouts=stereo,"
f"volume={musik_db}dB,atrim=duration={T:.3f},asetpts=PTS-STARTPTS,"
f"afade=t=in:d=1.5[mus]")
if duck:
g.append("[voice]asplit=2[vo][vsc]")
g.append("[mus][vsc]sidechaincompress=threshold=0.015:ratio=8:attack=50:release=800[musd]")
g.append("[vo][musd]amix=inputs=2:duration=first:normalize=0[mix]")
else:
g.append("[voice][mus]amix=inputs=2:duration=first:normalize=0[mix]")
else:
g.append("[voice]anull[mix]")
if ausblenden_s > 0:
g.append(f"[mix]afade=t=out:st={max(0.0, T - ausblenden_s):.3f}:d={ausblenden_s:.3f}[aout]")
else:
g.append("[mix]anull[aout]")
return ";".join(g)
def parse_loudnorm_json(text):
m = re.findall(r"\{[^{}]*\"input_i\"[^{}]*\}", text, re.S)
if not m:
return None
return json.loads(m[-1])
def mische_und_schreibe(video_datei, voice, musik, T, ausblenden_s, musik_db, ziel_datei,
tmpname, dry=False, duck=True):
"""Mischt Sprache (+Musik), misst/normalisiert auf -14 LUFS (zweistufig) und
schreibt Video (kopiert) + AAC-Audio in ziel_datei."""
eingaben = ["-i", str(video_datei), "-i", str(voice)]
idx_m = None
if musik:
eingaben += ["-stream_loop", "-1", "-i", str(musik)]
idx_m = 2
g = audio_graph(1, idx_m, T, ausblenden_s, musik_db, duck)
# Durchgang 1: messen
g1 = g + f";[aout]loudnorm=I={LUFS_ZIEL}:TP={TP_ZIEL}:LRA={LRA_ZIEL}:print_format=json[o]"
p1 = schreibe_graph(g1, tmpname + "_a1.txt")
mess = None
if not dry:
_, err = ffmpeg_lauf(eingaben + ["-/filter_complex", str(p1), "-map", "[o]",
"-t", f"{T:.3f}", "-f", "null", "-"],
info=True, was="Lautheit messen")
mess = parse_loudnorm_json(err)
if mess is None:
raise Fehler("Lautheitsmessung lieferte kein Ergebnis.")
ln = f"loudnorm=I={LUFS_ZIEL}:TP={TP_ZIEL}:LRA={LRA_ZIEL}"
if mess and mess["input_i"] not in ("-inf", "inf"):
ln += (f":measured_I={mess['input_i']}:measured_TP={mess['input_tp']}"
f":measured_LRA={mess['input_lra']}:measured_thresh={mess['input_thresh']}"
f":offset={mess['target_offset']}:linear=true")
g2 = g + f";[aout]{ln},aresample=48000[afinal]"
p2 = schreibe_graph(g2, tmpname + "_a2.txt")
ffmpeg_lauf(eingaben + ["-/filter_complex", str(p2), "-map", "0:v", "-map", "[afinal]",
"-c:v", "copy", "-c:a", "aac", "-b:a", "192k", "-ar", "48000",
"-ac", "2", "-t", f"{T:.3f}", "-movflags", "+faststart",
str(ziel_datei)], dry=dry, was="Mischen/Export")
return [p1, p2], mess
def finde_audio(ordner, basisname):
ordner = Path(ordner)
for e in AUDIO_ENDUNGEN:
p = ordner / (basisname + e)
if p.exists():
return p
return None
def parse_audio_args(liste, ordner, basis_fn, sprachen):
"""--audio de=pfad ... und/oder --audio-ordner. Gibt {lang: Path} zurueck."""
erg = {}
for eintrag in liste or []:
if "=" not in eintrag:
raise Fehler(f"--audio erwartet sprache=pfad, bekam: {eintrag}")
l, p = eintrag.split("=", 1)
l = l.strip().lower()
if not Path(p).exists():
raise Fehler(f"Audiodatei fehlt: {p}")
erg[l] = Path(p)
if ordner:
for l in sprachen:
if l not in erg:
p = finde_audio(ordner, basis_fn(l))
if p:
erg[l] = p
return erg
# ------------------------------------------------------------------------- CSV
def lese_csv(pfad):
pfad = Path(pfad)
if not pfad.exists():
raise Fehler(f"Datei fehlt: {pfad}")
text = pfad.read_text(encoding="utf-8-sig")
erste = text.splitlines()[0] if text.strip() else ""
delim = ";" if erste.count(";") >= erste.count(",") else ","
zeilen = []
for row in csv.DictReader(text.splitlines(), delimiter=delim):
werte = {(k or "").strip().lower(): (v or "").strip() for k, v in row.items()}
erste_spalte = next(iter(werte.values()), "")
if erste_spalte.startswith("#") or not any(werte.values()):
continue
zeilen.append(werte)
return zeilen
def zeit(sek):
return f"{int(sek // 60)}:{sek % 60:05.2f}"
Reference in a new issue View git blame Copy permalink