Découpe un mix MP3 selon une playlist.

megamix-cut.py

#!/usr/bin/env python3
"""
Découpe un long mix MP3 selon une playlist, en alignant chaque coupe sur le point
audio le plus calme (RMS minimum) autour du timestamp annoncé, puis en option
sur le zero-crossing le plus proche pour limiter les clics.

Dépendances :
- ffmpeg
- ffprobe
"""

import argparse
import array
import os
import re
import shutil
import subprocess
import sys
from typing import List, Tuple


PLAYLIST = [
    ("00:00:00", "Nattlys i Regnet"),
    ("00:03:39", "Skygger over Byen"),
    ("00:08:29", "Danser Gjennom Mørket"),
    ("00:12:27", "Flammer i Nordvinden"),
    ("00:16:53", "Sølvspor på Asfalt"),
    ("00:21:43", "Hjertet Slår i Neon"),
    ("00:25:23", "Måneskinn og Motorlyd"),
    ("00:28:23", "Gater av Rav"),
    ("00:31:14", "Drømmer under Brostein"),
    ("00:34:42", "Nattlys i Regnet"),
    ("00:38:48", "Vind over Fjorden"),
    ("00:42:00", "Aske og Sommerlyn"),
    ("00:48:10", "Ekko fra Bakgården"),
    ("00:51:58", "Blå Timer i Sør"),
    ("00:55:52", "Støv og Stjerneskinn"),
    ("01:01:04", "Langs Elva ved Midnatt"),
    ("01:05:49", "Byen Sover Aldri Helt"),
]


def timestamp_vers_secondes(ts: str) -> float:
    parts = [float(p) for p in ts.strip().split(":")]
    if len(parts) == 3:
        h, m, s = parts
    elif len(parts) == 2:
        h, m, s = 0, parts[0], parts[1]
    else:
        raise ValueError(f"Timestamp invalide : {ts}")
    return h * 3600 + m * 60 + s


def nettoyer_nom(nom: str) -> str:
    nom = nom.replace("’", "'").replace("“", '"').replace("”", '"')
    nom = re.sub(r'[<>:"/\\|?*\x00-\x1f]', "", nom)
    nom = re.sub(r"\s+", " ", nom)
    return nom.strip().rstrip(".")


def verifier_outils() -> None:
    manquants = [outil for outil in ("ffmpeg", "ffprobe") if not shutil.which(outil)]
    if manquants:
        sys.exit(f"Outil(s) introuvable(s) dans le PATH : {', '.join(manquants)}")


def executer(cmd: List[str], capture_output: bool = False) -> subprocess.CompletedProcess:
    try:
        return subprocess.run(
            cmd,
            check=True,
            capture_output=capture_output,
            text=capture_output,
        )
    except subprocess.CalledProcessError as e:
        if capture_output:
            msg = e.stderr or e.stdout or str(e)
        else:
            msg = str(e)
        sys.exit(f"Erreur lors de l'exécution de la commande :\n{' '.join(cmd)}\n\n{msg}")


def duree_fichier(fichier: str) -> float:
    result = executer(
        [
            "ffprobe",
            "-v", "error",
            "-show_entries", "format=duration",
            "-of", "default=noprint_wrappers=1:nokey=1",
            fichier,
        ],
        capture_output=True,
    )
    try:
        return float(result.stdout.strip())
    except ValueError:
        sys.exit(f"Impossible de lire la durée du fichier : {fichier}")


def extraire_pcm_mono(
    fichier: str,
    debut: float,
    duree: float,
    sample_rate: int,
) -> array.array:
    cmd = [
        "ffmpeg",
        "-v", "error",
        "-ss", f"{debut:.6f}",
        "-i", fichier,
        "-t", f"{duree:.6f}",
        "-ac", "1",
        "-ar", str(sample_rate),
        "-f", "s16le",
        "-",
    ]
    result = subprocess.run(cmd, check=True, capture_output=True)
    samples = array.array("h")
    samples.frombytes(result.stdout)
    return samples


def trouver_zero_crossing_proche(
    samples: array.array,
    centre_index: int,
    rayon_samples: int,
) -> int:
    if not samples:
        return centre_index

    debut = max(1, centre_index - rayon_samples)
    fin = min(len(samples) - 1, centre_index + rayon_samples)

    meilleur_index = centre_index
    meilleure_distance = None

    for i in range(debut, fin):
        a = samples[i - 1]
        b = samples[i]

        # Vrai passage par zéro ou changement de signe
        if (a <= 0 < b) or (a >= 0 > b) or a == 0 or b == 0:
            distance = abs(i - centre_index)
            if meilleure_distance is None or distance < meilleure_distance:
                meilleure_distance = distance
                meilleur_index = i
                if distance == 0:
                    break

    return meilleur_index


def trouver_point_le_plus_calme(
    fichier: str,
    timestamp: float,
    tolerance: float = 5.0,
    fenetre_ms: float = 50.0,
    sample_rate: int = 8000,
    ajuster_zero_crossing: bool = True,
    zero_crossing_ms: float = 10.0,
) -> Tuple[float, float]:
    """
    Retourne (position, delta) où :
    - position = meilleur point de coupe trouvé en secondes
    - delta = position - timestamp
    """
    debut_extraction = max(0.0, timestamp - tolerance)
    duree_extraction = max(0.01, tolerance * 2.0)

    samples = extraire_pcm_mono(fichier, debut_extraction, duree_extraction, sample_rate)
    if len(samples) == 0:
        return timestamp, 0.0

    taille_fenetre = max(1, int(sample_rate * fenetre_ms / 1000.0))
    pas = max(1, taille_fenetre // 4)

    if len(samples) < taille_fenetre:
        centre = len(samples) // 2
        position = debut_extraction + centre / sample_rate
        return position, position - timestamp

    # Somme glissante des carrés en O(n) :
    # on calcule l'énergie initiale sur la première fenêtre, puis on glisse
    # sample par sample en ajoutant l'entrant et en retirant le sortant.
    energie = 0
    for i in range(taille_fenetre):
        v = samples[i]
        energie += v * v

    min_energie = energie
    min_debut = 0

    for i in range(1, len(samples) - taille_fenetre + 1):
        sortant = samples[i - 1]
        entrant = samples[i + taille_fenetre - 1]
        energie += entrant * entrant - sortant * sortant

        # On n'évalue le minimum que toutes les 'pas' positions,
        # comme dans la version d'origine, pour rester cohérent.
        if i % pas == 0 and energie < min_energie:
            min_energie = energie
            min_debut = i

    centre_fenetre = min_debut + taille_fenetre // 2

    if ajuster_zero_crossing:
        rayon_zc = max(1, int(sample_rate * zero_crossing_ms / 1000.0))
        centre_fenetre = trouver_zero_crossing_proche(samples, centre_fenetre, rayon_zc)

    position = debut_extraction + (centre_fenetre / sample_rate)
    return position, position - timestamp


def construire_bornes(
    fichier_entree: str,
    playlist: List[Tuple[str, str]],
    duree_totale: float,
    tolerance: float,
    fenetre_ms: float,
    sample_rate: int,
    ajuster_zero_crossing: bool,
    zero_crossing_ms: float,
    duree_min_piste: float,
) -> List[float]:
    bornes = [0.0]

    print("Recherche des points de coupe optimaux…")
    for i in range(1, len(playlist)):
        ts_theorique = timestamp_vers_secondes(playlist[i][0])

        position, delta = trouver_point_le_plus_calme(
            fichier=fichier_entree,
            timestamp=ts_theorique,
            tolerance=tolerance,
            fenetre_ms=fenetre_ms,
            sample_rate=sample_rate,
            ajuster_zero_crossing=ajuster_zero_crossing,
            zero_crossing_ms=zero_crossing_ms,
        )

        # Garde-fou : pas de retour en arrière
        position = max(position, bornes[-1])

        # Garde-fou : pas au-delà de la durée totale
        position = min(position, duree_totale)

        # Garde-fou : respecter une durée minimale pour la piste précédente
        if position - bornes[-1] < duree_min_piste:
            position = min(duree_totale, bornes[-1] + duree_min_piste)

        signe = "+" if delta >= 0 else ""
        print(
            f"  Piste {i+1:2d} « {playlist[i][1]:30s} » : "
            f"{ts_theorique:8.2f}s → {position:8.2f}s ({signe}{delta:.2f}s)"
        )

        bornes.append(position)

    bornes.append(duree_totale)

    # Dernier garde-fou global
    for i in range(1, len(bornes)):
        if bornes[i] < bornes[i - 1]:
            bornes[i] = bornes[i - 1]

    return bornes


def exporter_piste(
    fichier_entree: str,
    chemin_sortie: str,
    debut: float,
    duree: float,
    reencoder: bool,
    bitrate: str,
    titre: str,
    numero_piste: int,
    total_pistes: int,
    album: str,
    artiste: str = None,
) -> None:
    cmd = [
        "ffmpeg",
        "-y",
        "-loglevel", "error",
        "-i", fichier_entree,
        "-ss", f"{debut:.6f}",
        "-t", f"{duree:.6f}",
    ]

    if reencoder:
        cmd += ["-c:a", "libmp3lame", "-b:a", bitrate]
    else:
        cmd += ["-c", "copy"]

    cmd += [
        "-metadata", f"title={titre}",
        "-metadata", f"track={numero_piste}/{total_pistes}",
        "-metadata", f"album={album}",
    ]

    if artiste:
        cmd += ["-metadata", f"artist={artiste}"]

    cmd.append(chemin_sortie)

    executer(cmd, capture_output=False)


def decouper(
    fichier_entree: str,
    playlist: List[Tuple[str, str]],
    dossier_sortie: str = None,
    reencoder: bool = True,
    album: str = None,
    artiste: str = None,
    tolerance: float = 5.0,
    fenetre_ms: float = 50.0,
    sample_rate: int = 8000,
    ajuster_zero_crossing: bool = True,
    zero_crossing_ms: float = 10.0,
    duree_min_piste: float = 1.0,
    bitrate: str = "192k",
) -> None:
    if not os.path.isfile(fichier_entree):
        sys.exit(f"Fichier introuvable : {fichier_entree}")

    verifier_outils()

    if not playlist:
        sys.exit("Playlist vide.")

    nom_base = os.path.splitext(os.path.basename(fichier_entree))[0]
    if dossier_sortie is None:
        dossier_sortie = os.path.join(
            os.path.dirname(os.path.abspath(fichier_entree)),
            nom_base,
        )
    os.makedirs(dossier_sortie, exist_ok=True)

    duree_totale = duree_fichier(fichier_entree)
    nom_album = album or nom_base

    print(f"Fichier      : {fichier_entree}")
    print(f"Durée totale : {duree_totale:.2f} s")
    print(f"Sortie       : {dossier_sortie}")
    print(f"Mode         : {'ré-encodage' if reencoder else 'copie directe'}")
    if reencoder:
        print(f"Bitrate      : {bitrate}")
    print()

    bornes = construire_bornes(
        fichier_entree=fichier_entree,
        playlist=playlist,
        duree_totale=duree_totale,
        tolerance=tolerance,
        fenetre_ms=fenetre_ms,
        sample_rate=sample_rate,
        ajuster_zero_crossing=ajuster_zero_crossing,
        zero_crossing_ms=zero_crossing_ms,
        duree_min_piste=duree_min_piste,
    )

    print(
        f"\nDécoupage en {len(playlist)} pistes "
        f"({'ré-encodage' if reencoder else 'copie directe'})…\n"
    )

    largeur = len(str(len(playlist)))

    for i, (_, titre) in enumerate(playlist):
        debut = bornes[i]
        fin = min(bornes[i + 1], duree_totale)
        duree = fin - debut

        if debut >= duree_totale:
            print(f"  ⚠ Piste {i+1:02d} ignorée : début au-delà de la durée totale.")
            continue

        if duree <= 0:
            print(f"  ⚠ Piste {i+1:02d} ignorée : durée nulle ou négative.")
            continue

        numero = str(i + 1).zfill(largeur)
        nom_fichier = f"{numero} - {nettoyer_nom(titre)}.mp3"
        chemin_sortie = os.path.join(dossier_sortie, nom_fichier)

        exporter_piste(
            fichier_entree=fichier_entree,
            chemin_sortie=chemin_sortie,
            debut=debut,
            duree=duree,
            reencoder=reencoder,
            bitrate=bitrate,
            titre=titre,
            numero_piste=i + 1,
            total_pistes=len(playlist),
            album=nom_album,
            artiste=artiste,
        )

        print(
            f"  ✔ {numero} : {debut:8.2f}s → {fin:8.2f}s "
            f"({duree:6.2f}s)  →  {nom_fichier}"
        )

    print(f"\nTerminé. Fichiers enregistrés dans : {dossier_sortie}")


def main() -> None:
    parser = argparse.ArgumentParser(
        description=(
            "Découpe un MP3 selon une playlist, en cherchant un point de coupe "
            "audio calme près de chaque timestamp."
        )
    )
    parser.add_argument("fichier", help="Fichier MP3 à découper")
    parser.add_argument("-o", "--sortie", help="Dossier de sortie")
    parser.add_argument(
        "--copie",
        action="store_true",
        help="Copier sans ré-encoder (plus rapide mais moins précis)",
    )
    parser.add_argument("--album", help="Nom de l'album (tag ID3)")
    parser.add_argument("--artiste", help="Nom de l'artiste (tag ID3)")
    parser.add_argument(
        "--tolerance",
        type=float,
        default=5.0,
        help="Demi-fenêtre de recherche en secondes autour du timestamp (défaut : 5)",
    )
    parser.add_argument(
        "--fenetre-ms",
        type=float,
        default=50.0,
        help="Taille de la fenêtre RMS en ms (défaut : 50)",
    )
    parser.add_argument(
        "--sample-rate",
        type=int,
        default=8000,
        help="Fréquence d'échantillonnage d'analyse (défaut : 8000)",
    )
    parser.add_argument(
        "--sans-zero-crossing",
        action="store_true",
        help="Désactive l'ajustement sur le zero-crossing le plus proche",
    )
    parser.add_argument(
        "--zero-crossing-ms",
        type=float,
        default=10.0,
        help="Rayon de recherche du zero-crossing en ms (défaut : 10)",
    )
    parser.add_argument(
        "--duree-min-piste",
        type=float,
        default=1.0,
        help="Durée minimale d'une piste en secondes (défaut : 1.0)",
    )
    parser.add_argument(
        "--bitrate",
        default="192k",
        help="Bitrate MP3 si ré-encodage, ex. 192k, 256k, 320k (défaut : 192k)",
    )

    args = parser.parse_args()

    decouper(
        fichier_entree=args.fichier,
        playlist=PLAYLIST,
        dossier_sortie=args.sortie,
        reencoder=not args.copie,
        album=args.album,
        artiste=args.artiste,
        tolerance=args.tolerance,
        fenetre_ms=args.fenetre_ms,
        sample_rate=args.sample_rate,
        ajuster_zero_crossing=not args.sans_zero_crossing,
        zero_crossing_ms=args.zero_crossing_ms,
        duree_min_piste=args.duree_min_piste,
        bitrate=args.bitrate,
    )


if __name__ == "__main__":
    main()