"""Artifact ingestion and signal extraction helpers for TalkingHeadBench. This module enables user-provided files (reference images, clips, and LoRA weights) to be uploaded once, converted into pre-extracted signal JSON, and reused in OpenEnv episodes through an ingestion ID. """ from __future__ import annotations import copy import json import logging import os import shutil import tempfile import time import urllib.error import urllib.request from pathlib import Path from threading import Lock from typing import Any, Optional from uuid import uuid4 import cv2 import numpy as np from fastapi import UploadFile from src.envs.subenv2.node4_clip_extractor import ( _BLUR_CALIBRATION_CEILING, extract_clip_signals, ) from src.envs.subenv3.node7_weight_extractor import extract_weight_signals from src.schemas.subenv1 import ImageDiagnosticsObservation from src.schemas.subenv2 import ClipSignalObservation from src.schemas.subenv3 import WeightSignalObservation log = logging.getLogger(__name__) _UPLOAD_ROOT = Path(tempfile.gettempdir()) / "talkingheadbench_uploads" _UPLOAD_ROOT.mkdir(parents=True, exist_ok=True) _BUNDLE_STORE: dict[str, dict[str, Any]] = {} _BUNDLE_STORE_LOCK = Lock() def _env_int(name: str, default: int, *, minimum: int) -> int: raw = os.getenv(name) if raw is None: return default try: return max(minimum, int(raw)) except ValueError: return default API_VERSION = "1.0" _UPLOAD_CHUNK_BYTES = 1024 * 1024 _MAX_UPLOAD_BYTES = _env_int("THB_MAX_UPLOAD_BYTES", 512 * 1024 * 1024, minimum=1) _MAX_CLIPS_PER_REQUEST = _env_int("THB_MAX_CLIPS_PER_REQUEST", 12, minimum=1) _MAX_BUNDLES_IN_MEMORY = _env_int("THB_MAX_BUNDLES_IN_MEMORY", 128, minimum=1) _UPLOAD_TTL_SECONDS = _env_int("THB_UPLOAD_TTL_SECONDS", 24 * 60 * 60, minimum=300) _ALLOWED_IMAGE_EXTS = {".png", ".jpg", ".jpeg", ".webp"} _ALLOWED_CLIP_EXTS = {".mp4", ".mov", ".avi", ".mkv", ".webm"} _ALLOWED_WEIGHT_EXTS = {".safetensors"} _ALLOWED_TOKENIZER_EXTS = {".json"} _DEFAULT_PARAM_CONFIG: dict[str, float] = { "cfg": 7.5, "denoise_alt": 0.30, "eta": 0.10, } _FACE_DETECTOR_URL = ( "https://storage.googleapis.com/mediapipe-models/face_detector/" "blaze_face_short_range/float16/latest/blaze_face_short_range.tflite" ) _DEFAULT_FACE_DETECTOR_MODEL_CANDIDATES: tuple[Path, ...] = ( Path(__file__).resolve().parents[1] / "data" / "models" / "face_detector.tflite", Path.home() / ".cache" / "talkingheadbench" / "models" / "face_detector.tflite", ) def _clamp(value: float, low: float, high: float) -> float: return max(low, min(high, value)) def _env_truthy(name: str, *, default: bool) -> bool: raw = os.getenv(name) if raw is None: return default return raw.strip().lower() in {"1", "true", "yes", "on"} def _detect_conflicting_descriptors(prompt: str) -> list[str]: text = prompt.lower() pairs = [ ("young", "old"), ("male", "female"), ("smiling", "serious"), ("frontal", "profile"), ("bright", "dark"), ] conflicts: list[str] = [] for left, right in pairs: if left in text and right in text: conflicts.append(f"{left}|{right}") return conflicts def _candidate_face_detector_model_paths() -> list[Path]: env_path = os.getenv("THB_FACE_DETECTOR_MODEL", "").strip() candidates: list[Path] = [] if env_path: candidates.append(Path(env_path).expanduser()) candidates.extend(_DEFAULT_FACE_DETECTOR_MODEL_CANDIDATES) deduped: list[Path] = [] seen: set[str] = set() for path in candidates: key = str(path) if key in seen: continue seen.add(key) deduped.append(path) return deduped def _download_face_detector_model(dest: Path) -> Path: dest.parent.mkdir(parents=True, exist_ok=True) urllib.request.urlretrieve(_FACE_DETECTOR_URL, dest) return dest def _resolve_face_detector_model_path() -> Path | None: for candidate in _candidate_face_detector_model_paths(): if candidate.exists() and candidate.is_file(): return candidate if not _env_truthy("THB_AUTO_DOWNLOAD_FACE_DETECTOR", default=True): return None cache_target = _DEFAULT_FACE_DETECTOR_MODEL_CANDIDATES[-1] try: return _download_face_detector_model(cache_target) except (OSError, urllib.error.URLError, ValueError) as exc: log.warning( "Unable to auto-download FaceDetector model to %s: %s", cache_target, exc, ) return None def _create_tasks_face_detector() -> Any | None: model_path = _resolve_face_detector_model_path() if model_path is None: return None try: from mediapipe.tasks.python import BaseOptions from mediapipe.tasks.python.vision import ( FaceDetector, FaceDetectorOptions, RunningMode, ) options = FaceDetectorOptions( base_options=BaseOptions(model_asset_path=str(model_path)), running_mode=RunningMode.IMAGE, min_detection_confidence=0.5, ) return FaceDetector.create_from_options(options) except Exception as exc: # noqa: BLE001 log.warning("Failed to initialize Tasks FaceDetector from %s: %s", model_path, exc) return None def _get_haar_face_cascade() -> cv2.CascadeClassifier | None: cascade_path = Path(cv2.data.haarcascades) / "haarcascade_frontalface_default.xml" if not cascade_path.exists(): return None cascade = cv2.CascadeClassifier(str(cascade_path)) if cascade.empty(): return None return cascade def _estimate_face_occupancy_ratio(image_bgr) -> tuple[float, bool]: """Estimate face occupancy as (ratio, is_measured). is_measured=False means no detector backend was available and a synthetic fallback ratio was returned. """ height, width = image_bgr.shape[:2] detector = _create_tasks_face_detector() if detector is not None: try: import mediapipe as mp rgb = cv2.cvtColor(image_bgr, cv2.COLOR_BGR2RGB) mp_image = mp.Image(image_format=mp.ImageFormat.SRGB, data=rgb.copy()) result = detector.detect(mp_image) detections = getattr(result, "detections", []) areas: list[float] = [] for detection in detections: bbox = getattr(detection, "bounding_box", None) if bbox is None: continue box_width = float(getattr(bbox, "width", 0.0)) box_height = float(getattr(bbox, "height", 0.0)) if box_width <= 0.0 or box_height <= 0.0: continue areas.append((box_width * box_height) / float(max(width * height, 1))) if areas: return float(_clamp(max(areas), 0.0, 1.0)), True # Detector ran but no face was detected. return 0.05, True except Exception as exc: # noqa: BLE001 log.warning("Tasks face occupancy detection failed; falling back to Haar: %s", exc) finally: if hasattr(detector, "close"): detector.close() cascade = _get_haar_face_cascade() if cascade is not None: gray = cv2.cvtColor(image_bgr, cv2.COLOR_BGR2GRAY) faces = cascade.detectMultiScale( gray, scaleFactor=1.1, minNeighbors=4, minSize=(max(24, width // 10), max(24, height // 10)), ) if len(faces) > 0: areas = [ (float(face_w) * float(face_h)) / float(max(width * height, 1)) for (_, _, face_w, face_h) in faces ] return float(_clamp(max(areas), 0.0, 1.0)), True # Haar backend was available but no face was detected. return 0.05, True log.warning( "Face occupancy ratio fell back to synthetic default because no detector backend was available." ) return 0.35, False def extract_image_signals(image_path: Path, prompt: str) -> tuple[ImageDiagnosticsObservation, bool]: """Extract a Node 1-compatible observation from a reference image file.""" image_bgr = cv2.imread(str(image_path), cv2.IMREAD_COLOR) if image_bgr is None or image_bgr.size == 0: raise ValueError(f"Failed to read reference image: {image_path}") height, width = image_bgr.shape[:2] gray = cv2.cvtColor(image_bgr, cv2.COLOR_BGR2GRAY) face_occupancy_ratio, face_detection_measured = _estimate_face_occupancy_ratio(image_bgr) lap_var = float(cv2.Laplacian(gray, cv2.CV_64F).var()) sharpness = _clamp((lap_var / float(max(width * height, 1))) / 0.001, 0.0, 1.0) lab = cv2.cvtColor(image_bgr, cv2.COLOR_BGR2LAB) lighting_uniformity = _clamp(1.0 - float(lab[:, :, 0].astype("float32").std() / 80.0), 0.0, 1.0) edges = cv2.Canny(gray, 50, 150) background_complexity = _clamp(float((edges.mean() / 255.0) * 3.0), 0.0, 1.0) tokens = [tok for tok in prompt.split() if tok.strip()] token_count = len(tokens) density = len({tok.lower() for tok in tokens}) / max(1, token_count) conflicts = _detect_conflicting_descriptors(prompt) anchoring = 0.9 if "ohwx" in prompt.lower() else _clamp(0.35 + 0.50 * density, 0.0, 1.0) return ImageDiagnosticsObservation( face_occupancy_ratio=round(face_occupancy_ratio, 4), estimated_yaw_degrees=0.0, estimated_pitch_degrees=0.0, background_complexity_score=round(background_complexity, 4), lighting_uniformity_score=round(lighting_uniformity, 4), skin_tone_bucket=3, occlusion_detected=False, image_resolution=(width, height), estimated_sharpness=round(sharpness, 4), prompt_token_count=token_count, prompt_semantic_density=round(float(density), 4), conflicting_descriptors=conflicts, identity_anchoring_strength=round(float(anchoring), 4), ), face_detection_measured def _parse_param_config(param_config_json: Optional[str]) -> dict[str, float]: config = dict(_DEFAULT_PARAM_CONFIG) if not param_config_json: return config try: payload = json.loads(param_config_json) except json.JSONDecodeError as exc: raise ValueError("param_config_json must be valid JSON") from exc if not isinstance(payload, dict): raise ValueError("param_config_json must decode to an object") for key, value in payload.items(): if isinstance(value, (int, float)): config[key] = float(value) return config def _validate_upload_extension( upload: UploadFile, *, fallback_name: str, allowed_extensions: set[str], field_name: str, ) -> None: suffix = Path(upload.filename or fallback_name).suffix.lower() if suffix not in allowed_extensions: allowed = ", ".join(sorted(allowed_extensions)) raise ValueError(f"{field_name} must use one of the following extensions: {allowed}") def _remove_upload_dir(ingestion_id: str) -> None: upload_dir = _UPLOAD_ROOT / str(ingestion_id) if upload_dir.exists(): shutil.rmtree(upload_dir, ignore_errors=True) def _prune_expired_assets() -> None: now = time.time() expired_ids: list[str] = [] active_ids: set[str] = set() with _BUNDLE_STORE_LOCK: for ingestion_id, bundle in list(_BUNDLE_STORE.items()): active_ids.add(ingestion_id) metadata = bundle.get("ingestion_metadata", {}) created_at = metadata.get("created_at_unix") if isinstance(created_at, (int, float)) and now - float(created_at) > _UPLOAD_TTL_SECONDS: expired_ids.append(ingestion_id) for ingestion_id in expired_ids: _BUNDLE_STORE.pop(ingestion_id, None) active_ids.discard(ingestion_id) for ingestion_id in expired_ids: _remove_upload_dir(ingestion_id) for child in _UPLOAD_ROOT.iterdir(): if not child.is_dir() or child.name in active_ids: continue try: age_seconds = now - child.stat().st_mtime except OSError: continue if age_seconds > _UPLOAD_TTL_SECONDS: shutil.rmtree(child, ignore_errors=True) def _update_phoneme_coverage(coverage: dict[str, int], clip_obs: ClipSignalObservation) -> None: for phoneme in clip_obs.phoneme_sequence: coverage[phoneme] = int(coverage.get(phoneme, 0)) + 1 def _extract_clip_signals_fallback( clip_path: Path, dataset_context: dict[str, Any], ) -> ClipSignalObservation: """Deterministic OpenCV-only fallback when full extractor deps are unavailable.""" cap = cv2.VideoCapture(str(clip_path)) if not cap.isOpened(): raise ValueError(f"OpenCV could not open clip: {clip_path}") frames = [] while True: ok, frame = cap.read() if not ok: break frames.append(frame) if len(frames) >= 180: break cap.release() if len(frames) < 2: raise ValueError(f"Clip has too few frames: {clip_path}") blur_scores: list[float] = [] exposure_scores: list[float] = [] diffs: list[float] = [] gray_frames: list[Any] = [] mouth_brightness_series: list[float] = [] eye_brightness_series: list[float] = [] height, width = frames[0].shape[:2] x1 = max(0, int(width * 0.30)) x2 = min(width, int(width * 0.70)) y1 = max(0, int(height * 0.20)) y2 = min(height, int(height * 0.80)) mouth_y1 = y1 + int((y2 - y1) * 0.66) eye_y1 = y1 eye_y2 = y1 + max(1, int((y2 - y1) * 0.25)) cascade = _get_haar_face_cascade() occlusion_frames = 0 for idx, frame in enumerate(frames): gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY) gray_frames.append(gray) face_roi = gray[y1:y2, x1:x2] if face_roi.size == 0: face_roi = gray lap_var = float(cv2.Laplacian(face_roi, cv2.CV_64F).var()) pixel_count = max(1, face_roi.shape[0] * face_roi.shape[1]) blur_scores.append(_clamp((lap_var / pixel_count) / _BLUR_CALIBRATION_CEILING, 0.0, 1.0)) brightness = float(face_roi.mean() / 255.0) exposure_scores.append(_clamp(1.0 - abs(brightness - 0.5) * 2.0, 0.0, 1.0)) mouth_roi = gray[mouth_y1:y2, x1:x2] eye_roi = gray[eye_y1:eye_y2, x1:x2] mouth_brightness_series.append(float(mouth_roi.mean()) if mouth_roi.size else float(gray.mean())) eye_brightness_series.append(float(eye_roi.mean()) if eye_roi.size else float(gray.mean())) if cascade is not None: faces = cascade.detectMultiScale( gray, scaleFactor=1.1, minNeighbors=4, minSize=(max(24, width // 8), max(24, height // 8)), ) if len(faces) == 0: occlusion_frames += 1 else: hsv = cv2.cvtColor(frame, cv2.COLOR_BGR2HSV) center_hsv = hsv[y1:y2, x1:x2] if center_hsv.size: skin_like = ( (center_hsv[..., 0] >= 0) & (center_hsv[..., 0] <= 25) & (center_hsv[..., 1] >= 30) & (center_hsv[..., 1] <= 200) & (center_hsv[..., 2] >= 60) ) skin_ratio = float(skin_like.mean()) if skin_ratio < 0.06: occlusion_frames += 1 if idx > 0: prev = frames[idx - 1].astype("float32") cur = frame.astype("float32") diffs.append(float(abs(cur - prev).mean())) frame_difference_mean = float(sum(diffs) / max(1, len(diffs))) identity_drift_proxy = _clamp(frame_difference_mean / 45.0, 0.0, 1.0) landmark_jitter_proxy = _clamp(frame_difference_mean / 500.0, 0.0, 1.0) if len(mouth_brightness_series) >= 2: mouth_delta_std = float(np.std(np.diff(np.array(mouth_brightness_series, dtype=np.float32)))) # Keep a computed proxy for diagnostics, but return a conservative fixed # fallback confidence because true A/V lip-sync is not measured here. _ = _clamp(mouth_delta_std / 2.5, 0.0, 1.0) else: _ = 0.0 # Fallback extractor does not perform true audio-visual sync analysis. lip_sync_confidence = 0.35 blink_count = 0 if eye_brightness_series: eye_arr = np.array(eye_brightness_series, dtype=np.float32) baseline = float(np.median(eye_arr)) threshold = baseline * 0.88 in_blink = False for value in eye_arr: if value < threshold: if not in_blink: blink_count += 1 in_blink = True else: in_blink = False optical_flow_magnitude = 1.0 if len(gray_frames) >= 2: face_flows: list[float] = [] bg_flows: list[float] = [] face_mask = np.zeros_like(gray_frames[0], dtype=bool) face_mask[y1:y2, x1:x2] = True for i in range(min(len(gray_frames) - 1, 30)): flow = cv2.calcOpticalFlowFarneback( gray_frames[i], gray_frames[i + 1], None, 0.5, 3, 15, 3, 5, 1.2, 0, ) mag = np.sqrt(flow[..., 0] ** 2 + flow[..., 1] ** 2) face_mean = float(mag[face_mask].mean()) if face_mask.any() else 0.0 bg_mean = float(mag[~face_mask].mean() + 1e-6) face_flows.append(face_mean) bg_flows.append(bg_mean) if face_flows and bg_flows: optical_flow_magnitude = ( float(sum(face_flows) / len(face_flows)) / float(sum(bg_flows) / len(bg_flows)) ) return ClipSignalObservation( clip_id=clip_path.stem, face_embedding_variance=round(identity_drift_proxy, 4), landmark_stability_score=round(landmark_jitter_proxy, 4), identity_cosine_drift=round(identity_drift_proxy, 4), frame_difference_mean=round(frame_difference_mean, 4), optical_flow_magnitude=round(optical_flow_magnitude, 4), blink_count=blink_count, lip_sync_confidence=round(lip_sync_confidence, 4), phoneme_sequence=[], phoneme_coverage_new=0.0, blur_score=round(float(sum(blur_scores) / max(1, len(blur_scores))), 4), exposure_score=round(float(sum(exposure_scores) / max(1, len(exposure_scores))), 4), occlusion_frames=occlusion_frames, clips_audited_so_far=int(dataset_context.get("clips_audited_so_far", 0)), current_phoneme_coverage=dataset_context.get("current_phoneme_coverage", {}), current_pose_distribution=dataset_context.get("current_pose_distribution", {}), similar_clips_accepted=int(dataset_context.get("similar_clips_accepted", 0)), ) def _extract_clip_signal_observations( clip_paths: list[Path], ) -> tuple[list[ClipSignalObservation], int, list[str]]: observations: list[ClipSignalObservation] = [] phoneme_coverage: dict[str, int] = {} fallback_count = 0 fallback_clip_ids: list[str] = [] for idx, clip_path in enumerate(clip_paths): dataset_context = { "clips_audited_so_far": idx, "current_phoneme_coverage": phoneme_coverage, "current_pose_distribution": {}, "similar_clips_accepted": 0, } try: clip_obs = extract_clip_signals( clip_path=clip_path, dataset_context=dataset_context, aligner_output=None, ) fallback_used = False except Exception as exc: # noqa: BLE001 log.warning( "Full clip extractor failed for %s; using fallback extractor: %s", clip_path, exc, ) fallback_count += 1 clip_obs = _extract_clip_signals_fallback(clip_path, dataset_context) fallback_used = True if fallback_used: fallback_clip_ids.append(str(clip_obs.clip_id)) observations.append(clip_obs) _update_phoneme_coverage(phoneme_coverage, clip_obs) return observations, fallback_count, fallback_clip_ids async def _save_upload( upload: UploadFile, output_dir: Path, fallback_name: str, *, max_upload_bytes: int = _MAX_UPLOAD_BYTES, ) -> Path: original_name = Path(upload.filename or fallback_name).name suffix = Path(original_name).suffix.lower() or Path(fallback_name).suffix.lower() or ".bin" destination = output_dir / f"{Path(fallback_name).stem}_{uuid4().hex[:8]}{suffix}" bytes_written = 0 try: with destination.open("wb") as handle: while True: chunk = await upload.read(_UPLOAD_CHUNK_BYTES) if not chunk: break bytes_written += len(chunk) if bytes_written > max_upload_bytes: raise ValueError( f"Uploaded file exceeds the size limit of {max_upload_bytes} bytes" ) handle.write(chunk) except Exception: destination.unlink(missing_ok=True) await upload.close() raise await upload.close() return destination async def ingest_artifacts_to_bundle( *, reference_image: UploadFile | None, clips: list[UploadFile] | None, lora_weights: UploadFile | None, tokenizer_config: UploadFile | None, prompt: str, param_config_json: str, ) -> dict[str, Any]: """Extract a custom signal bundle from uploaded user artifacts.""" clips = clips or [] _prune_expired_assets() if reference_image is None and not clips and lora_weights is None: raise ValueError("At least one artifact is required (image, clips, or lora_weights).") if len(clips) > _MAX_CLIPS_PER_REQUEST: raise ValueError(f"A maximum of {_MAX_CLIPS_PER_REQUEST} clips is allowed per request") if reference_image is not None: _validate_upload_extension( reference_image, fallback_name="reference_image.png", allowed_extensions=_ALLOWED_IMAGE_EXTS, field_name="reference_image", ) for clip_upload in clips: _validate_upload_extension( clip_upload, fallback_name="clip.mp4", allowed_extensions=_ALLOWED_CLIP_EXTS, field_name="clips", ) if lora_weights is not None: _validate_upload_extension( lora_weights, fallback_name="weights.safetensors", allowed_extensions=_ALLOWED_WEIGHT_EXTS, field_name="lora_weights", ) if tokenizer_config is not None: _validate_upload_extension( tokenizer_config, fallback_name="tokenizer_config.json", allowed_extensions=_ALLOWED_TOKENIZER_EXTS, field_name="tokenizer_config", ) ingestion_id = str(uuid4()) created_at = int(time.time()) output_dir = _UPLOAD_ROOT / ingestion_id output_dir.mkdir(parents=True, exist_ok=True) bundle: dict[str, Any] = { "case_id": ingestion_id, "prompt": prompt, "param_config": _parse_param_config(param_config_json), "source_files": {}, "ingestion_metadata": { "api_version": API_VERSION, "created_at_unix": created_at, "limits": { "max_upload_bytes": _MAX_UPLOAD_BYTES, "max_clips_per_request": _MAX_CLIPS_PER_REQUEST, "upload_ttl_seconds": _UPLOAD_TTL_SECONDS, "max_bundles_in_memory": _MAX_BUNDLES_IN_MEMORY, }, "extractor_metadata": { "opencv_version": cv2.__version__, "clip_extractor_fallback_count": 0, }, }, } if reference_image is not None: image_path = await _save_upload(reference_image, output_dir, "reference_image.png") image_obs, face_detection_measured = extract_image_signals(image_path, prompt) bundle["image_observation"] = image_obs.model_dump(mode="json") bundle["source_files"]["reference_image"] = image_path.name bundle["ingestion_metadata"]["extractor_metadata"][ "face_detection_measured" ] = face_detection_measured if clips: clip_paths: list[Path] = [] for idx, clip_upload in enumerate(clips): clip_path = await _save_upload(clip_upload, output_dir, f"clip_{idx:03d}.mp4") clip_paths.append(clip_path) clip_obs_list, fallback_count, fallback_clip_ids = _extract_clip_signal_observations(clip_paths) fallback_set = set(fallback_clip_ids) bundle["clip_signal_observations"] = [] for clip_obs in clip_obs_list: clip_payload = clip_obs.model_dump(mode="json") fallback_used = str(clip_obs.clip_id) in fallback_set clip_payload["fallback_extractor_used"] = fallback_used clip_payload["lip_sync_reliable"] = not fallback_used bundle["clip_signal_observations"].append(clip_payload) bundle["source_files"]["clips"] = [p.name for p in clip_paths] bundle["ingestion_metadata"]["extractor_metadata"][ "clip_extractor_fallback_count" ] = fallback_count bundle["ingestion_metadata"]["extractor_metadata"][ "fallback_clip_ids" ] = fallback_clip_ids if lora_weights is not None: weight_path = await _save_upload(lora_weights, output_dir, "weights.safetensors") tokenizer_path: Path | None = None if tokenizer_config is not None: tokenizer_path = await _save_upload(tokenizer_config, output_dir, "tokenizer_config.json") weight_obs: WeightSignalObservation = extract_weight_signals( weight_path=weight_path, tokenizer_config_path=tokenizer_path, ) bundle["weight_observation"] = weight_obs.model_dump(mode="json") bundle["source_files"]["lora_weights"] = weight_path.name if tokenizer_path is not None: bundle["source_files"]["tokenizer_config"] = tokenizer_path.name return bundle def store_ingested_bundle(bundle: dict[str, Any]) -> str: _prune_expired_assets() ingestion_id = str(bundle.get("case_id") or uuid4()) saved = copy.deepcopy(bundle) saved["case_id"] = ingestion_id evicted_ids: list[str] = [] with _BUNDLE_STORE_LOCK: _BUNDLE_STORE[ingestion_id] = saved while len(_BUNDLE_STORE) > _MAX_BUNDLES_IN_MEMORY: oldest_id = next(iter(_BUNDLE_STORE.keys())) _BUNDLE_STORE.pop(oldest_id, None) evicted_ids.append(oldest_id) for evicted_id in evicted_ids: _remove_upload_dir(evicted_id) return ingestion_id def get_ingested_bundle(ingestion_id: str) -> dict[str, Any] | None: _prune_expired_assets() with _BUNDLE_STORE_LOCK: bundle = _BUNDLE_STORE.get(str(ingestion_id)) if bundle is None: return None return copy.deepcopy(bundle) def list_ingested_bundle_ids() -> list[str]: _prune_expired_assets() with _BUNDLE_STORE_LOCK: return sorted(_BUNDLE_STORE.keys()) def delete_ingested_bundle(ingestion_id: str) -> bool: with _BUNDLE_STORE_LOCK: removed = _BUNDLE_STORE.pop(str(ingestion_id), None) if removed is not None: _remove_upload_dir(str(ingestion_id)) return removed is not None