merge_daily.py
Rama main · réplica de lectura del repositorio original.
"""Fusionne les clips Blink en une vidéo par caméra et par journée locale.
L'horodatage est désormais incrusté directement dans l'image (hardsub, via le
filtre drawtext) plutôt qu'ajouté comme piste de sous-titres "soft" (mov_text).
Ce choix corrige les échecs d'envoi WhatsApp : WhatsApp n'affiche jamais les
pistes de sous-titres embarquées, et l'ancienne piste (une entrée par seconde)
gonflait l'atome moov au point de faire échouer la validation du média côté
mobile sur certains fichiers longs.
Conséquence : chaque groupe est maintenant ré-encodé (plus de -c:v copy), et
la conversion de fuseau horaire reste entièrement gérée côté Python
(zoneinfo) pour rester fiable sur toutes les plateformes. ffmpeg ne reçoit
que des epochs déjà convertis en « horloge murale locale déguisée en UTC »
et tourne avec TZ=UTC0, ce qui évite toute dépendance à une base de fuseaux
horaires IANA côté ffmpeg/OS (peu fiable sous Windows).
"""
from __future__ import annotations # Python 3.8 (build Windows 7) : les annotations "X | None" ne s'évaluent qu'à l'écriture des chaînes, jamais à l'exécution.
import argparse
import datetime as dt
import functools
import hashlib
import os
import platform
import re
import shutil
import subprocess
import sys
import tempfile
import threading
import time
from collections import defaultdict, namedtuple
from pathlib import Path
try:
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
except ImportError: # Python 3.8 (build Windows 7, voir build-win7.yml) : pas de zoneinfo en stdlib.
from backports.zoneinfo import ZoneInfo, ZoneInfoNotFoundError
from nico579_commons import atomique
import runtime
LIBELLES = {
"fr": {
"aide_desc":
"Fusion incrémentale des clips Blink par caméra : une vidéo par "
"jour, puis agrégats par semaine ISO et par mois.",
"aide_normalized":
"stock des clips normalisés et horodatés, dont toutes les vidéos "
"assemblées sont issues",
"aide_no_weekly": "ne pas (re)construire les agrégats hebdomadaires",
"aide_no_monthly": "ne pas (re)construire les agrégats mensuels",
"aide_no_timestamp": "ne pas incruster la date et l'heure dans l'image",
"aide_excluded_output": "dossier où sont mis de côté les clips écartés",
"aide_exclude":
"écarter des clips (chemin sous Blink_Clips, Blink_Normalized ou "
"Blink_Excluded) : le brut part dans Blink_Excluded, le segment "
"est effacé, le clip n'est plus retéléchargé ni assemblé",
"aide_include":
"annuler une exclusion : le brut revient de Blink_Excluded et le "
"clip est re-normalisé",
"aide_date": "limiter à une date locale YYYY-MM-DD",
"aide_camera": "limiter à une caméra",
"aide_force": "reconstruire même si rien n'a changé",
"aide_font": "chemin vers une police .ttf pour l'horodatage incrusté",
"aide_font_size":
"taille en pixels de l'horodatage incrusté, 8 à 500 "
"(défaut : proportionnelle à la hauteur de la vidéo)",
"aide_font_color":
"couleur de l'horodatage incrusté : nom (white, yellow...) ou "
"hexadécimal (#RRGGBB, 0xRRGGBB), éventuellement @opacité "
"(ex. white@0.8), défaut white",
"aide_box_opacity":
"opacité du bandeau derrière l'horodatage, 0.0 (invisible) à "
"1.0 (opaque), défaut 0.55",
"aide_clock":
"format de l'heure incrustée : 24h (14:05:09, défaut) ou "
"12h (02:05:09 PM)",
"aide_preset": "preset libx264 (ultrafast..veryslow), défaut veryfast",
"aide_crf": "qualité libx264 (0-51, plus bas = meilleure qualité), défaut 21",
"registre_inconnu": " Inconnu du registre, ignoré : {identity}",
"clip_exclu": " Exclu : {identity}",
"clip_reintegre": " Réintégré : {identity}",
"clip_reintegre_brut_absent":
" Réintégré : {identity} (brut absent, relancer blink2video.py)",
"ffmpeg_introuvable":
"FFmpeg est introuvable. Installez imageio-ffmpeg ou ajoutez ffmpeg au PATH.",
"drawtext_absent":
"Le binaire ffmpeg utilisé n'inclut pas le filtre drawtext "
"(compilation sans libfreetype). Utilisez un build ffmpeg complet "
"(ex. gyan.dev sous Windows, ou 'apt install ffmpeg' sous Linux).",
"horodatage_echec_police":
"L'incrustation de l'horodatage échoue avec la police {police} :\n{detail}",
"horodatage_echec_sans_message": "échec sans message",
"horodatage_vide":
"L'incrustation de l'horodatage ne dessine rien avec la police "
"{police} (image de test entièrement noire). Police illisible "
"par drawtext, ou format de date non supporté par la libc.",
"police_introuvable": "Police introuvable : {chemin}",
"police_absente":
"Aucune police TrueType trouvée pour l'incrustation de l'horodatage. "
"Placez un fichier .ttf (ex. DejaVuSans-Bold.ttf) à côté de "
"{fichier}, ou passez --font /chemin/vers/police.ttf.",
"registre_illisible": "Registre de téléchargement illisible : {chemin}",
"duree_introuvable": "durée introuvable pour {nom}",
"duree_invalide": "durée invalide pour {nom}",
"flux_video_introuvable": "flux vidéo introuvable pour {nom}",
"resolution_introuvable": "résolution introuvable pour {nom}",
"chemin_hors_racines": "Chemin situé hors de {racines} : {valeur}",
"ffmpeg_silencieux":
"FFmpeg silencieux plus de {secondes} s, processus tué "
"(reprise au prochain passage)",
"mp4_invalide": "FFmpeg n'a pas produit un MP4 valide",
"copie_refusee": " Copie de flux refusée, ré-encodage de la période",
"periode_deja_a_jour": "Déjà à jour : {nom} ({jours} jour(s))",
"periode_assemblage": "Assemblage : {label} / {camera} / {jours} jour(s)",
"echec_generique": " Échec : {erreur}",
"cree": " Créé : {destination}",
"periode_supprimee": "Supprimé (période vide) : {periode}",
"periode_protegee":
"Conservé tel quel : {periode} (une journée a perdu sa source et sa journalière)",
"assemblage_deja_en_cours": "Assemblage déjà en cours ({erreur}). Rien à faire.",
"normalisation_titre": "Normalisation : {total} clip(s) à encoder",
"echec_normalisation": " Échec : {erreur}",
"normalisation_resume": "Normalisation : {encoded} clip(s) encodé(s), {reused} réutilisé(s).",
"erreur_generique": "Erreur : {erreur}",
"aucun_groupe": "Aucun groupe de clips à fusionner.",
"cache_supprime": "Ancien cache supprimé : {chemin}",
"jour_reportee_source":
" Reportée (source absente ou invalide) : {nom}",
"jour_supprimee": "Supprimée (plus aucun clip) : {nom}",
"jour_reportee_partielle": " Reportée (encodage partiel) : {nom}",
"jour_deja_a_jour": "Déjà à jour : {nom} ({clips} clip(s))",
"jour_assemblage": " [{index}/{total}] Assemblage : {jour} / {camera} / {clips} clip(s)",
"stock_normalise": "Stock normalisé : {n} segment(s) obsolète(s) supprimé(s)",
"journalieres_resume": "Journalières : {built} créée(s), {skipped} déjà à jour, {failed} échec(s).",
"agregats_reportes":
"Agrégats reportés : au moins une journée sélectionnée a une source indisponible.",
"label_hebdomadaires": "Hebdomadaires",
"label_mensuelles": "Mensuelles",
"periode_titre": "\n{label} :",
"periode_resume": "{label} : {built} créée(s), {skipped} déjà à jour, {failed} échec(s).",
"couleur_invalide":
"Couleur invalide : {valeur!r}. Un nom (white, yellow...) ou un "
"code hexadécimal (#RRGGBB, 0xRRGGBB), éventuellement suivi de "
"@opacité (ex. white@0.8).",
"taille_police_invalide": "Taille de police invalide : {valeur!r} (8 à 500).",
"opacite_invalide": "Opacité invalide : {valeur!r} (0.0 à 1.0).",
"motif_structure_mp4": "structure=boite MP4 absente, tronquee ou pas une video",
"motif_outil_absent": "outil=aucun ffprobe ni ffmpeg disponible",
"motif_aucun": "aucune (le fichier est valide)",
"motif_diagnostic_impossible": "diagnostic impossible ({erreur})",
"travail_preparation_clips": "Préparation des clips",
"travail_assemblage_videos": "Assemblage des vidéos",
"aide_keep_clips": "effacer les clips bruts après N jours, par journée entière une fois sa journalière faite (0 : garder toujours)",
"aide_keep_daily": "effacer les journalières après N jours, une fois leurs bruts effacés (0 : garder toujours)",
"aide_keep_weekly": "effacer les hebdomadaires après N jours, une fois leurs journalières effacées (0 : garder toujours)",
"aide_keep_monthly": "effacer les mensuelles après N jours, une fois leurs journalières effacées (0 : garder toujours)",
"aide_keep_live": "effacer les directs enregistrés après N jours (0 : garder toujours)",
"aide_keep_snapshots": "effacer les photos après N jours (0 : garder toujours)",
"aide_purge_only": "appliquer seulement la conservation, sans rien assembler",
"purge_echec": "Conservation non appliquée à ce passage : {erreur}",
"jour_purge_fige": "Figée (clips déjà effacés par la conservation) : {nom}",
"purge_resume": "Conservation : {n} fichier(s) effacé(s)",
"purges_manuelles": "Clips bruts effacés ou revenus à la main : {n} (journalières gardées telles quelles)",
},
"en": {
"aide_desc":
"Incremental merge of Blink clips per camera: one video per day, "
"then ISO-week and monthly aggregates.",
"aide_normalized":
"store of normalized, timestamped clips all assembled videos are "
"built from",
"aide_no_weekly": "do not (re)build the weekly aggregates",
"aide_no_monthly": "do not (re)build the monthly aggregates",
"aide_no_timestamp": "do not burn the date and time into the image",
"aide_excluded_output": "folder where discarded clips are set aside",
"aide_exclude":
"discard clips (path under Blink_Clips, Blink_Normalized or "
"Blink_Excluded): the raw file moves to Blink_Excluded, the segment "
"is deleted, the clip is no longer re-downloaded or assembled",
"aide_include":
"undo a discard: the raw file comes back from Blink_Excluded and "
"the clip is re-normalized",
"aide_date": "limit to one local date YYYY-MM-DD",
"aide_camera": "limit to one camera",
"aide_force": "rebuild even if nothing changed",
"aide_font": "path to a .ttf font for the burned-in timestamp",
"aide_font_size":
"burned-in timestamp size in pixels, 8 to 500 "
"(default: proportional to the video's height)",
"aide_font_color":
"burned-in timestamp color: a name (white, yellow...) or hex "
"(#RRGGBB, 0xRRGGBB), optionally @opacity (e.g. white@0.8), "
"default white",
"aide_box_opacity":
"opacity of the box behind the timestamp, 0.0 (invisible) to "
"1.0 (opaque), default 0.55",
"aide_clock":
"burned-in time format: 24h (14:05:09, default) or "
"12h (02:05:09 PM)",
"aide_preset": "libx264 preset (ultrafast..veryslow), default veryfast",
"aide_crf": "libx264 quality (0-51, lower = better quality), default 21",
"registre_inconnu": " Unknown to the registry, ignored: {identity}",
"clip_exclu": " Excluded: {identity}",
"clip_reintegre": " Reinstated: {identity}",
"clip_reintegre_brut_absent":
" Reinstated: {identity} (raw file missing, restart blink2video.py)",
"ffmpeg_introuvable":
"FFmpeg not found. Install imageio-ffmpeg or add ffmpeg to PATH.",
"drawtext_absent":
"The ffmpeg binary in use does not include the drawtext filter "
"(built without libfreetype). Use a full ffmpeg build "
"(e.g. gyan.dev on Windows, or 'apt install ffmpeg' on Linux).",
"horodatage_echec_police":
"Burning in the timestamp fails with font {police}:\n{detail}",
"horodatage_echec_sans_message": "failed with no message",
"horodatage_vide":
"Burning in the timestamp draws nothing with font "
"{police} (test image entirely black). Font unreadable "
"by drawtext, or date format unsupported by libc.",
"police_introuvable": "Font not found: {chemin}",
"police_absente":
"No TrueType font found to burn in the timestamp. "
"Place a .ttf file (e.g. DejaVuSans-Bold.ttf) next to "
"{fichier}, or pass --font /path/to/font.ttf.",
"registre_illisible": "Download registry unreadable: {chemin}",
"duree_introuvable": "duration not found for {nom}",
"duree_invalide": "invalid duration for {nom}",
"flux_video_introuvable": "video stream not found for {nom}",
"resolution_introuvable": "resolution not found for {nom}",
"chemin_hors_racines": "Path outside of {racines}: {valeur}",
"ffmpeg_silencieux":
"FFmpeg silent for more than {secondes}s, process killed "
"(will resume next round)",
"mp4_invalide": "FFmpeg did not produce a valid MP4",
"copie_refusee": " Stream copy refused, re-encoding the period",
"periode_deja_a_jour": "Already up to date: {nom} ({jours} day(s))",
"periode_assemblage": "Assembling: {label} / {camera} / {jours} day(s)",
"echec_generique": " Failed: {erreur}",
"cree": " Created: {destination}",
"periode_supprimee": "Deleted (empty period): {periode}",
"periode_protegee":
"Kept as is: {periode} (a day lost both its source and its daily video)",
"assemblage_deja_en_cours": "Assembly already in progress ({erreur}). Nothing to do.",
"normalisation_titre": "Normalizing: {total} clip(s) to encode",
"echec_normalisation": " Failed: {erreur}",
"normalisation_resume": "Normalizing: {encoded} clip(s) encoded, {reused} reused.",
"erreur_generique": "Error: {erreur}",
"aucun_groupe": "No clip group to merge.",
"cache_supprime": "Old cache deleted: {chemin}",
"jour_reportee_source":
" Postponed (source missing or invalid): {nom}",
"jour_supprimee": "Deleted (no clip left): {nom}",
"jour_reportee_partielle": " Postponed (partial encoding): {nom}",
"jour_deja_a_jour": "Already up to date: {nom} ({clips} clip(s))",
"jour_assemblage": " [{index}/{total}] Assembling: {jour} / {camera} / {clips} clip(s)",
"stock_normalise": "Normalized storage: {n} obsolete segment(s) deleted",
"journalieres_resume": "Daily videos: {built} created, {skipped} already up to date, {failed} failed.",
"agregats_reportes":
"Aggregates postponed: at least one selected day has an unavailable source.",
"label_hebdomadaires": "Weekly",
"label_mensuelles": "Monthly",
"periode_titre": "\n{label}:",
"periode_resume": "{label}: {built} created, {skipped} already up to date, {failed} failed.",
"couleur_invalide":
"Invalid color: {valeur!r}. A name (white, yellow...) or a "
"hex code (#RRGGBB, 0xRRGGBB), optionally followed by "
"@opacity (e.g. white@0.8).",
"taille_police_invalide": "Invalid font size: {valeur!r} (8 to 500).",
"opacite_invalide": "Invalid opacity: {valeur!r} (0.0 to 1.0).",
"motif_structure_mp4": "structure=MP4 box missing, truncated or not a video",
"motif_outil_absent": "tool=neither ffprobe nor ffmpeg available",
"motif_aucun": "none (the file is valid)",
"motif_diagnostic_impossible": "diagnosis impossible ({erreur})",
"travail_preparation_clips": "Preparing clips",
"travail_assemblage_videos": "Assembling videos",
"aide_keep_clips": "delete raw clips after N days, a whole day at a time once its daily video is made (0: keep forever)",
"aide_keep_daily": "delete daily videos after N days, once their raw clips are gone (0: keep forever)",
"aide_keep_weekly": "delete weekly videos after N days, once their daily videos are gone (0: keep forever)",
"aide_keep_monthly": "delete monthly videos after N days, once their daily videos are gone (0: keep forever)",
"aide_keep_live": "delete recorded live views after N days (0: keep forever)",
"aide_keep_snapshots": "delete snapshots after N days (0: keep forever)",
"aide_purge_only": "only apply retention, assemble nothing",
"purge_echec": "Retention not applied on this pass: {erreur}",
"jour_purge_fige": "Kept as is (clips already deleted by retention): {nom}",
"purge_resume": "Retention: {n} file(s) deleted",
"purges_manuelles": "Raw clips deleted or brought back by hand: {n} (daily videos kept as they are)",
},
}
def msg(cle: str, **valeurs) -> str:
return runtime.traduire(LIBELLES, cle, **valeurs)
# Les sorties, visibles ; l'état (merge.log) reste dans runtime.app_dir().
BASE_DIR = runtime.dossier_sorties()
DEFAULT_INPUT = BASE_DIR / "Blink_Clips"
DEFAULT_OUTPUT = BASE_DIR / "Blink_Daily"
DEFAULT_WEEKLY = BASE_DIR / "Blink_Weekly"
DEFAULT_MONTHLY = BASE_DIR / "Blink_Monthly"
DEFAULT_NORMALIZED = BASE_DIR / "Blink_Normalized"
DEFAULT_EXCLUDED = BASE_DIR / "Blink_Excluded"
# Mêmes dossiers que serve.py (DOSSIER_DIRECT, DOSSIER_SNAPSHOTS), lus ici
# seulement par la conservation (issue #120).
DEFAULT_LIVE = BASE_DIR / "Blink_Direct"
DEFAULT_SNAPSHOTS = BASE_DIR / "Blink_Snapshots"
DEFAUTS_CONSERVATION = {
"keep_clips_days": 0, "keep_daily_days": 0, "keep_weekly_days": 0,
"keep_monthly_days": 0, "keep_live_days": 0, "keep_snapshots_days": 0,
"purge_only": False, "live_output": DEFAULT_LIVE, "snapshots_output": DEFAULT_SNAPSHOTS,
}
DOWNLOAD_STATE = ".blink_download_state.json"
MERGE_STATE = ".blink_merge_state.json"
NORMALIZED_STATE = ".blink_normalized.json"
NORMALIZE_VERSION = "normalized-v1-hardsub"
RENDER_VERSION = "daily-v7-normalized"
PERIOD_VERSION = "period-v2-copy-first"
# Partagé par run_ffmpeg_batch et concat_copy : un ffmpeg bloqué (constaté en
# réel, AUDIT-2026-08-13 28.82) ne doit jamais laisser l'assemblage figé en
# silence indéfiniment.
SILENCE_MAX = 300.0 # 5 min : au-delà, ffmpeg est considéré bloqué.
# Nom d'une vidéo journalière : 2026-08-10_jardin.mp4. Les agrégats hebdo et
# mensuel sont construits à partir de ces fichiers (déjà horodatés et
# normalisés), pas à partir des clips bruts.
DAILY_NAME = re.compile(r"^(?P<day>\d{4}-\d{2}-\d{2})_(?P<camera>.+)\.mp4$")
# Ancien cache de segments, remplacé par Blink_Normalized. Supprimé au premier
# passage : c'était un dossier caché de fichiers jetables, le stock normalisé
# est maintenant une couche nommée du pipeline.
LEGACY_SEGMENT_DIR = ".blink_segments"
ClipInfo = namedtuple(
"ClipInfo", ["created", "source", "duration", "width", "height", "fps", "has_audio"]
)
# Apparence de l'horodatage incrusté (issue GitHub #7 : le fichier de police
# seul, via --font, ne suffisait pas à qui voulait aussi ajuster taille,
# couleur ou transparence du bandeau). size=None conserve l'ancien calcul
# automatique (proportionnel à la hauteur vidéo) plutôt que d'imposer une
# valeur fixe qui rendrait mal sur une autre résolution.
# clock : "24h" (défaut historique) ou "12h" avec AM/PM, demandé sur Reddit
# par un utilisateur nord-américain. Valeur par défaut dans le namedtuple pour
# que les appels à trois champs restent valides.
HORLOGES = runtime.HORLOGES
TimestampStyle = namedtuple("TimestampStyle", ["size", "color", "box_opacity", "clock"],
defaults=["24h"])
STYLE_PAR_DEFAUT = TimestampStyle(size=None, color="white", box_opacity=0.55)
def _tronquer_utf8(value: str, maximum: int) -> str:
brut = value.encode("utf-8")[:maximum]
return brut.decode("utf-8", errors="ignore")
def safe_name(value: str) -> str:
"""Produit un composant de chemin sûr à partir d'un nom libre (caméra).
Seule version conservée (revue de code du 0eab463, bug #5) : trois
copies quasi identiques de ce même nettoyage avaient dérivé, chacune
avec une partie du traitement mais pas toutes - celle-ci, la plus
complète (déjà dans blink_models.py), reste ici parce que blink_models
importe déjà ce module (l'inverse créerait un import circulaire), et
serve.py l'importait aussi. blink_models.safe_name et serve.safe_file
n'existent plus qu'en alias vers celle-ci.
Ne rend PAS deux noms distincts garantis différents : "A/B" et "A_B"
produisent le même résultat, un compromis assumé (voir I-16,
test_T_I16_A_slash_B_et_A_underscore_B_restent_distincts) - l'identité
réelle d'un clip ne repose jamais sur cette chaîne (state_key(),
target_path() y ajoutent chacun leur propre distinction), seul le
regroupement journalier/hebdo/mensuel de merge_daily.py peut donc
encore confondre deux caméras dont le nom brut diffère mais se
nettoie pareil - cas assez rare pour rester documenté plutôt que
corrigé dans l'immédiat (voir BACKLOG.md)."""
cleaned = re.sub(r"[^\w.-]+", "_", value, flags=re.UNICODE).strip("._")
cleaned = _tronquer_utf8(cleaned, 32).rstrip(". ") or "camera"
racine = cleaned.split(".", 1)[0].casefold()
reserves = {"con", "prn", "aux", "nul", *{f"com{i}" for i in range(1, 10)},
*{f"lpt{i}" for i in range(1, 10)}}
if racine in reserves:
cleaned = f"_{cleaned}"
return cleaned
def valid_mp4(path: Path) -> bool:
"""Valide rapidement la structure externe d'un MP4/MP4 fragmenté.
On parcourt uniquement les en-têtes des boîtes ISO-BMFF, en sautant leurs
contenus. C'est suffisamment peu coûteux pour les inventaires récurrents,
mais bien plus strict que la simple recherche de ``ftyp`` : chaque boîte
doit tenir entièrement dans le fichier et un média autonome doit contenir
``ftyp``, ``moov`` et ``mdat``. La validation approfondie d'un nouveau
téléchargement est assurée par :func:`valid_mp4_complet`.
"""
try:
taille_fichier = path.stat().st_size
# Trois en-têtes de boîte au minimum ; ftyp requiert en outre les
# champs major_brand et minor_version (16 octets boîte comprise).
if taille_fichier < 32:
return False
trouvees = set()
position = 0
with path.open("rb") as fichier:
while position < taille_fichier:
restant = taille_fichier - position
if restant < 8:
return False
fichier.seek(position)
entete = fichier.read(8)
if len(entete) != 8:
return False
taille_boite = int.from_bytes(entete[:4], "big")
type_boite = entete[4:8]
taille_entete = 8
if taille_boite == 1:
extension = fichier.read(8)
if len(extension) != 8:
return False
taille_boite = int.from_bytes(extension, "big")
taille_entete = 16
elif taille_boite == 0:
# Une boîte de taille nulle s'étend, par définition,
# jusqu'à EOF et doit donc nécessairement être la dernière.
taille_boite = restant
if taille_boite < taille_entete or taille_boite > restant:
return False
if type_boite == b"ftyp" and taille_boite < taille_entete + 8:
return False
if type_boite in {b"ftyp", b"moov", b"mdat"}:
trouvees.add(type_boite)
position += taille_boite
return position == taille_fichier and trouvees == {b"ftyp", b"moov", b"mdat"}
except (OSError, OverflowError):
return False
@functools.lru_cache(maxsize=1)
def _outil_validation_media() -> tuple[str, str] | None:
"""Trouve une sonde média une fois, hors du chemin des simples listings."""
candidats_ffprobe = []
for motif in ("ffprobe*.exe", "ffprobe-*", "ffprobe"):
candidats_ffprobe.extend(
str(candidat)
for candidat in sorted(runtime.resource_dir().glob(motif))
if candidat.is_file()
)
systeme = shutil.which("ffprobe")
if systeme:
candidats_ffprobe.append(systeme)
if candidats_ffprobe:
return "ffprobe", candidats_ffprobe[0]
try:
return "ffmpeg", find_ffmpeg()
except RuntimeError:
return None
# Lignes que ffprobe écrit sur des clips pourtant intacts. Issue #49 : chaque
# clip enregistré sur la clé USB d'un Sync Module contient une unité d'accès
# H.264 sans image, que l'analyseur H.264 signale (« missing picture in access
# unit with size 9 »), alors que la vidéo est complète ; les clips du cloud
# n'en ont pas. Refusée comme une erreur, cette ligne faisait rejeter en
# silence tous les clips USB partout où ffprobe est présent (paquet Linux
# depuis la 0.12.29, image Docker, ffprobe installé sur le système). Mesuré
# le 2026-09-30 sur 104 clips réels et 15 copies tronquées, avec le ffprobe du
# paquet 0.15.4 et FFmpeg stable 8.1.3 : aucune coupure ne se signalait par
# cette seule ligne, toujours accompagnée d'un « Invalid NAL unit size » ou
# d'une erreur du conteneur, qui restent des motifs de refus.
_BRUITS_FFPROBE = ("missing picture in access unit",)
def _erreurs_ffprobe(stderr: str) -> list:
"""Lignes d'erreur de ffprobe qui comptent, sans le bruit connu ci-dessus."""
erreurs = []
for ligne in (stderr or "").splitlines():
ligne = ligne.strip()
# « Last message repeated N times » répète la ligne précédente, déjà
# jugée pour elle-même : erreur si elle l'était, bruit sinon.
if not ligne or ligne.startswith("Last message repeated"):
continue
if any(bruit in ligne for bruit in _BRUITS_FFPROBE):
continue
erreurs.append(ligne)
return erreurs
def _ligne_sans_donnee_privee(ligne: str, path: Path) -> str:
"""Une ligne d'erreur de la sonde, sans chemin, nom de fichier ni adresse.
ffprobe cite parfois le fichier (« /chemin/clip.mp4: Invalid data ... ») ;
la raison d'un refus est imprimée dans le journal, que les utilisateurs
collent dans les issues publiques : jamais de chemin ni de nom de clip."""
for morceau in sorted({str(path), path.name, str(path.parent)}, key=len, reverse=True):
if morceau:
ligne = ligne.replace(morceau, "<fichier>")
ligne = re.sub(r"@ 0x[0-9a-fA-F]+", "@ ADR", ligne)
ligne = re.sub(r"[A-Za-z]:[\\/][^\s:]*|/[^\s:]+/[^\s:]*", "<chemin>", ligne)
return " ".join(ligne.split())[:140]
def _motif_refus_mp4_complet(path: Path) -> str:
"""Pourquoi ce téléchargement est refusé ; chaîne vide s'il est valide.
Source unique de la décision : ``valid_mp4_complet`` n'en garde que le
booléen, ``raison_refus_mp4`` rend le texte. Les motifs sont des mots-clés
stables plus, quand la sonde s'est exprimée, sa première ligne utile."""
if not valid_mp4(path):
return msg("motif_structure_mp4")
outil = _outil_validation_media()
if outil is None:
return msg("motif_outil_absent")
genre, executable = outil
if genre == "ffprobe":
commande = [
executable, "-v", "error", "-select_streams", "v:0",
"-count_packets", "-show_entries", "stream=nb_read_packets",
"-of", "csv=p=0", str(path),
]
sortie = subprocess.PIPE
else:
commande = [
executable, "-hide_banner", "-loglevel", "error", "-xerror",
"-i", str(path), "-map", "0:v:0", "-c", "copy",
"-f", "null", "-",
]
sortie = subprocess.DEVNULL
try:
resultat = runtime.lancer(
commande, stdin=subprocess.DEVNULL, stdout=sortie,
stderr=subprocess.PIPE, text=True, encoding="utf-8",
errors="replace", check=False, timeout=120,
)
except (OSError, subprocess.SubprocessError) as erreur:
return f"outil={genre} lancement={type(erreur).__name__}"
lignes = _erreurs_ffprobe(resultat.stderr) if genre == "ffprobe" else [
ligne.strip() for ligne in (resultat.stderr or "").splitlines() if ligne.strip()]
premiere = _ligne_sans_donnee_privee(lignes[0], path) if lignes else ""
if resultat.returncode != 0:
return (f"outil={genre} code={resultat.returncode}"
+ (f" ligne={premiere}" if premiere else ""))
if genre == "ffprobe":
if lignes:
return f"outil={genre} ligne={premiere}"
nombres = [ligne.strip() for ligne in (resultat.stdout or "").splitlines()]
try:
if not any(int(nombre) > 0 for nombre in nombres if nombre):
return f"outil={genre} paquets-video=0"
except ValueError:
return f"outil={genre} sortie=illisible"
return ""
def valid_mp4_complet(path: Path) -> bool:
"""Vérifie un téléchargement entier avant de le publier ou le supprimer.
La vérification structurelle élimine les réponses HTML et les fichiers
tronqués au niveau d'une boîte. Une sonde parcourt ensuite tous les paquets
vidéo : elle détecte notamment un ``mdat`` écourté malgré un ``ftyp`` et un
``moov`` encore lisibles. Faute de sonde, on échoue fermé afin qu'aucune
source distante ne soit supprimée sur la foi d'un contrôle incomplet.
Toute ligne d'erreur de ffprobe fait refuser le fichier, sauf le bruit
connu des clips intacts (_BRUITS_FFPROBE).
"""
return not _motif_refus_mp4_complet(path)
def raison_refus_mp4(path: Path) -> str:
"""La raison du refus de ``valid_mp4_complet`` pour ce fichier, sans chemin
ni nom de clip, prête à être imprimée. Ne lève jamais : un diagnostic ne
doit pas transformer un échec en plantage. Relance le contrôle, donc à ne
demander qu'après un refus."""
try:
return _motif_refus_mp4_complet(path) or msg("motif_aucun")
except Exception as erreur:
return msg("motif_diagnostic_impossible", erreur=type(erreur).__name__)
def load_json(path: Path, default: dict) -> dict:
"""Lit un objet JSON, ou ``default`` s'il est absent, corrompu ou d'un
autre type que dict.
La lecture elle-même (refus passager de Windows face à un remplacement
concurrent, antivirus, indexeur) est nico579_commons.atomique.lire_json :
un refus qui persiste lève au lieu de rendre ``default``, qui ferait
réécrire un contenu vidé à toute lecture-modification-écriture."""
value = atomique.lire_json(path, default)
return value if isinstance(value, dict) else default
def save_json(path: Path, value: dict) -> None:
"""Remplace ``path`` atomiquement (nico579_commons.atomique.ecrire_json).
Temporaire propre à chaque appel : un nom fixe était partagé par tous les
écrivains, et serve.py tourne sur un ThreadingHTTPServer dont deux routes
réécrivent ASSEMBLED_DURATIONS en parallèle (FileNotFoundError, reproduit
197 fois sur 600 écritures concurrentes, audit du 2026-09-24). Les refus
passagers de Windows (162 sur 600 en CI) sont retentés."""
atomique.ecrire_json(path, value)
def find_ffmpeg() -> str:
"""Choisit un ffmpeg capable d'incruster du texte.
Le premier trouvé ne convient pas forcément : les binaires livrés par
imageio-ffmpeg pour Linux sont compilés sans libfreetype, donc sans le
filtre drawtext, alors que ceux de Windows et de macOS l'ont. Un outil dont
la raison d'être est d'inscrire l'heure dans l'image ne peut pas se
contenter du premier venu : on essaie les candidats dans l'ordre et on
retient le premier qui sait le faire.
À défaut, on renvoie quand même le premier candidat : check_drawtext_available
produira ensuite un message clair, plutôt qu'un « ffmpeg introuvable »
trompeur alors qu'il y en a un."""
candidats = []
# En bundle, ffmpeg voyage avec l'exécutable : on le prend là en premier,
# pour ne pas dépendre de ce qui traîne sur la machine cible.
for motif in ("ffmpeg*.exe", "ffmpeg-*", "ffmpeg"):
candidats += [str(c) for c in sorted(runtime.resource_dir().glob(motif))
if c.is_file()]
systeme = shutil.which("ffmpeg")
if systeme:
candidats.append(systeme)
vendor = runtime.resource_dir() / "_vendor"
if vendor.is_dir():
sys.path.insert(0, str(vendor))
try:
import imageio_ffmpeg
candidats.append(imageio_ffmpeg.get_ffmpeg_exe())
except (ImportError, RuntimeError):
pass
if not candidats:
raise RuntimeError(msg("ffmpeg_introuvable"))
for candidat in candidats:
if has_drawtext(candidat):
return candidat
return candidats[0]
def extraire_vignette(ffmpeg: str, source: Path, pending: Path, timeout=None) -> bool:
"""Extrait la vignette d'un clip dans `pending` ; vrai si le fichier existe.
Prise un peu après le début, jamais sur la première image : une caméra qui
vient de se déclencher livre souvent une ou deux images noires ou
surexposées, le temps que l'exposition s'ajuste. Un clip plus court que la
position demandée se rabat sur sa toute première image. Partagée par
serve.py (à la demande) et le téléchargeur (à l'arrivée du clip) : une seule
manière de fabriquer la vignette, donc un seul aspect."""
for entree in (["-ss", "1.5", "-i", str(source)], ["-i", str(source)]):
# -ss avant -i : ffmpeg saute directement à la position demandée au
# lieu de décoder tout ce qui précède.
runtime.lancer(
[ffmpeg, "-hide_banner", "-loglevel", "error", "-y", *entree,
"-frames:v", "1", "-vf", "scale=480:-2", "-q:v", "5", str(pending)],
stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL, check=False, timeout=timeout,
)
if pending.is_file() and pending.stat().st_size > 0:
return True
return False
def has_drawtext(ffmpeg: str) -> bool:
try:
result = runtime.lancer(
[ffmpeg, "-hide_banner", "-filters"],
stdin=subprocess.DEVNULL, stdout=subprocess.PIPE,
stderr=subprocess.DEVNULL, text=True, encoding="utf-8",
errors="replace", check=False, timeout=30,
)
except (OSError, subprocess.SubprocessError):
return False
return "drawtext" in (result.stdout or "")
def check_drawtext_available(ffmpeg: str) -> None:
"""Certains builds ffmpeg minimalistes (dont certains binaires portables)
omettent libfreetype et donc le filtre drawtext. On le détecte tôt avec
un message clair plutôt que de laisser échouer chaque fusion."""
result = runtime.lancer(
[ffmpeg, "-hide_banner", "-filters"],
stdin=subprocess.DEVNULL,
stdout=subprocess.PIPE,
stderr=subprocess.DEVNULL,
text=True,
encoding="utf-8",
errors="replace",
check=False,
)
if "drawtext" not in result.stdout:
raise RuntimeError(msg("drawtext_absent"))
def check_timestamp_rendering(ffmpeg: str, font_path: Path,
style: TimestampStyle = STYLE_PAR_DEFAUT) -> None:
"""Incruste un horodatage de test sur une image noire et vérifie qu'il en
reste quelque chose.
C'est un garde-fou de bout en bout : drawtext ne signale ni un chemin de
police mal encodé dans le filtergraph (fontconfig prend le relais en
silence), ni un format strftime que la libc locale ignore (le texte rendu
est alors simplement vide). Sans ce test, ces deux pannes ne se voient
qu'en regardant la vidéo finale.
La couleur de style est reprise ici, pas seulement la police : un texte
choisi noir sur ce fond noir de test échouerait sinon à tort ce garde-fou
(image entièrement noire par construction, cause confondue avec une
police illisible)."""
graph = (
"color=c=black:s=640x120:d=0.1,"
+ drawtext_chain(quote_filter_path(font_path), 40, 0, style.color,
clock=style.clock)
+ ",format=gray"
)
result = runtime.lancer(
[ffmpeg, "-hide_banner", "-loglevel", "error", "-filter_complex", graph,
"-frames:v", "1", "-f", "rawvideo", "-"],
stdin=subprocess.DEVNULL,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
env=dict(os.environ, TZ="UTC0"),
check=False,
)
stderr = result.stderr.decode("utf-8", "replace").strip()
if result.returncode != 0 or stderr:
raise RuntimeError(msg(
"horodatage_echec_police", police=font_path,
detail=stderr or msg("horodatage_echec_sans_message"),
))
if not any(result.stdout):
raise RuntimeError(msg("horodatage_vide", police=font_path))
def find_font(explicit: Path | None) -> Path:
if explicit is not None:
if explicit.is_file():
return explicit
raise RuntimeError(msg("police_introuvable", chemin=explicit))
# À côté du programme, comme le dit le message police_absente (avant
# 0.14, c'était aussi le dossier des données par défaut).
programme = runtime._dossier_ancre()
candidates = [
programme / "DejaVuSans-Bold.ttf",
programme / "fonts" / "DejaVuSans-Bold.ttf",
]
system = platform.system()
if system == "Windows":
windir = Path(os.environ.get("WINDIR", r"C:\Windows"))
candidates += [
windir / "Fonts" / "arialbd.ttf",
windir / "Fonts" / "arial.ttf",
windir / "Fonts" / "consolab.ttf",
]
elif system == "Darwin":
candidates += [
Path("/System/Library/Fonts/Supplemental/Arial Bold.ttf"),
Path("/System/Library/Fonts/Supplemental/Arial.ttf"),
Path("/Library/Fonts/Arial Bold.ttf"),
]
else:
candidates += [
Path("/usr/share/fonts/truetype/dejavu/DejaVuSans-Bold.ttf"),
Path("/usr/share/fonts/truetype/liberation/LiberationSans-Bold.ttf"),
Path("/usr/share/fonts/TTF/DejaVuSans-Bold.ttf"),
]
for candidate in candidates:
if candidate.is_file():
return candidate
raise RuntimeError(msg("police_absente", fichier=Path(__file__).name))
def quote_filter_path(path: Path) -> str:
"""Encode un chemin pour une option de filtre ffmpeg (ex. drawtext:fontfile).
Le texte d'un filtergraph subit DEUX passes de dé-échappement successives :
d'abord le découpage de la chaîne de filtres, puis celui des arguments d'un
filtre. Un chemin Windows ('C:\\Windows\\Fonts\\arialbd.ttf') doit donc :
- être entre quotes simples, ce qui le protège de la 1re passe ;
- avoir ses ':' échappés en '\\:', consommés par la 2e passe ;
- n'utiliser que des '/' comme séparateur, sinon les '\\' sont consommés
de façon incohérente entre les deux passes (Windows accepte '/').
"""
text = path.as_posix().replace("'", "'\\''").replace(":", "\\:")
return f"'{text}'"
def parse_created_at(value: str) -> dt.datetime:
created = dt.datetime.fromisoformat(value.replace("Z", "+00:00"))
if created.tzinfo is None:
created = created.replace(tzinfo=dt.timezone.utc)
return created.astimezone(dt.timezone.utc)
def read_registry(state_path: Path) -> dict:
"""Clips connus du téléchargeur, vide tant que rien n'a été récupéré.
Un registre absent est l'état normal d'une installation neuve, pas une
panne : l'interface doit pouvoir s'ouvrir et l'assemblage se dire sans
matière avant le premier téléchargement. Un registre présent mais illisible
est autre chose, une corruption qu'il faut annoncer au lieu de la confondre
avec « rien à montrer »."""
if not state_path.exists():
return {}
state = load_json(state_path, None)
entries = state.get("clips") if isinstance(state, dict) else None
if not isinstance(entries, dict):
raise RuntimeError(msg("registre_illisible", chemin=state_path))
return entries
def _cles_camera_par_collision(entries: dict) -> dict:
"""(nom, network_id) -> clé de regroupement, seulement pour les noms où
une vraie collision existe.
Le registre distingue les appareils par network_id/device_id, mais le
regroupement journalier n'utilisait que le nom : deux caméras portant le
même nom sur deux hubs finissaient dans le même assemblage (reproduit
avec deux "jardin" sur des réseaux distincts, mélangées dans une seule
journalière).
Le fix littéral (regrouper par identité stable partout) fragmenterait à
tort les données réelles : une même caméra physique peut avoir plusieurs
network_id/device_id incohérents dans le registre pour des raisons
historiques (vieilles entrées USB sans network_id, bug de rapprochement
déjà corrigé par ailleurs mais pas rétroactif sur l'existant) - vérifié
sur le registre réel, "jardin" et "Terrasse1" y ont chacun plusieurs
couples (network_id, device_id) mais un seul network_id CONNU (toujours
le même quand renseigné, jamais deux valeurs différentes). Ce n'est donc
pas un vrai homonyme, juste une incohérence, et ne doit rien changer.
Ne sépare donc que les noms où au moins deux network_id NON VIDES et
DIFFÉRENTS coexistent (signal positif de deux hubs réels, même principe
que _hub_compatible() dans blink_registre.py) : ce cas ne s'est jamais
produit dans le registre réel, seulement dans la reproduction du bug.
Le network_id trié en premier garde le nom nu (n'invalide pas
l'historique déjà assemblé sous ce nom, quel qu'il soit) ; les autres
reçoivent un suffixe lisible, conservant le nom pour l'affichage comme
demandé plutôt qu'un identifiant opaque.
Même règle pour deux noms DIFFÉRENTS qui se nettoient pareil (« Garage »
et « Garage! » donnent le même dossier par safe_name) : sans elle, leurs
journalières s'écrivaient au même chemin, l'une écrasant l'autre. Le nom
déjà propre garde son dossier (l'historique déjà assemblé n'est pas
déplacé), les autres reçoivent « (2) », « (3) »... ; la casse ne distingue
pas deux dossiers, Windows et macOS les confondant."""
reseaux_par_nom = defaultdict(set)
paires = set()
for entry in entries.values():
camera = str(entry.get("camera") or "camera").strip() or "camera"
network_id = str(entry.get("network_id") or "")
paires.add((camera, network_id))
if network_id:
reseaux_par_nom[camera].add(network_id)
cles = {}
for camera, reseaux in reseaux_par_nom.items():
if len(reseaux) <= 1:
continue
_principal, *autres = sorted(reseaux)
for rang, reseau in enumerate(autres, start=2):
cles[(camera, reseau)] = f"{camera} ({rang})"
def dossier(cle: str) -> str:
return safe_name(cle).casefold()
cles_par_paire = {paire: cles.get(paire, paire[0]) for paire in paires}
noms = sorted(set(cles_par_paire.values()),
key=lambda nom: (safe_name(nom) != nom, nom.casefold(), nom))
pris, renommes = set(), {}
for nom in noms:
if dossier(nom) not in pris:
pris.add(dossier(nom))
continue
rang = 2
while dossier(f"{nom} ({rang})") in pris or f"{nom} ({rang})" in noms:
rang += 1
renommes[nom] = f"{nom} ({rang})"
pris.add(dossier(renommes[nom]))
for paire, nom in cles_par_paire.items():
if nom in renommes:
cles[paire] = renommes[nom]
return cles
def _camera_key(cles_collision: dict, entry: dict, camera: str) -> str:
return cles_collision.get((camera, str(entry.get("network_id") or "")), camera)
def load_groups(input_dir: Path, timezone: ZoneInfo) -> dict:
"""Regroupe les fichiers enregistrés par le téléchargeur incrémental.
Une journée sans aucun clip survivant (tous écartés, ou plus aucun
valide) reste présente dans le résultat, avec une liste vide, plutôt que
de disparaître : sans cette entrée, la journée n'était plus jamais revue
une fois son dernier clip écarté, ni reconstruite (elle devrait se vider)
ni supprimée - sa vidéo déjà assemblée restait sur disque indéfiniment et
continuait d'être agrégée dans l'hebdomadaire et la mensuelle (bug #2,
revue de code du 0eab463, voir AUDIT-2026-08-13.md 28.61)."""
entries = read_registry(input_dir / DOWNLOAD_STATE)
cles_collision = _cles_camera_par_collision(entries)
root = input_dir.resolve()
groups = defaultdict(list)
for entry in entries.values():
try:
created = parse_created_at(entry["created_at"])
camera = str(entry.get("camera") or "camera").strip() or "camera"
except (KeyError, TypeError, ValueError):
continue
camera = _camera_key(cles_collision, entry, camera)
if entry.get("purged_at"):
# Purgé par la conservation : ni clip à assembler, ni journée à
# vider (sa journalière reste telle quelle, voir journees_purgees).
continue
local_day = created.astimezone(timezone).date().isoformat()
clips = groups[(camera, local_day)]
try:
if entry.get("excluded"):
continue
source = (root / entry["path"]).resolve()
if not runtime.est_relatif_a(source, root) or not valid_mp4(source):
continue
except (KeyError, TypeError, OSError):
continue
clips.append((created, source))
for clips in groups.values():
clips.sort(key=lambda item: (item[0], str(item[1]).casefold()))
return dict(groups)
def journees_a_source_indisponible(input_dir: Path, timezone: ZoneInfo) -> tuple[set, set]:
"""(jours, identités) pour les clips NON écartés dont le fichier brut est
absent ou invalide.
load_groups() traite un clip manquant/invalide exactement comme un clip
explicitement écarté : les deux disparaissent de sa liste, sans laisser
de trace de la raison. Nécessaire pour ne PAS confondre en aval « tous
les clips de cette journée ont été écartés par choix » (la journalière
déjà assemblée, et les segments normalisés associés, doivent disparaître)
et « au moins un clip existe encore dans le registre mais son fichier a
disparu » (l'un et les autres doivent au contraire être préservés, bug
constaté en réel : source USB débranchée pendant un passage, dossier
déplacé).
jours : (caméra, jour local) ayant au moins un tel clip, à exempter de la
suppression de journalière dans _executer(). identités : clip_identity()
de chacun (calculable sans que le fichier existe, c'est un chemin relatif
- voir clip_identity()), à ajouter à used_segments avant prune_normalized()
pour ne pas effacer un segment déjà encodé et toujours valide du seul
fait que son brut source a disparu."""
entries = read_registry(input_dir / DOWNLOAD_STATE)
cles_collision = _cles_camera_par_collision(entries)
root = input_dir.resolve()
jours, identites = set(), set()
for entry in entries.values():
try:
created = parse_created_at(entry["created_at"])
camera = str(entry.get("camera") or "camera").strip() or "camera"
except (KeyError, TypeError, ValueError):
continue
camera = _camera_key(cles_collision, entry, camera)
if entry.get("excluded") or entry.get("purged_at"):
continue
local_day = created.astimezone(timezone).date().isoformat()
try:
source = (root / entry["path"]).resolve()
if runtime.est_relatif_a(source, root) and valid_mp4(source):
continue # chargé normalement par load_groups, rien à signaler
except (KeyError, TypeError, OSError):
jours.add((camera, local_day))
continue
jours.add((camera, local_day))
try:
identites.add(clip_identity(input_dir, source))
except ValueError:
pass
return jours, identites
def group_fingerprint(keys: list) -> str:
"""Empreinte d'une journée : la liste ordonnée des clés de rendu de ses
segments normalisés. La journalière n'étant qu'une concaténation de ces
segments, elle est à jour exactement quand cette liste n'a pas bougé."""
digest = hashlib.sha256()
digest.update((RENDER_VERSION + "\n").encode("utf-8"))
for key in keys:
digest.update((key + "\n").encode("utf-8"))
return digest.hexdigest()
def probe_clip_info(ffmpeg: str, source: Path) -> tuple[float, int, int, float, bool]:
"""Lit durée, résolution, fps et présence d'audio depuis l'en-tête du MP4."""
result = runtime.lancer(
[ffmpeg, "-hide_banner", "-i", str(source)],
stdin=subprocess.DEVNULL,
stdout=subprocess.DEVNULL,
stderr=subprocess.PIPE,
text=True,
encoding="utf-8",
errors="replace",
check=False,
)
stderr = result.stderr
duration_match = re.search(r"Duration:\s*(\d+):(\d+):(\d+(?:\.\d+)?)", stderr)
if not duration_match:
raise RuntimeError(msg("duree_introuvable", nom=source.name))
hours, minutes, seconds = duration_match.groups()
duration = int(hours) * 3600 + int(minutes) * 60 + float(seconds)
if duration <= 0:
raise RuntimeError(msg("duree_invalide", nom=source.name))
video_line_match = re.search(r"Stream #\d+:\d+[^\n]*Video:[^\n]*", stderr)
if not video_line_match:
raise RuntimeError(msg("flux_video_introuvable", nom=source.name))
video_line = video_line_match.group(0)
size_match = re.search(r"(\d{2,5})x(\d{2,5})", video_line)
if not size_match:
raise RuntimeError(msg("resolution_introuvable", nom=source.name))
width, height = int(size_match.group(1)), int(size_match.group(2))
fps_match = re.search(r"([\d.]+)\s*fps", video_line)
fps = float(fps_match.group(1)) if fps_match else 15.0
has_audio = bool(re.search(r"Stream #\d+:\d+[^\n]*Audio:", stderr))
return duration, width, height, fps, has_audio
def resolve_identity(
input_dir: Path, normalized_dir: Path, excluded_dir: Path, value: str
) -> str:
"""Retrouve l'identité d'un clip à partir d'un chemin donné à la main.
On accepte le brut, le normalisé ou l'écarté indifféremment : les trois
arborescences sont des miroirs, la partie relative est la même. En pratique
on désigne le clip qu'on vient de regarder."""
# Espaces et fins de ligne rognés : un chemin arrive souvent d'un
# copier-coller ou d'une liste, et un « \r » traînant est invisible à
# l'affichage tout en faisant échouer la correspondance.
path = Path(value.strip()).resolve()
roots = (input_dir.resolve(), normalized_dir.resolve(), excluded_dir.resolve())
for root in roots:
if runtime.est_relatif_a(path, root):
return path.relative_to(root).as_posix()
raise RuntimeError(msg(
"chemin_hors_racines",
racines=", ".join(root.name for root in roots), valeur=value,
))
def move_aside(source: Path, destination: Path) -> None:
"""Déplace un fichier en créant l'arborescence, sans jamais écraser."""
if not source.exists():
return
runtime.creer_sous_sorties(destination.parent)
if destination.exists():
source.unlink()
return
shutil.move(str(source), str(destination))
def set_excluded(
input_dir: Path,
normalized_dir: Path,
excluded_dir: Path,
values: list,
excluded: bool,
) -> int:
"""Écarte des clips, ou les réintègre.
Exclure ne détruit rien : le brut est déplacé dans Blink_Excluded, sa
version normalisée est effacée (elle se refabrique à partir du brut), et
une pierre tombale est posée dans le registre de téléchargement pour que
blink2video.py ne le rapatrie plus. La journalière concernée voit alors sa liste
de segments changer, donc son empreinte, donc elle est réassemblée au
passage suivant, et les agrégats avec elle.
Réintégrer fait le chemin inverse et ramène le brut à sa place : le clip
est re-normalisé au même passage, sans dépendre de la rétention du hub.
Vider Blink_Excluded reste un geste séparé et explicite."""
state_path = input_dir / DOWNLOAD_STATE
changed = 0
# Même verrou que le téléchargement (runtime.verrou ne distingue les
# ressources que par ce premier nom, "exclusion" n'est qu'une étiquette).
# Lecture, décision et écriture doivent se tenir sous un seul verrou tenu
# en continu : lire puis écrire séparément laissait une fenêtre où un
# téléchargement concurrent pouvait ajouter un clip entre les deux, que
# notre propre copie, déjà périmée, effaçait ensuite en écrasant toute la
# table des clips.
with runtime.verrou("registre", "exclusion", stale_after=60, attente=10):
clips = read_registry(state_path)
for value in values:
identity = resolve_identity(input_dir, normalized_dir, excluded_dir, value)
keys = [
key
for key, entry in clips.items()
if isinstance(entry, dict) and entry.get("path") == identity
]
if not keys:
print(msg("registre_inconnu", identity=identity))
continue
for key in keys:
if excluded:
clips[key]["excluded"] = True
clips[key]["excluded_at"] = dt.datetime.now(dt.timezone.utc).isoformat()
move_aside(input_dir / identity, excluded_dir / identity)
(normalized_dir / identity).unlink(missing_ok=True)
print(msg("clip_exclu", identity=identity))
else:
# Un horodatage plutôt qu'un simple retrait du champ :
# _ecrire_registre() (blink_registre.py) s'en sert pour
# savoir que CETTE décision de réintégration est plus
# récente qu'un excluded=True qu'un downloader périmé
# tenterait de réécrire après coup (revue du 27/08, bug 2).
clips[key]["excluded"] = False
clips[key]["excluded_at"] = dt.datetime.now(dt.timezone.utc).isoformat()
move_aside(excluded_dir / identity, input_dir / identity)
if (input_dir / identity).exists():
print(msg("clip_reintegre", identity=identity))
else:
print(msg("clip_reintegre_brut_absent", identity=identity))
changed += 1
if changed:
# Le registre porte aussi sa version, et un jour d'autres champs,
# que read_registry() ne renvoie pas : on repart de l'état complet
# pour n'en remplacer que la table des clips. Cet état est garanti
# inchangé depuis notre lecture ci-dessus, le verrou étant resté
# tenu sans interruption entre les deux.
state = load_json(state_path, {})
state["clips"] = clips
save_json(state_path, state)
return changed
def clip_identity(input_dir: Path, source: Path) -> str:
"""Identité stable d'un clip : son chemin relatif au dossier de
téléchargement. C'est aussi le chemin de sa version normalisée, les deux
arborescences étant des miroirs : la correspondance entre un brut et son
segment se lit à l'œil, sans consulter de registre."""
return source.resolve().relative_to(input_dir.resolve()).as_posix()
def stat_tag(source: Path) -> str:
stat = source.stat()
return f"{stat.st_size}:{stat.st_mtime_ns}"
def clip_info(ffmpeg: str, registry: dict, identity: str, created, source: Path) -> ClipInfo:
"""Caractéristiques d'un clip, mémorisées dans le registre.
Analyser un clip coûte un lancement de ffmpeg ; comme le résultat ne change
pas tant que le fichier ne change pas, on le garde. C'est aussi ce qui
permettra plus tard de raisonner sur un clip dont le brut a été effacé."""
entry = registry["clips"].get(identity) or {}
tag = stat_tag(source)
probe = entry.get("probe")
if entry.get("stat") != tag or not isinstance(probe, dict):
duration, width, height, fps, has_audio = probe_clip_info(ffmpeg, source)
probe = {
"duration": duration, "width": width, "height": height,
"fps": fps, "has_audio": has_audio,
}
entry.update({"stat": tag, "probe": probe})
registry["clips"][identity] = entry
return ClipInfo(
created=created,
source=source,
duration=probe["duration"],
width=probe["width"],
height=probe["height"],
fps=probe["fps"],
has_audio=probe["has_audio"],
)
def camera_target(registry: dict, camera: str, infos: list) -> tuple:
"""Résolution et cadence cibles d'une caméra, mémorisées et monotones.
Les calculer par journée les ferait osciller : un jour sans clip haute
définition ramènerait la cible vers le bas, ce qui invaliderait tous les
segments déjà encodés puis les ré-invaliderait au clip suivant. On les fixe
donc par caméra, et on ne les laisse que monter. Un clip de meilleure
définition relève la cible et provoque un ré-encodage complet de la caméra,
ce qui est le comportement voulu : mieux vaut letterboxer les anciens que
rétrécir les nouveaux."""
stored = registry["cameras"].get(camera) or {}
width = max([info.width for info in infos] + [int(stored.get("width", 0))])
height = max([info.height for info in infos] + [int(stored.get("height", 0))])
# Cadence arrondie à l'image entière. Mesurée au millième, elle varie d'un
# clip à l'autre sans que rien ne change pour l'œil : un clip venu du cloud
# mesuré à 30,11 au lieu de 30,0 a suffi à relever la cible, donc à
# ré-encoder les trente-deux segments de la caméra. La montée doit répondre
# à une vraie différence, 15 contre 30, pas à un artefact de mesure.
fps = float(round(max([info.fps for info in infos]
+ [float(stored.get("fps", 0))])))
registry["cameras"][camera] = {"width": width, "height": height, "fps": fps}
return width, height, fps
def wall_clock_epoch(moment_utc: dt.datetime, timezone: ZoneInfo) -> int:
"""Convertit un instant UTC en epoch « déguisé » : les chiffres de
l'heure locale (calculés correctement via zoneinfo côté Python) sont
réinterprétés comme si c'était de l'UTC. En lançant ensuite ffmpeg avec
TZ=UTC0, drawtext affiche exactement cette heure locale sans jamais
consulter de base de fuseaux horaires côté ffmpeg/OS."""
local = moment_utc.astimezone(timezone)
naive_as_utc = local.replace(tzinfo=dt.timezone.utc)
return int(naive_as_utc.timestamp())
# Format strftime de l'heure, tel qu'il doit figurer DANS le filtergraph.
# 24h : %X plutôt que %T ou %H:%M:%S, voir drawtext_chain. 12h : les ':' sont
# échappés par trois antislashs, seule forme qui survive aux trois passes de
# dé-échappement (graphe, option, arguments de %{pts}) ; vérifié sur une image
# rendue par le ffmpeg 7.1 embarqué : « 01/11/2026 02:59:59 PM ». %r, qui
# éviterait l'échappement, rend une chaîne vide avec le strftime de Windows.
FORMAT_HEURE = {
"24h": "%X",
"12h": "%I\\\\\\:%M\\\\\\:%S %p",
}
def drawtext_chain(font_value: str, fontsize: int, epoch: int,
color: str = "white", box_opacity: float = 0.55,
clock: str = "24h") -> str:
"""Filtre drawtext affichant une horloge murale partant de `epoch`.
%X (heure complète) plutôt que %T ou %H:%M:%S : d'une part le strftime de
MSVC ne connaît pas %T et rend alors une chaîne vide sans la moindre
erreur, d'autre part un ':' littéral ne survit pas aux dé-échappements
successifs du filtergraph. ffmpeg n'appelant pas setlocale, %X reste en
locale « C », soit HH:MM:SS partout. Le format 12h (`clock`) passe par
%I, %M, %S et %p, connus de tous les strftime, avec des ':' échappés
(FORMAT_HEURE) ; %p vaut AM ou PM en locale « C ».
`color` a déjà été validé (_couleur_ffmpeg_valide, appelé par argparse) :
seuls alphanumériques, « # » et un éventuel « @opacité » peuvent
l'atteindre, aucun caractère qui romprait ce filtergraph."""
heure = FORMAT_HEURE.get(clock, FORMAT_HEURE["24h"])
text_expr = f"'%{{pts\\:localtime\\:{epoch}\\:%d/%m/%Y {heure}}}'"
return (
f"drawtext=fontfile={font_value}:fontsize={fontsize}:fontcolor={color}:"
f"box=1:boxcolor=black@{box_opacity}:boxborderw=8:x=20:y=h-th-20:text={text_expr}"
)
def build_batch_filter(
batch: list,
target_w: int,
target_h: int,
target_fps: float,
timezone: ZoneInfo,
font_value: str | None,
style: TimestampStyle = STYLE_PAR_DEFAUT,
) -> str:
"""Construit le filtergraph d'un lot : normalisation + horodatage incrusté
sur chaque clip, puis concaténation.
Avec font_value=None l'horodatage est omis : c'est le mode utilisé pour
recoller des vidéos déjà horodatées (agrégats hebdo / mensuel) dont les
paramètres diffèrent et qui doivent donc être ré-encodées.
Les clips sans piste audio reçoivent un silence généré *dans* le graphe par
une source anullsrc dédiée. Une entrée lavfi partagée ne conviendrait pas :
une sortie de filtre ne peut être branchée qu'une seule fois, donc deux
clips muets dans le même lot rendraient le graphe invalide.
"""
fontsize = style.size if style.size is not None else max(18, target_h // 18)
chains = []
for i, clip in enumerate(batch):
overlay = ""
if font_value is not None:
epoch = wall_clock_epoch(clip.created, timezone)
overlay = "," + drawtext_chain(font_value, fontsize, epoch,
style.color, style.box_opacity,
style.clock)
chains.append(
f"[{i}:v]scale={target_w}:{target_h}:force_original_aspect_ratio=decrease,"
f"pad={target_w}:{target_h}:(ow-iw)/2:(oh-ih)/2:color=black,setsar=1,"
f"fps={target_fps},setpts=PTS-STARTPTS"
+ overlay
+ f"[v{i}]"
)
if clip.has_audio:
chains.append(
f"[{i}:a]aresample=48000,aformat=channel_layouts=stereo,"
f"asetpts=PTS-STARTPTS[a{i}]"
)
else:
chains.append(
f"anullsrc=channel_layout=stereo:sample_rate=48000,"
f"atrim=0:{clip.duration:.3f},asetpts=PTS-STARTPTS[a{i}]"
)
concat_inputs = "".join(f"[v{i}][a{i}]" for i in range(len(batch)))
chains.append(f"{concat_inputs}concat=n={len(batch)}:v=1:a=1[outv][outa]")
return ";".join(chains)
def run_ffmpeg_batch(
ffmpeg: str,
batch: list,
target_w: int,
target_h: int,
target_fps: float,
timezone: ZoneInfo,
font_value: str | None,
preset: str,
crf: int,
output_path: Path,
on_progress=None,
style: TimestampStyle = STYLE_PAR_DEFAUT,
) -> tuple[bool, str]:
filter_graph = build_batch_filter(
batch, target_w, target_h, target_fps, timezone, font_value, style
)
command = [ffmpeg, "-hide_banner", "-loglevel", "error", "-y"]
for clip in batch:
command += ["-i", str(clip.source)]
command += [
"-filter_complex", filter_graph,
"-map", "[outv]",
"-map", "[outa]",
"-c:v", "libx264",
"-profile:v", "high",
"-pix_fmt", "yuv420p",
"-preset", preset,
"-crf", str(crf),
"-c:a", "aac",
"-b:a", "128k",
"-ar", "48000",
"-movflags", "+faststart",
"-max_muxing_queue_size", "9999",
# -progress écrit un flux clé=valeur lisible par un programme, sans
# rien changer au reste. C'est l'option prévue pour ça depuis 2012 ;
# analyser la ligne d'état habituelle serait fragile (elle change de
# forme selon la version et n'est pas faite pour être relue).
"-progress", "pipe:1", "-nostats",
str(output_path),
]
env = dict(os.environ, TZ="UTC0")
process = runtime.demarrer(
command,
stdin=subprocess.DEVNULL,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
encoding="utf-8",
errors="replace",
# bufsize=1 : lecture ligne par ligne. Sans lui Python remplit un
# tampon avant de rendre la main, et l'avancement arrive par paquets.
bufsize=1,
env=env,
# Sa propre session hors Windows, comme les verbes lancés côte à côte
# (voir blink_cli.executer) : avec_descendance=True sur ce PID fera
# os.killpg(getpgid(pid)) (runtime.arreter_processus), qui viserait
# sinon le groupe entier du terminal ayant lancé le parent.
start_new_session=(sys.platform != "win32"),
)
# Inscrit pour qu'un « stop » ailleurs le tue directement : le PID
# principal qui nous a lancés n'a lui-même jamais droit au taskkill /T
# (protection du navigateur, voir arreter_processus), donc sans ça cet
# ffmpeg resterait orphelin et tournerait jusqu'à sa fin.
runtime.inscrire_travailleur(process.pid)
# stderr est vidé par un fil dédié : s'il se remplit pendant qu'on lit
# stdout, ffmpeg se bloque sur son écriture et le tout s'immobilise.
captured = []
drain = threading.Thread(target=lambda: captured.append(process.stderr.read()))
drain.start()
total = sum(clip.duration for clip in batch) or 1.0
# Chien de garde : un ffmpeg qui se bloque (constaté en réel, AUDIT-2026-
# 08-13 28.82) laissait sinon `process.wait()` - et la lecture de
# `process.stdout` avant lui - attendre indéfiniment. La boucle merge
# entière restait alors figée en silence, sans erreur ni log, jusqu'à un
# arrêt manuel.
#
# Le seuil porte sur l'ABSENCE de progression, pas sur une durée totale
# fixe : `-progress pipe:1` écrit plusieurs lignes par seconde tant que
# l'encodage avance, quelle que soit la vitesse de la machine ou sa
# charge du moment. Un délai fixe proportionnel à la durée du lot aurait
# fallu deviner un facteur de vitesse (PC lent, machine chargée par
# ailleurs...) forcément arbitraire ; ici, peu importe que ce lot prenne
# 10 secondes ou 2 heures du moment que ça continue d'avancer, seul un
# silence prolongé est suspect. SILENCE_MAX (module) : même seuil que
# concat_copy, qui n'a pas de -progress à observer.
dernier_signe = time.monotonic()
signe_verrou = threading.Lock()
termine = threading.Event()
tue_par_silence = False
def surveiller() -> None:
nonlocal tue_par_silence
while not termine.wait(10):
with signe_verrou:
silence = time.monotonic() - dernier_signe
if silence > SILENCE_MAX:
tue_par_silence = True
process.kill()
return
chien_de_garde = threading.Thread(target=surveiller, daemon=True)
chien_de_garde.start()
try:
for line in process.stdout:
with signe_verrou:
dernier_signe = time.monotonic()
if on_progress is None:
continue
name, _, value = line.strip().partition("=")
# out_time_us et out_time_ms sont tous deux en microsecondes ; le
# second est mal nommé et conservé par compatibilité.
if name in ("out_time_us", "out_time_ms") and value.isdigit():
on_progress(min(int(value) / 1_000_000 / total, 1.0))
process.wait()
finally:
termine.set()
chien_de_garde.join(timeout=5)
runtime.retirer_travailleur(process.pid)
drain.join(timeout=5)
stderr = "".join(part for part in captured if part).strip()
if tue_par_silence:
return False, msg("ffmpeg_silencieux", secondes=int(SILENCE_MAX))
if process.returncode != 0 or not valid_mp4(output_path):
return False, stderr or msg("mp4_invalide")
return True, ""
def concat_copy(ffmpeg: str, parts: list, destination: Path) -> tuple[bool, str]:
"""Recolle des lots déjà ré-encodés (même codec/résolution/fps) par
simple copie de flux : rapide, sans perte."""
list_path = None
try:
with tempfile.NamedTemporaryFile(
mode="w",
suffix=".ffconcat",
prefix="blink_",
dir=destination.parent,
encoding="utf-8",
newline="\n",
delete=False,
) as concat_file:
list_path = Path(concat_file.name)
for part in parts:
escaped = part.resolve().as_posix().replace("'", "'\\''")
concat_file.write(f"file '{escaped}'\n")
command = [
ffmpeg, "-hide_banner", "-loglevel", "error", "-y",
"-f", "concat", "-safe", "0", "-i", str(list_path),
"-c", "copy", "-movflags", "+faststart",
str(destination),
]
process = runtime.demarrer(
command,
stdin=subprocess.DEVNULL,
stdout=subprocess.DEVNULL,
stderr=subprocess.PIPE,
text=True,
encoding="utf-8",
errors="replace",
# Même raison que run_ffmpeg_batch : sans session propre hors
# Windows, le killpg de arreter_processus viserait le groupe du
# terminal ayant lancé le parent plutôt que ce seul processus.
start_new_session=(sys.platform != "win32"),
)
# Même inscription que run_ffmpeg_batch, même raison : ce concat final
# est un vrai enfant du processus courant, mais taskkill /T ne
# s'applique jamais au PID principal (protection du navigateur).
runtime.inscrire_travailleur(process.pid)
try:
try:
_, stderr = process.communicate(timeout=SILENCE_MAX)
except subprocess.TimeoutExpired:
# Même risque documenté que run_ffmpeg_batch (AUDIT-2026-08-13
# 28.82) : un ffmpeg bloqué laisserait sinon communicate()
# attendre indéfiniment, gelant l'assemblage hebdomadaire ou
# mensuel en silence. Pas de -progress ici (simple copie de
# flux, rien à observer) : un délai fixe fait donc office de
# chien de garde, faute de mieux. kill() puis un second
# communicate() sans timeout : la doc subprocess prévient que
# TimeoutExpired ne tue pas l'enfant, sans quoi ses tubes
# restent ouverts et un futur wait() peut se bloquer à son tour.
process.kill()
process.communicate()
return False, msg("ffmpeg_silencieux", secondes=int(SILENCE_MAX))
finally:
runtime.retirer_travailleur(process.pid)
if process.returncode != 0 or not valid_mp4(destination):
message = (stderr or "").strip() or msg("mp4_invalide")
return False, message
return True, ""
finally:
if list_path is not None:
list_path.unlink(missing_ok=True)
def progress_printer(label: str):
"""Fabrique un rapporteur d'avancement pour un encodage.
Sur un terminal la ligne est réécrite en place, comme n'importe quel outil
en ligne de commande. Dans un tube (serve.py qui relit la sortie), un
simple retour chariot n'émettrait aucune ligne et rien n'arriverait avant
la fin : on écrit alors des lignes complètes, que le lecteur reconnaît à
leur forme et n'ajoute pas au journal.
Sous pythonw.exe (raccourcis de démarrage automatique, voir
autostart.py), sys.stdout vaut None : chaque fusion y échouait dès le
premier lot, avant même d'encoder quoi que ce soit, `sys.stdout.isatty()`
levant une AttributeError aussitôt rattrapée par la boucle --loop puis
retentée sans jamais aboutir (repéré en réel le 27 août 2026 : « tour
interrompu par une erreur » toutes les 5 min pendant des heures). Aucun
flux vers lequel écrire : le rapporteur ne fait rien plutôt que de
retenter un print() tout aussi voué à échouer."""
if sys.stdout is None:
return lambda fraction: None
interactive = sys.stdout.isatty()
state = {"percent": -1, "when": 0.0}
def report(fraction: float) -> None:
percent = int(fraction * 100)
now = time.monotonic()
if percent < state["percent"] + 2 and percent < 100:
return
if now - state["when"] < 0.5 and percent < 100:
return
if percent == state["percent"]:
return
state["percent"], state["when"] = percent, now
if interactive:
print(f"\r {label} {percent:3d}%", end="", flush=True)
if percent >= 100:
print()
else:
print(f" {label} {percent}%", flush=True)
return report
def render_key(
identity: str,
clip: ClipInfo,
target: tuple,
epoch: int,
font_path: Path | None,
preset: str,
crf: int,
style: TimestampStyle = STYLE_PAR_DEFAUT,
) -> str:
"""Empreinte de rendu d'un segment normalisé : tout ce qui change son
contenu doit figurer ici, sinon un segment périmé serait réutilisé
silencieusement. Bascule avec/sans horodatage comprise : sans elle, un
segment déjà encodé horodaté resterait tel quel après désactivation du
réglage, l'horodatage figé dedans plutôt que réellement retiré.
Le style (taille/couleur/opacité, issue GitHub #7) ne compte que si un
horodatage est effectivement dessiné : sans ça, changer de couleur alors
que --no-timestamp est actif réencoderait tout pour rien, la valeur
n'ayant jamais atteint l'image.
L'horloge 12h n'est ajoutée que lorsqu'elle est choisie : en 24h,
l'empreinte reste celle d'avant ce réglage, sans quoi la mise à jour
réencoderait tous les segments existants pour une image identique."""
target_w, target_h, target_fps = target
style_tag = (
f"{style.size}/{style.color}/{style.box_opacity}"
+ ("/12h" if style.clock == "12h" else "")
if font_path is not None else "sans-horodatage"
)
payload = "|".join(
[
NORMALIZE_VERSION,
identity,
stat_tag(clip.source),
f"{target_w}x{target_h}@{target_fps}",
str(epoch),
font_path.as_posix() if font_path is not None else "sans-horodatage",
preset,
str(crf),
style_tag,
]
)
return hashlib.sha256(payload.encode("utf-8")).hexdigest()
def normalize_clip(
ffmpeg: str,
timezone: ZoneInfo,
registry: dict,
normalized_dir: Path,
identity: str,
clip: ClipInfo,
target: tuple,
key: str,
font_path: Path | None,
preset: str,
crf: int,
force: bool,
on_progress=None,
style: TimestampStyle = STYLE_PAR_DEFAUT,
) -> tuple[bool, str, bool]:
"""Produit la version normalisée (et horodatée si `font_path` est donné)
d'un clip, si nécessaire.
Renvoie (ok, erreur, ré-encodé). Un segment déjà présent dont la clé de
rendu correspond est laissé tel quel : c'est ce qui rend la fusion
réellement incrémentale."""
destination = normalized_dir / identity
entry = registry["clips"].get(identity) or {}
if not force and entry.get("key") == key and valid_mp4(destination):
return True, "", False
runtime.creer_sous_sorties(destination.parent)
# Extension .mp4 conservée : ffmpeg choisit son format de sortie d'après
# elle, un simple suffixe .tmp le laisserait sans muxer.
pending = destination.with_name(destination.stem + ".tmp.mp4")
pending.unlink(missing_ok=True)
target_w, target_h, target_fps = target
font_value = quote_filter_path(font_path) if font_path is not None else None
ok, error = run_ffmpeg_batch(
ffmpeg, [clip], target_w, target_h, target_fps, timezone,
font_value, preset, crf, pending, on_progress, style=style,
)
if not ok:
pending.unlink(missing_ok=True)
return False, error, False
pending.replace(destination)
entry["key"] = key
entry["normalized_at"] = dt.datetime.now(dt.timezone.utc).isoformat()
registry["clips"][identity] = entry
return True, "", True
def merge_group(
ffmpeg: str,
segments: list,
destination: Path,
) -> tuple[bool, str]:
"""Assemble la vidéo d'une journée à partir de ses segments normalisés.
Il n'y a plus rien à encoder ici : les segments partagent codec, résolution
et cadence, donc une simple copie de flux suffit. La journalière est une
vue du stock normalisé, au même titre que l'hebdomadaire et la mensuelle."""
runtime.creer_sous_sorties(destination.parent)
temporary_video = destination.with_name(destination.stem + ".tmp.mp4")
temporary_video.unlink(missing_ok=True)
try:
ok, error = concat_copy(ffmpeg, segments, temporary_video)
if not ok:
return False, error
if not valid_mp4(temporary_video):
return False, msg("mp4_invalide")
temporary_video.replace(destination)
return True, ""
finally:
temporary_video.unlink(missing_ok=True)
def prune_normalized(normalized_dir: Path, registry: dict, keep: set) -> int:
"""Supprime les segments qu'aucun clip connu ne revendique plus, ainsi que
les entrées de registre correspondantes.
On ne le fait qu'après un passage complet : avec un filtrage --camera ou
--date, les groupes non traités n'auraient pas déclaré leurs segments et
seraient effacés à tort."""
if not normalized_dir.is_dir():
return 0
removed = 0
for candidate in normalized_dir.rglob("*.mp4*"):
if candidate.is_file() and candidate not in keep:
candidate.unlink(missing_ok=True)
removed += 1
for directory in sorted(
(p for p in normalized_dir.rglob("*") if p.is_dir()),
key=lambda p: len(p.parts),
reverse=True,
):
if not any(directory.iterdir()):
directory.rmdir()
known = {
identity
for identity in registry["clips"]
if (normalized_dir / identity) in keep
}
for identity in list(registry["clips"]):
if identity not in known:
del registry["clips"][identity]
return removed
def collect_dailies(output_dir: Path) -> dict:
"""Inventorie les vidéos journalières déjà produites, par caméra.
On lit le disque plutôt que le registre de fusion : les agrégats restent
ainsi corrects même si les journalières ont été construites par plusieurs
exécutions filtrées (--date, --camera)."""
dailies = defaultdict(list)
if not output_dir.is_dir():
return {}
for camera_dir in sorted(
p for p in output_dir.iterdir() if p.is_dir() and not p.name.startswith(".")
):
for video in sorted(camera_dir.glob("*.mp4")):
match = DAILY_NAME.match(video.name)
if not match or not valid_mp4(video):
continue
try:
day = dt.date.fromisoformat(match.group("day"))
except ValueError:
continue
dailies[camera_dir.name].append((day, video))
return dict(dailies)
def period_label(day: dt.date, period: str) -> str:
"""Étiquette de la période contenant `day`.
Semaine ISO 8601 (lundi-dimanche, format YYYY-Www) : c'est la convention
des dates ISO déjà utilisée partout ailleurs dans le projet, et elle évite
l'ambiguïté du « début de semaine » selon les pays."""
if period == "weekly":
year, week, _ = day.isocalendar()
return f"{year}-W{week:02d}"
return f"{day.year}-{day.month:02d}"
def period_fingerprint(parts: list) -> str:
digest = hashlib.sha256()
digest.update((PERIOD_VERSION + "\n").encode("utf-8"))
for part in parts:
stat = part.stat()
digest.update(f"{part.name}|{stat.st_size}|{stat.st_mtime_ns}\n".encode("utf-8"))
return digest.hexdigest()
def concat_videos(
ffmpeg: str,
timezone: ZoneInfo,
parts: list,
destination: Path,
preset: str,
crf: int,
) -> tuple[bool, str]:
"""Recolle des vidéos déjà horodatées.
Chemin rapide : si toutes partagent résolution et cadence, une simple copie
de flux suffit (pas de perte, quelques secondes). Sinon on ré-encode via le
filtre concat, seule façon fiable d'assembler des sources hétérogènes."""
runtime.creer_sous_sorties(destination.parent)
temporary = destination.with_name(destination.stem + ".tmp.mp4")
temporary.unlink(missing_ok=True)
try:
# On tente la copie de flux d'abord, sans rien mesurer : les
# journalières d'une caméra sortent toutes du même stock normalisé,
# donc du même codec, de la même résolution et de la même cadence.
#
# Surtout, ne pas décider d'après les cadences mesurées : la cadence
# moyenne d'un fichier concaténé vaut images / durée, et elle varie de
# quelques centièmes d'une journée à l'autre (30.00, 30.03, 30.04 ici)
# parce qu'une caméra Blink ne livre pas un flux parfaitement régulier.
# Ce test déclarait donc « hétérogène » un stock parfaitement homogène,
# et ré-encodait la semaine et le mois entiers à chaque passage.
ok, error = concat_copy(ffmpeg, parts, temporary)
if not ok:
print(msg("copie_refusee"))
try:
infos = []
for part in parts:
duration, width, height, fps, has_audio = probe_clip_info(ffmpeg, part)
infos.append(
ClipInfo(None, part, duration, width, height, fps, has_audio)
)
except RuntimeError as error:
return False, str(error)
ok, error = run_ffmpeg_batch(
ffmpeg, infos,
max(i.width for i in infos),
max(i.height for i in infos),
float(round(max(i.fps for i in infos))),
timezone, None, preset, crf, temporary,
)
if not ok:
return False, error
if not valid_mp4(temporary):
return False, msg("mp4_invalide")
temporary.replace(destination)
return True, ""
finally:
temporary.unlink(missing_ok=True)
def build_periods(
ffmpeg: str,
timezone: ZoneInfo,
output_dir: Path,
period_dir: Path,
period: str,
force: bool,
preset: str,
crf: int,
proteges=frozenset(),
) -> tuple[int, int, int]:
"""Assemble les vidéos journalières en agrégats hebdomadaires ou mensuels.
``proteges`` : clés « caméra|période » dont une journée a perdu à la fois
ses clips bruts et sa journalière (voir periodes_a_proteger()). Leur
agrégat déjà assemblé la contient peut-être encore : il n'est ni
reconstruit à partir des autres jours, ni supprimé comme période vide."""
dailies = collect_dailies(output_dir)
# Pas de retour anticipé si dailies est vide : une caméra qui n'a plus
# aucune journalière (tout exclu) doit quand même atteindre le nettoyage
# des périodes obsolètes en bas de fonction, sans quoi ses anciennes
# hebdomadaires/mensuelles ne seraient jamais revisitées ni supprimées.
state_path = period_dir / MERGE_STATE
state = load_json(state_path, {"version": 1, "groups": {}})
state.setdefault("groups", {})
built = skipped = failed = 0
# Clés (camera|label) encore couvertes par au moins une journalière : le
# reste de state["groups"] correspond à des périodes devenues vides
# (tous leurs jours exclus ou disparus), jamais revisitées par la boucle
# ci-dessous puisqu'elle ne parcourt que collect_dailies() - sans ce
# suivi, leur agrégat déjà assemblé restait sur disque indéfiniment.
labels_attendus = set()
for camera, entries in sorted(dailies.items()):
buckets = defaultdict(list)
for day, video in entries:
buckets[period_label(day, period)].append(video)
for label, parts in sorted(buckets.items()):
destination = period_dir / camera / f"{label}_{camera}.mp4"
key = f"{camera}|{label}"
labels_attendus.add(key)
if key in proteges:
print(msg("periode_protegee", periode=key))
skipped += 1
continue
fingerprint = period_fingerprint(parts)
if (
not force
and state["groups"].get(key, {}).get("fingerprint") == fingerprint
and valid_mp4(destination)
):
print(msg("periode_deja_a_jour", nom=destination.name, jours=len(parts)))
skipped += 1
continue
print(msg("periode_assemblage", label=label, camera=camera, jours=len(parts)))
ok, error = concat_videos(
ffmpeg, timezone, parts, destination, preset, crf
)
if not ok:
print(msg("echec_generique", erreur=error))
failed += 1
continue
state["groups"][key] = {
"fingerprint": fingerprint,
"path": destination.relative_to(period_dir).as_posix(),
"days": len(parts),
"updated_at": dt.datetime.now(dt.timezone.utc).isoformat(),
}
save_json(state_path, state)
print(msg("cree", destination=destination))
built += 1
obsoletes = [key for key in state["groups"]
if key not in labels_attendus and key not in proteges]
if obsoletes:
for key in obsoletes:
entree = state["groups"].pop(key)
chemin = entree.get("path")
if chemin:
(period_dir / chemin).unlink(missing_ok=True)
print(msg("periode_supprimee", periode=key))
save_json(state_path, state)
return built, skipped, failed
def journees_purgees(input_dir: Path, timezone: ZoneInfo) -> set:
"""(caméra, jour local) ayant au moins un clip purgé par la conservation.
À distinguer d'une source indisponible (clé USB débranchée, dossier
déplacé) : la journée n'est pas en échec, sa journalière est simplement
figée, et ses agrégats aussi tant qu'elle manque (periodes_a_proteger)."""
entries = read_registry(input_dir / DOWNLOAD_STATE)
cles_collision = _cles_camera_par_collision(entries)
jours = set()
for entry in entries.values():
if not isinstance(entry, dict) or not entry.get("purged_at"):
continue
try:
created = parse_created_at(entry["created_at"])
camera = str(entry.get("camera") or "camera").strip() or "camera"
except (KeyError, TypeError, ValueError):
continue
camera = _camera_key(cles_collision, entry, camera)
jours.add((camera, created.astimezone(timezone).date().isoformat()))
return jours
def suivre_purges_manuelles(input_dir: Path) -> int:
"""Note comme purgés les bruts effacés à la main, et lève la note de ceux
qui sont revenus ; rend le nombre de changements.
Vider Blink_Clips à la main est permis depuis #118 : le téléchargement ne
les reprend pas. Mais l'assemblage prenait chaque brut absent pour une
source débranchée : un échec à chaque passage, et plus jamais
d'hebdomadaire ni de mensuelle. Un brut absent alors que son dossier est
là est traité comme une purge de la conservation (journalière figée,
agrégats protégés), avec la marque purge_manuelle qui la rend
réversible : un dossier déplacé puis remis en place retrouve ses clips au
passage suivant. Un fichier présent mais illisible reste une source
indisponible."""
if not input_dir.is_dir():
return 0
state_path = input_dir / DOWNLOAD_STATE
root = input_dir.resolve()
changes = 0
# Même verrou et même lecture-écriture d'un tenant que set_excluded().
with runtime.verrou("registre", "purge manuelle", stale_after=60, attente=10):
clips = read_registry(state_path)
maintenant = dt.datetime.now(dt.timezone.utc).isoformat()
for entry in clips.values():
if not isinstance(entry, dict) or entry.get("excluded") or not entry.get("path"):
continue
try:
source = (root / entry["path"]).resolve()
except (TypeError, OSError):
continue
if not runtime.est_relatif_a(source, root):
continue
present = source.exists()
if entry.get("purge_manuelle"):
if present:
entry.pop("purged_at", None)
entry.pop("purge_manuelle", None)
changes += 1
elif not entry.get("purged_at") and not present:
entry["purged_at"] = maintenant
entry["purge_manuelle"] = True
changes += 1
if changes:
state = load_json(state_path, {})
state["clips"] = clips
save_json(state_path, state)
if changes:
print(msg("purges_manuelles", n=changes))
return changes
def fin_de_periode(label: str, period: str):
"""Dernier jour d'une période (« 2026-W07 », « 2026-02 »), None si
l'étiquette n'en est pas une."""
try:
if period == "weekly":
annee, semaine = label.split("-W")
return dt.date.fromisocalendar(int(annee), int(semaine), 7)
annee, mois = label.split("-")
suivant = dt.date(int(annee) + int(mois) // 12, int(mois) % 12 + 1, 1)
return suivant - dt.timedelta(days=1)
except ValueError:
return None
def _retirer_dossiers_vides(racine: Path) -> None:
"""Retire les sous-dossiers vidés par la purge, jamais la racine."""
if not racine.is_dir():
return
for dossier in sorted((p for p in racine.rglob("*") if p.is_dir()),
key=lambda p: len(p.parts), reverse=True):
try:
dossier.rmdir()
except OSError:
pass # pas vide, ou tenu ouvert : rien à faire
def purger_anciens(args, timezone: ZoneInfo, aujourdhui: dt.date | None = None) -> int:
"""Conservation par catégorie (issue #120) : efface ce qui a dépassé sa
durée, sans jamais laisser l'assemblage refaire ou raccourcir le reste.
Chaque vidéo assemblée est refaite à partir de ce dont elle est faite
tant que cette matière existe. D'où l'ordre et les gardes ci-dessous :
- bruts : par journée entière (une journée à moitié purgée serait
réassemblée plus courte), seulement quand sa journalière existe et est
plus récente que chacun de ses clips (sinon elle ne les contient pas
tous) ; sans assemblage (--purge-only), les bruts sont la seule copie
et la durée choisie s'applique telle quelle. Le registre garde le clip
marqué purgé : il n'est ni retéléchargé (#118), ni pris pour une
source débranchée ;
- journalières : seulement quand plus aucun brut de la journée ne reste,
et que chaque agrégat activé qui la contient existe et est plus récent
qu'elle ;
- hebdomadaires et mensuelles : seulement quand plus aucune journalière
de la période ne reste ;
- directs et photos : d'après la date du fichier, rien n'en dépend.
Une durée de 0 garde pour toujours."""
if not any(getattr(args, nom) for nom in DEFAUTS_CONSERVATION if nom.startswith("keep_")):
return 0
aujourdhui = aujourdhui or dt.datetime.now(timezone).date()
input_dir = args.input.resolve()
output_dir = args.output.resolve()
periodes = (("weekly", args.weekly_output.resolve(), args.no_weekly,
args.keep_weekly_days),
("monthly", args.monthly_output.resolve(), args.no_monthly,
args.keep_monthly_days))
def perime(jour: dt.date, jours: int) -> bool:
return bool(jours) and (aujourdhui - jour).days > jours
def mtime(chemin: Path) -> float:
try:
return chemin.stat().st_mtime
except OSError:
return 0.0
def chemin_journaliere(camera_dir: str, jour: dt.date) -> Path:
return output_dir / camera_dir / f"{jour.isoformat()}_{camera_dir}.mp4"
supprimes = 0
state_path = input_dir / DOWNLOAD_STATE
if args.keep_clips_days:
root = input_dir
with runtime.verrou("registre", "conservation", stale_after=60, attente=10):
clips = read_registry(state_path)
cles_collision = _cles_camera_par_collision(clips)
par_jour = defaultdict(list)
for cle, entry in clips.items():
if (not isinstance(entry, dict) or entry.get("purged_at")
or entry.get("excluded") or not entry.get("path")):
continue
try:
created = parse_created_at(entry["created_at"])
camera = str(entry.get("camera") or "camera").strip() or "camera"
except (KeyError, TypeError, ValueError):
continue
camera = _camera_key(cles_collision, entry, camera)
par_jour[(camera, created.astimezone(timezone).date())].append(cle)
maintenant = dt.datetime.now(dt.timezone.utc).isoformat()
purges = 0
for (camera, jour), cles in sorted(par_jour.items()):
if not perime(jour, args.keep_clips_days):
continue
sources = [(root / clips[cle]["path"]).resolve() for cle in cles]
if any(not runtime.est_relatif_a(source, root) for source in sources):
continue
if not args.purge_only:
journaliere = chemin_journaliere(safe_name(camera), jour)
if (not valid_mp4(journaliere)
or mtime(journaliere) < max(map(mtime, sources))):
continue
for cle, source in zip(cles, sources):
source.unlink(missing_ok=True)
clips[cle]["purged_at"] = maintenant
purges += 1
if purges:
# Comme set_excluded() : état complet relu sous le verrou
# tenu sans interruption, seule la table des clips change.
state = load_json(state_path, {})
state["clips"] = clips
save_json(state_path, state)
_retirer_dossiers_vides(input_dir)
supprimes += purges
if args.keep_daily_days and output_dir.is_dir():
clips = read_registry(state_path)
cles_collision = _cles_camera_par_collision(clips)
# Une journée dont un brut reste (même momentanément illisible) garde
# sa journalière : elle serait refaite, ou c'est peut-être sa seule copie.
avec_bruts = set()
for entry in clips.values():
if (not isinstance(entry, dict) or entry.get("purged_at")
or entry.get("excluded")):
continue
try:
created = parse_created_at(entry["created_at"])
camera = str(entry.get("camera") or "camera").strip() or "camera"
except (KeyError, TypeError, ValueError):
continue
camera = _camera_key(cles_collision, entry, camera)
avec_bruts.add((safe_name(camera), created.astimezone(timezone).date()))
for camera_dir, entrees in collect_dailies(output_dir).items():
for jour, video in entrees:
if not perime(jour, args.keep_daily_days) or (camera_dir, jour) in avec_bruts:
continue
agregats = [period_dir / camera_dir / f"{period_label(jour, period)}_{camera_dir}.mp4"
for period, period_dir, desactive, _ in periodes if not desactive]
if any(not valid_mp4(agregat) or mtime(agregat) < mtime(video)
for agregat in agregats):
continue
video.unlink(missing_ok=True)
supprimes += 1
_retirer_dossiers_vides(output_dir)
for period, period_dir, _, jours_garde in periodes:
if not jours_garde or not period_dir.is_dir():
continue
restantes = {(camera_dir, period_label(jour, period))
for camera_dir, entrees in collect_dailies(output_dir).items()
for jour, _ in entrees}
for video in sorted(period_dir.glob("*/*.mp4")):
camera_dir = video.parent.name
suffixe = f"_{camera_dir}.mp4"
if not video.name.endswith(suffixe):
continue
label = video.name[:-len(suffixe)]
fin = fin_de_periode(label, period)
if fin is None or not perime(fin, jours_garde) or (camera_dir, label) in restantes:
continue
video.unlink(missing_ok=True)
supprimes += 1
_retirer_dossiers_vides(period_dir)
for racine, motif, jours_garde in ((args.live_output.resolve(), "*.mp4", args.keep_live_days),
(args.snapshots_output.resolve(), "*.jpg",
args.keep_snapshots_days)):
if not jours_garde or not racine.is_dir():
continue
limite = time.time() - (jours_garde + 1) * 86400
for fichier in racine.rglob(motif):
if fichier.is_file() and mtime(fichier) < limite:
fichier.unlink(missing_ok=True)
supprimes += 1
_retirer_dossiers_vides(racine)
if supprimes:
print(msg("purge_resume", n=supprimes))
journal(msg("purge_resume", n=supprimes))
return supprimes
def periodes_a_proteger(indisponibles: set, output_dir: Path, period: str) -> set:
"""Clés « caméra|période » que build_periods() ne doit ni reconstruire ni
supprimer : celles d'une journée dont un clip brut est indisponible ET
dont la journalière manque (audit du 26/09/2026, B02). La reconstruire à
partir des autres jours raccourcirait un agrégat qui contient peut-être
encore cette journée, parfois devenu sa seule copie.
Calculées sur toutes les journées, pas seulement celles d'un filtre
--date ou --camera : build_periods() reconstruit toutes les périodes,
quel que soit ce filtre. Une journée indisponible dont la journalière
existe encore ne protège rien, l'agrégat la reprend telle quelle."""
presentes = {
(dossier, jour)
for dossier, entrees in collect_dailies(output_dir).items()
for jour, _ in entrees
}
proteges = set()
for camera, jour_texte in indisponibles:
try:
jour = dt.date.fromisoformat(jour_texte)
except (TypeError, ValueError):
continue
dossier = safe_name(camera)
if (dossier, jour) not in presentes:
proteges.add(f"{dossier}|{period_label(jour, period)}")
return proteges
COULEUR_FFMPEG_RE = re.compile(r"^#?[A-Za-z0-9]+(@(0(\.[0-9]+)?|1(\.0+)?))?$")
def _couleur_ffmpeg_valide(valeur: str) -> str:
"""N'admet qu'un nom de couleur ou un code hexadécimal, avec un éventuel
@opacité (0.0-1.0) : la syntaxe complète que sait lire fontcolor= dans
drawtext_chain(). Rejette tout le reste, en particulier « : » et « , »
qui romprait ce filtergraph (une nouvelle option, ou un nouveau filtre)."""
if not COULEUR_FFMPEG_RE.match(valeur):
raise argparse.ArgumentTypeError(msg("couleur_invalide", valeur=valeur))
return valeur
def _taille_police_valide(valeur: str) -> int:
try:
nombre = int(valeur)
except ValueError:
nombre = -1
if not 8 <= nombre <= 500:
raise argparse.ArgumentTypeError(msg("taille_police_invalide", valeur=valeur))
return nombre
def _opacite_valide(valeur: str) -> float:
try:
nombre = float(valeur)
except ValueError:
nombre = -1.0
if not 0.0 <= nombre <= 1.0:
raise argparse.ArgumentTypeError(msg("opacite_invalide", valeur=valeur))
return nombre
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(
prog="blink2video merge",
description=msg("aide_desc"),
)
parser.add_argument("--input", type=Path, default=DEFAULT_INPUT)
parser.add_argument("--output", type=Path, default=DEFAULT_OUTPUT)
parser.add_argument("--weekly-output", type=Path, default=DEFAULT_WEEKLY)
parser.add_argument("--monthly-output", type=Path, default=DEFAULT_MONTHLY)
parser.add_argument(
"--normalized-output", type=Path, default=DEFAULT_NORMALIZED,
help=msg("aide_normalized"),
)
parser.add_argument("--no-weekly", action="store_true", help=msg("aide_no_weekly"))
parser.add_argument("--no-monthly", action="store_true", help=msg("aide_no_monthly"))
parser.add_argument("--no-timestamp", action="store_true", help=msg("aide_no_timestamp"))
# Conservation par catégorie (issue #120) : 0, le défaut, garde toujours.
for option, cle_aide in (("--keep-clips-days", "aide_keep_clips"),
("--keep-daily-days", "aide_keep_daily"),
("--keep-weekly-days", "aide_keep_weekly"),
("--keep-monthly-days", "aide_keep_monthly"),
("--keep-live-days", "aide_keep_live"),
("--keep-snapshots-days", "aide_keep_snapshots")):
parser.add_argument(option, type=runtime.jours_non_negatifs, default=0,
metavar="N", help=msg(cle_aide))
parser.add_argument("--purge-only", action="store_true", help=msg("aide_purge_only"))
parser.add_argument("--live-output", type=Path, default=DEFAULT_LIVE)
parser.add_argument("--snapshots-output", type=Path, default=DEFAULT_SNAPSHOTS)
parser.add_argument(
"--excluded-output", type=Path, default=DEFAULT_EXCLUDED,
help=msg("aide_excluded_output"),
)
parser.add_argument(
"--exclude", nargs="+", metavar="CLIP", default=[], help=msg("aide_exclude"),
)
parser.add_argument(
"--include", nargs="+", metavar="CLIP", default=[], help=msg("aide_include"),
)
parser.add_argument("--timezone", default="Europe/Paris")
runtime.ajouter_boucle(parser)
parser.add_argument("--date", help=msg("aide_date"))
parser.add_argument("--camera", help=msg("aide_camera"))
parser.add_argument("--force", action="store_true", help=msg("aide_force"))
parser.add_argument("--font", type=Path, default=None, help=msg("aide_font"))
parser.add_argument(
"--font-size", type=_taille_police_valide, default=None,
help=msg("aide_font_size"),
)
parser.add_argument(
"--font-color", type=_couleur_ffmpeg_valide, default="white",
help=msg("aide_font_color"),
)
parser.add_argument(
"--box-opacity", type=_opacite_valide, default=0.55,
help=msg("aide_box_opacity"),
)
parser.add_argument("--clock", choices=HORLOGES, default="24h", help=msg("aide_clock"))
parser.add_argument("--preset", default="veryfast", help=msg("aide_preset"))
parser.add_argument("--crf", type=int, default=21, help=msg("aide_crf"))
return parser.parse_args()
def journal(ligne: str) -> None:
"""Trace horodatée : lancé par le planificateur, le script n'a pas de
console (AUDIT-2026-08-23, blocage silencieux de la boucle merge)."""
moment = dt.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
try:
with (runtime.app_dir() / "merge.log").open("a", encoding="utf-8") as fichier:
fichier.write(f"{moment} {ligne}\n")
except OSError:
pass
def main() -> int:
args = parse_args()
def travail():
# Sans ce contrôle, un disque démonté se lirait comme des clips
# disparus : le tour échouerait sur chacun, purge comprise (issue #134).
try:
runtime.exiger_dossier_sorties(args.input)
except runtime.DossierSortiesAbsent as erreur:
print(erreur)
journal(str(erreur))
return 1
try:
with runtime.verrou("merge", "assemblage", stale_after=7200,
attente=30):
try:
code = _executer(args)
finally:
# Toujours retirer la marque d'avancement, même après une
# erreur : l'interface croirait sinon qu'un calcul dure.
runtime.fin_travail()
except runtime.BusyError as erreur:
# Un assemblage déjà en cours fait le même travail : le doubler
# réencoderait les mêmes clips et brouillerait le registre.
print(msg("assemblage_deja_en_cours", erreur=erreur))
journal(f"assemblage deja en cours : {erreur}")
return 0
runtime.marquer("merge")
return code
return runtime.repeter(travail, args.loop, journal)
def _calculer_cibles_encodage(groups: dict, registry: dict, input_dir: Path,
ffmpeg: str) -> dict:
"""Calcule un format par caméra sur tous ses clips, indépendamment des filtres."""
targets = {}
for camera in sorted({key[0] for key in groups}):
infos = [
clip_info(ffmpeg, registry, clip_identity(input_dir, source), created, source)
for key, clips in groups.items()
if key[0] == camera
for created, source in clips
]
targets[camera] = camera_target(registry, camera, infos)
return targets
def _planifier_normalisation(args, groups: dict, selected: dict, targets: dict,
registry: dict, *, ffmpeg: str, timezone: ZoneInfo,
input_dir: Path, normalized_dir: Path,
font_path: Path | None,
style: TimestampStyle = STYLE_PAR_DEFAUT) -> tuple:
"""Liste les journées à traiter, les segments à garder et les clips à encoder."""
plan: dict = {}
used_segments: set = set()
pending: set = set()
for (camera, day), clips in sorted(groups.items(), key=lambda item: item[0]):
target = targets[camera]
entries = []
for created, source in clips:
identity = clip_identity(input_dir, source)
info = clip_info(ffmpeg, registry, identity, created, source)
key = render_key(
identity, info, target, wall_clock_epoch(created, timezone),
font_path, args.preset, args.crf, style=style,
)
# Déclaré même hors sélection, pour échapper au nettoyage.
used_segments.add(normalized_dir / identity)
if (camera, day) not in selected:
continue
entries.append((identity, info, key))
known = (registry["clips"].get(identity) or {}).get("key")
if args.force or known != key or not valid_mp4(normalized_dir / identity):
pending.add(identity)
if (camera, day) in selected:
plan[(camera, day)] = (target, entries)
return plan, used_segments, pending
def _normaliser_plan(args, plan: dict, pending: set, registry: dict,
registry_path: Path, *, ffmpeg: str, timezone: ZoneInfo,
normalized_dir: Path, font_path: Path | None,
style: TimestampStyle = STYLE_PAR_DEFAUT) -> tuple:
"""Encode le plan et rend les journées préparées ainsi que le nombre d'échecs."""
normalized: dict = {}
encoded = reused = failed = 0
position, total = 0, len(pending)
if total:
print(msg("normalisation_titre", total=total))
for (camera, day), (target, entries) in sorted(plan.items(), key=lambda i: i[0]):
keys, segments = [], []
journee_incomplete = False
for identity, info, key in entries:
report = None
if identity in pending:
position += 1
print(f" [{position}/{total}] {identity}", flush=True)
report = progress_printer(f"[{position}/{total}]")
runtime.travail(msg("travail_preparation_clips"), position - 1, total,
cle="phase.prepare_clips")
ok, error, did_encode = normalize_clip(
ffmpeg, timezone, registry, normalized_dir, identity, info,
target, key, font_path, args.preset, args.crf, args.force,
report, style=style,
)
if not ok:
print(msg("echec_normalisation", erreur=error))
failed += 1
# Une journée dont un seul clip échoue à normaliser ne doit
# pas voir sa liste de segments amputée servir quand même à
# l'assemblage (étape 3) : ça remplacerait une journalière
# complète par une version partielle, silencieusement plus
# courte. journee_incomplete bloque cet assemblage.
journee_incomplete = True
continue
encoded += did_encode
reused += not did_encode
keys.append(key)
segments.append(normalized_dir / identity)
if did_encode:
# Sauvegarde par clip fraîchement encodé, pas seulement en
# fin de journée : un arrêt en cours de route (redémarrage,
# plantage) ne doit perdre que le clip en train d'être
# encodé, pas tout le progrès déjà acquis sur les
# précédents du même jour. Sans ça, une journée de
# plusieurs dizaines de clips recommençait entièrement à
# chaque interruption, même après avoir déjà correctement
# encodé la plupart d'entre eux - vécu en réel : plusieurs
# redémarrages consécutifs sur une même journée d'environ
# trente clips, qui relançait ffmpeg sur la totalité à
# chaque fois.
save_json(registry_path, registry)
normalized[(camera, day)] = (keys, segments, journee_incomplete)
save_json(registry_path, registry)
print(msg("normalisation_resume", encoded=encoded, reused=reused))
return normalized, failed
def _executer(args) -> int:
input_dir = args.input.resolve()
output_dir = args.output.resolve()
normalized_dir = args.normalized_output.resolve()
excluded_dir = args.excluded_output.resolve()
# Les exclusions s'appliquent avant tout le reste : la suite du passage
# reconstruit alors naturellement ce qui en dépend.
if args.exclude or args.include:
try:
if args.exclude:
set_excluded(input_dir, normalized_dir, excluded_dir, args.exclude, True)
if args.include:
set_excluded(input_dir, normalized_dir, excluded_dir, args.include, False)
except RuntimeError as error:
print(msg("erreur_generique", erreur=error))
return 1
# Conservation (issue #120) : avant l'assemblage, sous le même verrou
# « merge ». Jamais sur un passage filtré, qui ne voit qu'une partie.
# Un Namespace construit sans ces options (appelant direct, tests) vaut
# « aucune conservation ».
for nom, defaut in DEFAUTS_CONSERVATION.items():
if not hasattr(args, nom):
setattr(args, nom, defaut)
if not (args.date or args.camera):
try:
suivre_purges_manuelles(args.input.resolve())
purger_anciens(args, ZoneInfo(args.timezone))
except (OSError, runtime.BusyError, ZoneInfoNotFoundError) as erreur:
print(msg("purge_echec", erreur=erreur))
if args.purge_only:
return 0
style = TimestampStyle(args.font_size, args.font_color, args.box_opacity, args.clock)
try:
timezone = ZoneInfo(args.timezone)
ffmpeg = find_ffmpeg()
font_path = None
if not args.no_timestamp:
check_drawtext_available(ffmpeg)
font_path = find_font(args.font)
check_timestamp_rendering(ffmpeg, font_path, style)
groups = load_groups(input_dir, timezone)
indisponibles, identites_indisponibles = journees_a_source_indisponible(
input_dir, timezone
)
purgees = journees_purgees(input_dir, timezone)
except (RuntimeError, ZoneInfoNotFoundError) as error:
print(msg("erreur_generique", erreur=error))
return 1
selected = {
key: clips
for key, clips in groups.items()
if (not args.date or key[1] == args.date)
and (not args.camera or key[0].casefold() == args.camera.casefold())
}
if not selected:
print(msg("aucun_groupe"))
merge_state_path = output_dir / MERGE_STATE
merge_state = load_json(merge_state_path, {"version": 1, "groups": {}})
merge_state.setdefault("version", 1)
merge_state.setdefault("groups", {})
registry_path = normalized_dir / NORMALIZED_STATE
registry = load_json(registry_path, {"version": 1, "cameras": {}, "clips": {}})
registry.setdefault("version", 1)
registry.setdefault("cameras", {})
registry.setdefault("clips", {})
legacy = output_dir / LEGACY_SEGMENT_DIR
if legacy.is_dir():
shutil.rmtree(legacy, ignore_errors=True)
print(msg("cache_supprime", chemin=legacy))
# Étape 1 : cibles d'encodage. Elles se calculent sur l'ensemble des clips
# d'une caméra, y compris ceux qu'un filtre --date ou --camera exclut de la
# construction, sinon la cible dépendrait du filtre employé.
try:
targets = _calculer_cibles_encodage(groups, registry, input_dir, ffmpeg)
except RuntimeError as error:
print(msg("erreur_generique", erreur=error))
return 1
save_json(registry_path, registry)
# Étape 2 : plan de normalisation. On établit la liste complète avant
# d'encoder quoi que ce soit, pour pouvoir annoncer « [3/24] » : un
# compteur sans total connu n'apprend rien sur le temps restant, et c'est
# ce total qui alimente la barre de progression de serve.py.
plan, used_segments, pending = _planifier_normalisation(
args, groups, selected, targets, registry, ffmpeg=ffmpeg,
timezone=timezone, input_dir=input_dir, normalized_dir=normalized_dir,
font_path=font_path, style=style,
)
save_json(registry_path, registry)
# Étape 2 bis : encodage des seuls segments manquants ou périmés.
normalized, failed = _normaliser_plan(
args, plan, pending, registry, registry_path, ffmpeg=ffmpeg,
timezone=timezone, normalized_dir=normalized_dir, font_path=font_path,
style=style,
)
# Étape 3 : assemblage des journalières, par simple copie de flux.
built = skipped = 0
todo = sorted(normalized.items(), key=lambda item: item[0])
for index, ((camera, day), (keys, segments, journee_incomplete)) in enumerate(todo, start=1):
camera_path = safe_name(camera)
destination = output_dir / camera_path / f"{day}_{camera_path}.mp4"
state_key = f"{camera}|{day}"
if (camera, day) in purgees:
# Un clip arrivé après la purge de sa journée : la journalière
# existante contient déjà le reste, la refaire avec ce seul clip
# l'amputerait. Pas un échec, elle est simplement figée.
print(msg("jour_purge_fige", nom=destination.name))
skipped += 1
continue
if (camera, day) in indisponibles:
# load_groups() omet les bruts absents/invalides. La liste peut
# donc être non vide ET incomplète : ne jamais remplacer une
# journalière complète par les seuls clips encore accessibles.
print(msg("jour_reportee_source", nom=destination.name))
failed += 1
continue
if not segments:
# Rien à assembler : soit la journée n'a plus aucun clip valide
# (tout écarté), soit l'encodage a échoué pour ceux qui
# existaient. Seul le premier cas doit faire disparaître une
# vidéo déjà assemblée - le second ne doit surtout pas l'effacer
# sur un échec passager (réseau, ffmpeg) : elle reste à jour et
# sera retentée au prochain passage.
_, entries = plan.get((camera, day), (None, ()))
# (camera, day) dans indisponibles : un clip de cette journée
# existe encore dans le registre mais son fichier a disparu -
# à distinguer d'un choix d'exclusion (voir
# journees_a_source_indisponible()). Préserver la journalière
# déjà assemblée plutôt que la supprimer sur cette seule absence.
if not entries and (camera, day) not in indisponibles and destination.exists():
destination.unlink()
merge_state["groups"].pop(state_key, None)
save_json(merge_state_path, merge_state)
print(msg("jour_supprimee", nom=destination.name))
continue
if journee_incomplete:
# Au moins un clip de cette journée a échoué à normaliser :
# segments est amputé par rapport aux clips réellement présents,
# l'assembler remplacerait une journalière complète par une
# version partielle plus courte, silencieusement. failed est déjà
# compté (étape 2) ; la journée reste à jour et sera retentée au
# prochain passage, comme un échec total (voir plus haut).
print(msg("jour_reportee_partielle", nom=destination.name))
continue
fingerprint = group_fingerprint(keys)
previous = merge_state["groups"].get(state_key, {})
if (
not args.force
and previous.get("fingerprint") == fingerprint
and valid_mp4(destination)
):
print(msg("jour_deja_a_jour", nom=destination.name, clips=len(segments)))
skipped += 1
continue
print(msg("jour_assemblage", index=index, total=len(todo), jour=day,
camera=camera, clips=len(segments)))
runtime.travail(msg("travail_assemblage_videos"), index - 1, len(todo),
cle="phase.assemble_videos")
ok, error = merge_group(ffmpeg, segments, destination)
if not ok:
print(msg("echec_generique", erreur=error))
failed += 1
continue
merge_state["groups"][state_key] = {
"fingerprint": fingerprint,
"path": destination.relative_to(output_dir).as_posix(),
"clips": len(segments),
"updated_at": dt.datetime.now(dt.timezone.utc).isoformat(),
}
save_json(merge_state_path, merge_state)
print(msg("cree", destination=destination))
built += 1
if not args.date and not args.camera and not failed:
# Un segment déjà encodé pour un clip dont le brut a depuis disparu
# (source USB débranchée, dossier déplacé) reste valide : le
# préserver malgré son absence de used_segments, sans quoi
# prune_normalized() l'effacerait comme un simple clip écarté par
# choix (voir journees_a_source_indisponible()).
keep = used_segments | {normalized_dir / identity
for identity in identites_indisponibles}
removed = prune_normalized(normalized_dir, registry, keep)
if removed:
print(msg("stock_normalise", n=removed))
save_json(registry_path, registry)
print(msg("journalieres_resume", built=built, skipped=skipped, failed=failed))
if indisponibles & selected.keys():
# Une journalière reportée peut elle-même ne pas être présente.
# Reconstruire ses périodes à partir des seuls autres jours les
# raccourcirait aussi. Conserver les agrégats jusqu'à réparation.
print(msg("agregats_reportes"))
return 1
for period, period_dir, desactive in (
("weekly", args.weekly_output.resolve(), args.no_weekly),
("monthly", args.monthly_output.resolve(), args.no_monthly),
):
if desactive:
continue
label = msg("label_hebdomadaires") if period == "weekly" else msg("label_mensuelles")
print(msg("periode_titre", label=label))
p_built, p_skipped, p_failed = build_periods(
ffmpeg, timezone, output_dir, period_dir, period,
args.force, args.preset, args.crf,
proteges=periodes_a_proteger(indisponibles | purgees, output_dir, period),
)
failed += p_failed
print(msg("periode_resume", label=label, built=p_built, skipped=p_skipped,
failed=p_failed))
return 1 if failed else 0
if __name__ == "__main__":
raise SystemExit(main())