#!/usr/bin/env python3
"""Guarda de escrituras SQL para `scripts/consultar-pruebas.sh` (#24603, a raíz de #24601).

POR QUÉ EXISTE. El 28-sep-2026 un UPDATE sin WHERE pisó `dic_campo.descripcion` (1,995 de
1,995 filas) y `dic_tabla.descripcion` (65 de 65) en la base de pruebas. El freno que había
era un `grep -q ' WHERE '` sobre el texto entero, y se midió el 01-oct-2026 que deja pasar
un UPDATE que NO tiene WHERE real cuando la palabra aparece:

    - en un comentario        (`-- WHERE id=1`, `# WHERE id=1`)
    - en una subconsulta      (`SET x = (SELECT 1 FROM t WHERE ...)`)
    - dentro de un texto      (`SET d = 'aplica WHERE se use'`)
    - o es tautología         (`WHERE 1=1`)

Un grep no puede distinguir eso: hay que leer el SQL como lo lee el motor. Este archivo lo
parte en piezas (textos, comentarios, paréntesis) y sólo cuenta el WHERE que está en el
nivel de la sentencia.

QUÉ HACE, EN ORDEN (cada paso sólo corre si el anterior dejó pasar):

  1. Parte el texto en sentencias (`;` fuera de textos/comentarios/paréntesis).
  2. Bloquea lo que no se deshace: DROP DATABASE/SCHEMA, TRUNCATE, un UPDATE/DELETE sin
     WHERE real, un WHERE tautológico, y lo que no se puede analizar (PREPARE de texto
     armado, comandos del cliente como `\\!`, WITH ... UPDATE).
  3. Para cada UPDATE/DELETE de una sola tabla, cuenta cuántas filas toca (COUNT real
     contra el servidor) y cuántas tiene la tabla. Frena si toca más del umbral
     (MB_UMBRAL_FILAS, 100 por defecto) o si toca TODAS (con WHERE o sin él: dic_tabla
     tiene 65 filas, así que un umbral de 100 solo no la habría salvado).
  4. Antes de escribir, guarda las filas que se van a tocar en un archivo
     (`~/.config/mb/snapshots/`), para poder deshacer sin depender de PITR (que el
     28-sep falló 4 veces por un error de OCI).
  5. Saltarse un freno exige `--se-lo-que-hago <tiquete>` y queda una línea en
     `~/.config/mb/excepciones-sql.log`.

⚠️ ALCANCE. Es un freno contra el DESCUIDO (un UPDATE sin WHERE, un WHERE que no es el que se
cree, una tabla entera), no una barrera contra alguien que quiera esquivarlo a propósito: con
tiempo se le encuentran más formas de escribir SQL que lo engañen. Por eso lo que no se puede
analizar (CALL, PREPARE armado, EXPLAIN ANALYZE, comandos del cliente, sesiones que cambian de
estado) se frena en vez de adivinarse, y por eso el cierre de fondo es quitar la escritura de
las cuentas personales (#24603, decisión 2).

⚠️ LÍMITE CONOCIDO: el conteo, el respaldo y la escritura corren en sesiones distintas. Si otra
persona cambia las filas que calzan con el WHERE entre un paso y otro, la escritura puede tocar
filas que no se contaron ni se respaldaron. Cerrarlo de verdad exige correr todo en UNA transacción
con las filas bloqueadas, o sea rehacer la herramienta; no se hizo porque la base de pruebas la usa
poca gente a la vez y el freno es contra el descuido, no contra la concurrencia. Lo mismo vale
para un WHERE con reloj (NOW(), CURRENT_TIMESTAMP): entre el conteo y la escritura pasan segundos;
frenarlo impediría toda limpieza por antigüedad (`WHERE fecha < NOW() - INTERVAL 7 DAY`).

Uso (lo llama el guion; también sirve suelto para las pruebas):

    sql_guarda.py analizar "<sql>"             sólo lee el texto, no toca ninguna base
    sql_guarda.py previo [--tiquete N] "<sql>" analizar + contar + snapshot + bitácora

Códigos de salida: 0 deja pasar, 2 frena. Nunca escribe en la base.
"""

import datetime
import hashlib
import json
import os
import re
import subprocess
import sys
from urllib.parse import quote

UMBRAL = int(os.environ.get("MB_UMBRAL_FILAS", "100"))
CARPETA = os.path.expanduser(os.environ.get("MB_CARPETA_CONFIG", "~/.config/mb"))

# ───────────────────────────── 1. Partir el texto en piezas ─────────────────────────────

class Pieza:
    __slots__ = ("tipo", "texto", "nivel", "src", "ini", "fin", "cmt")

    def __init__(self, tipo, texto, nivel, src=None, ini=0, fin=0, cmt=None):
        self.tipo, self.texto, self.nivel = tipo, texto, nivel
        self.src, self.ini, self.fin = src, ini, fin   # dónde está en el SQL original
        self.cmt = cmt   # (ini, fin) del /*! ... */ ejecutable que la contiene, o None

    def __repr__(self):
        return "{}:{}@{}".format(self.tipo, self.texto, self.nivel)


def piezas(sql):
    """Devuelve (lista de Pieza, error|None). Los comentarios no aparecen: se descartan,
    salvo `/*! ... */`, que el servidor SÍ ejecuta, así que su contenido cuenta como código."""
    out, i, n, nivel = [], 0, len(sql), 0
    while i < n:
        c = sql[i]
        if c.isspace():
            i += 1
        elif c == "#" or (c == "-" and sql.startswith("--", i) and (i + 2 >= n or ord(sql[i + 2]) <= 32 or ord(sql[i + 2]) == 127)):
            while i < n and sql[i] != "\n":
                i += 1
        elif sql.startswith("/*", i):
            fin = sql.find("*/", i + 2)
            if fin < 0:
                return out, "comentario /* sin cerrar"
            if sql.startswith("/*!", i):
                # /*!50000 código */ : se ejecuta. Se re-analiza el contenido, sin el número.
                version = re.match(r"\d{0,6}", sql[i + 3:fin]).group()
                if version and int(version) > 80000:
                    # El servidor sólo ejecuta el cuerpo si su versión es >= ésta, y acá no se sabe cuál es.
                    return out, "comentario /*!{} condicionado por la versión del servidor: no se sabe si se ejecuta".format(version)
                dentro = re.sub(r"^\d{0,6}", "", sql[i + 3:fin])
                sub, err = piezas(dentro)
                if err:
                    return out, err
                prof = 0
                for p in sub:   # un ( que se abre adentro y se cierra afuera (o al revés) cambia a quién pertenece cada WHERE
                    if p.tipo == "sym" and p.texto in "()":
                        prof += 1 if p.texto == "(" else -1
                        if prof < 0:
                            break
                if prof != 0:
                    return out, "paréntesis que cruza el borde de un comentario /*! */: no se puede saber a qué sentencia pertenece cada parte"
                base = i + 3 + (len(sql[i + 3:fin]) - len(dentro))   # dónde empieza `dentro` en el SQL original
                for p in sub:
                    out.append(Pieza(p.tipo, p.texto, p.nivel + nivel, sql, base + p.ini, base + p.fin, (i, fin + 2)))
            i = fin + 2
        elif c in "'\"`":
            j = i + 1
            while True:
                if j >= n:
                    return out, "texto o identificador sin cerrar ({})".format(c)
                if sql[j] == "\\" and c != "`":
                    j += 2
                    continue
                if sql[j] == c:
                    if j + 1 < n and sql[j + 1] == c:   # '' o `` dobles
                        j += 2
                        continue
                    break
                j += 1
            out.append(Pieza("id" if c == "`" else "str", sql[i:j + 1], nivel, sql, i, j + 1))
            i = j + 1
        elif c.isalnum() or c in "_$@.":
            j = i
            while j < n and (sql[j].isalnum() or sql[j] in "_$@."):
                j += 1
            out.append(Pieza("pal", sql[i:j], nivel, sql, i, j))
            i = j
        else:
            if c == "(":
                out.append(Pieza("sym", c, nivel, sql, i, i + 1))
                nivel += 1
            elif c == ")":
                nivel = max(0, nivel - 1)
                out.append(Pieza("sym", c, nivel, sql, i, i + 1))
            else:
                out.append(Pieza("sym", c, nivel, sql, i, i + 1))
            i += 1
    return out, None


def sentencias(ps):
    """Parte en sentencias por `;` de nivel 0. Descarta las vacías."""
    res, cur = [], []
    salto = False
    for k, p in enumerate(ps):
        if salto:   # la `G` de un `\G`: ya se consumió como parte del delimitador
            salto = False
            continue
        # `\G` y `\g` TERMINAN la sentencia en mysqlsh/mysql (igual que `;`): lo que venga después es otra.
        es_barra_g = (p.tipo == "sym" and p.texto == "\\" and p.nivel == 0 and k + 1 < len(ps)
                      and ps[k + 1].tipo == "pal" and ps[k + 1].texto in ("G", "g") and ps[k + 1].ini == p.fin)
        if es_barra_g:
            salto = True
        if es_barra_g or (p.tipo == "sym" and p.texto == ";" and p.nivel == 0):
            if cur:
                res.append(cur)
            cur = []
        else:
            cur.append(p)
    if cur:
        res.append(cur)
    return res


def mayus(p):
    return p.texto.upper() if p.tipo == "pal" else None


def unir(ps):
    """Piezas pegadas con espacios. SÓLO para mensajes y comparaciones: `MIN ( id )` con
    espacios deja de ser una función para MySQL. Para armar SQL se usa `fuente()`."""
    return " ".join(p.texto for p in ps)


def fuente(ps):
    """El trozo del SQL ORIGINAL que cubren estas piezas, tal como lo escribió quien lo
    escribió (con sus saltos de línea; los comentarios de afuera no entran)."""
    if not ps:
        raise ValueError("fuente() sin piezas")
    out = [ps[0].src[ps[0].ini:ps[0].fin]]
    for a, b in zip(ps, ps[1:]):
        # Entre dos piezas del MISMO contexto va lo que el autor escribió; si una está dentro de un /*! */ y la
        # otra no (o en otro), lo del medio son los delimitadores del comentario y NO van en el SQL generado.
        out.append(a.src[a.fin:b.ini] if a.cmt == b.cmt else " ")
        out.append(b.src[b.ini:b.fin])
    return "".join(out)

# ───────────────────────────── 2. Mirar cada sentencia ─────────────────────────────

TAUTOLOGIA = re.compile(
    r"^\(*\s*(1\s*=\s*1|0\s*=\s*0|1|TRUE|2\s*>\s*1|1\s*<>\s*0|'1'\s*=\s*'1'|'a'\s*=\s*'a'|NOT\s+FALSE)\s*\)*$",
    re.I)


def es_tautologia(cond):
    """`WHERE 1=1`, `WHERE TRUE`, `WHERE x=1 OR 1=1`: se lleva la tabla entera igual."""
    ramas, cur = [], []
    for p in cond:
        if p.nivel == 0 and mayus(p) == "OR":
            ramas.append(cur)
            cur = []
        else:
            cur.append(p)
    ramas.append(cur)
    return any(TAUTOLOGIA.match(unir(r).replace(" ", "")) or TAUTOLOGIA.match(unir(r)) for r in ramas)


class Frena(Exception):
    def __init__(self, motivo, info=None):
        Exception.__init__(self, motivo)
        self.info = info


def cortar(ps, claves, desde=0):
    """Índice de la primera pieza de nivel 0 cuya palabra esté en `claves`, o -1."""
    for k in range(desde, len(ps)):
        if ps[k].nivel == 0 and mayus(ps[k]) in claves:
            return k
    return -1


_ESCAPES = {"0": "\0", "b": "\b", "n": "\n", "r": "\r", "t": "\t", "Z": "\x1a", "%": "\\%", "_": "\\_"}


def decodificar(crudo):
    """El valor de un literal 'texto' o "texto" tal como lo entrega el motor: sin las comillas de
    afuera, con `\\x` y la comilla doblada ya resueltas. (PREPARE lo vuelve a leer como SQL.)"""
    delim, cuerpo, out, i = crudo[0], crudo[1:-1], [], 0
    while i < len(cuerpo):
        c = cuerpo[i]
        if c == "\\" and i + 1 < len(cuerpo):
            out.append(_ESCAPES.get(cuerpo[i + 1], cuerpo[i + 1]))
            i += 2
        elif c == delim and i + 1 < len(cuerpo) and cuerpo[i + 1] == delim:
            out.append(delim)
            i += 2
        else:
            out.append(c)
            i += 1
    return "".join(out)


def _empieza_linea(p):
    """¿Antes de la pieza, en su línea, sólo hay espacios (o es el inicio del texto)?"""
    antes = p.src[:p.ini] if p.src is not None else ""
    return antes.rstrip(" \t").endswith("\n") or not antes.strip()


def mirar(st, profundidad=0):
    """Una sentencia → dict {tipo, ...}. Lanza Frena con el porqué si no debe correr."""
    if not st:
        return {"tipo": "VACIA"}
    primera = mayus(st[0])
    segunda = mayus(st[1]) if len(st) > 1 else None

    # Comandos del cliente (mysqlsh/mysql): `\!`, `\sql`, `\source`, DELIMITER, SOURCE, `\.`
    if st[0].texto == "\\" or primera in ("DELIMITER", "SOURCE"):
        raise Frena("comando del cliente (\\ / DELIMITER / SOURCE): no se puede analizar")
    # Los comandos del cliente (`\.`, `\source`, `\!`...) no llevan `;` delante: mysqlsh los toma al
    # empezar una LÍNEA aunque esa línea sea la continuación de otra sentencia. Con `st[0]` solo se
    # veían los del inicio de la sentencia.
    for p in st:
        if p.texto == "\\" and p.tipo == "sym" and _empieza_linea(p):
            raise Frena("comando del cliente (\\ ...) al empezar una línea: mysqlsh lo ejecuta sin `;` y no se puede analizar")
    # `source archivo` y `DELIMITER x` (sin barra) también son comandos del cliente al empezar una línea.
    for k, p in enumerate(st[1:], 1):
        if p.tipo == "pal" and p.texto.upper() in ("SOURCE", "DELIMITER") and _empieza_linea(p):
            # Sólo es una COLUMNA llamada así si la estructura lo dice: viene tras una coma (o SELECT/DISTINCT) y
            # lo que sigue es FROM, AS, una coma o el fin. Nunca por el nombre del archivo: `source from` es un comando.
            previo_ = st[k - 1]
            sig = st[k + 1] if k + 1 < len(st) else None
            tras_lista = previo_.texto == "," or mayus(previo_) in ("SELECT", "DISTINCT")
            if tras_lista and (sig is None or sig.texto == "," or mayus(sig) in ("FROM", "AS")):
                continue
            raise Frena("comando del cliente ({}) al empezar una línea: mysqlsh lo ejecuta sin `;` y no se puede analizar"
                        .format(p.texto.upper()))
    if primera == "BINLOG":
        raise Frena("BINLOG aplica eventos de filas crudos (UPDATE/DELETE incluidos): no se cuenta ni se respalda")

    if primera in ("DROP",) and segunda in ("DATABASE", "SCHEMA"):
        raise Frena("borra una BASE ENTERA (DROP {}), y en pruebas hay copias de clientes reales".format(segunda))
    if primera == "SET" and any(p.tipo == "pal" and "SQL_MODE" in p.texto.upper() for p in st):
        raise Frena("SET sql_mode cambia cómo se leen los textos del resto del envío y el análisis ya no es fiable. "
                    "Mandalo por separado")
    if primera == "LOAD" and any(mayus(p) == "REPLACE" for p in st):
        raise Frena("LOAD DATA/XML ... REPLACE borra y reemplaza las filas que ya existan con esa llave: no se cuenta ni se respalda")
    if primera in ("CREATE", "ALTER") and any(mayus(p) in ("EVENT", "TRIGGER", "PROCEDURE", "FUNCTION")
                                              for p in st[1:12]):
        raise Frena("CREATE/ALTER de un EVENT, TRIGGER, PROCEDURE o FUNCTION guarda SQL que corre después "
                    "(cada vez que alguien escriba, o a una hora), sin pasar por este análisis")
    if primera == "ALTER" and any(mayus(st[k]) == "DISCARD" and k + 1 < len(st) and mayus(st[k + 1]) == "TABLESPACE"
                                  for k in range(len(st))):
        raise Frena("ALTER TABLE ... DISCARD TABLESPACE tira el archivo .ibd de la tabla: sin una copia compatible las filas se pierden")
    if primera == "ALTER" and any(mayus(st[k]) in ("TRUNCATE", "DROP") and k + 1 < len(st)
                                  and mayus(st[k + 1]) == "PARTITION" for k in range(len(st))):
        raise Frena("ALTER TABLE ... TRUNCATE/DROP PARTITION descarta para siempre todas las filas de la partición")
    if primera == "REPLACE":
        raise Frena("REPLACE borra y vuelve a insertar las filas que ya existan con esa llave: no se cuenta ni se respalda")
    if primera == "INSERT" and any(st[k].nivel == 0 and mayus(st[k]) == "DUPLICATE" and k + 1 < len(st)
                                   and mayus(st[k + 1]) == "KEY" for k in range(len(st))):
        raise Frena("INSERT ... ON DUPLICATE KEY UPDATE pisa las filas que ya existan con esa llave: no se cuenta ni se respalda")
    if primera == "CALL":
        raise Frena("CALL ejecuta un procedimiento almacenado: no se puede saber qué escribe ni cuántas filas")
    if primera in ("EXPLAIN", "DESCRIBE", "DESC") and any(mayus(p) == "ANALYZE" for p in st[1:6]):
        raise Frena("EXPLAIN ANALYZE EJECUTA la sentencia (UPDATE/DELETE incluidos): no se analiza como lectura")
    if primera == "CLONE":
        raise Frena("CLONE INSTANCE borra TODAS las bases del servidor receptor antes de copiar el donante y lo reinicia")
    if primera == "TRUNCATE":
        raise Frena("TRUNCATE vacía la tabla ENTERA y no se deshace")

    if primera in ("PREPARE", "EXECUTE"):
        if primera == "EXECUTE":
            return {"tipo": "EXECUTE"}   # lo que importa es el PREPARE que lo armó
        k = cortar(st, ("FROM",))
        if k >= 0 and k + 1 < len(st) and st[k + 1].tipo == "str" and len(st) == k + 2:
            if profundidad >= 3:
                raise Frena("PREPARE anidado demasiado profundo")
            crudo = st[k + 1].texto
            interior = decodificar(crudo)
            sub, err = piezas(interior)
            if err:
                raise Frena("PREPARE con texto que no se puede leer: " + err)
            anidadas = [mirar(s2, profundidad + 1) for s2 in sentencias(sub)]
            return {"tipo": "PREPARE", "anidadas": anidadas, "llamadas": _llamadas(sub)}
        raise Frena("PREPARE con SQL armado (variable o concatenación): no se puede saber qué ejecuta")

    if primera == "WITH":
        # CTE: sólo se deja pasar si en el nivel 0 no hay una escritura.
        if cortar(st, ("UPDATE", "DELETE")) >= 0:
            raise Frena("WITH ... UPDATE/DELETE: no se puede analizar el WHERE. Pasalo como UPDATE normal")
        return {"tipo": "WITH-LECTURA"}

    if primera == "UPDATE":
        iset = cortar(st, ("SET",))
        if iset < 0:
            raise Frena("UPDATE sin SET: no se entiende")
        tablas = [p for p in st[1:iset] if mayus(p) not in ("LOW_PRIORITY", "IGNORE")]
        iwhere = cortar(st, ("WHERE",), iset)
        return _con_where("UPDATE", st, tablas, iwhere, iset)

    if primera == "DELETE":
        ifrom = cortar(st, ("FROM",))
        if ifrom < 0:
            raise Frena("DELETE sin FROM: no se entiende")
        iwhere = cortar(st, ("WHERE",), ifrom)
        fin_tablas = iwhere if iwhere >= 0 else len(st)
        quita = cortar(st, ("ORDER", "LIMIT", "USING"), ifrom)
        if quita >= 0 and (iwhere < 0 or quita < iwhere):
            fin_tablas = quita
        tablas = st[ifrom + 1:fin_tablas]
        return _con_where("DELETE", st, tablas, iwhere, ifrom)

    if primera == "USE":
        if len(st) != 2 or st[1].tipo not in ("pal", "id") or not re.fullmatch(r"[A-Za-z0-9_$]+", st[1].texto.strip("`")):
            raise Frena("USE con un nombre de base que no se entiende")
        return {"tipo": "USE", "esquema": st[1].texto.strip("`")}
    if primera == "INSERT":
        return _insert(st)
    if primera == "LOAD":
        return _load(st)

    return {"tipo": primera or "OTRA"}


def _insert(st):
    """INSERT: destino y si es MASIVO (`... SELECT`, `... TABLE` o varias filas en VALUES). Importa por los
    triggers de INSERT de la tabla, que pueden escribir en otras filas (p. ej. detmov_insert → existencias_tienda)."""
    k = 1
    while k < len(st) and mayus(st[k]) in ("LOW_PRIORITY", "DELAYED", "HIGH_PRIORITY", "IGNORE", "INTO"):
        k += 1
    fin = k
    while fin < len(st) and not (st[fin].texto == "(" or mayus(st[fin]) in ("VALUES", "VALUE", "SELECT", "SET",
                                                                           "PARTITION", "TABLE", "WITH")):
        fin += 1
    ref = _destino(st[k:fin]) if fin > k else None
    masivo = any(mayus(q) in ("SELECT", "TABLE") for q in st[fin:])
    iv = cortar(st, ("VALUES", "VALUE"))
    if iv >= 0 and any(q.nivel == 0 and q.texto == "," for q in st[iv + 1:]):
        masivo = True
    return {"tipo": "INSERT", "tabla": ref, "masivo": masivo}


def _load(st):
    """LOAD DATA/XML ... INTO TABLE t: dispara los triggers de INSERT de `t` por CADA fila cargada, así que se
    trata como un INSERT masivo. Si no se encuentra el destino, se devuelve sin tabla (como antes)."""
    k = next((j + 1 for j in range(len(st) - 1) if mayus(st[j]) == "INTO" and mayus(st[j + 1]) == "TABLE"), None)
    if k is None:
        return {"tipo": "LOAD"}
    k += 1
    fin = k
    while fin < len(st) and not (st[fin].texto == "(" or mayus(st[fin]) in (
            "PARTITION", "CHARACTER", "CHARSET", "FIELDS", "COLUMNS", "LINES", "IGNORE", "SET", "ROWS", "FORMAT")):
        fin += 1
    ref = _destino(st[k:fin]) if fin > k else None
    return {"tipo": "INSERT", "tabla": ref, "masivo": True} if ref else {"tipo": "LOAD"}


def _destino(tablas):
    """Si el UPDATE/DELETE es de UNA tabla (`t`, `db.t`, `` `db`.`t` ``, con alias `t a` o `t AS a`)
    devuelve su nombre tal como se escribió; con varias tablas, JOIN o subconsultas devuelve None."""
    if not tablas or any(mayus(p) in ("JOIN", "STRAIGHT_JOIN", "NATURAL", "USING") or p.texto in (",", "(")
                         for p in tablas):
        return None
    ref, k = [tablas[0]], 1
    # `db` . `tabla` llegan como piezas pegadas (sin espacio entre ellas): se juntan.
    # MySQL toma `db . tabla` igual que `db.tabla`: el punto puede ir suelto, con espacios a los lados.
    while k < len(tablas) and (tablas[k].texto.startswith(".") or ref[-1].texto.endswith(".") or
                               tablas[k].ini == ref[-1].fin) and (tablas[k].texto.startswith(".") or tablas[k].tipo in ("pal", "id")):
        ref.append(tablas[k])
        k += 1
    resto = tablas[k:]
    if len(resto) == 0 or (len(resto) == 1 and resto[0].tipo in ("pal", "id")) or \
            (len(resto) == 2 and mayus(resto[0]) == "AS" and resto[1].tipo in ("pal", "id")):
        return fuente(ref)
    return None


def _con_where(tipo, st, tablas, iwhere, hasta):
    ref = _destino(tablas)
    sin_where = None
    if ref:
        # Para poder respaldar la tabla entera si alguien se salta el freno con un tiquete.
        sin_where = {"tipo": tipo, "tabla": ref, "forzada": True,
                     "total_sql": "SELECT COUNT(*) FROM {}".format(ref),
                     "snapshot_sql": "SELECT * FROM {}".format(fuente(tablas))}
    if iwhere < 0:
        raise Frena("un {} sin WHERE toca la tabla ENTERA".format(tipo), sin_where)
    # El WHERE llega hasta un ORDER BY / LIMIT de nivel 0.
    fin = cortar(st, ("ORDER", "LIMIT"), iwhere + 1)
    cond = st[iwhere + 1: fin if fin >= 0 else len(st)]
    cola = st[fin:] if fin >= 0 else []
    if not cond:
        raise Frena("un {} con WHERE vacío toca la tabla ENTERA".format(tipo), sin_where)
    if es_tautologia(cond) and not (fin >= 0 and cortar(cola, ("LIMIT",)) >= 0):
        # Con LIMIT el WHERE siempre-verdadero no toca la tabla entera: se cuenta y respalda lo que el LIMIT deja.
        raise Frena("el WHERE ({}) siempre es verdadero: toca la tabla ENTERA".format(unir(cond)), sin_where)
    # Conteo, respaldo y escritura son tres sentencias: un predicado que da otro resultado cada vez
    # (RAND, UUID...) no identificaría las mismas filas en las tres.
    if any(p.tipo == "pal" and p.texto.upper() in VOLATILES for p in cond):
        raise Frena("el WHERE usa una función que da otro valor cada vez o que depende de la sesión (RAND, UUID, "
                    "LAST_INSERT_ID...): el conteo y el respaldo no serían de las mismas filas que se escriben", sin_where)
    # El conteo corre en OTRA sesión: una variable (@x) o un `?` de PREPARE valdría otra cosa ahí.
    if any(p.texto == "?" or (p.tipo == "pal" and p.texto.startswith("@") and not p.texto.startswith("@@"))
           for p in cond + list(tablas)):
        raise Frena("el WHERE usa una variable (@x) o un ? de PREPARE: no se puede contar de antemano", sin_where)
    if not ref:
        raise Frena("un {} de varias tablas o con sintaxis que no se puede contar (JOIN, FORCE INDEX, PARTITION, "
                    "subconsulta en el FROM): no se sabe cuántas filas toca ni se puede respaldar".format(tipo))
    info = {"tipo": tipo, "tabla": None, "llamadas": _llamadas(cond + list(tablas))}
    if ref:
        where = "{}\nWHERE {}".format(fuente(tablas), fuente(cond))
        limite = bool(cola) and cortar(cola, ("LIMIT",)) >= 0
        orden = None
        if limite and mayus(cola[0]) == "ORDER":
            fin_orden = cortar(cola, ("LIMIT",))
            items, cur = [], []
            for q in cola[2:fin_orden]:
                if q.nivel == 0 and q.texto == ",":
                    items.append(cur)
                    cur = []
                else:
                    cur.append(q)
            items.append(cur)
            orden = []
            for it in items:
                if it and mayus(it[-1]) in ("ASC", "DESC"):
                    it = it[:-1]
                if len(it) == 1 and it[0].tipo in ("pal", "id"):
                    orden.append(it[0].texto.replace("`", "").split(".")[-1].lower())
                else:
                    orden = None   # una expresión: no se puede saber si ordena de forma única
                    break
        if limite:   # con LIMIT sólo se toca lo que el LIMIT deja: se cuenta eso, no todo lo que calza
            where += "\n" + fuente(cola)
            cuenta = "SELECT COUNT(*) FROM (SELECT 1 FROM {}) AS limitado".format(where)
        else:
            cuenta = "SELECT COUNT(*) FROM {}".format(where)
        info.update({
            "tabla": ref,
            "total_sql": "SELECT COUNT(*) FROM {}".format(ref),
            "cuenta_sql": cuenta,
            "snapshot_sql": "SELECT * FROM {}".format(where),
            "limite": limite, "orden": orden,
        })
    return info


def _llamadas(ps):
    """Nombres con forma de llamada a función (`f(`, `db.f(`, `` `f`( ``) en estas piezas. El conteo y el respaldo
    previos las ejecutan: si alguna es una función ALMACENADA con efectos, escribiría antes de autorizar nada."""
    nombres = set()
    for k in range(1, len(ps)):
        if ps[k].texto == "(" and ps[k].tipo == "sym":
            q = ps[k - 1]
            if q.tipo == "id":
                n = q.texto[1:-1].replace("``", "`")   # entre comillas vale cualquier carácter
                if n:
                    nombres.add(n)
            elif q.tipo == "pal":
                n = q.texto.split(".")[-1]
                if re.fullmatch(r"[A-Za-z_][A-Za-z0-9_$]*", n):
                    nombres.add(n)
    return sorted(nombres)


# Funciones que dan otro valor cada vez, o que dependen de lo que hizo ESTA conexión antes: el conteo y el
# respaldo corren en otras conexiones y verían otra cosa que la escritura.
SESION = ("LAST_INSERT_ID", "ROW_COUNT", "FOUND_ROWS", "CONNECTION_ID", "GET_LOCK", "RELEASE_LOCK",
          "RELEASE_ALL_LOCKS", "IS_FREE_LOCK", "IS_USED_LOCK", "SLEEP", "BENCHMARK")
VOLATILES = ("RAND", "UUID", "UUID_SHORT", "SYSDATE", "RANDOM_BYTES", "UUID_TO_BIN") + SESION


def tiene_efecto(st):
    """Un SELECT que cambia el estado de la sesión: usa funciones de sesión, asigna (`:=`) o vuelca (`INTO`)."""
    for k, q in enumerate(st):
        if q.tipo == "pal" and q.texto.upper() in SESION:
            return True
        if q.nivel == 0 and mayus(q) == "INTO":
            return True
        if q.texto == ":" and k + 1 < len(st) and st[k + 1].texto == "=":
            return True
    return False

LECTURA = ("SELECT", "SHOW", "DESCRIBE", "DESC", "EXPLAIN", "VACIA", "WITH-LECTURA")


def aplanar(info):
    """Una sentencia puede traer escrituras anidadas (PREPARE): todas reciben el mismo trato."""
    yield info
    for hija in info.get("anidadas", []):
        for x in aplanar(hija):
            yield x


def analizar(sql):
    """→ (lista de dicts por escritura, motivo|None). `motivo` ≠ None: hay que frenar."""
    ps, err = piezas(sql)
    if err:
        return [], "no se puede leer el SQL: " + err
    res, sucio, preparadas, ejecuciones, motivo = [], False, 0, 0, None
    esquema, llamadas = None, set()
    for st in sentencias(ps):
        # Tras un rechazo se SIGUE mirando: con tiquete el envío entero corre igual, y lo que viene después
        # tiene que quedar contado y respaldado. Ya no hay estado limpio: todo lo siguiente es «posterior».
        try:
            llamadas.update(_llamadas(st))
            info = mirar(st)
            llamadas.update(info.get("llamadas_prep", []) if info["tipo"] == "PREPARE" else [])
            if info["tipo"] == "PREPARE":
                llamadas.update(info.get("llamadas", []))
            if info["tipo"] == "USE":
                esquema = info["esquema"]   # los pasos previos se conectan con esta base por omisión
                continue
            if info["tipo"] == "EXECUTE":
                ejecuciones += 1
            if info["tipo"] == "PREPARE" and any(x["tipo"] in ("UPDATE", "DELETE") for x in aplanar(info)):
                preparadas += 1
            if preparadas and ejecuciones > 1:
                raise Frena("EXECUTE repetido de una escritura preparada: cada ejecución puede tocar otras filas "
                            "y sólo se contó y respaldó una vez. Mandá un EXECUTE por envío")
            for x in aplanar(info):
                if esquema and x["tipo"] in ("UPDATE", "DELETE", "INSERT"):
                    x["esquema"] = esquema
                if x["tipo"] in ("UPDATE", "DELETE"):
                    if sucio:
                        raise Frena("un {} que viene DESPUÉS de otra sentencia que cambia datos o la sesión "
                                    "(SET, USE, INSERT, otro UPDATE...) en el mismo envío: el conteo previo corre "
                                    "en otra sesión y vería otro estado. Mandalas por separado".format(x["tipo"]), x)
                    res.append(x)
                elif x["tipo"] == "INSERT":
                    res.append(x)
            if info["tipo"] not in LECTURA or tiene_efecto(st):
                sucio = True
        except Frena as e:
            if e.info:
                if esquema:
                    e.info.setdefault("esquema", esquema)
                res.append(e.info)
            if motivo is None:
                motivo = str(e)
            sucio = True
    if llamadas:
        res.insert(0, {"tipo": "LLAMADAS", "nombres": sorted(llamadas)})
    return res, motivo

# ───────────────────────────── 3. Contar, guardar, anotar ─────────────────────────────

ESQUEMA = [None]   # base por omisión de los pasos previos (la del último USE del envío)


def mysqlsh(sql, formato):
    uri = "{}:{}@{}:{}".format(quote(os.environ["BD_USUARIO"], safe=""), quote(os.environ["BD_CLAVE"], safe=""),
                               os.environ.get("SERVIDOR", "132.226.40.48"), os.environ.get("PUERTO", "3310"))
    arg = ["--json=raw"] if formato == "json" else ["--result-format=tabbed"]
    if ESQUEMA[0]:
        arg += ["--schema", ESQUEMA[0]]
    r = subprocess.run(["mysqlsh", "--sql", "--uri", uri] + arg + ["-e", sql],
                       capture_output=True, text=True)
    lineas = [x for x in r.stdout.splitlines() if x.strip()]
    if r.returncode != 0:
        raise RuntimeError((r.stderr or r.stdout).strip().splitlines()[-1] if (r.stderr or r.stdout).strip() else "mysqlsh falló")
    return lineas


def contar(sql):
    ultima = mysqlsh(sql, "tabbed")[-1].strip()
    if not re.fullmatch(r"\d+", ultima):   # nada de adivinar un número a partir de otro texto
        raise RuntimeError("el servidor no devolvió un número: «{}»".format(ultima[:60]))
    return int(ultima)


def hijas_en_cascada(tabla, tipo="DELETE"):
    """Tablas (esquema.tabla) que pierden o cambian filas si se borra (o se actualiza) `tabla`
    (ON DELETE / ON UPDATE CASCADE, SET NULL, SET DEFAULT). Se busca sólo en el esquema de la tabla destino."""
    regla = "UPDATE_RULE" if tipo == "UPDATE" else "DELETE_RULE"
    partes = [x.strip() for x in tabla.replace("`", "").split(".")]
    if not 1 <= len(partes) <= 2 or not all(re.fullmatch(r"[A-Za-z0-9_$]+", x) for x in partes):
        raise RuntimeError("nombre de tabla raro: " + tabla[:60])
    esquema = "'{}'".format(partes[0]) if len(partes) == 2 else "DATABASE()"
    lineas = mysqlsh("SELECT DISTINCT CONSTRAINT_SCHEMA, TABLE_NAME FROM information_schema.REFERENTIAL_CONSTRAINTS "
                     "WHERE REFERENCED_TABLE_NAME='{}' AND UNIQUE_CONSTRAINT_SCHEMA = {} "
                     "AND {} IN ('CASCADE','SET NULL','SET DEFAULT')".format(partes[-1], esquema, regla), "tabbed")
    if lineas and lineas[0].startswith("CONSTRAINT_SCHEMA"):
        lineas = lineas[1:]
    return sorted(set(".".join(x.split()) for x in lineas))


def huella(sql):
    """Qué se autorizó EXACTAMENTE: largo y SHA-256 del SQL completo (la bitácora sólo guarda el principio)."""
    return {"sql_largo": len(sql), "sql_sha256": hashlib.sha256(sql.encode("utf-8", "replace")).hexdigest()}


def triggers_de(tabla, eventos=("UPDATE", "DELETE")):
    """Triggers de la tabla destino para esos eventos: escriben en OTRAS filas/tablas y eso no se cuenta ni se respalda."""
    partes = [x.strip() for x in tabla.replace("`", "").split(".")]
    if not 1 <= len(partes) <= 2 or not all(re.fullmatch(r"[A-Za-z0-9_$]+", x) for x in partes):
        return []
    esquema = "'{}'".format(partes[0]) if len(partes) == 2 else "DATABASE()"
    lineas = mysqlsh("SELECT DISTINCT TRIGGER_NAME FROM information_schema.TRIGGERS WHERE EVENT_OBJECT_TABLE='{}' "
                     "AND EVENT_OBJECT_SCHEMA = {} AND EVENT_MANIPULATION IN ({})"
                     .format(partes[-1], esquema, ",".join("'{}'".format(e) for e in eventos)), "tabbed")
    if lineas and lineas[0].startswith("TRIGGER_NAME"):
        lineas = lineas[1:]
    return sorted(set(x.strip() for x in lineas))


def funciones_almacenadas(nombres):
    """Cuáles de estos nombres son funciones almacenadas en el servidor (en cualquier esquema)."""
    lineas = mysqlsh("SELECT DISTINCT ROUTINE_NAME FROM information_schema.ROUTINES WHERE ROUTINE_TYPE='FUNCTION' "
                     "AND ROUTINE_NAME IN ({})".format(",".join("'{}'".format(x.replace("\\", "\\\\").replace("'", "\\'"))
                                                                for x in nombres)), "tabbed")
    if lineas and lineas[0].startswith("ROUTINE_NAME"):
        lineas = lineas[1:]
    return sorted(set(x.strip() for x in lineas))


def es_vista(tabla):
    """¿El destino es una VISTA? Un UPDATE/DELETE a través de una vista cambia la tabla base y el respaldo
    (SELECT * de la vista) sólo trae las columnas que la vista proyecta."""
    partes = [x.strip() for x in tabla.replace("`", "").split(".")]
    if not 1 <= len(partes) <= 2 or not all(re.fullmatch(r"[A-Za-z0-9_$]+", x) for x in partes):
        raise RuntimeError("nombre de tabla raro: " + tabla[:60])
    esquema = "'{}'".format(partes[0]) if len(partes) == 2 else "DATABASE()"
    ultima = mysqlsh("SELECT COUNT(*) AS vistas FROM information_schema.VIEWS WHERE TABLE_SCHEMA = {} "
                     "AND TABLE_NAME = '{}'".format(esquema, partes[-1]), "tabbed")[-1].strip()
    if not re.fullmatch(r"\d+", ultima):
        raise RuntimeError("el servidor no devolvió un número: «{}»".format(ultima[:60]))
    return int(ultima) > 0


def columnas_invisibles(tabla):
    """Cuántas columnas INVISIBLES tiene la tabla: `SELECT *` no las trae y el respaldo quedaría incompleto."""
    partes = [x.strip() for x in tabla.replace("`", "").split(".")]
    if not 1 <= len(partes) <= 2 or not all(re.fullmatch(r"[A-Za-z0-9_$]+", x) for x in partes):
        raise RuntimeError("nombre de tabla raro: " + tabla[:60])
    esquema = "'{}'".format(partes[0]) if len(partes) == 2 else "DATABASE()"
    ultima = mysqlsh("SELECT COUNT(*) AS invisibles FROM information_schema.COLUMNS WHERE TABLE_SCHEMA = {} "
                     "AND TABLE_NAME = '{}' AND EXTRA LIKE '%INVISIBLE%'".format(esquema, partes[-1]), "tabbed")[-1].strip()
    if not re.fullmatch(r"\d+", ultima):
        raise RuntimeError("el servidor no devolvió un número: «{}»".format(ultima[:60]))
    return int(ultima)


def pk_de(tabla):
    """Columnas de la llave primaria de la tabla destino (vacío si no tiene)."""
    partes = [x.strip() for x in tabla.replace("`", "").split(".")]
    if not 1 <= len(partes) <= 2 or not all(re.fullmatch(r"[A-Za-z0-9_$]+", x) for x in partes):
        raise RuntimeError("nombre de tabla raro: " + tabla[:60])
    esquema = "'{}'".format(partes[0]) if len(partes) == 2 else "DATABASE()"
    lineas = mysqlsh("SELECT COLUMN_NAME FROM information_schema.STATISTICS WHERE TABLE_SCHEMA = {} "
                     "AND TABLE_NAME = '{}' AND INDEX_NAME = 'PRIMARY'".format(esquema, partes[-1]), "tabbed")
    if lineas and lineas[0].startswith("COLUMN_NAME"):
        lineas = lineas[1:]
    return [x.strip().lower() for x in lineas]


def anotar(tiquete, detalle):
    os.makedirs(CARPETA, mode=0o700, exist_ok=True)
    ruta = os.path.join(CARPETA, "excepciones-sql.log")
    fd = os.open(ruta, os.O_WRONLY | os.O_APPEND | os.O_CREAT, 0o600)
    with os.fdopen(fd, "a") as f:
        f.write(json.dumps({"cuando_utc": datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
                            "quien": os.environ.get("BD_USUARIO", "?"), "tiquete": tiquete,
                            **detalle}, ensure_ascii=False) + "\n")


def _es_aviso(linea):
    """Un aviso de mysqlsh es un objeto JSON cuya ÚNICA clave es `warning`. Una fila de datos viaja anidada en
    `rows`, así que una columna llamada `warning` no se confunde con él."""
    try:
        d = json.loads(linea)
    except ValueError:
        return False
    return isinstance(d, dict) and set(d) == {"warning"}


def guardar_filas(info):
    carpeta = os.path.join(CARPETA, "snapshots")
    os.makedirs(carpeta, mode=0o700, exist_ok=True)
    marca = datetime.datetime.now(datetime.timezone.utc).strftime("%Y%m%d-%H%M%S-%f") + "-" + str(os.getpid())
    nombre = re.sub(r"[^A-Za-z0-9_.-]", "_", info["tabla"])
    ruta = os.path.join(carpeta, "{}-{}.jsonl".format(marca, nombre))
    lineas = [x for x in mysqlsh(info["snapshot_sql"], "json") if not _es_aviso(x)]
    fd = os.open(ruta, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
    with os.fdopen(fd, "w") as f:
        f.write("\n".join(lineas) + "\n")
    return ruta


SNAPSHOT_MAX = int(os.environ.get("MB_SNAPSHOT_MAX_FILAS", "100000"))


def _frena(motivo):
    print("✋ {}.".format(motivo), file=sys.stderr)
    print("   Si es lo que querés:  consultar-pruebas.sh --se-lo-que-hago <N° de tiquete> \"...\"", file=sys.stderr)
    return 2


def previo(sql, tiquete):
    """Corre los pasos 1-5. Devuelve 0 (deja pasar) o 2 (frena); imprime el porqué."""
    excepcion = tiquete is not None
    infos, motivo = analizar(sql)

    # Con NO_BACKSLASH_ESCAPES el motor no lee `\'` como este analizador: un texto podría cerrarse
    # antes y esconder una sentencia, y con ANSI_QUOTES las comillas dobles son identificadores y no textos. Sólo
    # importa si el SQL trae una barra o comillas dobles; si no se puede saber, se frena.
    if not motivo and ("\\" in sql or '"' in sql):
        try:
            modo = mysqlsh("SELECT @@SESSION.sql_mode", "tabbed")[-1]
        except Exception as e:
            modo, motivo = "", "no se pudo leer el sql_mode del servidor ({}) y el SQL trae una barra invertida o comillas dobles".format(e)
        if "NO_BACKSLASH_ESCAPES" in modo or "ANSI_QUOTES" in modo:
            motivo = ("el servidor usa NO_BACKSLASH_ESCAPES o ANSI_QUOTES: la barra invertida y las comillas dobles se "
                      "leen distinto y el análisis no es fiable")
        if motivo:
            infos = []

    if motivo:
        if not excepcion:
            return _frena(motivo)
        anotar(tiquete, {"frenado_por": motivo, "tablas": [i["tabla"] for i in infos if i.get("tabla")],
                         "sql": sql[:500], **huella(sql)})
    for info in infos:
        ESQUEMA[0] = info.get("esquema")
        if info["tipo"] == "LLAMADAS":
            # Una función ALMACENADA puede escribir en cualquier tabla sin que este análisis lo vea.
            try:
                alm = funciones_almacenadas(info["nombres"])
            except Exception as e:
                alm = ["?"]
                print("⚠️  No se pudo saber si el SQL llama funciones almacenadas ({}).".format(e), file=sys.stderr)
            if alm and not excepcion:
                return _frena("El SQL llama a la función almacenada {}: puede escribir en otras tablas y eso no se "
                              "cuenta ni se respalda aquí".format(", ".join(alm[:3])))
            if alm:
                anotar(tiquete, {"funciones_almacenadas": alm, "sql": sql[:500], **huella(sql)})
            continue
        if info["tipo"] == "INSERT" and info.get("tabla"):
            try:
                trg = triggers_de(info["tabla"], ("INSERT",))
            except Exception as e:
                trg = ["?"] if info.get("masivo") else []   # sin saber si hay triggers, un INSERT masivo se trata como si los hubiera
                if info.get("masivo"):
                    print("⚠️  No se pudo leer los triggers de {} ({}).".format(info["tabla"], e), file=sys.stderr)
            if trg and info.get("masivo") and not excepcion:
                return _frena("Este INSERT masivo (SELECT o varias filas) en {} dispara triggers de INSERT ({}) que pueden "
                              "cambiar filas que ya existían, y eso no se cuenta ni se respalda aquí".format(
                                  info["tabla"], ", ".join(trg[:3])))
            if trg:
                print("⚠️  {} tiene {} trigger(s) de INSERT ({}): lo que escriban en otras filas no se cuenta ni se "
                      "respalda aquí.".format(info["tabla"], len(trg), ", ".join(trg[:3])), file=sys.stderr)
                if excepcion:
                    anotar(tiquete, {"tabla": info["tabla"], "triggers_insert": trg, "sql": sql[:500], **huella(sql)})
            continue
        if info["tipo"] not in ("UPDATE", "DELETE"):
            continue
        if info.get("forzada"):
            # El freno se saltó con un tiquete: igual se respalda la tabla si cabe.
            try:
                total = contar(info["total_sql"])
                if 0 < total <= SNAPSHOT_MAX:
                    print("↩️  Respaldo de {} fila(s) de {} en {} (para deshacer)."
                          .format(total, info["tabla"], guardar_filas(info)), file=sys.stderr)
                elif total > SNAPSHOT_MAX:
                    print("⚠️  {} tiene {} filas (más de {}): NO se respaldó.".format(info["tabla"], total, SNAPSHOT_MAX),
                          file=sys.stderr)
            except Exception as e:
                print("⚠️  No se pudo respaldar {} ({}).".format(info["tabla"], e), file=sys.stderr)
            continue
        if not info.get("cuenta_sql"):
            return _frena("Este {} no se puede contar de antemano".format(info["tipo"]))
        if info.get("llamadas"):
            # Antes de CUALQUIER consulta previa: el conteo y el respaldo EJECUTAN el WHERE, y una función
            # almacenada puede escribir. Con tiquete tampoco: la ejecutaría una vez de más sin que nadie lo cuente.
            try:
                alm = funciones_almacenadas(info["llamadas"])
            except Exception as e:
                return _frena("No se pudo saber si el WHERE llama funciones almacenadas ({}). Sin eso no se cuenta".format(e))
            if alm:
                return _frena("El WHERE llama a la función almacenada {}: el conteo y el respaldo previos la ejecutarían "
                              "(puede escribir) sin pasar por este freno. Usá valores literales".format(", ".join(alm[:3])))
        try:
            n, total = contar(info["cuenta_sql"]), contar(info["total_sql"])
            hijas = hijas_en_cascada(info["tabla"], info["tipo"]) if info["tipo"] in ("DELETE", "UPDATE") else []
        except Exception as e:   # sin saber cuánto toca, no se escribe
            return _frena("No se pudo contar cuántas filas toca ({}). Sin eso no se escribe".format(e))
        try:
            vista = es_vista(info["tabla"])
        except Exception as e:
            vista = not excepcion   # sin saber si es una vista, sin tiquete no se escribe
            if excepcion:
                print("⚠️  No se pudo saber si {} es una vista ({}).".format(info["tabla"], e), file=sys.stderr)
        if vista and not excepcion:
            return _frena("{} es una VISTA (o no se pudo saber): escribir a través de ella cambia la tabla base y el "
                          "respaldo no traería sus columnas completas. Usá la tabla base".format(info["tabla"]))
        if vista:
            print("⚠️  {} es una vista: el respaldo NO trae la fila completa de la tabla base.".format(info["tabla"]),
                  file=sys.stderr)
        try:
            invisibles = columnas_invisibles(info["tabla"])
        except Exception as e:
            invisibles = -1 if not excepcion else 0
            if excepcion:
                print("⚠️  No se pudo saber si {} tiene columnas invisibles ({}).".format(info["tabla"], e), file=sys.stderr)
        if invisibles and not excepcion:
            return _frena("{} tiene columnas INVISIBLES (o no se pudo saber): el respaldo con SELECT * no las guardaría "
                          "y no se podría deshacer".format(info["tabla"]))
        if invisibles:
            print("⚠️  {} tiene columnas invisibles: el respaldo NO las incluye.".format(info["tabla"]), file=sys.stderr)
        try:
            triggers = triggers_de(info["tabla"])
        except Exception:
            triggers = []
        if triggers:
            print("⚠️  {} tiene {} trigger(s) de UPDATE/DELETE ({}): lo que escriban en OTRAS tablas no se cuenta ni "
                  "se respalda aquí.".format(info["tabla"], len(triggers), ", ".join(triggers[:3])), file=sys.stderr)
        if info.get("limite") and not excepcion:
            # Con LIMIT, el respaldo y la escritura eligen cada uno SU subconjunto: sólo es el mismo
            # si el ORDER BY incluye toda la llave primaria (orden único).
            try:
                pk = pk_de(info["tabla"])
            except Exception as e:
                return _frena("No se pudo leer la llave primaria de {} ({})".format(info["tabla"], e))
            if not pk or not info.get("orden") or not set(pk) <= set(info["orden"]):
                return _frena("Un {} con LIMIT necesita ORDER BY por toda la llave primaria de {} ({}) para que el "
                              "respaldo y la escritura sean las mismas filas".format(
                                  info["tipo"], info["tabla"], ", ".join(pk) or "no tiene"))
        entera = total > 0 and n >= total
        if hijas and not excepcion:
            return _frena("{} {} también {} filas de {} (ON {} CASCADE / SET NULL) y eso no se cuenta ni se respalda aquí"
                          .format("Borrar de" if info["tipo"] == "DELETE" else "Actualizar", info["tabla"],
                                  "borra o cambia" if info["tipo"] == "DELETE" else "cambia", ", ".join(hijas), info["tipo"]))
        if (n > UMBRAL or entera) and not excepcion:
            que = "TODAS las filas de {} ({})".format(info["tabla"], total) if entera else \
                  "{} filas de {} (el umbral es {})".format(n, info["tabla"], UMBRAL)
            return _frena("Este {} toca {}".format(info["tipo"], que))
        if n > SNAPSHOT_MAX and not excepcion:
            return _frena("Este {} toca {} filas de {}, más que el tope de respaldo ({}): no se podría deshacer"
                          .format(info["tipo"], n, info["tabla"], SNAPSHOT_MAX))
        if n > SNAPSHOT_MAX:
            print("⚠️  {} filas de {} son más de {}: NO se respaldó.".format(n, info["tabla"], SNAPSHOT_MAX), file=sys.stderr)
        elif n > 0:
            ruta = guardar_filas(info)
            print("↩️  Respaldo de {} fila(s) de {} en {} (para deshacer).".format(n, info["tabla"], ruta), file=sys.stderr)
        if excepcion:
            anotar(tiquete, {"tabla": info["tabla"], "filas": n, "de_total": total, "hijas_en_cascada": hijas,
                             "triggers": triggers, "sql": sql[:500], **huella(sql)})
    return 0


def main(argv):
    if len(argv) >= 3 and argv[1] == "analizar":
        infos, motivo = analizar(argv[2])
        print(json.dumps({"frena": motivo, "sentencias": infos}, ensure_ascii=False))
        return 2 if motivo else 0
    if len(argv) >= 3 and argv[1] == "previo":
        resto, tiquete = argv[2:], None
        if resto[0] == "--tiquete":
            tiquete, resto = resto[1], resto[2:]
            if not re.fullmatch(r"\d{3,7}", tiquete):
                print("✋ El tiquete tiene que ser un número (ej. 24603), no «{}».".format(tiquete), file=sys.stderr)
                return 2
        return previo(resto[0], tiquete)
    print(__doc__, file=sys.stderr)
    return 1


if __name__ == "__main__":
    sys.exit(main(sys.argv))