"""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")) )