"""
organizer_service.py — ingestión de las fotos del fotógrafo (lector de QR), por colegio.

Port de `organizer_service.py` de hec-photo-app. Flujo:

  1. El fotógrafo sube las fotos por FTP a `incoming/{colegio}/`.
  2. El organizer (watchdog + sondeo de seguridad) las analiza en paralelo: QR de la tarjeta de la
     familia (cascada completa en `qr_detect`: puerta doble, CLAHE, Otsu, enfoque, OCR de rescate).
  3. El fotógrafo se deduce del PREFIJO del archivo (`DSC…` → "xavier"). Cada fotógrafo tiene su propio
     "QR pegajoso" persistido en base: una foto de tarjeta lo fija; las que siguen cuelgan de él.
  4. Destino: `photos/{colegio}/{fotógrafo}/{CÓDIGO}/` (la foto de la tarjeta, en `.qr/`).
     Prefijo desconocido o imagen ilegible → `incoming/…/_errors/`, sin perderla.

⚠️ El organizer no debe morir en silencio. Se murió de tres maneras en HEC y cada regla de abajo sale
de ese día (el test `test_organizer.py` las reproduce):
  - `process_pending` va dentro de `try/finally`: `processing` es su primera guarda, y una excepción que
    lo dejaba en True hacía que TODAS las llamadas siguientes salieran de inmediato.
  - Un archivo malo no tumba el lote: el fallo de un `future` se atrapa por archivo.
  - El `while True` vive FUERA del `try` de cada pasada, con backoff exponencial: una caída de la base
    retrasa el trabajo, no lo cancela. Solo `CancelledError` (apagado) atraviesa.
  - `_move` devuelve False sin lanzar y se comprueba: si no, la foto seguía en incoming y el contador
    decía que estaba organizada.
  - El estado es POR COLEGIO: uno solo compartido dejaba `processing=True` en todos a la vez.
  - watchdog llama desde SU hilo: todo entra al event loop por `call_soon_threadsafe`.
  - El análisis (CPU) y la espera de estabilidad (sleep) van en hilos, no bloquean la API.

Diferencias deliberadas con HEC: sin IA anticipada ni detección de ojos cerrados (en este proyecto la IA
la pide la familia, con cuota), y los fotógrafos/estado se guardan en la misma base que el resto, con otro esquema.
"""
import asyncio
import json
import logging
import re
import shutil
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import datetime
from pathlib import Path
from typing import Optional

import cv2
from watchdog.events import FileSystemEventHandler
from watchdog.observers import Observer

from core import config
from core.db import get_db
from core.timeutil import utcnow, utcnow_iso
from services import photo_service, photographer_service, qr_detect, school_service, shoot_service

logger = logging.getLogger(__name__)

CV2_AVAILABLE = True  # opencv es dependencia obligatoria (se conserva el nombre por compatibilidad)

_WORKERS = min(8, (__import__("os").cpu_count() or 4))
IMAGE_EXTENSIONS = config.IMAGE_EXTENSIONS
_IMAGE_EXTS_LOWER = {e.lower() for e in IMAGE_EXTENSIONS}
# Extensiones no-imagen que NO se purgan: la cámara las re-sube cada sesión (índice Nikon NC_FLLST.DAT).
_PURGE_KEEP_EXTS = {".dat"}
_PREFIX_TTL = 300.0
# A photo no scan claims is kept this long in incoming (a scan made a little late can still claim it), then it
# is parked in `_unassigned/` where nobody sees it and the operator is told.
_UNASSIGNED_GRACE_SECS = 600.0
_BACKOFF_MAX = 120.0


# ─── Directorios ──────────────────────────────────────────────────────────────

def incoming_dir_for(school_id: str) -> Path:
    """`incoming/{colegio}/`. La instalación anterior (un solo colegio) subía a la raíz de `incoming/`:
    esa carpeta sigue siendo la de ese colegio para no obligar a cambiar el FTP de la cámara."""
    try:
        with get_db() as conn:
            row = conn.execute("SELECT value FROM app_kv WHERE key='legacy_school_id'").fetchone()
        if row and row["value"] == school_id:
            return config.INCOMING_DIR
    except Exception:
        pass
    return config.INCOMING_DIR / school_id


# ─── Estado por colegio ───────────────────────────────────────────────────────

_status_by_school: dict[str, dict] = {}


def _new_status() -> dict:
    return {
        "processing": False, "pending_count": 0, "last_run_at": None, "last_processed_count": 0,
        "last_elapsed_ms": 0,
        # Salud: sin esto un organizer muerto es indistinguible de uno ocioso.
        "alive": False, "started_at": None, "last_ok_at": None, "last_error": None, "last_error_at": None,
        "consecutive_errors": 0, "failed_files": [], "sticky_by_photographer": {},
    }


def _status_for(school_id: str) -> dict:
    return _status_by_school.setdefault(school_id, _new_status())


def _record_error(school_id: str, message: str, files: Optional[list] = None) -> None:
    st = _status_for(school_id)
    st["last_error"] = message
    st["last_error_at"] = utcnow_iso()
    st["consecutive_errors"] += 1
    if files:
        seen = set(st["failed_files"])
        st["failed_files"] = (st["failed_files"] + [f for f in files if f not in seen])[-20:]
    _publish_status(school_id)


def _publish_status(school_id: str) -> None:
    """Deja el estado en base para que cualquier worker lo pueda mostrar (solo uno es leader)."""
    try:
        payload = dict(_status_for(school_id))
        payload["published_at"] = utcnow_iso()
        with get_db() as conn:
            conn.execute("INSERT OR REPLACE INTO app_kv (key, value) VALUES (?,?)",
                         (f"organizer_status:{school_id}", json.dumps(payload)))
            conn.commit()
    except Exception as exc:
        logger.debug("No se pudo publicar el estado del organizer: %s", exc)


def _read_published(school_id: str) -> Optional[dict]:
    try:
        with get_db() as conn:
            row = conn.execute("SELECT value FROM app_kv WHERE key=?", (f"organizer_status:{school_id}",)).fetchone()
        if not row:
            return None
        data = json.loads(row["value"])
        age = (utcnow() - datetime.fromisoformat(data["published_at"])).total_seconds()
        # Si el leader dejó de publicar, "vivo" ya no es creíble.
        data["alive"] = bool(data.get("alive")) and age < max(180.0, config.ORGANIZER_POLL_SECS * 3)
        return data
    except Exception:
        return None


def get_organizer_status(school_id: str) -> dict:
    """Estado + diagnóstico. `stalled` responde con datos a «¿hay fotos esperando que nadie mueve?»."""
    st = _status_for(school_id)
    if not st["alive"]:                        # lo ejecuta otro worker: leemos lo que publica
        published = _read_published(school_id)
        if published:
            st = published
    out = dict(st)
    out["failed_files"] = list(st.get("failed_files", []))

    pending = _count_pending(incoming_dir_for(school_id))
    out["pending_count"] = pending
    try:
        out["unassigned_count"] = _count_pending(incoming_dir_for(school_id) / "_unassigned")
    except Exception:
        out["unassigned_count"] = 0
    last_move = st.get("last_run_at") or st.get("started_at")
    age = None
    if last_move:
        try:
            age = (utcnow() - datetime.fromisoformat(last_move)).total_seconds()
        except ValueError:
            age = None
    out["seconds_since_last_run"] = int(age) if age is not None else None
    # Atascado = el bucle murió, o hay pendientes, no trabaja ahora y hace rato que no se mueve nada.
    out["stalled"] = bool(
        not out.get("alive")
        or (pending > 0 and not out.get("processing") and age is not None and age > config.ORGANIZER_STALL_SECS)
    )
    return out


def get_status(school_id: Optional[str] = None) -> dict:
    """Alias para el monitor: sin colegio, el primero que haya arrancado."""
    if school_id is None:
        school_id = next(iter(_status_by_school), None)
        if school_id is None:
            return {**_new_status(), "stalled": False, "seconds_since_last_run": None}
    return get_organizer_status(school_id)


def _count_pending(incoming: Path) -> int:
    try:
        return sum(1 for f in incoming.iterdir() if f.is_file() and f.suffix.lower() in _IMAGE_EXTS_LOWER)
    except Exception:
        return 0


# ─── Helpers de proceso ───────────────────────────────────────────────────────

def _filter_stable(files: list) -> list:
    """Difiere el LOTE ENTERO si algún archivo sigue creciendo (subida FTP en curso)."""
    if not files:
        return []
    snapshot: dict[Path, int] = {}
    for f in files:
        try:
            snapshot[f] = f.stat().st_size
        except OSError:
            snapshot[f] = -1
    time.sleep(config.FTP_STABLE_SECS)
    uploading = []
    for f in files:
        try:
            size = f.stat().st_size
            if size != snapshot.get(f) or size == 0:
                uploading.append(f.name)
        except OSError:
            uploading.append(f.name)
    if uploading:
        logger.info("Archivos en transferencia (FTP): %s…", ", ".join(uploading[:3]))
        return []
    return files


def _exif_datetime(path: Path) -> Optional[datetime]:
    try:
        from PIL import Image
        with Image.open(path) as img:
            exif = img.getexif()
            raw = None
            if exif:
                sub = exif.get_ifd(0x8769) if hasattr(exif, "get_ifd") else {}
                raw = sub.get(36867) or sub.get(36868) or exif.get(306)
            if raw:
                return datetime.strptime(raw, "%Y:%m:%d %H:%M:%S")
    except Exception:
        pass
    return None


def _analyze_file(path: Path, prefixes: "set[str] | None" = None) -> tuple:
    """(ruta, texto QR | None, fecha EXIF | None, ilegible). Resolución completa: a 1/4 fallaban los QR
    que ocupan <17 % del encuadre y a 1/2 los <6 %; el volumen (lotes de 1-2 fotos) lo permite."""
    try:
        img = cv2.imread(str(path), cv2.IMREAD_COLOR)
        if img is None:
            return path, None, None, True
        return path, qr_detect.detect_qr(img, path, prefixes), _exif_datetime(path), False
    except Exception as exc:
        logger.error("Error analizando %s: %s", path.name, exc)
        return path, None, None, False


# ─── Core ─────────────────────────────────────────────────────────────────────

class PhotoOrganizer:
    def __init__(self, school_id: str, incoming_dir: Path, output_dir: Path):
        self.school_id = school_id
        self.incoming = incoming_dir
        self.output = output_dir          # photos/{colegio}
        self.error_dir = self.incoming / "_errors"
        self.error_dir.mkdir(parents=True, exist_ok=True)
        self.output.mkdir(parents=True, exist_ok=True)
        self.lock = asyncio.Lock()
        self._prefixes: "set[str] | None" = None
        self._prefixes_at = 0.0

    # ── prefijos del OCR de rescate ──
    def _code_prefixes(self) -> "set[str]":
        now = time.monotonic()
        if self._prefixes is None or now - self._prefixes_at > _PREFIX_TTL:
            configured = None
            try:
                configured = school_service.get_settings(self.school_id).get("qr_code_prefix") or None
            except Exception as exc:
                logger.warning("[%s] No se pudo leer qr_code_prefix: %s", self.school_id, exc)
            self._prefixes = qr_detect.valid_prefixes(configured, self.output)
            self._prefixes_at = now
            logger.info("[%s] Prefijos del OCR de rescate: %s", self.school_id, ", ".join(sorted(self._prefixes)) or "(ninguno)")
        return self._prefixes

    def _purge_unprocessable(self) -> None:
        """Quita de incoming lo que no se puede procesar (RAW .NEF, restos de FTP): si no, se acumula para
        siempre. Conserva .dat y solo toca ficheros 'asentados' (no una subida en curso)."""
        now = time.time()
        try:
            entries = list(self.incoming.iterdir())
        except Exception as exc:
            logger.error("[%s] No se pudo leer incoming para limpieza: %s", self.school_id, exc)
            return
        for f in entries:
            try:
                ext = f.suffix.lower()
                if not f.is_file() or ext in _IMAGE_EXTS_LOWER or ext in _PURGE_KEEP_EXTS or f.name.startswith("."):
                    continue
                if now - f.stat().st_mtime < config.PURGE_MIN_AGE_SECS:
                    continue
                f.unlink()
                logger.info("[%s] No procesable, eliminado de incoming: %s", self.school_id, f.name)
            except FileNotFoundError:
                continue
            except Exception as exc:
                logger.error("[%s] No se pudo eliminar %s: %s", self.school_id, f.name, exc)

    def _analyze_batch(self, candidates: list, prefixes: "set[str]") -> tuple[dict, list]:
        """Análisis en paralelo (en un hilo). Un archivo que revienta se queda fuera sin tumbar al resto."""
        cache: dict[Path, tuple] = {}
        failed: list[str] = []
        with ThreadPoolExecutor(max_workers=_WORKERS) as pool:
            futures = {pool.submit(_analyze_file, f, prefixes): f for f in candidates}
            for fut in as_completed(futures):
                src = futures[fut]
                try:
                    path, qr, dt, unreadable = fut.result()
                    cache[path] = (qr, dt, unreadable)
                except Exception as exc:
                    logger.error("[%s] Fallo analizando %s: %s", self.school_id, src.name, exc, exc_info=True)
                    failed.append(src.name)
        return cache, failed

    async def process_pending(self) -> None:
        """Organiza lo que haya en incoming. Todo el cuerpo va en try/finally (ver docstring del módulo)."""
        st = _status_for(self.school_id)
        if st["processing"]:
            return
        st["processing"] = True
        t0 = time.monotonic()
        failed: list[str] = []
        try:
            # The drop folder is the camera's FTP target: if something removed it, bring it back rather than
            # failing every cycle. (Ownership matters for the FTP user: see the incident note in cleanup_service.)
            try:
                self.incoming.mkdir(parents=True, exist_ok=True)
            except OSError as exc:
                logger.error("[%s] Could not recreate %s: %s", self.school_id, self.incoming, exc)
            await asyncio.to_thread(self._purge_unprocessable)
            try:
                candidates = [f for f in self.incoming.iterdir()
                              if f.is_file() and f.suffix.lower() in _IMAGE_EXTS_LOWER and not f.name.startswith(".")]
            except Exception as exc:
                logger.error("[%s] No se pudo leer incoming: %s", self.school_id, exc)
                _record_error(self.school_id, f"No se pudo leer incoming: {exc}")
                return

            st["pending_count"] = len(candidates)
            if not candidates:
                st["last_ok_at"] = utcnow_iso()
                st["consecutive_errors"] = 0
                st["last_error"] = None
                return

            logger.info("[%s] Detectados %d archivos en %s", self.school_id, len(candidates), self.incoming)
            candidates = await asyncio.to_thread(_filter_stable, candidates)
            if not candidates:
                return

            shoot_on, gap, max_child = await asyncio.to_thread(self._shoot_settings)
            if shoot_on:
                moved, failed = await self._process_shoot(candidates)
                self._finish_pass(st, t0, moved, failed)
                return

            prefixes = await asyncio.to_thread(self._code_prefixes)
            cache, analysis_failed = await asyncio.to_thread(self._analyze_batch, candidates, prefixes)
            failed += analysis_failed
            files = [f for f in candidates if f in cache]

            def _sort_key(f: Path):
                dt = cache[f][1]
                if dt:
                    return dt
                try:
                    return datetime.fromtimestamp(f.stat().st_mtime)
                except OSError:
                    return datetime.max          # sin fecha legible, al final

            files.sort(key=_sort_key)             # orden de disparo: la tarjeta va antes que sus fotos

            moved = 0
            for f in files:
                qr, _dt, unreadable = cache.get(f, (None, None, False))
                try:
                    if await self._process_one(f, qr, unreadable):
                        moved += 1
                    else:
                        failed.append(f.name)      # `_move` falló: la foto sigue en incoming
                except Exception as exc:
                    logger.error("[%s] Fallo organizando %s: %s", self.school_id, f.name, exc, exc_info=True)
                    failed.append(f.name)

            st["last_run_at"] = utcnow_iso()
            st["last_processed_count"] = moved
            st["last_elapsed_ms"] = int((time.monotonic() - t0) * 1000)
            if failed:
                _record_error(self.school_id,
                              f"{len(failed)} archivo(s) no se pudieron organizar: {', '.join(failed[:3])}", files=failed)
            else:
                st["last_ok_at"] = utcnow_iso()
                st["consecutive_errors"] = 0
                st["last_error"] = None
                st["failed_files"] = []
        finally:
            # Pase lo que pase, el organizer vuelve a estar disponible.
            st["processing"] = False
            st["pending_count"] = _count_pending(self.incoming)
            _publish_status(self.school_id)

    # ── Photographer-app mode (see services/shoot_service.py: counts and a fence, never clocks) ──────────

    def _shoot_settings(self) -> tuple[bool, float, int]:
        try:
            s = school_service.get_settings(self.school_id)
            return (s.get("shoot_mode") == "true", 0.0, int(s.get("shoot_max_per_child") or 3))
        except Exception as exc:
            logger.warning("[%s] Could not read shoot settings, using the card-in-photo mode: %s", self.school_id, exc)
            return False, 0.0, 3

    def _finish_pass(self, st: dict, t0: float, moved: int, failed: list) -> None:
        st["last_run_at"] = utcnow_iso()
        st["last_processed_count"] = moved
        st["last_elapsed_ms"] = int((time.monotonic() - t0) * 1000)
        if failed:
            _record_error(self.school_id,
                          f"{len(failed)} archivo(s) no se pudieron organizar: {', '.join(failed[:3])}", files=failed)
        else:
            st["last_ok_at"] = utcnow_iso()
            st["consecutive_errors"] = 0
            st["last_error"] = None
            st["failed_files"] = []

    async def _process_shoot(self, candidates: list, _gap: float = 0.0, _max_child: int = 0) -> tuple[int, list]:
        moved, failed = 0, []
        by_photographer: dict[str, list] = {}
        for f in candidates:
            ph = await asyncio.to_thread(photographer_service.resolve_photographer, self.school_id, f.name)
            if not ph:
                if self._move(f, self.error_dir / f.name):
                    moved += 1
                else:
                    failed.append(f.name)
                continue
            by_photographer.setdefault(ph, []).append(f)
        for ph, files in by_photographer.items():
            async with self.lock:
                try:
                    m, fl = await asyncio.to_thread(self._shoot_photographer, ph, files)
                    moved += m
                    failed += fl
                except Exception as exc:
                    logger.error("[%s] Shoot pass failed for %s: %s", self.school_id, ph, exc, exc_info=True)
                    failed += [f.name for f in files]
        return moved, failed

    def _shoot_photographer(self, ph: str, files: list) -> tuple[int, list]:
        """Turn this photographer's closed classes into children, in order. Blocking: runs in a thread."""
        school = self.school_id
        runs = shoot_service.build_runs(shoot_service.events_for(school, ph))
        state = shoot_service.get_state(school, ph)
        by_name = {f.name: f for f in files}
        moved, failed = 0, []

        for run in runs:
            if not run.closed or run.id <= state["processed"]:
                continue
            outcome, m, fl = self._materialize_run(ph, run, by_name, state)
            moved += m
            failed += fl
            if outcome == "wait":
                break                      # files still uploading / a move failed: retry on the next pass
            state["processed"] = run.id
            shoot_service.set_state(school, ph, state)

        # Strays: files no class claims (taken before the first scan, or between two classes). They are never
        # guessed onto a child; after a short wait they are parked where nobody sees them.
        claimed = shoot_service.claimed_names(runs, state["processed"])
        opened = runs[-1] if runs and not runs[-1].closed else None
        parked = 0
        now = time.time()
        for f in files:
            if f.name in claimed or not f.exists():
                continue
            if opened is not None and f.name not in set(opened.before):
                continue                   # it belongs to the class being shot right now
            try:
                age = now - f.stat().st_mtime
            except OSError:
                continue
            if age >= _UNASSIGNED_GRACE_SECS and self._move(f, self._unique(self.incoming / "_unassigned" / f.name)):
                parked += 1
        if parked:
            from services import alert_service
            alert_service.notify(
                f"shoot-unassigned:{school}", "Photos taken with no class active",
                f"{parked} photo(s) from {ph} were taken while no class was open on the phone and were parked in "
                f"incoming/_unassigned/. Nobody can see them; scan the class QR before shooting.")
        return moved, failed

    def _materialize_run(self, ph: str, run, by_name: dict, state: dict) -> tuple[str, int, list]:
        """("done" | "wait", moved, failed). A run is turned into children only if what was declared on the
        phone equals the files of its fence; otherwise nothing is assigned (see _block_run)."""
        school = self.school_id
        moved, failed = 0, []
        names = run.files
        if names is None:
            return self._block_run(ph, run, [], "the class was never closed with a file list", state)

        progress = shoot_service.get_progress(school, ph, run.id)
        if progress is None:
            missing = [n for n in names if n not in by_name]
            if missing:
                if any((self.incoming / n).exists() for n in missing):
                    return "wait", 0, []   # still uploading
                return self._block_run(ph, run, names, f"{len(missing)} file(s) of the class are gone", state)
            if run.declared != len(names):
                return self._block_run(
                    ph, run, names, f"declared {run.declared} photo(s) but {len(names)} arrived", state)
            ordered = shoot_service.sort_capture_order(
                [by_name[n] for n in names], lambda p: (_exif_datetime(p).timestamp() if _exif_datetime(p) else None))
            progress = {"order": [p.name for p in ordered], "i": 0, "pos": 0, "codes": []}
            shoot_service.set_progress(school, ph, run.id, progress)

        order = progress["order"]
        for idx in range(progress["i"], len(run.items)):
            kind, n = run.items[idx]
            group = order[progress["pos"]:progress["pos"] + n]
            code = None
            if kind == shoot_service.KIND_CHILD:
                # A retry after a failed move must reuse the code it already created, never make a second one.
                code = progress.setdefault("pending", {}).get(str(idx))
                if not code:
                    code = shoot_service.create_child_family(school, run.class_name)
                    progress["pending"][str(idx)] = code
                    shoot_service.set_progress(school, ph, run.id, progress)
                dest_dir = self.output / ph / code
                logger.info("[%s] %s: %s -> %s (%d photo(s))", school, ph, run.class_name, code, len(group))
            else:
                dest_dir = self.incoming / "_ignored"
            ok = True
            for name in group:
                src = self.incoming / name
                if not src.exists():
                    continue                      # already placed by an earlier, interrupted attempt
                if self._move(src, self._unique(dest_dir / name)):
                    moved += 1
                else:
                    failed.append(name)
                    ok = False
            if not ok:
                return "wait", moved, failed
            progress["i"], progress["pos"] = idx + 1, progress["pos"] + n
            if code:
                progress["codes"].append(code)
            shoot_service.set_progress(school, ph, run.id, progress)
            photo_service.invalidate_families_cache(school)
        return "done", moved, failed

    def _block_run(self, ph: str, run, names: list, reason: str, state: dict) -> tuple[str, int, list]:
        """The phone and the folder disagree: assign NOTHING of this class. Its files are parked (not lost) and
        the operator is told. Only this class is affected — the fence keeps the next one independent."""
        school = self.school_id
        moved, failed = 0, []
        for n in names:
            src = self.incoming / n
            if src.exists():
                if self._move(src, self._unique(self.incoming / "_unassigned" / f"class-{run.id}" / n)):
                    moved += 1
                else:
                    failed.append(n)
        if failed:
            return "wait", moved, failed
        state.setdefault("blocked", []).append({
            "run": run.id, "class": run.class_name, "declared": run.declared, "received": len(names), "reason": reason})
        state["blocked"] = state["blocked"][-20:]
        from services import alert_service
        alert_service.notify(
            f"shoot-blocked:{school}:{run.id}", f"Class {run.class_name} was not assigned",
            f"{ph}: {reason}. Its {len(names)} photo(s) were parked in incoming/_unassigned/class-{run.id}/ and "
            f"no family code was created for this class.")
        logger.error("[%s] %s: class %s blocked — %s", school, ph, run.class_name, reason)
        return "done", moved, failed

    async def _process_one(self, file_path: Path, qr_detected: Optional[str], unreadable: bool) -> bool:
        """Coloca una foto. False si se quedó donde estaba (`_move` no lanza: hay que mirar su resultado)."""
        if unreadable:
            self._move(file_path, self.error_dir / file_path.name)
            return True

        photographer = await asyncio.to_thread(photographer_service.resolve_photographer, self.school_id, file_path.name)
        if not photographer:
            self._move(file_path, self.error_dir / file_path.name)
            return True

        async with self.lock:
            card = qr_detect.extract_code(qr_detected) if qr_detected else None
            if qr_detected and not card:
                logger.warning("[%s] QR sin código válido ignorado (%r) en %s", self.school_id, qr_detected[:60], file_path.name)
            if card:
                code = card
                await asyncio.to_thread(photographer_service.set_sticky_qr, self.school_id, photographer, code)
                _status_for(self.school_id)["sticky_by_photographer"][photographer] = code
            else:
                code = await asyncio.to_thread(photographer_service.get_sticky_qr, self.school_id, photographer)

            dest_dir = self.output / photographer / code
            dest_dir.mkdir(parents=True, exist_ok=True)
            if card:
                qr_sub = dest_dir / ".qr"          # la foto de la tarjeta no entra en la galería
                qr_sub.mkdir(parents=True, exist_ok=True)
                dest = self._unique(qr_sub / file_path.name)
                logger.info("[%s] Tarjeta detectada: %s (%s)", self.school_id, code, file_path.name)
            else:
                dest = self._unique(dest_dir / file_path.name)
            if not self._move(file_path, dest):
                return False
            photo_service.invalidate_families_cache(self.school_id)
            logger.info("[%s] %s → %s/%s", self.school_id, file_path.name, photographer, code)
            return True

    @staticmethod
    def _unique(path: Path) -> Path:
        p, n = path, 1
        while p.exists():
            p = path.parent / f"{path.stem}_{n}{path.suffix}"
            n += 1
        return p

    def _move(self, src: Path, dst: Path) -> bool:
        try:
            dst.parent.mkdir(parents=True, exist_ok=True)
            shutil.move(str(src), str(dst))
            return True
        except PermissionError:
            logger.error("[%s] Error de PERMISOS al mover %s (carpeta %s)", self.school_id, src.name, src.parent)
        except Exception as exc:
            logger.error("[%s] Error moviendo %s: %s", self.school_id, src.name, exc)
        return False


# ─── Watchdog ─────────────────────────────────────────────────────────────────

class _Handler(FileSystemEventHandler):
    """Rebota los eventos de disco al event loop. watchdog los llama desde SU hilo, donde `call_later` y
    `create_task` no son thread-safe: todo entra por `call_soon_threadsafe`."""

    def __init__(self, org: PhotoOrganizer, loop: asyncio.AbstractEventLoop):
        self._org = org
        self._loop = loop
        self._timer = None

    def on_modified(self, event):
        if not event.is_directory:
            self._reset_timer()

    def on_created(self, event):
        if not event.is_directory:
            self._reset_timer()

    def _reset_timer(self):
        try:
            self._loop.call_soon_threadsafe(self._schedule)
        except RuntimeError:
            pass                      # el loop se está cerrando: el sondeo ya no hace falta

    def _schedule(self):               # ya en el hilo del event loop
        if self._timer:
            self._timer.cancel()
        self._timer = self._loop.call_later(config.WATCHDOG_DEBOUNCE_SECS, self._fire)

    def _fire(self):
        task = self._loop.create_task(self._org.process_pending())
        task.add_done_callback(self._on_done)

    def _on_done(self, task: asyncio.Task):
        if task.cancelled():
            return
        exc = task.exception()
        if exc:
            logger.error("[%s] Pasada del watchdog falló: %s", self._org.school_id, exc, exc_info=exc)
            _record_error(self._org.school_id, f"Pasada del watchdog falló: {exc}")


async def run_organizer(school_id: str) -> None:
    """Bucle de un colegio. **No termina nunca por un error** (ver docstring del módulo)."""
    st = _status_for(school_id)
    st["alive"] = True
    st["started_at"] = utcnow_iso()
    try:
        incoming = incoming_dir_for(school_id)
        incoming.mkdir(parents=True, exist_ok=True)
        organizer = PhotoOrganizer(school_id, incoming, photo_service.school_root(school_id))
    except Exception as exc:
        logger.error("[%s] No se pudo crear el organizer: %s", school_id, exc, exc_info=True)
        _record_error(school_id, f"No se pudo iniciar el organizer: {exc}")
        st["alive"] = False
        _publish_status(school_id)
        return

    logger.info("[%s] Organizer iniciado en %s", school_id, incoming)
    _publish_status(school_id)

    observer = None
    try:
        observer = Observer()
        observer.schedule(_Handler(organizer, asyncio.get_running_loop()), str(incoming), recursive=False)
        observer.start()
    except Exception as exc:
        # Degradación, no muerte: sin watchdog el sondeo recoge las fotos, con hasta ORGANIZER_POLL_SECS de retraso.
        logger.error("[%s] Watchdog no disponible, solo sondeo cada %ss: %s", school_id, config.ORGANIZER_POLL_SECS, exc)
        _record_error(school_id, f"Watchdog no disponible (sondeo cada {int(config.ORGANIZER_POLL_SECS)}s): {exc}")

    delay = 0.0
    try:
        while True:
            if delay:
                await asyncio.sleep(delay)
            try:
                await organizer.process_pending()
                delay = config.ORGANIZER_POLL_SECS
            except asyncio.CancelledError:
                raise                                # apagado: este sí se propaga
            except Exception as exc:
                errors = _status_for(school_id)["consecutive_errors"] + 1
                _record_error(school_id, f"Ciclo del organizer falló: {exc}")
                delay = min(config.ORGANIZER_RETRY_SECS * (2 ** min(errors - 1, 4)), _BACKOFF_MAX)
                logger.error("[%s] Ciclo del organizer falló (%d seguidos), reintento en %.0fs: %s",
                             school_id, errors, delay, exc, exc_info=True)
    except asyncio.CancelledError:
        logger.info("[%s] Organizer detenido", school_id)
        raise
    finally:
        st["alive"] = False
        _publish_status(school_id)
        if observer is not None:
            try:
                observer.stop()
            except Exception:
                pass


async def run_all_organizers() -> None:
    """Un organizer por colegio activo. Detecta colegios nuevos o desactivados sin reiniciar el servicio."""
    tasks: dict[str, asyncio.Task] = {}
    try:
        while True:
            try:
                active = {s["id"] for s in await asyncio.to_thread(school_service.list_schools, False)}
            except Exception as exc:
                logger.error("Gestor de organizers: no se pudo listar colegios: %s", exc)
                active = set(tasks)
            for sid in sorted(active):
                task = tasks.get(sid)
                if task is None or task.done():
                    if task is not None and not task.cancelled() and task.exception():
                        logger.error("[%s] Organizer terminó con error: %s — se reinicia", sid, task.exception())
                    tasks[sid] = asyncio.create_task(run_organizer(sid), name=f"organizer-{sid}")
            for sid in [s for s in tasks if s not in active]:
                tasks.pop(sid).cancel()
                logger.info("[%s] Colegio desactivado: organizer detenido", sid)
            await asyncio.sleep(config.ORGANIZER_RESCAN_SECS)
    except asyncio.CancelledError:
        for t in tasks.values():
            t.cancel()
        await asyncio.gather(*tasks.values(), return_exceptions=True)
        raise