"""Altas conservadoras fuera de la ingesta; habilitadas tras preparar la instancia."""
import copy
import fcntl
import os
import queue
import re
import threading
import time
import unicodedata
from contextlib import contextmanager
from datetime import datetime

from dateutil.relativedelta import relativedelta
import pymongo
from pymongo import MongoClient
from pymongo.errors import DuplicateKeyError

from shared_db_utils import insert_log_in_user_history, log_centros as log

SERVICIOS = ("enviro", "nanoox", "scapp", "pss", "flux")
ITG = ("itg", "itg7")
LOCK_PATH = "/var/lib/itgmanager/centros-auto.lock"
PENDIENTE = {"origen": "auto", "estado_alta": "pendiente"}
_cola = queue.Queue(maxsize=500)
_memoria = threading.Lock()
_en_cola = set()
_cache = {}
_hilo = None
_cliente = None
_habilitado_hasta = 0
_habilitado_valor = False
_guardar_permiso = None
_ultimo_aviso_cola = 0


def configurar(guardar_permiso):
    global _guardar_permiso
    _guardar_permiso = guardar_permiso


def auditar(db, slug, tipo):
    """Historial visible en 'Logs Usuarios' para acciones automáticas; su fallo no revierte nada."""
    try:
        insert_log_in_user_history(db.historial_users, "$System", slug, tipo)
    except Exception as error:
        log.warning("autodeteccion.auditoria_fallida centro=%s tipo=%s error=%s", slug, tipo, error)


def campos_reservados(datos):
    return isinstance(datos, dict) and any(
        key.split(".", 1)[0] in {"origen", "estado_alta", "autodeteccion"} for key in datos)


def sin_pendientes(permisos, ciclos, username):
    if username in ITG:
        return permisos
    pendientes = {d["centro_id"] for d in ciclos.find(PENDIENTE, {"centro_id": 1})}
    if not pendientes:
        return permisos
    resultado = copy.deepcopy(permisos)
    excluidos = set()
    for region, areas in list(resultado.items()):
        if not isinstance(areas, dict) or not areas:
            continue
        for area, centros in list(areas.items()):
            if isinstance(centros, dict) and pendientes.intersection(centros):
                excluidos.update(pendientes.intersection(centros))
                for slug in pendientes:
                    centros.pop(slug, None)
                if not centros:
                    del areas[area]
        # Solo vacíos causados por excluir pendientes: no cambia la validación manual existente.
        if not areas:
            del resultado[region]
    if excluidos:
        log.info("permisos.pendientes_excluidos usuario_destino=%s centros=%s",
                 username, ",".join(sorted(excluidos)))
    return resultado


def identidad(servicio, raw):
    if servicio not in ("nanoox", "flux", "scapp") or len(raw) != 3:
        return None
    region, area, centro = raw
    if servicio == "flux":
        region, area = "flux", "barcos"
    partes = (centro, area, region)
    if not all(isinstance(p, str) and re.fullmatch(r"[A-Za-z0-9]{1,60}", p) and
               p.lower().title() == p.lower().capitalize() for p in partes):
        return None
    return "-".join(p.lower() for p in partes)


def plegar(nombre):
    normal = unicodedata.normalize("NFKD", nombre).casefold()
    return re.sub(r"[^a-z0-9]", "", normal)


def conflicto(db, slug):
    nombre = plegar(slug.split("-")[0])
    for doc in db.ciclos.find({}, {"centro_id": 1}, max_time_ms=2000):
        otro = doc.get("centro_id", "")
        if isinstance(otro, str) and otro != slug and plegar(otro.rsplit("-", 2)[0]) == nombre:
            return True
    if db.ciclos.find_one({"centro_id": slug}, {"_id": 1}, max_time_ms=2000):
        return False
    for user in db.users.find({}, {"permisos": 1}, max_time_ms=2000):
        for areas in (user.get("permisos") or {}).values():
            if not isinstance(areas, dict):
                continue
            for centros in areas.values():
                if isinstance(centros, dict) and any(
                        plegar(c.rsplit("-", 2)[0]) == nombre for c in centros):
                    return True
    return False


def hay_datos(client, servicio, id_alm):
    if servicio == "flux":
        return client.FLUX__TS[f"{id_alm.lower()}_flux_ts"].find_one({}, {"_id": 1}, max_time_ms=2000) is not None
    if servicio == "scapp":
        return client.ScappDB.data_analytic.find_one({"centro_id": id_alm}, {"_id": 1}, max_time_ms=2000) is not None
    centro = id_alm.split("-")[-1]
    db = client.NANOOX__TS
    for doc in db[f"{centro}_estructura_logica"].find(
            {"_id": {"$in": ["modulos", "aviots", "cans", "bombas"]}}, {"centro_id": 1}, max_time_ms=2000):
        if doc.get("centro_id") != id_alm:
            return False
    sufijos = ("sensores", "cans", "generador_comap_timeseries")
    for sufijo in ("metadata",) + sufijos:
        if db[f"{centro}_{sufijo}"].find_one({"centro_id": {"$ne": id_alm}}, {"_id": 1}, max_time_ms=2000):
            return False
    # {C}_generador_timeseries (generador_id: None) no prueba ni bloquea: su único escritor,
    # guardar_estructura_en_mongodb, escribe antes la estructura lógica (y cans/sensores, si vienen)
    # con centro_id, así que un homónimo que dejó generadores ya fue detectado arriba.
    return any(db[f"{centro}_{sufijo}"].find_one({"centro_id": id_alm}, {"_id": 1}, max_time_ms=2000)
               is not None for sufijo in sufijos)


@contextmanager
def bloqueo_automaticos():
    # ponytail: un host escritor; varios hosts requieren coordinación antes de habilitar la función.
    if not callable(getattr(pymongo, "timeout", None)):
        raise RuntimeError("La autodetección requiere PyMongo con timeout")
    descriptor = os.open(LOCK_PATH, os.O_CREAT | os.O_RDWR, 0o600)
    with os.fdopen(descriptor, "a") as archivo:
        fcntl.flock(archivo, fcntl.LOCK_EX | fcntl.LOCK_NB)
        try:
            with pymongo.timeout(10):
                yield
        finally:
            fcntl.flock(archivo, fcntl.LOCK_UN)


def habilitado(client):
    global _habilitado_hasta, _habilitado_valor
    if time.monotonic() < _habilitado_hasta:
        return _habilitado_valor
    anterior = _habilitado_valor
    _habilitado_valor = False
    _habilitado_hasta = time.monotonic() + 600
    if not callable(getattr(pymongo, "timeout", None)):
        log.warning("autodeteccion.desactivada motivo=pymongo_sin_timeout version=%s", pymongo.version)
        return False
    db = client.users
    indice = any(x.get("unique") and list(x.get("key", {}).items()) == [("centro_id", 1)]
                 and not x.get("partialFilterExpression") and not x.get("sparse")
                 for x in db.ciclos.list_indexes())
    usuarios = {u["username"] for u in db.users.find(
        {"username": {"$in": list(ITG)}, "cargo": "Superusuario"}, {"username": 1}, max_time_ms=2000)} == set(ITG)
    _habilitado_valor = indice and usuarios
    if not _habilitado_valor:
        log.warning("autodeteccion.desactivada motivo=%s",
                    "falta_indice_unico" if not indice else "faltan_usuarios_itg_superusuario")
    elif not anterior:
        log.info("autodeteccion.habilitada indice=centro_id_unico usuarios=%s", ",".join(ITG))
    return _habilitado_valor


def preparar_instancia(client):
    """Preflight de despliegue: nunca modifica ciclos, permisos ni datos de equipos."""
    global _habilitado_hasta
    if not callable(getattr(pymongo, "timeout", None)):
        raise RuntimeError("La autodetección requiere PyMongo con timeout")
    _habilitado_hasta = 0
    with pymongo.timeout(30):
        if habilitado(client):
            log.info("autodeteccion.preparacion indice=existente")
        else:
            try:
                client.users.ciclos.create_index("centro_id", unique=True)
            except Exception as error:
                # Típicamente centro_id duplicados: no se corrigen datos para forzar el índice.
                log.error("autodeteccion.preparacion_fallida motivo=indice_unico error=%s", error)
                raise
            log.info("autodeteccion.preparacion indice=creado coleccion=users.ciclos clave=centro_id")
            _habilitado_hasta = 0
            if not habilitado(client):
                log.error("autodeteccion.preparacion_fallida motivo=usuarios_itg")
                raise RuntimeError("Instancia sin preparar: comprobar usuarios ITG")
    # No truncar ni reemplazar el archivo: otros procesos pueden seguir bloqueando su inode.
    descriptor = os.open(LOCK_PATH, os.O_CREAT | os.O_RDWR, 0o600)
    os.close(descriptor)
    log.info("autodeteccion.preparacion exclusion=%s", LOCK_PATH)


def ventana(recibido):
    inicio = recibido.strftime("%Y-%m-%d")
    fin = (recibido + relativedelta(months=18)).strftime("%Y-%m-%d")
    return {"init": inicio, "fin": fin,
            "status": datetime.strptime(inicio, "%Y-%m-%d") <= datetime.now() <= datetime.strptime(fin, "%Y-%m-%d")}


def nuevo_ciclo(slug, servicio, recibido):
    doc = {s: {"init": None, "fin": None, "status": False} for s in SERVICIOS}
    doc.update(centro_id=slug, origen="auto", estado_alta="pendiente", cicle_info={"specie": ""},
               gps_coordinates={"active": "auto", "manual": {"lat": None, "lon": None, "timestamp": None},
                                "auto": {"lat": None, "lon": None}},
               autodeteccion={"servicio": servicio, "fecha": recibido, "alta_completa": False})
    doc[servicio] = ventana(recibido)
    return doc


def completar_alta(db, doc, guardar_permiso=None):
    """Devuelve True solo si esta llamada dejó el alta completa."""
    if doc.get("origen") != "auto" or doc.get("autodeteccion", {}).get("alta_completa"):
        return False
    centro, area, region = doc["centro_id"].split("-")
    path = f"permisos.{region}.{area}.{doc['centro_id']}"
    db.estados.update_one({"centro_id": doc["centro_id"]}, {"$setOnInsert": {
        "enviro": "B", "nanoox": "B", "flux": "B"}}, upsert=True)
    (guardar_permiso or _guardar_permiso)(db.users, region, area, centro)
    presentes = {u["username"] for u in db.users.find(
        {"username": {"$in": list(ITG)}, path: {"$exists": True}}, {"username": 1})}
    if presentes != set(ITG):
        log.warning("autodeteccion.alta_incompleta centro=%s motivo=faltan_permisos_itg usuarios=%s",
                    doc["centro_id"], ",".join(sorted(set(ITG) - presentes)))
        raise RuntimeError("Alta incompleta: faltan permisos ITG")
    if not db.estados.find_one({"centro_id": doc["centro_id"]}):
        log.warning("autodeteccion.alta_incompleta centro=%s motivo=falta_estado", doc["centro_id"])
        raise RuntimeError("Alta incompleta: falta estado")
    result = db.ciclos.update_one({"_id": doc["_id"]}, {"$set": {"autodeteccion.alta_completa": True}})
    if not result.matched_count:
        log.warning("autodeteccion.alta_incompleta centro=%s motivo=ciclo_eliminado", doc["centro_id"])
        raise RuntimeError("El ciclo ya no existe")
    if result.modified_count:
        log.info("autodeteccion.alta_completada centro=%s permisos=%s estado=creado_o_existente",
                 doc["centro_id"], ",".join(ITG))
    return bool(result.modified_count)


def abrir_ventana(db, doc, servicio, recibido):
    """Devuelve True solo si abrió la ventana; registra el resultado y la auditoría."""
    bloque = doc.get(servicio)
    if bloque is not None and (not isinstance(bloque, dict) or bloque.get("init") not in (None, "")):
        return False
    filtro = {"_id": doc["_id"], f"{servicio}.init": {"$in": [None, ""]}}
    campos = {f"{servicio}.{key}": val for key, val in ventana(recibido).items()}
    # No reemplazar el bloque: puede contener campos adicionales del cliente.
    if servicio in doc and bloque is None:
        filtro = {"_id": doc["_id"], servicio: {"$type": "null"}}
        campos = {servicio: ventana(recibido)}
    result = db.ciclos.update_one(filtro, {"$set": campos})
    if not result.modified_count:
        log.info("autodeteccion.ventana_sin_cambios centro=%s servicio=%s motivo=editada_en_paralelo",
                 doc["centro_id"], servicio)
        return False
    abierta = ventana(recibido)
    log.info("autodeteccion.ventana_abierta centro=%s servicio=%s init=%s fin=%s origen=%s",
             doc["centro_id"], servicio, abierta["init"], abierta["fin"], doc.get("origen", "manual"))
    auditar(db, doc["centro_id"], "edit_center")
    return True


def procesar(client, servicio, raw, id_alm, recibido):
    slug = identidad(servicio, raw)
    if not slug:
        log.info("autodeteccion.descartado motivo=nombre_no_apto servicio=%s raw=%s (alta manual)",
                 servicio, "/".join(map(str, raw)))
        return
    if not habilitado(client):
        return  # habilitado() registra el motivo
    db = client.users
    doc = db.ciclos.find_one({"centro_id": slug}, max_time_ms=2000)
    incompleto = doc and doc.get("origen") == "auto" and not doc.get("autodeteccion", {}).get("alta_completa")
    bloque = (doc or {}).get(servicio)
    if isinstance(bloque, dict) and bloque.get("init") and not incompleto:
        return  # camino normal: centro con ventana; sin log para no repetirlo cada 10 minutos
    if not incompleto:
        if conflicto(db, slug):
            log.warning("autodeteccion.descartado motivo=conflicto centro=%s servicio=%s (homónimo o permiso previo; alta manual)",
                        slug, servicio)
            return
        if not hay_datos(client, servicio, id_alm):
            log.info("autodeteccion.descartado motivo=sin_datos_de_identidad centro=%s servicio=%s id_almacenamiento=%s",
                     slug, servicio, id_alm)
            return
    if doc and doc.get("origen") != "auto":
        abrir_ventana(db, doc, servicio, recibido)
        return
    with bloqueo_automaticos():
        actual = db.ciclos.find_one({"centro_id": slug}, max_time_ms=2000)
        if doc and (not actual or actual["_id"] != doc["_id"]):
            log.info("autodeteccion.descartado motivo=ciclo_cambiado_en_paralelo centro=%s", slug)
            return
        if actual is None:
            if conflicto(db, slug):
                log.warning("autodeteccion.descartado motivo=conflicto centro=%s servicio=%s (homónimo o permiso previo; alta manual)",
                            slug, servicio)
                return
            try:
                resultado = db.ciclos.update_one(
                    {"centro_id": slug}, {"$setOnInsert": nuevo_ciclo(slug, servicio, recibido)}, upsert=True)
                if resultado.upserted_id is not None:
                    log.info("autodeteccion.alta_creada centro=%s servicio=%s estado=pendiente id_almacenamiento=%s",
                             slug, servicio, id_alm)
                    auditar(db, slug, "add_center")
            except DuplicateKeyError:
                log.info("autodeteccion.alta_concurrente centro=%s (otro escritor creó el ciclo)", slug)
            actual = db.ciclos.find_one({"centro_id": slug})
        if not actual:
            log.warning("autodeteccion.descartado motivo=ciclo_ausente_tras_alta centro=%s", slug)
            return
        completar_alta(db, actual)
        # Un alta incompleta acredita su servicio original, no otro servicio recibido después.
        if not incompleto:
            abrir_ventana(db, actual, servicio, recibido)


def _trabajar():
    global _cliente
    while True:
        clave, servicio, raw, id_alm, recibido = _cola.get()
        try:
            if os.environ.get("AUTODETECCION_CENTROS", "1") == "1":
                if _cliente is None:
                    _cliente = MongoClient("mongodb://localhost:27017/", serverSelectionTimeoutMS=3000,
                                           connectTimeoutMS=3000, socketTimeoutMS=10000)
                procesar(_cliente, servicio, raw, id_alm, recibido)
        except BlockingIOError:
            log.warning("autodeteccion.pospuesto motivo=exclusion_ocupada servicio=%s raw=%s (reintento en 10 min)",
                        servicio, "/".join(map(str, raw)))
        except Exception:
            log.exception("autodeteccion.error servicio=%s raw=%s (reintento en 10 min)",
                          servicio, "/".join(map(str, raw)))
        finally:
            with _memoria:
                _en_cola.discard(clave)
                ahora = time.monotonic()
                for key in list(_cache):
                    if _cache[key] <= ahora:
                        del _cache[key]
                if len(_cache) >= 1000:
                    _cache.pop(next(iter(_cache)))
                _cache[clave] = ahora + 600
            _cola.task_done()


def encolar_dato_recibido(servicio, raw, id_alm, recibido=None):
    global _hilo, _ultimo_aviso_cola
    try:
        if os.environ.get("AUTODETECCION_CENTROS", "1") != "1" or not _memoria.acquire(blocking=False):
            return
        try:
            clave = (servicio, tuple(raw))
            if clave in _en_cola or _cache.get(clave, 0) > time.monotonic():
                return
            if _hilo is None or not _hilo.is_alive():
                _hilo = threading.Thread(target=_trabajar, daemon=True, name="centros-autodetectados")
                _hilo.start()
            _cola.put_nowait((clave, servicio, tuple(raw), id_alm, recibido or datetime.now().astimezone()))
            _en_cola.add(clave)
        finally:
            _memoria.release()
    except Exception as error:
        # Ingesta disponible incluso con cola llena o fallo de arranque; aviso limitado a 1/min.
        try:
            if time.monotonic() - _ultimo_aviso_cola > 60:
                _ultimo_aviso_cola = time.monotonic()
                log.warning("autodeteccion.no_encolado motivo=%s servicio=%s", type(error).__name__, servicio)
        except Exception:
            pass  # El log nunca puede afectar la ingesta.