""" protection_service.py — protección de caras en servidor (port de hec-photo-app). Dos ejes independientes y combinables, como en HEC (ambos apagables por separado): - `protection_heads_enabled`: el archivo que SE SERVIRÁ lleva las caras difuminadas, horneadas en servidor con StoryFace KIS (`POST /api/v2/kis/apply_effect`, 1 crédito por llamada). - `protection_overlay` (`none` | `lens` | `puzzle`): capa animada en el NAVEGADOR sobre lo que se sirva, que estropea las capturas de pantalla. Aquí solo se guarda y valida su configuración; la dibujan los componentes de `frontend/src/components/protection/`. No se precalcula nada al ingerir la foto: proteger por adelantado CADA foto de CADA familia gastaría un crédito aunque esa familia no abra nunca el portal. Se protege bajo demanda, la primera vez que se pide la vista previa, y el resultado queda en disco (`.protect/` junto al original; las visitas siguientes son gratis e instantáneas). ⚠️ Diferencia deliberada con HEC: allí, si KIS falla, esa foto se sirve SIN proteger (fail-open, para no bloquear la venta). Aquí el respaldo es la vista previa reducida, sin texto encima (`media_service`): la familia sigue viendo y comprando la foto, pero nunca un original sin proteger. """ import asyncio import hashlib import logging from pathlib import Path from typing import Optional import httpx from core import config from core.db import get_db from services import ai_health, photo_service from utils.retry import with_retries logger = logging.getLogger(__name__) _sem = asyncio.Semaphore(config.KIS_PARALLEL_MAX) DEFAULT_HEADS_CONFIG = {"intensity": 1.0, "max_heads": 0, "invert": False} DEFAULT_LENS_CONFIG = {"numBalls": 4, "radius": 90, "zoom": 2.5, "speed": 2.0} DEFAULT_PUZZLE_CONFIG = { "shape": "mix", "cut_style": "classic", "num_pieces": 6, "missing": 1, "swaps": 3, "circle_size": 0.90, "mix_ratio": 0.45, "border_width": 10, "animate": True, "seed": 42, "gap": 4, } DEFAULT_PROTECTION = { "protection_heads_enabled": False, "protection_heads": DEFAULT_HEADS_CONFIG, "protection_overlay": "none", "protection_lens": DEFAULT_LENS_CONFIG, "protection_puzzle": DEFAULT_PUZZLE_CONFIG, } # (mínimo, máximo) — los mismos rangos que los deslizadores del panel de HEC _RANGES = { "heads": {"intensity": (0.0, 2.0), "max_heads": (0, 50)}, "lens": {"numBalls": (1, 10), "radius": (30, 160), "zoom": (1.0, 5.0), "speed": (0.5, 5.0)}, "puzzle": {"num_pieces": (4, 12), "missing": (0, 6), "swaps": (0, 6), "circle_size": (0.6, 1.0), "mix_ratio": (0.0, 1.0), "border_width": (0, 28), "gap": (0, 20), "seed": (0, 1_000_000)}, } _ENUMS = { "overlay": {"none", "lens", "puzzle"}, "shape": {"puzzle", "circle", "mix"}, "cut_style": {"classic", "wavy", "sharp", "straight"}, } _STORE_KEY = "portal_protection" # ─── Configuración (validada) ───────────────────────────────────────────────── def _number(section: str, key: str, value, default): lo, hi = _RANGES[section][key] try: num = float(value) except (TypeError, ValueError): raise ValueError(f"{section}.{key}: invalid number") if not lo <= num <= hi: raise ValueError(f"{section}.{key}: debe estar entre {lo} y {hi}") return int(num) if isinstance(default, int) and not isinstance(default, bool) else round(num, 3) def normalize_config(raw: Optional[dict]) -> dict: """Mezcla `raw` con los valores por defecto y valida rangos. ValueError si algo no es válido.""" raw = raw or {} out = { "protection_heads_enabled": bool(raw.get("protection_heads_enabled", False)), "protection_heads": dict(DEFAULT_HEADS_CONFIG), "protection_overlay": raw.get("protection_overlay", "none"), "protection_lens": dict(DEFAULT_LENS_CONFIG), "protection_puzzle": dict(DEFAULT_PUZZLE_CONFIG), } if out["protection_overlay"] not in _ENUMS["overlay"]: raise ValueError("protection_overlay: none, lens o puzzle") for k, v in (raw.get("protection_heads") or {}).items(): if k == "invert": out["protection_heads"]["invert"] = bool(v) elif k in _RANGES["heads"]: out["protection_heads"][k] = _number("heads", k, v, DEFAULT_HEADS_CONFIG[k]) for k, v in (raw.get("protection_lens") or {}).items(): if k in _RANGES["lens"]: out["protection_lens"][k] = _number("lens", k, v, DEFAULT_LENS_CONFIG[k]) for k, v in (raw.get("protection_puzzle") or {}).items(): if k in _ENUMS: if v not in _ENUMS[k]: raise ValueError(f"puzzle.{k}: invalid value") out["protection_puzzle"][k] = v elif k == "animate": out["protection_puzzle"]["animate"] = bool(v) elif k in _RANGES["puzzle"]: out["protection_puzzle"][k] = _number("puzzle", k, v, DEFAULT_PUZZLE_CONFIG[k]) return out def get_protection(school_id: str) -> dict: import json with get_db() as conn: row = conn.execute("SELECT value FROM school_settings WHERE school_id=? AND key=?", (school_id, _STORE_KEY)).fetchone() try: return normalize_config(json.loads(row["value"]) if row and row["value"] else {}) except Exception as exc: # configuración corrupta: nunca bloquear el portal logger.warning("[%s] Configuración de protección ilegible, se usan los valores por defecto: %s", school_id, exc) return normalize_config({}) def set_protection(school_id: str, raw: dict) -> dict: import json cfg = normalize_config(raw) with get_db() as conn: conn.execute("INSERT OR REPLACE INTO school_settings (school_id, key, value) VALUES (?,?,?)", (school_id, _STORE_KEY, json.dumps(cfg))) conn.commit() return cfg def public_view(cfg: dict) -> dict: """Lo que ve el navegador del portal: el overlay y su configuración. Lo de caras es cosa del servidor.""" return { "overlay": cfg["protection_overlay"], "lens": cfg["protection_lens"], "puzzle": cfg["protection_puzzle"], "heads": cfg["protection_heads_enabled"], } # ─── Caras (KIS) ────────────────────────────────────────────────────────────── def config_fingerprint(cfg: dict) -> str: """Huella de los parámetros que cambian el resultado visual (normaliza tipos: `1` y `1.0` son lo mismo).""" key = ( round(float(cfg.get("intensity", DEFAULT_HEADS_CONFIG["intensity"])), 2), int(cfg.get("max_heads", DEFAULT_HEADS_CONFIG["max_heads"])), bool(cfg.get("invert", DEFAULT_HEADS_CONFIG["invert"])), ) return hashlib.sha1(repr(key).encode()).hexdigest()[:10] def protect_paths(photo_path: Path, cfg: dict) -> tuple[Path, Path]: """(carpeta de caché, archivo protegido). El nombre lleva la huella de `cfg` a propósito: sin ella, cambiar los ajustes no tenía efecto, porque se servía el archivo cacheado con el ajuste ANTERIOR.""" protect_dir = photo_path.parent / ".protect" return protect_dir, protect_dir / f"{photo_path.stem}_heads_{config_fingerprint(cfg)}.jpg" def _forget_stale_variants(protect_dir: Path, stem: str, keep: Path) -> None: """Borra las variantes cacheadas de OTRAS configuraciones. Best-effort: nunca propaga.""" try: for stale in protect_dir.glob(f"{stem}_heads_*"): if stale != keep: stale.unlink(missing_ok=True) except Exception as exc: logger.warning("heads: no se pudo limpiar la caché vieja de %s: %s", stem, exc) def kis_form_data(cfg: dict) -> dict: """Campos del formulario de `apply_effect`. ⚠️ El `protect_invert` de KIS está al revés de lo que promete nuestro interruptor: con `false` (su valor por defecto) el efecto se aplica AL FONDO y la cara queda nítida; con `true` se aplica A LA CARA. Nuestro `invert` significa lo contrario (sin marcar: protege las caras), así que se manda invertido. Bug real en producción de HEC: sin esto el interruptor hacía lo opuesto a su etiqueta. """ return { "protect_effect": "blur", "protect_intensity": str(cfg.get("intensity", DEFAULT_HEADS_CONFIG["intensity"])), "protect_max_heads": str(cfg.get("max_heads", DEFAULT_HEADS_CONFIG["max_heads"])), "protect_invert": "false" if cfg.get("invert") else "true", } async def _call_kis(photo_bytes: bytes, filename: str, api_token: str, cfg: dict) -> bytes: async with httpx.AsyncClient(timeout=120.0) as client: response = await client.post( config.KIS_API_URL, files={"image": (filename, photo_bytes, "image/jpeg")}, data=kis_form_data(cfg), headers={"X-API-Key": api_token}, ) if response.status_code != 200: logger.error("KIS apply_effect error %s: %s", response.status_code, response.text[:500]) raise Exception(f"KIS apply_effect failed (HTTP {response.status_code}): {response.text[:200]}") url = response.json().get("url") if not url: raise Exception("KIS apply_effect: respuesta sin url") download = await client.get(url) if download.status_code != 200: raise Exception("KIS apply_effect: error al descargar el resultado") return download.content async def ensure_heads(photo_path: Path, api_token: str, cfg: dict, school_id: str = "") -> Path: """Versión con caras protegidas de una foto, cacheada en disco. Idempotente por foto Y por configuración. Propaga la excepción si falla: el llamante decide qué servir.""" protect_dir, out = protect_paths(photo_path, cfg) if out.exists(): return out async with _sem: if out.exists(): # otra tarea pudo generarla mientras esperábamos return out protect_dir.mkdir(exist_ok=True) photo_bytes = await asyncio.to_thread(photo_path.read_bytes) try: raw = await with_retries(lambda: _call_kis(photo_bytes, photo_path.name, api_token, cfg), label=f"kis-protect {photo_path.name}") except Exception as exc: ai_health.record_failure(school_id, ai_health.TASK_HEADS, exc) # solo el fallo DEFINITIVO raise ai_health.record_success(school_id) tmp = out.with_suffix(".tmp") await asyncio.to_thread(tmp.write_bytes, raw) await asyncio.to_thread(tmp.replace, out) await asyncio.to_thread(_forget_stale_variants, protect_dir, photo_path.stem, out) return out def _resize_webp(src: Path, dst: Path, max_px: int) -> None: from PIL import Image, ImageOps dst.parent.mkdir(parents=True, exist_ok=True) with Image.open(src) as img: img = ImageOps.exif_transpose(img) img.thumbnail((max_px, max_px), Image.LANCZOS) if img.mode not in ("RGB", "RGBA"): img = img.convert("RGB") tmp = dst.with_suffix(".tmp") img.save(tmp, format="WEBP", quality=80, method=4) tmp.replace(dst) async def protected_preview(photo_path: Path, school_id: str, max_px: int, cfg: Optional[dict] = None) -> Optional[Path]: """Vista previa de la foto con las caras protegidas, o None si la protección está apagada / sin clave. Lanza si KIS falla (el llamante cae a la vista previa reducida).""" cfg = cfg or get_protection(school_id) if not cfg["protection_heads_enabled"] or not config.STORYFACE_API_KEY: return None heads_cfg = cfg["protection_heads"] kis_out = await ensure_heads(photo_path, config.STORYFACE_API_KEY, heads_cfg, school_id) fp = config_fingerprint(heads_cfg) dst = photo_service.mirror_path(photo_path, f".h{fp}.{max_px}.webp") if not dst.exists() or dst.stat().st_mtime < kis_out.stat().st_mtime: await asyncio.to_thread(_resize_webp, kis_out, dst, max_px) return dst