""" school_service.py — multi-colegio (multitenant). Un « colegio » (school) agrupa familias, alumnos, pedidos, precios y configuración. Todo lo que antes estaba fijo en el código (nombre, fechas, precios, retención RGPD, reparto, fotógrafo…) vive ahora en `school_settings` y se edita desde el panel de admin. - `schools.id` es el slug (inmutable, ej. "sainte-marie"): se envía en la cabecera X-School y se guarda en families.school_id / orders.school_id. - El código de familia sigue siendo único globalmente (PK de families), así que el QR de la tarjeta identifica de forma inequívoca familia → colegio. - Contraseña de director: una por colegio (hash scrypt). El operador Meeconic (ADMIN_PASSWORD) administra todos los colegios. """ import logging import re import unicodedata from datetime import date from typing import Any, Optional from core import config from core.db import get_db, TABLE_OPTIONS from core.security import hash_password from core.timeutil import utcnow_iso log = logging.getLogger("meeconic.schools") class SettingsError(ValueError): pass # ── Especificación de los ajustes editables ───────────────────────────────────── # tipo: str | date | bool | int | float | email SETTINGS_SPEC: dict[str, dict[str, Any]] = { "school_name": {"type": "str", "default": "", "max": 120}, "shoot_date": {"type": "date", "default": ""}, "closure_date": {"type": "date", "default": ""}, "delivery_week": {"type": "date", "default": ""}, "revenue_share_pct": {"type": "float", "default": "0", "min": 0, "max": 100}, "partner_name": {"type": "str", "default": "", "max": 120}, "portal_url": {"type": "str", "default": "", "max": 200}, "online_payment": {"type": "bool", "default": "true"}, "onsite_payment": {"type": "bool", "default": "false"}, "postal_delivery": {"type": "bool", "default": "false"}, "school_delivery": {"type": "bool", "default": "true"}, "rgpd_days": {"type": "int", "default": str(config.PHOTO_RETENTION_DAYS), "min": 1, "max": 365}, "photographer_name": {"type": "str", "default": "", "max": 120}, "photographer_id": {"type": "str", "default": "", "max": 40}, "ftp_url": {"type": "str", "default": "", "max": 200}, "notification_email": {"type": "email", "default": ""}, # Prefijo de los códigos de familia de este colegio (MAR → MAR-4JC-IBD). Vacío = 3 primeras letras del nombre. "qr_code_prefix": {"type": "prefix", "default": ""}, # Accent of the family card header. Must stay dark enough for the white text on it (contrast >= 4.5). "brand_color": {"type": "color", "default": "#0A1F44"}, # Photographer app (/shoot): the class QR is scanned on the phone and each child is declared with a tap, # instead of a QR card photographed before each child. Off keeps the card-in-photo behaviour of HEC. "shoot_mode": {"type": "bool", "default": "false"}, "shoot_max_per_child": {"type": "int", "default": "3", "min": 1, "max": 10}, # TEST ONLY: lets the phone take photos itself and drop them in incoming/ as the FTP camera would. Off in # production; the real flow is camera -> FTP, the phone only counts. "shoot_test_camera": {"type": "bool", "default": "false"}, "ai_enabled": {"type": "bool", "default": "true"}, "ai_max_per_family": {"type": "int", "default": "5", "min": 0, "max": 50}, } for _k, _v in config.DEFAULT_PRICES.items(): SETTINGS_SPEC[_k] = {"type": "float", "default": str(_v).rstrip("0").rstrip(".") if "." in str(_v) else str(_v), "min": 0, "max": 1000} PRICE_KEYS = list(config.DEFAULT_PRICES.keys()) _EMAIL_RE = re.compile(r"^[^@\s]+@[^@\s]+\.[^@\s]+$") def contrast_with_white(hex_color: str) -> float: """WCAG contrast ratio between `#RRGGBB` and white.""" h = hex_color.lstrip("#") lin = [(c / 255 / 12.92) if c / 255 <= 0.03928 else (((c / 255) + 0.055) / 1.055) ** 2.4 for c in (int(h[i:i + 2], 16) for i in (0, 2, 4))] lum = 0.2126 * lin[0] + 0.7152 * lin[1] + 0.0722 * lin[2] return 1.05 / (lum + 0.05) def _normalize(key: str, raw: Any) -> str: spec = SETTINGS_SPEC[key] t = spec["type"] s = "" if raw is None else str(raw).strip() if t == "bool": if isinstance(raw, bool): return "true" if raw else "false" if s.lower() in ("true", "1", "yes", "on"): return "true" if s.lower() in ("false", "0", "no", "off", ""): return "false" raise SettingsError(f"{key}: invalid boolean value.") if t == "date": if s == "": return "" try: return date.fromisoformat(s).isoformat() except ValueError: raise SettingsError(f"{key}: invalid date (format YYYY-MM-DD).") if t in ("int", "float"): try: num = int(s) if t == "int" else float(s.replace(",", ".")) except ValueError: raise SettingsError(f"{key}: invalid number.") if num < spec["min"] or num > spec["max"]: raise SettingsError(f"{key}: must be between {spec['min']} and {spec['max']}.") return str(num) if t == "int" else (str(int(num)) if num == int(num) else str(num)) if t == "prefix": s = s.upper() if s and not re.match(r"^[A-Z]{2,6}$", s): raise SettingsError(f"{key}: 2 to 6 letters (A-Z).") return s if t == "color": if not re.match(r"^#[0-9a-fA-F]{6}$", s): raise SettingsError(f"{key}: use a #RRGGBB colour.") if contrast_with_white(s) < 4.5: raise SettingsError(f"{key}: colour too light — white text on it would be unreadable.") return s.upper() if t == "email": if s and not _EMAIL_RE.match(s): raise SettingsError(f"{key}: invalid e-mail address.") return s # str if len(s) > spec.get("max", 200): raise SettingsError(f"{key}: too long (max {spec.get('max', 200)}).") return s # ── Esquema y migración ───────────────────────────────────────────────────────── def init_school_tables() -> None: with get_db() as conn: conn.execute(f""" CREATE TABLE IF NOT EXISTS schools ( id VARCHAR(64) NOT NULL PRIMARY KEY, name VARCHAR(200) NOT NULL, director_password_hash VARCHAR(255), active TINYINT NOT NULL DEFAULT 1, created_at VARCHAR(40) NOT NULL ) {TABLE_OPTIONS}""") conn.execute(f""" CREATE TABLE IF NOT EXISTS school_settings ( school_id VARCHAR(64) NOT NULL, `key` VARCHAR(64) NOT NULL, value MEDIUMTEXT, PRIMARY KEY (school_id, `key`) ) {TABLE_OPTIONS}""") conn.commit() # ── CRUD colegios ──────────────────────────────────────────────────────────────── def slugify(name: str) -> str: base = "".join(c for c in unicodedata.normalize("NFD", name) if unicodedata.category(c) != "Mn") base = re.sub(r"[^a-zA-Z0-9]+", "-", base).strip("-").lower() return (base or "ecole")[:40] def _unique_slug(conn, wanted: str) -> str: slug, i = wanted, 2 while conn.execute("SELECT 1 FROM schools WHERE id=?", (slug,)).fetchone(): slug = f"{wanted}-{i}" i += 1 return slug def create_school(name: str, slug: Optional[str] = None, director_password: Optional[str] = None, initial_settings: Optional[dict] = None) -> dict: name = (name or "").strip() if not name: raise SettingsError("The school name is required.") if director_password is not None and director_password != "" and len(director_password) < 8: raise SettingsError("The head teacher password must be at least 8 characters.") with get_db() as conn: if slug: slug = slugify(slug) if conn.execute("SELECT 1 FROM schools WHERE id=?", (slug,)).fetchone(): raise SettingsError("This school code already exists.") else: slug = _unique_slug(conn, slugify(name)) conn.execute( "INSERT INTO schools (id, name, director_password_hash, active, created_at) VALUES (?,?,?,1,?)", (slug, name, hash_password(director_password) if director_password else None, utcnow_iso()), ) values = {k: spec["default"] for k, spec in SETTINGS_SPEC.items()} values["school_name"] = name for k, v in (initial_settings or {}).items(): if k in SETTINGS_SPEC: try: values[k] = _normalize(k, v) except SettingsError: pass for k, v in values.items(): conn.execute("INSERT INTO school_settings (school_id, key, value) VALUES (?,?,?)", (slug, k, v)) conn.commit() log.info("School created: %s (%s)", name, slug) return get_school(slug) # type: ignore[return-value] def _row(r) -> dict: return { "id": r["id"], "name": r["name"], "active": bool(r["active"]), "has_director_password": bool(r["director_password_hash"]), "created_at": r["created_at"], } def get_school(school_id: str) -> Optional[dict]: with get_db() as conn: r = conn.execute("SELECT * FROM schools WHERE id=?", (school_id,)).fetchone() return _row(r) if r else None def get_school_with_hash(school_id: str) -> Optional[dict]: with get_db() as conn: r = conn.execute("SELECT * FROM schools WHERE id=?", (school_id,)).fetchone() return dict(r) if r else None def list_schools(include_inactive: bool = True) -> list[dict]: with get_db() as conn: q = "SELECT * FROM schools" + ("" if include_inactive else " WHERE active=1") + " ORDER BY name COLLATE NOCASE" return [_row(r) for r in conn.execute(q).fetchall()] def update_school(school_id: str, *, name: Optional[str] = None, active: Optional[bool] = None, director_password: Optional[str] = None) -> dict: if not get_school(school_id): raise SettingsError("School not found.") with get_db() as conn: if name is not None: name = name.strip() if not name: raise SettingsError("The school name is required.") conn.execute("UPDATE schools SET name=? WHERE id=?", (name, school_id)) conn.execute("INSERT OR REPLACE INTO school_settings (school_id, key, value) VALUES (?,?,?)", (school_id, "school_name", name)) if active is not None: conn.execute("UPDATE schools SET active=? WHERE id=?", (1 if active else 0, school_id)) if director_password is not None: if director_password == "": conn.execute("UPDATE schools SET director_password_hash=NULL WHERE id=?", (school_id,)) else: if len(director_password) < 8: raise SettingsError("The head teacher password must be at least 8 characters.") conn.execute("UPDATE schools SET director_password_hash=? WHERE id=?", (hash_password(director_password), school_id)) conn.commit() return get_school(school_id) # type: ignore[return-value] # ── Ajustes ────────────────────────────────────────────────────────────────────── def get_settings(school_id: str, conn=None) -> dict[str, str]: """Ajustes del colegio fusionados con los valores por defecto.""" out = {k: spec["default"] for k, spec in SETTINGS_SPEC.items()} def _load(c): for r in c.execute("SELECT key, value FROM school_settings WHERE school_id=?", (school_id,)): if r["key"] in SETTINGS_SPEC and r["value"] is not None: out[r["key"]] = r["value"] if conn is not None: _load(conn) else: with get_db() as c: _load(c) return out def update_settings(school_id: str, values: dict) -> dict[str, str]: """Valida y guarda. Claves desconocidas → error (nada se guarda a medias).""" if not get_school(school_id): raise SettingsError("School not found.") clean: dict[str, str] = {} for k, v in values.items(): if k not in SETTINGS_SPEC: raise SettingsError(f"Unknown setting: {k}") clean[k] = _normalize(k, v) if "school_name" in clean and not clean["school_name"]: raise SettingsError("The school name is required.") with get_db() as conn: for k, v in clean.items(): conn.execute("INSERT OR REPLACE INTO school_settings (school_id, key, value) VALUES (?,?,?)", (school_id, k, v)) if "school_name" in clean: conn.execute("UPDATE schools SET name=? WHERE id=?", (clean["school_name"], school_id)) conn.commit() return get_settings(school_id) def pricing_from_settings(settings: dict) -> dict[str, float]: """Mismas claves que el antiguo PRICING (print_10x15, digital, magic_pack…).""" return {k[len("price_"):]: float(settings[k]) for k in PRICE_KEYS} # Identificadores de formato enviados por el frontend → clave de precio. FORMAT_PRICE_KEY = { "print-10x15": "price_print_10x15", "print-20x30": "price_print_20x30", "digital": "price_digital", "pack": "price_pack", "magic-print": "price_magic_print", "magic-digital": "price_magic_digital", "magic-pack": "price_magic_pack", } # Formatos que dan derecho a descargar el archivo original. DIGITAL_FORMATS = {"digital", "pack", "magic-digital", "magic-pack"} MAGIC_FORMATS = {"magic-print", "magic-digital", "magic-pack"} def school_id_for_family(code: str) -> Optional[str]: with get_db() as conn: r = conn.execute("SELECT school_id FROM families WHERE code=?", (code,)).fetchone() return r["school_id"] if r else None def retention_days_for_family(code: str) -> int: sid = school_id_for_family(code) if sid: try: return int(get_settings(sid)["rgpd_days"]) except Exception: pass return config.PHOTO_RETENTION_DAYS def code_prefix_for(school_id: str) -> str: """Prefijo de los códigos de familia: el configurado, o las 3 primeras letras del nombre.""" from utils.code_generator import prefix_from_name school = get_school(school_id) configured = get_settings(school_id).get("qr_code_prefix", "") return configured or prefix_from_name(school["name"] if school else "")