"""Nombres de presentación. Nunca modifica IDs, ciclos ni permisos.""" import time import unicodedata import pymongo from flask import Blueprint, g, has_request_context, jsonify, request from flask_jwt_extended import get_jwt_identity, jwt_required from shared_db_utils import get_db, insert_log_in_user_history, log_centros as log etiquetas_bp = Blueprint("etiquetas", __name__) _aviso_sin_timeout = False TIPOS = {"region": 1, "area": 2, "centro": 3} MAPAS = {"region": "regiones", "area": "areas", "centro": "centros"} def clave_etiqueta(tipo, slug): if tipo not in TIPOS or not isinstance(slug, str): return None partes = slug.split("-") if len(partes) != TIPOS[tipo] or not all(p.strip() for p in partes): return None return "-".join("".join(unicodedata.normalize("NFC", p).strip().strip('"').lower().split()) for p in partes) def catalogo_etiquetas(db): entradas = {} for doc in db.ciclos.find({}, {"centro_id": 1}): slug = doc.get("centro_id") if not isinstance(slug, str): continue entradas[("centro", slug)] = True partes = slug.split("-") if len(partes) == 3: _, area, region = partes entradas[("region", region)] = True entradas[("area", f"{region}-{area}")] = True grupos = {} for tipo, slug in entradas: clave = clave_etiqueta(tipo, slug) grupos.setdefault((tipo, clave), set()).add(slug) labels = {doc["_id"]: doc.get("label", "") for doc in db.etiquetas.find({})} return [{"tipo": tipo, "slug": slug, "clave": clave_etiqueta(tipo, slug), "conflicto": clave_etiqueta(tipo, slug) is None or len(grupos[(tipo, clave_etiqueta(tipo, slug))]) > 1, "label": labels.get(f"{tipo}:{clave_etiqueta(tipo, slug)}", "")} for tipo, slug in entradas] def usuario_actual(db): return db.users.find_one({"username": get_jwt_identity()}, {"cargo": 1, "permisos": 1}) or {} def entidades_permitidas(permisos): permitidas = set() if not isinstance(permisos, dict): return permitidas for region, areas in permisos.items(): permitidas.add(("region", region)) if not isinstance(areas, dict): continue for area, centros in areas.items(): permitidas.add(("area", f"{region}-{area}")) if isinstance(centros, dict): permitidas.update(("centro", centro) for centro in centros) return permitidas @etiquetas_bp.route("/etiquetas", methods=["GET"]) @jwt_required() def obtener_etiquetas(): db = get_db().users usuario = usuario_actual(db) permitidas = {(tipo, clave_etiqueta(tipo, slug)) for tipo, slug in entidades_permitidas(usuario.get("permisos"))} resultado = {nombre: {} for nombre in MAPAS.values()} for item in catalogo_etiquetas(db): if (not item["conflicto"] and item["label"] and (usuario.get("cargo") == "Superusuario" or (item["tipo"], item["clave"]) in permitidas)): resultado[MAPAS[item["tipo"]]][item["clave"]] = item["label"] return jsonify(resultado) @etiquetas_bp.route("/etiquetas/catalogo", methods=["GET"]) @jwt_required() def obtener_catalogo(): db = get_db().users if usuario_actual(db).get("cargo") != "Superusuario": log.warning("etiquetas.rechazado motivo=sin_permiso accion=catalogo usuario=%s", get_jwt_identity()) return jsonify(codigo="sin_permiso"), 403 return jsonify(catalogo_etiquetas(db)) @etiquetas_bp.route("/etiquetas", methods=["PUT"]) @jwt_required() def guardar_etiqueta(): db = get_db().users usuario = get_jwt_identity() if usuario_actual(db).get("cargo") != "Superusuario": log.warning("etiquetas.rechazado motivo=sin_permiso accion=guardar usuario=%s", usuario) return jsonify(codigo="sin_permiso"), 403 data = request.get_json(silent=True) if not isinstance(data, dict): log.warning("etiquetas.rechazado motivo=cuerpo_invalido usuario=%s", usuario) return jsonify(codigo="etiqueta_invalida"), 400 tipo, slug, label = data.get("tipo"), data.get("slug"), data.get("label") if not isinstance(tipo, str) or not isinstance(label, str) or clave_etiqueta(tipo, slug) is None: log.warning("etiquetas.rechazado motivo=tipo_o_slug_invalido tipo=%s slug=%s usuario=%s", tipo, slug, usuario) return jsonify(codigo="etiqueta_invalida"), 400 label = label.strip() if len(label) > 60 or any(c in label for c in "\r\n/"): log.warning("etiquetas.rechazado motivo=label_invalido tipo=%s slug=%s usuario=%s", tipo, slug, usuario) return jsonify(codigo="etiqueta_invalida"), 400 item = next((x for x in catalogo_etiquetas(db) if x["tipo"] == tipo and x["slug"] == slug), None) if item is None: log.warning("etiquetas.rechazado motivo=entidad_no_encontrada tipo=%s slug=%s usuario=%s", tipo, slug, usuario) return jsonify(codigo="entidad_no_encontrada"), 404 if item["conflicto"]: log.warning("etiquetas.rechazado motivo=clave_ambigua tipo=%s slug=%s clave=%s usuario=%s", tipo, slug, item["clave"], usuario) return jsonify(codigo="etiqueta_ambigua"), 409 key = f"{tipo}:{item['clave']}" if label: from datetime import datetime db.etiquetas.update_one({"_id": key}, {"$set": { "tipo": tipo, "slug": slug, "label": label, "actualizado": datetime.now(), "actualizado_por": get_jwt_identity(), }}, upsert=True) else: db.etiquetas.delete_one({"_id": key}) log.info("etiquetas.%s tipo=%s slug=%s clave=%s label=%r usuario=%s", "guardada" if label else "eliminada", tipo, slug, item["clave"], label, usuario) try: insert_log_in_user_history(db.historial_users, usuario, slug, "edit_label") except Exception as error: log.warning("etiquetas.auditoria_fallida slug=%s usuario=%s error=%s", slug, usuario, error) return jsonify(ok=True) def etiqueta_centro(client, region, area, centro, respaldo): """Presupuesto compartido de 300 ms por request, incluida selección de servidor y red.""" if not has_request_context(): return respaldo if not callable(getattr(pymongo, "timeout", None)): global _aviso_sin_timeout if not _aviso_sin_timeout: _aviso_sin_timeout = True log.warning("etiquetas.respaldo motivo=pymongo_sin_timeout version=%s (notificaciones y reportes sin labels)", pymongo.version) return respaldo if not hasattr(g, "etiquetas_limite"): g.etiquetas_limite = time.monotonic() + .3 g.etiquetas_resueltas = None if g.etiquetas_resueltas is None: g.etiquetas_resueltas = {} restante = g.etiquetas_limite - time.monotonic() if restante <= 0: log.warning("etiquetas.respaldo motivo=presupuesto_agotado centro=%s-%s-%s", centro, area, region) return respaldo try: with pymongo.timeout(restante): g.etiquetas_resueltas = { item["clave"]: item["label"] for item in catalogo_etiquetas(client.users) if item["tipo"] == "centro" and not item["conflicto"] and item["label"] } except Exception as error: # Una vez por request: las siguientes llamadas reutilizan el mapa vacío y el respaldo. log.warning("etiquetas.respaldo motivo=%s centro=%s-%s-%s", type(error).__name__, centro, area, region) return respaldo return g.etiquetas_resueltas.get(clave_etiqueta("centro", f"{centro}-{area}-{region}"), respaldo)