blink_registre.py
Rama main · réplica de lectura del repositorio original.
"""Registre incrémental des clips acquis : chargement, fusion, identité,
corrélation USB/cloud et historique durable des téléchargements.
Extrait de blink2video.py à l'étape 8 (AUDIT-2026-08-13.md, section 20, 8.3).
Ne connaît ni la session Blink, ni comment un clip se télécharge : uniquement
comment le savoir, l'identifier et l'écrire de façon durable."""
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 bisect
import datetime as dt
import hashlib
import json
import os
import stat as stat_module
import time
from pathlib import Path
from nico579_commons import atomique
import runtime
from blink_models import (
_apparier_evenements,
_identifiant_camera,
_identifiant_reseau,
_meme_camera,
clip_datetime_utc,
msg,
)
OUTPUT = runtime.dossier_sorties() / "Blink_Clips"
STATE_FILENAME = ".blink_download_state.json"
STATE_V1_BACKUP_FILENAME = ".blink_download_state.v1.backup.json"
def _copie_registre(output: Path) -> Path:
"""Copie hors du dossier de bruts, qui peut être purgé entièrement."""
return output.parent / f".{output.name}{STATE_FILENAME}"
# Registre absent : distinct de tout contenu JSON possible, « null » compris.
_ABSENT = object()
# Trou observé jusqu'à 125 ms (2026-11-06) : voir _lire_registre().
TENTATIVES_ABSENCE = 10
PAUSE_ABSENCE_S = 0.05
REMPLACEMENT_NON_ATOMIQUE = os.name == "nt"
def _lire_registre(output: Path):
"""Contenu du registre ou de sa copie, _ABSENT seulement si absence avérée.
Sous Windows, os.replace n'est pas atomique pour un lecteur : pendant le
remplacement, la cible n'existe plus du tout. Sonde du 2026-11-06, à
l'instant d'une lecture vide : le dossier ne contenait que le temporaire,
jusqu'à 125 ms durant. Cette absence n'est pas un refus, lire_json n'a
aucune raison de la réessayer, et le registre passait pour vide en silence.
On insiste donc avant d'y croire, sauf sans dossier de bruts : une
installation neuve n'attend jamais, et ailleurs le remplacement est atomique."""
for _ in range(TENTATIVES_ABSENCE):
for chemin in (output / STATE_FILENAME, _copie_registre(output)):
state = atomique.lire_json(chemin, _ABSENT, tolerer_corrompu=False)
if state is not _ABSENT:
return state
if not REMPLACEMENT_NON_ATOMIQUE or not output.is_dir():
break
time.sleep(PAUSE_ABSENCE_S)
return _ABSENT
def load_download_state(output: Path) -> dict:
"""Charge le registre incrémental placé dans le dossier de destination.
Absent, il est vide : c'est une installation neuve. Présent mais
momentanément inaccessible, il ne l'est surtout pas. Sous Windows, une
lecture échoue en PermissionError pendant qu'un autre processus remplace
le fichier (la boucle cloud le réécrit chaque minute), et ce refus passait
ici pour une corruption : le 2026-10-17, une passe USB a ainsi vu un
registre vide, puis retéléchargé et annoncé comme nouveaux six clips déjà
acquis. nico579_commons.atomique.lire_json réessaie ce refus passager et
lève s'il persiste : une passe interrompue (_faire_passage le signale, la
suivante réessaie) vaut mieux qu'un registre vide, qui fait tout
retélécharger et, relu avant une écriture, effacerait l'historique.
Si le dossier de bruts a été supprimé, la copie placée à côté fournit
l'historique ; le prochain enregistrement recrée le registre principal.
Seul un contenu invalide suit encore la stratégie du registre vide (I-10)."""
try:
state = _lire_registre(output)
if state is _ABSENT:
return {"version": 1, "clips": {}}
if not isinstance(state, dict):
raise ValueError(msg("registre_racine_objet"))
if state.get("version") not in (1, 2) or not isinstance(state.get("clips"), dict):
raise ValueError(msg("registre_format_inconnu"))
valides = {}
ignores = 0
for cle, entree in state["clips"].items():
if not isinstance(cle, str) or not isinstance(entree, dict):
ignores += 1
continue
instant = entree.get("created_at")
if not isinstance(instant, str):
ignores += 1
continue
try:
dt.datetime.fromisoformat(instant)
except ValueError:
ignores += 1
continue
# Migration v1 additive : les anciennes clés et archives restent
# intactes. Les métadonnées nouvelles permettent les recherches
# sans dépendre du nom de fichier ni de l'ID USB renuméroté.
entree.setdefault("source", "usb")
entree.setdefault("network_id", "")
entree.setdefault("device_id", "")
entree.setdefault("remote_id", "")
entree.setdefault("sync_id", str(cle).split(":", 1)[0])
# .strip() : même raison que dans state_key() plus bas (nom USB
# avec espace parasite sans device_id). N'affecte que les vieilles
# entrées v1 sans camera_identity du tout : setdefault ne touche
# jamais une entrée qui en a déjà un, migrée ou non.
camera_identite = (
f"device:{entree['device_id']}" if entree["device_id"]
else f"name:{str(entree.get('camera') or 'camera').strip().casefold()}"
)
entree.setdefault("camera_identity", camera_identite)
empreinte = json.dumps(
[entree["network_id"], camera_identite, instant],
ensure_ascii=False, separators=(",", ":"),
)
entree.setdefault(
"correlation_id",
hashlib.sha256(empreinte.encode("utf-8")).hexdigest(),
)
valides[cle] = entree
if ignores:
print(msg("registre_entrees_invalides", ignores=ignores))
state["clips"] = valides
state["version"] = 2
return state
except json.JSONDecodeError:
categorie = msg("registre_categorie_donnees")
except ValueError:
categorie = msg("registre_categorie_schema")
print(msg("registre_illisible", categorie=categorie))
return {"version": 1, "clips": {}}
def save_download_state(output: Path, state: dict) -> None:
"""Enregistre le registre incrémental, sans écraser le travail d'un autre.
Deux boucles écrivent désormais ici, l'une pour la clé USB toutes les dix
minutes, l'autre pour le cloud toutes les minutes. Écrire sa propre copie
en bloc ferait perdre les clips que l'autre vient d'ajouter : on relit donc
le fichier juste avant d'écrire, et on superpose ses propres entrées. Une
entrée n'est jamais retirée par ce chemin, seulement ajoutée ou mise à
jour, ce qui rend la superposition sûre.
Le remplacement reste atomique : un plantage en cours d'écriture laisse
l'ancien registre intact plutôt qu'un fichier tronqué."""
runtime.creer_sous_sorties(output)
state_file = output / STATE_FILENAME
with runtime.verrou("registre", "ecriture", stale_after=60, attente=10):
_sauvegarder_registre_v1(state_file)
_ecrire_registre(state_file, state)
# La copie survit à la suppression du dossier Blink_Clips. L'écriture
# atomique et le même verrou empêchent une copie partielle ou ancienne.
atomique.ecrire_json(_copie_registre(output), state)
def _sauvegarder_registre_v1(state_file: Path) -> None:
"""Conserve une copie unique du registre v1 avant sa première migration."""
backup = state_file.with_name(STATE_V1_BACKUP_FILENAME)
# Même lecture retentée que load_download_state(). Un refus qui persiste
# remonte tel quel : rien ne dit alors qu'une migration était en jeu.
try:
ancien = atomique.lire_json(state_file, None, tolerer_corrompu=False)
except json.JSONDecodeError as erreur:
raise OSError(msg("registre_migration_annulee")) from erreur
if not isinstance(ancien, dict) or ancien.get("version") != 1:
return
try:
brut = state_file.read_bytes()
if backup.exists():
sauvegarde = backup.read_bytes()
chargee = json.loads(sauvegarde)
if not isinstance(chargee, dict) or chargee.get("version") != 1:
raise OSError(msg("registre_sauvegarde_invalide"))
return
temporaire = backup.with_suffix(backup.suffix + ".tmp")
temporaire.write_bytes(brut)
temporaire.replace(backup)
if backup.read_bytes() != brut:
raise OSError(msg("registre_sauvegarde_non_verifiable"))
except (OSError, json.JSONDecodeError) as erreur:
raise OSError(msg("registre_migration_annulee")) from erreur
def _exclusion_a_retenir(precedente: dict, entree: dict) -> tuple:
"""État d'exclusion (excluded, excluded_at) à retenir pour un clip que
la fusion trouve des deux côtés.
La décision d'exclusion/réintégration (posée par set_excluded(), jamais
par le téléchargeur) est un état à part : sur ce champ précis, seule la
décision la plus récente doit l'emporter, dans les deux sens - une
priorité fixe à excluded=True protégeait une exclusion contre une copie
de downloader périmée, mais laissait une réintégration se faire défaire
par cette même copie arrivée après coup (revue du 27/08, bug 2).
Un excluded_at absent d'un côté ne prouve rien sur l'ordre réel : il
peut s'agir d'une entrée neuve qui ne connaît pas ce champ, ou d'une
donnée antérieure à son introduction. Un côté daté l'emporte donc
toujours sur un côté muet ; seul le cas où aucun des deux n'est daté
retombe sur le repli historique (B-04) : ne jamais laisser une copie
muette annuler une exclusion active."""
disque_exclu = bool(precedente.get("excluded"))
entrant_exclu = bool(entree.get("excluded"))
depuis_disque = precedente.get("excluded_at")
depuis_entrant = entree.get("excluded_at")
# Même lorsque les booléens coïncident, la date la plus récente doit
# survivre : la faire reculer autoriserait ensuite une décision opposée
# périmée à gagner lors d'une autre sauvegarde de téléchargement.
if depuis_disque and depuis_entrant:
if depuis_disque > depuis_entrant:
return disque_exclu, depuis_disque
return entrant_exclu, depuis_entrant
if depuis_disque and not depuis_entrant:
return disque_exclu, depuis_disque
if depuis_entrant and not depuis_disque:
return entrant_exclu, depuis_entrant
return disque_exclu or entrant_exclu, None
def _ecrire_registre(state_file: Path, state: dict) -> None:
"""Superpose ses propres entrées à celles déjà sur disque, atomiquement."""
output = state_file.parent
fusionne = dict(state)
disque = load_download_state(output)
clips = dict(disque.get("clips") or {})
for cle, entree in (state.get("clips") or {}).items():
precedente = clips.get(cle)
if isinstance(precedente, dict) and isinstance(entree, dict):
excluded, excluded_at = _exclusion_a_retenir(precedente, entree)
if (excluded, excluded_at) != (bool(entree.get("excluded")), entree.get("excluded_at")):
entree = dict(entree)
entree["excluded"] = excluded
if excluded_at is not None:
entree["excluded_at"] = excluded_at
# Purge par la conservation (issue #120, merge_daily.purger_anciens) :
# le brut n'existe plus, le marqueur ne doit jamais reculer devant
# une copie de téléchargeur antérieure à la purge.
# Une purge manuelle (merge_daily.suivre_purges_manuelles) garde sa
# marque avec elle : c'est ce qui la rend réversible.
purge = precedente.get("purged_at")
if purge and not entree.get("purged_at"):
entree = dict(entree)
entree["purged_at"] = purge
if precedente.get("purge_manuelle"):
entree["purge_manuelle"] = True
clips[cle] = entree
fusionne["clips"] = clips
fusionne["version"] = max(
2, int(disque.get("version") or 1), int(state.get("version") or 1),
)
# Remplacement retenté : sous Windows, il est refusé tant qu'un lecteur
# (la page, l'autre boucle) tient le registre ouvert.
atomique.ecrire_json(state_file, fusionne)
_invalider_index_registre(state)
state["version"] = fusionne["version"]
state["clips"] = fusionne.get("clips", state.get("clips"))
def state_key(sync, clip, source: str = "usb") -> str:
"""Identifie la provenance sans la confondre avec la corrélation.
Une identité USB contient module, réseau, caméra et instant, mais jamais
l'ID du manifeste : le Sync Module le renumérote lors d'une réindexation.
Une identité cloud ajoute au contraire l'ID distant du média, immuable
lorsqu'il est fourni par l'API. Le rapprochement USB/cloud reste une
opération séparée, tolérante sur l'instant.
Faute de ``device_id`` dans les objets USB de blinkpy 0.25, leur nom API
original est le repli documenté ; un renommage de caméra USB ne peut donc
pas être reconnu avec certitude."""
created = clip_datetime_utc(clip).isoformat()
# .strip() avant .casefold() : cohérent avec camera_setting_key ci-dessous
# et CloudClip.__init__ (blink_models.py). Un objet USB blinkpy 0.25 sans
# device_id peut renvoyer un nom avec un espace parasite en tête/fin (vu
# en réel sur le registre de production, 38 clips "jardin " isolés du
# reste de l'historique de la même caméra entre le 07 et le 11/08) ; sans
# ce trim, cet espace crée une identité de corrélation séparée.
camera = _identifiant_camera(clip) or f"name:{str(clip.name).strip().casefold()}"
identite = json.dumps(
[
source,
str(getattr(sync, "sync_id", "")),
_identifiant_reseau(clip, sync),
camera,
created,
str(getattr(clip, "id", "")) if source == "cloud" else "",
],
ensure_ascii=False,
separators=(",", ":"),
)
return "v2:" + hashlib.sha256(identite.encode("utf-8")).hexdigest()
def camera_setting_key(sync, clip) -> str:
"""Clé stable d'un réglage destructif propre à une caméra."""
network_id = _identifiant_reseau(clip, sync)
device_id = _identifiant_camera(clip)
material = (
["device", network_id, device_id]
if device_id
else [
"legacy", network_id, str(getattr(sync, "sync_id", "")),
str(getattr(clip, "name", "camera")).strip().casefold(),
]
)
digest = hashlib.sha256(
json.dumps(material, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
).hexdigest()
return f"camera-v2-{digest[:32]}"
def camera_setting_key_from_entry(entry: dict) -> str:
"""Même clé à partir d'une entrée persistée du registre."""
network_id = str(entry.get("network_id") or "")
device_id = str(entry.get("device_id") or "")
material = (
["device", network_id, device_id]
if device_id
else [
"legacy", network_id, str(entry.get("sync_id") or ""),
str(entry.get("camera") or "camera").strip().casefold(),
]
)
digest = hashlib.sha256(
json.dumps(material, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
).hexdigest()
return f"camera-v2-{digest[:32]}"
def remember_download(state: dict, sync, hub_name: str, clip, output: Path,
target: Path, source: str = "usb") -> None:
"""Marque un clip comme acquis uniquement lorsque son fichier existe.
La provenance est notée : une caméra couverte par un abonnement enregistre
dans le cloud, une autre sur la clé du module. L'interface le montre par
caméra, ce qui évite de chercher pourquoi telle caméra ne produit rien."""
_invalider_index_registre(state)
state["version"] = 2
state.setdefault("clips", {})[state_key(sync, clip, source)] = {
"hub": hub_name,
"camera": clip.name,
"created_at": clip_datetime_utc(clip).isoformat(),
"path": target.relative_to(output).as_posix(),
"bytes": target.stat().st_size,
"source": source,
"network_id": _identifiant_reseau(clip, sync),
"device_id": _identifiant_camera(clip),
"remote_id": str(getattr(clip, "id", "")),
"sync_id": str(getattr(sync, "sync_id", "")),
# .strip() : même raison que dans state_key() (nom USB avec espace
# parasite sans device_id) - écrit à chaque nouveau téléchargement,
# le site le plus important à protéger des trois de ce fichier.
"camera_identity": (
f"device:{_identifiant_camera(clip)}" if _identifiant_camera(clip)
else f"name:{str(clip.name).strip().casefold()}"
),
}
entree = state["clips"][state_key(sync, clip, source)]
empreinte = json.dumps(
[entree["network_id"], entree["camera_identity"], entree["created_at"]],
ensure_ascii=False, separators=(",", ":"),
)
entree["correlation_id"] = hashlib.sha256(empreinte.encode("utf-8")).hexdigest()
def _chemin_entree(output: Path, entry: dict, *, index=None,
cle: str | None = None) -> Path | None:
"""Résout un média du registre uniquement à l'intérieur de ``output``."""
brut = entry.get("path")
if not isinstance(brut, str) or not brut.strip():
return None
cache_cle = (cle, os.fspath(output))
if index is not None and cle is not None:
memorise = index.chemins.get(cache_cle)
if memorise is not None and memorise[0] == brut:
return memorise[1]
relatif = Path(brut)
if relatif.is_absolute() or ".." in relatif.parts:
return None
if index is None:
racine = output.resolve()
else:
racine = index.racines.get(os.fspath(output))
if racine is None:
racine = output.resolve()
index.racines[os.fspath(output)] = racine
chemin = (racine / relatif).resolve()
try:
chemin.relative_to(racine)
except ValueError:
return None
if index is not None and cle is not None:
index.chemins[cache_cle] = (brut, chemin)
return chemin
class _ClipConnu:
"""Entrée du registre présentée comme un clip, pour le rapprochement."""
def __init__(self, entree: dict):
self.name = entree.get("camera") or "camera"
self.created_at = dt.datetime.fromisoformat(entree["created_at"])
self.id = entree.get("remote_id") or 0
self.device_id = entree.get("device_id") or ""
self.network_id = entree.get("network_id") or ""
# Provenance et identifiant distant, pour meme_evenement_cloud().
self.source = entree.get("source") or "usb"
self.remote_id = str(entree.get("remote_id") or "")
def _identifiants_cloud_distincts(connu, clip) -> bool:
"""Vrai si l'entrée cloud ``connu`` et le clip cloud ``clip`` portent deux
identifiants distants différents : ce sont alors deux médias, quel que
soit leur écart dans le temps (audit du 26/09/2026, B01). Une entrée USB
garde dans remote_id son numéro de manifeste, que le Sync Module
renumérote : il n'est jamais comparable à un identifiant cloud. Une
vieille entrée sans identifiant reste rapprochée par l'instant seul."""
if getattr(connu, "source", "") != "cloud":
return False
gauche = str(getattr(connu, "remote_id", "") or "")
droite = str(getattr(clip, "id", "") or "")
return bool(gauche and droite and gauche != droite)
def meme_evenement_cloud(connu, clip, sync=None) -> bool:
"""Une entrée connue peut-elle être le même événement que ce clip cloud ?
Même caméra à la tolérance près, sauf deux identifiants cloud différents."""
return (_meme_camera(connu, clip, sync_droite=sync)
and not _identifiants_cloud_distincts(connu, clip))
class _IndexRegistre:
"""Vue temporelle éphémère d'un registre, jamais incluse dans son JSON."""
def __init__(self, state: dict):
self.state = state
self.clips = state.get("clips")
self.taille = len(self.clips) if isinstance(self.clips, dict) else 0
self.valides = []
self.racines = {}
self.chemins = {}
self.fichiers = {}
temporels = []
tombstones = []
for position, (cle, entry) in enumerate((self.clips or {}).items()):
if not isinstance(cle, str) or not isinstance(entry, dict):
continue
if not entry.get("created_at"):
continue
try:
connu = _ClipConnu(entry)
instant = clip_datetime_utc(connu)
except (TypeError, ValueError):
continue
self.valides.append((cle, entry, connu, position))
element = (instant, position, cle, entry, connu)
temporels.append(element)
if entry.get("excluded"):
tombstones.append(element)
self.temporels = sorted(temporels)
self.instants = [element[0] for element in self.temporels]
self.tombstones = sorted(tombstones)
self.instants_tombstones = [element[0] for element in self.tombstones]
_CACHE_INDEX_REGISTRE = None
def _invalider_index_registre(state: dict) -> None:
"""Oublie la vue éphémère avant toute mutation interne du registre."""
global _CACHE_INDEX_REGISTRE
if (
_CACHE_INDEX_REGISTRE is not None
and _CACHE_INDEX_REGISTRE.state is state
):
_CACHE_INDEX_REGISTRE = None
def _index_registre(state: dict) -> _IndexRegistre:
"""Réutilise un unique index hors JSON pour les recherches successives."""
global _CACHE_INDEX_REGISTRE
clips = state.get("clips")
taille = len(clips) if isinstance(clips, dict) else 0
if (
_CACHE_INDEX_REGISTRE is None
or _CACHE_INDEX_REGISTRE.state is not state
or _CACHE_INDEX_REGISTRE.clips is not clips
or _CACHE_INDEX_REGISTRE.taille != taille
):
_CACHE_INDEX_REGISTRE = _IndexRegistre(state)
return _CACHE_INDEX_REGISTRE
def _trouver_entree(state: dict, sync, clip,
consumed: set[str] | None = None,
index: _IndexRegistre | None = None,
source: str = "usb",
) -> tuple[str | None, dict | None]:
"""Trouve au plus une entrée corrélée, avec priorité aux tombstones."""
cle_exacte = state_key(sync, clip, source)
entry = None if cle_exacte in (consumed or ()) else state["clips"].get(cle_exacte)
if isinstance(entry, dict) and entry.get("excluded"):
return cle_exacte, entry
index = index or _index_registre(state)
instant = clip_datetime_utc(clip)
marge = dt.timedelta(seconds=2)
# Avec une clé exacte ordinaire, seule une ancienne tombstone peut avoir
# priorité. Ne pas rescanner les médias ordinaires préserve le chemin rapide.
if isinstance(entry, dict):
temporels = index.tombstones
instants = index.instants_tombstones
else:
temporels = index.temporels
instants = index.instants
debut = bisect.bisect_left(instants, instant - marge)
fin = bisect.bisect_right(instants, instant + marge)
meilleur = None
for indice in range(debut, fin):
instant_connu, position, cle, candidat, connu = temporels[indice]
if cle in (consumed or ()):
continue
ecart = abs((instant_connu - instant).total_seconds())
compatible = (meme_evenement_cloud(connu, clip, sync) if source == "cloud"
else _meme_camera(connu, clip, sync_droite=sync))
if compatible:
classement = (
not bool(candidat.get("excluded")),
cle != cle_exacte,
ecart,
position,
)
if meilleur is None or classement < meilleur[0]:
meilleur = (classement, cle, candidat)
if meilleur is None:
return (cle_exacte, entry) if isinstance(entry, dict) else (None, None)
return meilleur[1], meilleur[2]
def _hub_compatible(entry: dict, sync) -> bool:
"""Faux seulement si les deux hubs sont connus et clairement différents.
_meme_camera() compare des network_id, absents des vieilles entrées USB
(blinkpy 0.25 ne les exposait pas) : sans second signal, le rapprochement
retombe alors sur le seul nom de caméra et confondrait deux hubs
homonymes. Le nom de hub, lui, est renseigné dès le premier
remember_download() et reste disponible sur ces mêmes vieilles entrées."""
hub_connu = str(entry.get("hub") or "").strip().casefold()
hub_courant = str(getattr(sync, "name", "") or "").strip().casefold()
return not hub_connu or not hub_courant or hub_connu == hub_courant
def _apparier_registre(state: dict, sync, clips: list,
index: _IndexRegistre | None = None) -> dict:
"""Associe en lot chaque clip USB à au plus une entrée du registre."""
index = index or _index_registre(state)
correspondances = {}
cles_prises = set()
clips_pris = set()
# Une exclusion est une décision utilisateur absolue, même si une entrée
# ordinaire possède par ailleurs une clé de source exacte pour le clip.
tombstones = [
(cle, entry, connu)
for cle, entry, connu, _ in index.valides
if entry.get("excluded") and _hub_compatible(entry, sync)
]
if tombstones:
paires_exclues = _apparier_evenements(
[element[2] for element in tombstones],
clips,
2,
compatibles=lambda connu, clip: _meme_camera(
connu, clip, sync_droite=sync,
),
)
for indice_entree, indice_clip in paires_exclues:
cle, entry, _ = tombstones[indice_entree]
correspondances[indice_clip] = (cle, entry)
cles_prises.add(cle)
clips_pris.add(indice_clip)
# Pour le reste, une identité de source exacte est plus forte qu'une
# corrélation tolérante.
for indice_clip, clip in enumerate(clips):
if indice_clip in clips_pris:
continue
cle = state_key(sync, clip)
entry = state.get("clips", {}).get(cle)
if (
isinstance(entry, dict)
and not entry.get("excluded")
and cle not in cles_prises
):
correspondances[indice_clip] = (cle, entry)
cles_prises.add(cle)
clips_pris.add(indice_clip)
restants = [indice for indice in range(len(clips)) if indice not in clips_pris]
entrees = [
(cle, entry, connu)
for cle, entry, connu, _ in index.valides
if cle not in cles_prises and not entry.get("excluded")
and _hub_compatible(entry, sync)
]
if not entrees or not restants:
return correspondances
connus = [element[2] for element in entrees]
clips_restants = [clips[indice] for indice in restants]
paires = _apparier_evenements(
connus,
clips_restants,
2,
compatibles=lambda connu, clip: _meme_camera(
connu, clip, sync_droite=sync,
),
)
for indice_entree, indice_restant in paires:
cle, entry, _ = entrees[indice_entree]
correspondances[restants[indice_restant]] = (cle, entry)
return correspondances
def is_downloaded(state: dict, sync, clip, target: Path,
consumed: set[str] | None = None,
index: _IndexRegistre | None = None,
source: str = "usb") -> bool:
"""Un clip est acquis si le registre et le fichier non vide sont présents.
Exception : un clip marqué « exclu » compte comme acquis même sans fichier.
C'est une pierre tombale, posée par `merge_daily.py --exclude`, qui dit
« écarté volontairement, ne pas rapatrier » ; sans elle, supprimer le
fichier ne ferait que provoquer un nouveau téléchargement. Même principe
que le fichier d'archive de yt-dlp (--download-archive, hérité de
youtube-dl) : on retient l'identifiant, pas la présence du média."""
index = index or _index_registre(state)
cle_entree, entry = _trouver_entree(
state, sync, clip, consumed, index, source,
)
if isinstance(entry, dict) and entry.get("excluded"):
acquis = True
elif not isinstance(entry, dict):
acquis = False
else:
chemin = _chemin_entree(
target.parents[2], entry, index=index, cle=cle_entree,
)
empreinte_fichier = None
if chemin is not None:
try:
stat = chemin.stat()
empreinte_fichier = (
os.fspath(chemin), stat.st_size, stat.st_mtime_ns,
entry.get("bytes"),
)
except OSError:
stat = None
else:
stat = None
memorise = index.fichiers.get(cle_entree)
if memorise is not None and memorise[0] == empreinte_fichier:
acquis = memorise[1]
elif stat is None or not stat_module.S_ISREG(stat.st_mode):
acquis = False
else:
taille = stat.st_size
annoncee = entry.get("bytes")
acquis = taille > 0 and (
not isinstance(annoncee, int) or annoncee <= 0 or taille == annoncee
)
if cle_entree is not None:
index.fichiers[cle_entree] = (empreinte_fichier, acquis)
if acquis and consumed is not None:
consumed.add(cle_entree)
return acquis
def _entree_acquise(output: Path, entree: dict) -> bool:
"""Une tombstone reste connue ; un média ordinaire doit exister et être entier."""
if not isinstance(entree, dict):
return False
if entree.get("excluded"):
return True
chemin = _chemin_entree(output, entree)
if chemin is None or not chemin.is_file():
return False
taille = chemin.stat().st_size
annoncee = entree.get("bytes")
return taille > 0 and (
not isinstance(annoncee, int) or annoncee <= 0 or taille == annoncee
)
def _entree_deja_traitee(entree: dict) -> bool:
"""Un clip déjà téléchargé reste connu même après une purge des bruts."""
return isinstance(entree, dict) and (
bool(entree.get("excluded")) or bool(entree.get("path"))
)