# -*- coding: utf-8 -*-
"""Prueba funcional completa: levanta una copia del programa sobre un cliente de ensayo con de todo
(cámara, exportados, proyectos, música, stock de Envato, basura, proxies, repetidos, huecos, un mes mal
escrito y un clip de video de verdad) y ejercita TODAS las funciones por su API, en el mismo orden en
que las usaría una persona. Imprime una tabla: qué se probó, qué salió y en cuánto tiempo."""
import os
import sys
import json
import time
import shutil
import socket
import tempfile
import subprocess
import urllib.parse
import urllib.error
import urllib.request

AQUI = os.path.dirname(os.path.abspath(__file__))
TMP = tempfile.mkdtemp(prefix='organilab-smoke-')
APP = os.path.join(TMP, 'app')
os.makedirs(APP)
for f in ('organizador.py', 'reglas.py', 'asistente.py', 'actualizador.py', 'resolve_bridge.py', 'braw.py', 'firma.py', 'ventana.py', 'instalar.py'):
    shutil.copy(os.path.join(AQUI, f), APP)
shutil.copytree(os.path.join(AQUI, 'static'), os.path.join(APP, 'static'))
RAIZ = os.path.join(TMP, 'Cliente Smoke')
RESP = os.path.join(TMP, 'Respaldo')
FILAS = []


def prueba(nombre, fn):
    t = time.time()
    try:
        det = fn()
        FILAS.append((nombre, 'OK', '%.2f s' % (time.time() - t), det or ''))
    except Exception as ex:  # noqa
        FILAS.append((nombre, 'FALLA', '%.2f s' % (time.time() - t), str(ex)[:160]))


def mk(rel, y, m, d, tam=4096):
    p = os.path.join(RAIZ, rel.replace('/', os.sep))
    os.makedirs(os.path.dirname(p), exist_ok=True)
    with open(p, 'wb') as fh:
        fh.write(os.urandom(4096) * max(1, tam // 4096))
    t = time.mktime(time.struct_time((y, m, d, 12, 0, 0, 0, 0, -1)))
    os.utime(p, (t, t))
    return p


# ---------------------------------------------------------------- el cliente de ensayo
mk('2026/03- Marzo/Grabacion/A001_03051200_C001.braw', 2026, 3, 5, 3_000_000)
mk('2026/03- Marzo/Grabacion/A001_03051200_C002.braw', 2026, 3, 5, 3_000_000)
mk('2026/03- Marzo/Grabacion/C0010.MP4', 2026, 3, 12, 2_000_000)          # sony, otro día
mk('2026/03- Marzo/Grabacion/C0010M01.XML', 2026, 3, 12, 500)
mk('2026/03- Marzo/Grabacion/Proxy/C0010.MOV', 2026, 3, 12, 100_000)      # proxy
mk('2026/03- Marzo/Grabacion/Thumbs.db', 2026, 3, 12, 100)                # basura
mk('2026/03- Marzo/01- Botox/Reel botox.mp4', 2026, 3, 20, 6_000_000)
mk('2026/03- Marzo/01- Botox/Reel botox.prproj', 2026, 3, 20, 50_000)
mk('2026/03- Marzo/01- Botox/Visual/Elements/clinic-2025-01-22-22-34-00-utc.mp4', 2026, 3, 20, 900_000)   # stock viejo
mk('2026/10- octubre/Exportados/Final.mp4', 2026, 10, 2, 700_000)          # mes en minúsculas
mk('Musica/tema.wav', 2026, 1, 1, 300_000)
mk('Sueltos/Portada.jpg', 2026, 2, 2, 60_000)
mk('Sueltos/Pieza.aep', 2026, 2, 2, 40_000)
shutil.copy2(os.path.join(RAIZ, '2026', '03- Marzo', '01- Botox', 'Reel botox.mp4'), os.path.join(RAIZ, 'Sueltos', 'Reel botox.mp4'))   # repetido de verdad: mismo contenido
mk('Raro/algo.xyz', 2026, 2, 2, 10)
os.makedirs(os.path.join(RAIZ, 'Vacia', 'dentro'), exist_ok=True)

# dos videos de verdad para el convertidor (6 s, 1280x720, H.264): uno con tanto grano y tanta tasa que sí
# adelgaza, y otro ya comprimido como sale de una cámara, que el convertidor no debe ofrecer
FF = shutil.which('ffmpeg')
CLIP = os.path.join(RAIZ, '2026', '03- Marzo', 'Grabacion', 'C0020.MP4')
CLIP2 = os.path.join(RAIZ, '2026', '03- Marzo', 'Grabacion', 'C0021.MP4')
if FF:
    for destino, imagen, crf in ((CLIP, 'testsrc=duration=6:size=1280x720:rate=25,noise=alls=40:allf=t', '8'),
                                 (CLIP2, 'testsrc=duration=6:size=1280x720:rate=25', '30')):
        subprocess.run([FF, '-hide_banner', '-loglevel', 'error', '-y', '-f', 'lavfi', '-i', imagen,
                        '-f', 'lavfi', '-i', 'sine=frequency=440:duration=6', '-c:v', 'libx264', '-preset', 'ultrafast', '-crf', crf,
                        '-pix_fmt', 'yuv420p', '-c:a', 'aac', '-shortest', destino], check=True)
        t = time.mktime(time.struct_time((2025, 6, 10, 12, 0, 0, 0, 0, -1)))
        os.utime(destino, (t, t))
# y, si se da uno, un BRAW de verdad (variable ORGANILAB_BRAW_PRUEBA): los de arriba son bytes al azar
BRAW = os.environ.get('ORGANILAB_BRAW_PRUEBA') or ''
if BRAW:
    shutil.copy(BRAW, os.path.join(RAIZ, '2026', '03- Marzo', 'Grabacion', 'A001_03051200_C003.braw'))
    t = time.mktime(time.struct_time((2026, 3, 5, 12, 0, 0, 0, 0, -1)))
    os.utime(os.path.join(RAIZ, '2026', '03- Marzo', 'Grabacion', 'A001_03051200_C003.braw'), (t, t))

# ---------------------------------------------------------------- el servidor
with socket.socket() as s_:
    s_.bind(('127.0.0.1', 0))
    PUERTO = s_.getsockname()[1]
ENSAYO = os.path.join(TMP, 'equipo')
os.makedirs(ENSAYO)
env = dict(os.environ, ORGANIZADOR_PUERTO=str(PUERTO), PYTHONIOENCODING='utf-8', ORGANILAB_ENSAYO=ENSAYO)
log = open(os.path.join(TMP, 'servidor.log'), 'w', encoding='utf-8')
proc = subprocess.Popen([sys.executable, 'organizador.py', '--sin-navegador'], cwd=APP, env=env, stdout=log, stderr=subprocess.STDOUT)
BASE = f'http://127.0.0.1:{PUERTO}'


def get(p):
    return json.loads(urllib.request.urlopen(BASE + p, timeout=60).read())


def post(p, d):
    rq = urllib.request.Request(BASE + p, json.dumps(d).encode(), {'Content-Type': 'application/json', 'Origin': BASE})
    try:
        r = urllib.request.urlopen(rq, timeout=120)
        return r.status, json.loads(r.read())
    except urllib.error.HTTPError as e:
        return e.code, json.loads(e.read())


def q(**kw):
    return '?' + urllib.parse.urlencode(kw)


for _ in range(80):
    try:
        urllib.request.urlopen(BASE + '/api/estado', timeout=1)
        break
    except Exception:  # noqa
        time.sleep(0.25)
else:
    raise SystemExit('el servidor no arrancó')


def esperar(cond, seg=90):
    fin = time.time() + seg
    while time.time() < fin:
        if cond():
            return True
        time.sleep(0.3)
    return False


def raiz_estado():
    return next((x for x in get('/api/estado')['raices'] if x['ruta'].lower() == RAIZ.lower()), None)


def arbol():
    return sorted(os.path.relpath(os.path.join(b, f), RAIZ).lower() for b, _, fs in os.walk(RAIZ) for f in fs
                  if '.organilab' not in b.lower())


# ---------------------------------------------------------------- las pruebas, en orden de uso
def p_estatico():
    html = urllib.request.urlopen(BASE + '/', timeout=10).read().decode('utf-8')
    assert 'googleapis' not in html, 'la interfaz sigue pidiendo la fuente a Google'
    r = urllib.request.urlopen(BASE + '/static/manrope.woff2', timeout=10)
    assert r.headers.get('Content-Type') == 'font/woff2', r.headers.get('Content-Type')
    assert r.read(4) == b'wOF2'
    return 'sin nada externo; fuente local servida como font/woff2'


def p_registrar():
    c, r = post('/api/raices', {'raices': [{'ruta': RAIZ, 'cliente': 'Cliente Smoke', 'tipo': 'contenido'}]})
    assert c == 200, r
    # anidada: se rechaza
    c2, r2 = post('/api/raices', {'raices': [{'ruta': RAIZ, 'cliente': 'A', 'tipo': 'contenido'},
                                             {'ruta': os.path.join(RAIZ, 'Sueltos'), 'cliente': 'B', 'tipo': 'contenido'}]})
    assert c2 != 200 or r2.get('error'), 'aceptó una carpeta dentro de otra'
    return 'registra, y rechaza una carpeta anidada'


def p_escanear():
    c, r = post('/api/escanear', {'raiz': RAIZ})
    assert c == 200 and r.get('lanzado'), r
    assert esperar(lambda: (raiz_estado() or {}).get('escaneo') and not ((raiz_estado() or {}).get('progreso') or {}).get('estado') == 'corriendo')
    e = raiz_estado()['escaneo']
    esperados = len(arbol())
    assert e['n'] == esperados, 'inventarió %d de %d' % (e['n'], esperados)
    return '%d archivos inventariados, los mismos que hay en el disco' % e['n']


def p_carpeta():
    e = get('/api/carpeta/estado' + q(raiz=RAIZ))
    assert e.get('raiz') and 'escaneo' in e, e
    a = get('/api/carpeta/acciones' + q(raiz=RAIZ))
    claves = [x['clave'] for x in a['acciones']]
    for k in ('limpiar', 'nombrar', 'fechas', 'resto', 'vacias', 'stock', 'repetidos', 'respaldar'):
        assert k in claves, 'falta la acción ' + k
    toca = [x['clave'] for x in a['acciones'] if x['estado'] == 'toca']
    assert 'limpiar' in toca and 'vacias' in toca and 'repetidos' in toca and 'stock' in toca, toca
    return 'lista de acciones completa; pendientes: ' + ', '.join(toca)


def p_que_hay():
    m = get('/api/mapa' + q(raiz=RAIZ))
    assert m['basura']['n'] >= 1 and m['proxies']['n'] >= 1, (m['basura'], m['proxies'])
    assert m.get('arbol'), 'sin árbol de peso'
    d = get('/api/discos')
    assert isinstance(d, (list, dict))
    return 'basura %d · proxies %d · %d ramas' % (m['basura']['n'], m['proxies']['n'], len(m['arbol']))


def p_tipo_y_pesa():
    # «Qué pesa» entra nivel a nivel
    d = get('/api/mapa/carpeta' + q(raiz=RAIZ))
    assert d['hijos'] and d['n'] == raiz_estado()['escaneo']['n'], (d['n'], len(d['hijos']))
    d2 = get('/api/mapa/carpeta' + q(raiz=RAIZ, ruta=os.path.join(RAIZ, '2026')))
    assert d2['rel'] == '2026' and d2['niveles'] >= 2 and d2['roles'], d2
    # el tipo de trabajo de la carpeta y el del cliente no cuadran: Inicio lo dice, y elegir uno los iguala
    cl = get('/api/estado')['config'].get('clientes') or {}
    cl['Cliente Smoke'] = {'tipo': 'proyecto'}
    c, r = post('/api/clientes', {'clientes': cl})
    assert c == 200, r
    a = {x['clave']: x for x in get('/api/carpeta/acciones' + q(raiz=RAIZ))['acciones']}
    assert 'tipo' in a and a['tipo']['tipos'] == ['contenido', 'proyecto'], a.get('tipo')
    c, r = post('/api/raiz/tipo', {'raiz': RAIZ, 'tipo': 'contenido'})
    assert c == 200, r
    e = get('/api/estado')
    assert e['config']['clientes']['Cliente Smoke']['tipo'] == 'contenido' and raiz_estado()['tipo'] == 'contenido'
    # y la ficha se queda, hecha, con lo que se eligió y cuándo: no desaparece
    t = {x['clave']: x for x in get('/api/carpeta/acciones' + q(raiz=RAIZ))['acciones']}['tipo']
    assert t['estado'] == 'hecho' and 'Piezas mes a mes' in (t.get('hecho') or ''), t
    # buscar en todos los clientes, y las fechas raras (aquí no hay ninguna)
    b = get('/api/buscar' + q(q='reel botox'))
    assert b['total'] >= 1 and all('reel botox' in x['nombre'].lower() for x in b['resultados']), b
    rl = get('/api/relojes' + q(raiz=RAIZ))
    assert rl['grupos'] == [] and rl['corregidos'] == [], rl
    return '«Qué pesa» entra hasta %d niveles bajo 2026; el tipo de la carpeta y el del cliente se igualan; buscar encuentra %d' % (d2['niveles'], b['total'])


def p_nombrar():
    v = get('/api/vocabulario' + q(raiz=RAIZ))
    toks = {t['token']: t for t in v['tokens']}
    assert 'raro' in toks and not toks['raro'].get('cubeta'), 'debería salir «Raro» como no reconocida'
    c, r = post('/api/diccionario', {'token': 'raro', 'cubeta': 'DOCUMENTOS'})
    assert c == 200, r
    v2 = get('/api/vocabulario' + q(raiz=RAIZ))
    assert {t['token']: t for t in v2['tokens']}['raro'].get('cubeta') == 'DOCUMENTOS'
    c, r = post('/api/diccionario', {'token': 'raro', 'cubeta': None})     # y se olvida
    # las carpetas de la propia app no se enseñan: «basura» es su _Para borrar\basura, no una del cliente
    c, r = post('/api/diccionario', {'token': 'Basura', 'cubeta': 'GRABACION'})
    assert c == 400 and 'propia app' in r.get('error', ''), (c, r)
    return 'reconoce, aprende y olvida un nombre de carpeta; no deja enseñar las de la app'


PLAN = {}


def p_plan():
    pl = get('/api/plan' + q(raiz=RAIZ))
    PLAN.update(pl)
    filas = pl['filas']
    assert filas, 'plan vacío'
    cub = {f['cubeta'] for f in filas}
    for c in ('GRABACION', 'BASURA', 'PROXY'):
        assert c in cub, 'el plan no propone nada de ' + c
    mes = [f for f in filas if f['accion'] == 'renombrar' and '10- Octubre' in f['destino']]
    assert mes, 'no propone normalizar «10- octubre»'
    dias = {f['destino_rel'].split(os.sep)[3] for f in filas if f['cubeta'] == 'GRABACION' and f['destino_rel'].count(os.sep) >= 3}
    assert any(d.startswith(('05-', '12-')) for d in dias), 'marzo tuvo dos rodajes y no sale carpeta de día: %s' % dias
    return '%d filas · %d marcadas · reparte marzo por día de rodaje' % (len(filas), sum(1 for f in filas if f['marcado']))


LOTES = {}


def acciones_quietas():
    """La lista de Inicio cuando el programa ha terminado de volver a pensar el plan tras mover algo."""
    assert esperar(lambda: not get('/api/carpeta/acciones' + q(raiz=RAIZ))['calculando'], 60), 'el plan no se volvió a pensar'
    a = get('/api/carpeta/acciones' + q(raiz=RAIZ))
    return a, {x['clave']: x for x in a['acciones']}


def p_aplicar():
    antes = arbol()
    c, r = post('/api/aplicar', {'raiz': RAIZ, 'ids': [f['id'] for f in PLAN['filas'] if f['marcado']]})
    assert c == 200, r
    assert not r['errores'], r['errores']
    assert r['movimientos'] > 0, 'no movió nada'
    LOTES['orden'] = r['lote']
    av = get('/api/aplicando')
    assert av['lote'] == r['lote'] and av['fin'] and av['unidades'] == av['unidades_total'] > 0 and av['movidos'] == r['movimientos'], av
    despues = arbol()
    assert sorted(os.path.basename(x) for x in antes) == sorted(os.path.basename(x) for x in despues), 'se perdió algún archivo'
    assert '10- Octubre' in os.listdir(os.path.join(RAIZ, '2026')), 'el mes sigue en minúsculas: %s' % os.listdir(os.path.join(RAIZ, '2026'))
    assert not any('thumbs.db' in x and '_para borrar' not in x for x in despues), 'la basura sigue fuera'
    # Inicio, sin que nadie vuelva a calcular: lo hecho sale como hecho (y cuándo), no como pendiente
    a, por = acciones_quietas()
    assert por['resto']['estado'] == 'hecho' and por['resto']['detalle'].startswith('Hecho hoy'), por['resto']
    assert por['fechas']['estado'] != 'toca', por['fechas']
    assert por['limpiar']['estado'] == 'hecho' and 'ya en' in por['limpiar']['detalle'], por['limpiar']
    assert a['ultimo'] and a['ultimo']['lote'] == r['lote'], a['ultimo']
    e = get('/api/carpeta/estado' + q(raiz=RAIZ))
    assert e['plan'] and e['plan']['marcadas'] == 0, e['plan']
    fr = e['diagnostico']['frases']
    assert any('ya apartados' in f for f in fr) and not any('de basura de sistema' in f for f in fr), fr
    return '%d movimientos, sin errores, «10- octubre» pasó a «10- Octubre»; el avance acabó en %d de %d; Inicio lo da por hecho' % (r['movimientos'], av['unidades'], av['unidades_total'])


def p_deshacer():
    c, r = post('/api/deshacer', {'lote': LOTES['orden']})
    assert c == 200 and not r['errores'], r
    assert '10- octubre' in os.listdir(os.path.join(RAIZ, '2026')), 'deshacer no devolvió las minúsculas: %s' % os.listdir(os.path.join(RAIZ, '2026'))
    assert os.path.isfile(os.path.join(RAIZ, '2026', '03- Marzo', 'Grabacion', 'Thumbs.db')), 'la basura no volvió'
    # y en Inicio vuelve a estar todo por hacer (antes se quedaba con «lo seguro ya está hecho»)
    a, por = acciones_quietas()
    assert all(por[k]['estado'] == 'toca' for k in ('limpiar', 'fechas', 'resto')), {k: por[k]['estado'] for k in ('limpiar', 'fechas', 'resto')}
    assert not a['ultimo'], a['ultimo']
    return '%d movimientos deshechos, e Inicio vuelve a ofrecerlo todo' % r['deshechos']


def p_reaplicar():
    pl = get('/api/plan' + q(raiz=RAIZ))
    c, r = post('/api/aplicar', {'raiz': RAIZ, 'ids': [f['id'] for f in pl['filas'] if f['marcado']]})
    assert c == 200 and not r['errores'], r
    LOTES['orden2'] = r['lote']
    return 'vuelve a aplicar sin errores'


def p_historial():
    b = get('/api/bitacora' + q(raiz=RAIZ))
    lotes = b.get('lotes') or b
    assert any(LOTES['orden2'] in json.dumps(l) for l in lotes), 'el lote aplicado no sale en el historial'
    p = get('/api/papelera' + q(raiz=RAIZ))
    assert any(x['nombre'] == 'basura' and x['n'] >= 1 for x in p['subcarpetas']), p
    return 'historial y _Para borrar reflejan lo hecho'


def p_repetidos():
    d = get('/api/duplicados' + q(raiz=RAIZ))
    assert d['grupos'], 'no ve el Reel botox repetido'
    c, r = post('/api/duplicados/apartar', {'raiz': RAIZ})
    assert c == 200 and r['movimientos'] >= 1, r
    return '%d grupo(s), %d copia(s) apartada(s) a Duplicados' % (len(d['grupos']), r['movimientos'])


def p_stock():
    s = get('/api/stock' + q(raiz=RAIZ, meses=3, tipos='Video'))
    assert s['grupos'], 'no ve el stock de Envato viejo'
    rutas = [x for g in s['grupos'] for x in g['rutas']]
    c, r = post('/api/stock/papelera', {'raiz': RAIZ, 'rutas': rutas})
    assert c == 200 and r['movimientos'] == len(rutas), r
    return '%d archivo(s) de stock apartados' % r['movimientos']


def p_vacias():
    v = get('/api/vacias' + q(raiz=RAIZ))
    assert any(x['rel'] == 'Vacia' for x in v['carpetas']), v
    c, r = post('/api/vacias/borrar', {'raiz': RAIZ, 'rutas': [x['ruta'] for x in v['carpetas']]})
    assert c == 200 and r['borradas'] >= 2, r
    c, d = post('/api/deshacer', {'lote': r['lote']})
    assert os.path.isdir(os.path.join(RAIZ, 'Vacia', 'dentro')), 'deshacer no recreó la carpeta'
    post('/api/vacias/borrar', {'raiz': RAIZ, 'rutas': [x['ruta'] for x in v['carpetas']]})
    return '%d huecos borrados, deshechos y vueltos a borrar' % r['borradas']


def p_mapa():
    e = get('/api/estructura' + q(raiz=RAIZ))
    assert e['cubetas'] and e['arbol']['hijos'], 'mapa vacío'
    a = get('/api/estructura/archivo' + q(nombre='Proyecto.aep', tipo='contenido'))
    assert a['destino'].endswith('After Effects'), a
    ex = get('/api/estructura/existen' + q(raiz=RAIZ, ruta=json.dumps(['{anio}', '{mes}', '00_Grabacion'])))
    assert ex['n'] >= 1, ex
    c, r = post('/api/estructura/renombrar', {'tipo': 'contenido', 'ruta': ['{anio}', '{mes}', '00_Grabacion'], 'nombre': 'Grabacion', 'raiz': RAIZ, 'en_disco': True})
    assert c == 200 and r.get('ok') and r['movidas'] >= 1, r
    c, r2 = post('/api/estructura/renombrar', {'tipo': 'contenido', 'ruta': ['{anio}', '{mes}', 'Grabacion'], 'nombre': '00_Grabacion', 'raiz': RAIZ, 'en_disco': True})
    assert r2.get('ok'), r2
    c, r3 = post('/api/plantillas', {'plantillas': {'contenido': {'MUSICA': '_Recursos/Musica/{sub}'}}})
    assert c == 200
    return 'árbol, dónde iría, existen, renombrar ida y vuelta, guardar plantilla'


def p_colores():
    c, r = post('/api/colores/pintar', {'raiz': RAIZ})
    assert c == 200, r
    if r.get('error'):
        return 'este disco no lo permite (' + r['error'][:50] + ')'
    assert r['pintadas'] >= 3, r
    c, qq = post('/api/colores/quitar', {'raiz': RAIZ})
    assert qq['quitadas'] == r['pintadas'], (r, qq)
    return '%d carpetas pintadas y despintadas' % r['pintadas']


IDS = {}


def p_motores():
    m = get('/api/optimizar/motores')
    assert m.get('ffmpeg'), 'sin ffmpeg'
    disp = [k for k, v in m['formatos'].items() if v]
    assert 'h265_10bit' in disp or 'h264' in disp, disp
    assert 'ok' in (m.get('braw') or {}), m.get('braw')
    assert get('/api/estado')['herramientas']['braw']['ok'] in (True, False)
    return 'formatos disponibles: ' + ', '.join(disp) + ' · GPU: ' + (m['gpu'].get('nombre') or 'no') + \
        ' · lector de BRAW: ' + ('sí' if m['braw']['ok'] else 'no (' + (m['braw'].get('motivo') or '')[:40] + ')')


def p_analizar():
    post('/api/config', {'optimizar': {'meses_minimos': 0, 'min_mb': 0}})
    c, r = post('/api/optimizar/analizar', {'raiz': RAIZ, 'min_mb': 0})
    assert c == 200, r
    assert esperar(lambda: ((raiz_estado() or {}).get('analisis') or {}).get('estado') not in (None, 'corriendo'), 120)
    cand = get('/api/optimizar/candidatos' + q(raiz=RAIZ, meses=0, min_mb=0, formato='h265_10bit'))['candidatos']
    clip = [x for x in cand if x['ruta'].lower().endswith('c0020.mp4')]
    assert clip and not clip[0]['excluido'], (clip or cand)[:1]
    IDS['ruta'] = clip[0]['ruta']
    comprimido = [x for x in cand if x['ruta'].lower().endswith('c0021.mp4')]
    assert comprimido and comprimido[0]['excluido'].startswith('convertido ahorraría solo un'), (comprimido or cand)[:1]
    return 'leyó %d clip(s) · C0020.MP4 %s %s a %d kb/s: se ofrece · C0021.MP4 a %d kb/s: no' % (
        len(cand), clip[0]['codec'], clip[0]['res'], clip[0]['kbps'], comprimido[0]['kbps'])


def p_convertir():
    m = get('/api/optimizar/motores')
    fmt = 'h265_10bit' if m['formatos'].get('h265_10bit') else 'h264'
    c, r = post('/api/optimizar/encolar', {'raiz': RAIZ, 'rutas': [IDS['ruta']], 'perfil': 'equilibrado', 'formato': fmt, 'chroma': '420', 'color': 'rec709', 'iniciar': True})
    assert c == 200 and r['ids'], r
    IDS['trabajo'] = r['ids'][0]
    assert esperar(lambda: not get('/api/estado')['cola']['activa'], 240), 'la cola no terminó'
    cola = get('/api/optimizar/cola' + q(raiz=RAIZ))
    t = next(x for x in cola['trabajos'] if x['id'] == IDS['trabajo'])
    assert t['estado'] in ('listo', 'verificado'), (t['estado'], t.get('log', '')[-200:])
    c, v = post('/api/optimizar/verificar', {'ids': [IDS['trabajo']]})
    assert v['resultados'][0]['ok'], v
    c, rr = post('/api/optimizar/reemplazar', {'ids': [IDS['trabajo']]})
    assert rr['resultados'][0]['ok'], rr
    assert os.path.isfile(t['salida']) and not os.path.isfile(IDS['ruta'])
    return '%s: convertido (%s), verificado y reemplazado; original apartado' % (fmt, t['motor'])


def p_comparar():
    # el original ya se reemplazó (está en _Para borrar/originales): el cuadro tiene que salir de ahí
    for cual in ('original', 'convertido'):
        r = urllib.request.urlopen(BASE + '/api/optimizar/cuadro' + q(id=IDS['trabajo'], cual=cual, t=1.5), timeout=120)
        assert r.headers.get('Content-Type') == 'image/jpeg' and len(r.read()) > 2000, 'no salió el cuadro %s' % cual
    c, r = post('/api/optimizar/calidad', {'id': IDS['trabajo']})
    assert c == 200 and 30 < r['vmaf'] <= 100 and r['cuadros'] > 10, r
    t = next(x for x in get('/api/optimizar/cola')['trabajos'] if x['id'] == IDS['trabajo'])
    assert t['nota'] == r['vmaf'] and t.get('dur'), t
    return 'dos cuadros JPEG y nota VMAF %s sobre %d cuadros (%.0f s)' % (r['vmaf'], r['cuadros'], r['seg'])


def p_tanda():
    # al acabar la cola queda el resumen de la tanda: clips, pesos, tiempo; y se exporta en CSV
    u = get('/api/optimizar/cola')['cola']['ultima']
    assert u and u['clips'] >= 1 and u['bytes_in'] > u['bytes_out'] > 0 and u['lote'].startswith('tanda-'), u
    assert IDS['trabajo'] in u['ids'] and u['raiz'].lower() == RAIZ.lower(), u
    r = urllib.request.urlopen(BASE + '/api/optimizar/tanda.csv' + q(lote=u['lote']), timeout=60)
    csv_ = r.read().decode('utf-8-sig')
    assert r.headers.get('Content-Type', '').startswith('text/csv') and csv_.startswith('clip;carpeta;estado') and 'reemplazado' in csv_, csv_[:200]
    lotes = get('/api/bitacora')['lotes']
    l = next((x for x in lotes if x['lote'] == u['lote']), None)
    assert l and l['tanda']['clips'] == u['clips'], l
    # el horario: fuera de él la cola espera (se comprueba la regla, no esperando de verdad)
    c, r2 = post('/api/config', {'optimizar': {'horario': {'activo': True, 'desde': '20:00', 'hasta': '07:00'}, 'pausar_si_edito': False, 'avisar': False}})
    assert c == 200 and r2['config']['optimizar']['horario']['desde'] == '20:00'
    c, r3 = post('/api/config', {'optimizar': {'horario': {'activo': False, 'desde': '20:00', 'hasta': '07:00'}}})
    assert c == 200 and not r3['config']['optimizar']['horario']['activo']
    return 'resumen de %d clip(s), %s, CSV con %d líneas, y en Historial' % (u['clips'], u['lote'], csv_.count('\n'))


def p_braw():
    m = get('/api/optimizar/motores')['braw']
    if not BRAW:
        return 'sin clip de prueba (ORGANILAB_BRAW_PRUEBA): solo se comprueba que el lector responde (%s)' % ('sí' if m['ok'] else 'no')
    assert m['ok'], m
    cand = get('/api/optimizar/candidatos' + q(raiz=RAIZ, meses=0, min_mb=0, formato='h265_10bit'))['candidatos']
    clip = [x for x in cand if x['ruta'].lower().endswith('a001_03051200_c003.braw')]
    assert clip and clip[0]['motor'] == 'braw' and not clip[0]['excluido'], (clip or cand)[:1]
    c, r = post('/api/optimizar/encolar', {'raiz': RAIZ, 'rutas': [clip[0]['ruta']], 'perfil': 'equilibrado', 'formato': 'h265_10bit', 'chroma': '420', 'color': 'rec709', 'iniciar': True})
    assert c == 200 and r['ids'], r
    assert esperar(lambda: not get('/api/estado')['cola']['activa'], 240), 'la cola no terminó'
    t = next(x for x in get('/api/optimizar/cola' + q(raiz=RAIZ))['trabajos'] if x['id'] == r['ids'][0])
    assert t['estado'] == 'verificado' and t['motor'] == 'braw', (t['estado'], t['motor'], (t.get('log') or '')[-300:])
    assert os.path.isfile(t['salida']) and os.path.isfile(clip[0]['ruta'])           # el original sigue en su sitio
    c, rr = post('/api/optimizar/reemplazar', {'ids': r['ids']})
    assert rr['resultados'][0]['ok'], rr
    assert os.path.isfile(t['salida']) and not os.path.isfile(clip[0]['ruta'])
    return 'BRAW %s leído sin Resolve, convertido (%s), verificado y reemplazado' % (clip[0]['res'], (t.get('log') or '').split(' · ')[3][:40])


def p_cola_control():
    c, r = post('/api/optimizar/control', {'accion': 'quitar', 'ids': [IDS['trabajo']]})
    assert c == 200
    return 'control de cola responde'


def p_respaldo():
    pl = get('/api/respaldo/plan' + q(raiz=RAIZ, destino=RESP, por_tipo=1, sin_stock=1, incluir_raw=0, meses=0))
    assert pl.get('n', 0) > 0, pl
    c, r = post('/api/respaldo/copiar', {'raiz': RAIZ})
    assert c == 200, r
    assert esperar(lambda: get('/api/respaldo/estado')['estado'] != 'corriendo', 120)
    e = get('/api/respaldo/estado')
    assert e['hechos'] == pl['n'], (e['hechos'], pl['n'])
    assert os.path.isfile(os.path.join(RESP, 'Cliente Smoke', 'INDICE.csv')), 'falta INDICE.csv'
    return '%d archivos copiados a %s · INDICE.csv' % (e['hechos'], os.path.basename(RESP))


def p_asistente():
    pz = get('/api/asistente/pasos' + q(raiz=RAIZ))
    assert 'pasos' in pz and 'diagnostico' in pz, list(pz)
    c, r = post('/api/acciones/hacer', {'raiz': RAIZ})
    assert c == 200, r
    assert esperar(lambda: not get('/api/estado')['asistente'].get('activo'), 120)
    a = get('/api/estado')['asistente']
    assert not a.get('error'), a.get('error')
    # después, Inicio no vuelve a ofrecer lo que acaba de hacer
    _, por = acciones_quietas()
    assert por['fechas']['estado'] != 'toca' and por['resto']['estado'] != 'toca', {k: por[k] for k in ('fechas', 'resto')}
    return 'pasos + «hacer las N de una vez» terminan sin error, e Inicio no lo vuelve a ofrecer'


def p_pieza():
    hoy = time.localtime()
    v = get('/api/pieza/previa' + q(raiz=RAIZ, titulo='Reel de prueba', anio=hoy.tm_year, mes=hoy.tm_mon))
    assert v['rel'].endswith('Reel de prueba') and not v['existe'], v
    assert not os.path.exists(v['ruta']), 'la vista previa creó la carpeta'
    c, r = post('/api/pieza', {'raiz': RAIZ, 'titulo': 'Reel de prueba', 'anio': hoy.tm_year, 'mes': hoy.tm_mon})
    assert c == 200 and r.get('ok'), r
    assert sorted(os.listdir(r['ruta'])) == sorted(r['subcarpetas']) and len(r['subcarpetas']) >= 5, os.listdir(r['ruta'])
    c2, r2 = post('/api/pieza', {'raiz': RAIZ, 'titulo': 'reel de prueba', 'anio': hoy.tm_year, 'mes': hoy.tm_mon})
    assert c2 == 400 and 'ya existe' in r2.get('error', ''), (c2, r2)
    vac = [x['rel'] for x in get('/api/vacias' + q(raiz=RAIZ))['carpetas']]
    assert not any('Reel de prueba' in x for x in vac), 'ofrece borrar una pieza recién creada: %s' % vac
    c3, r3 = post('/api/deshacer', {'lote': r['lote']})
    assert c3 == 200 and not r3['errores'] and not os.path.exists(r['ruta']), r3
    return 'crea «%s» con %d subcarpetas, no repite, y se deshace' % (r['nombre'], len(r['subcarpetas']))


def p_entrada():
    tarjeta = os.path.join(TMP, 'tarjeta')
    clip = os.path.join(tarjeta, 'DCIM', '100MEDIA', 'DJI_0007.MP4')
    os.makedirs(os.path.dirname(clip))
    with open(clip, 'wb') as fh:
        fh.write(os.urandom(700_000))
    t = time.mktime(time.struct_time((2026, 8, 10, 12, 0, 0, 0, 0, -1)))
    os.utime(clip, (t, t))
    c, p = post('/api/entrada/plan', {'raiz': RAIZ, 'origen': tarjeta})
    assert c == 200 and p['n'] == 1 and p['bytes'] == 700_000, p
    c, mal = post('/api/entrada/plan', {'raiz': RAIZ, 'origen': os.path.join(RAIZ, '2026')})
    assert c == 400 and mal.get('error'), mal
    c, r = post('/api/entrada/traer', {'raiz': RAIZ, 'origen': tarjeta})
    assert c == 200 and r.get('lanzado'), r
    assert esperar(lambda: get('/api/estado')['traer'].get('estado') not in ('copiando', 'ordenando'), 120)
    tr = get('/api/estado')['traer']
    assert tr['estado'] == 'listo' and tr['copiados'] == 1 and not tr['errores_n'], tr
    final = os.path.join(RAIZ, '2026', '08- Agosto', '00_Grabacion', '10- Lunes', 'DJI', 'DJI_0007.MP4')
    assert os.path.isfile(final), 'no llegó a su sitio: %s' % [x for x in arbol() if 'dji' in x]
    assert os.path.getsize(final) == 700_000 and abs(os.path.getmtime(final) - t) < 2, 'la copia no conserva peso o fecha'
    assert os.path.isfile(clip), 'tocó la tarjeta'
    e = get('/api/entrada' + q(raiz=RAIZ))
    assert e['n'] == 0 and e['existe'], e
    # y lo que se deja a mano en la bandeja sale en la lista y se reparte con su botón
    suelto = os.path.join(RAIZ, '_Entrada', 'C0099.MP4')
    with open(suelto, 'wb') as fh:
        fh.write(os.urandom(4096))
    os.utime(suelto, (t, t))
    a = get('/api/carpeta/acciones' + q(raiz=RAIZ))
    assert any(x['clave'] == 'entrada' and x['estado'] == 'toca' for x in a['acciones']), [x['clave'] for x in a['acciones']]
    c, o = post('/api/entrada/ordenar', {'raiz': RAIZ})
    assert c == 200 and o['movimientos'] >= 1 and not o['errores'] and o['quedan'] == 0, o
    assert os.path.isfile(os.path.join(RAIZ, '2026', '08- Agosto', '00_Grabacion', '10- Lunes', 'SONY', 'C0099.MP4')), [x for x in arbol() if 'c0099' in x]
    return 'trae de una tarjeta con copia comprobada y la reparte; lo dejado a mano en la bandeja también'


def p_automatico():
    c, r = post('/api/automatico/config', {'activo': True, 'hora': '03:15'})
    assert c == 200 and r['activo'] and r['hora'] == '03:15', r
    tarea = json.load(open(os.path.join(ENSAYO, 'tarea.json'), encoding='utf-8'))
    assert tarea['activo'] and tarea['orden'].endswith(' automatico'), tarea
    c, mal = post('/api/automatico/config', {'activo': True, 'hora': 'a las tres'})
    assert c == 400, mal
    c, r = post('/api/automatico/ahora', {})
    assert c == 200 and r.get('lanzado'), r
    assert esperar(lambda: not get('/api/estado')['automatico'].get('activo'), 180)
    u = get('/api/automatico')['ultimo']
    assert u and len(u['carpetas']) == 1 and u['carpetas'][0]['estado'] in ('al dia', 'con dudas'), u
    assert not u['carpetas'][0]['errores'], u['carpetas'][0]
    # lo que lanza la tarea de Windows cuando el programa ya está abierto: se lo pide a él y se va, para
    # que no haya dos procesos moviendo lo mismo
    r = subprocess.run([sys.executable, 'organizador.py', 'automatico'], cwd=APP, env=env, timeout=120)
    assert r.returncode == 0, 'la orden «automatico» salió con %s' % r.returncode
    assert esperar(lambda: not get('/api/estado')['automatico'].get('activo'), 180)
    u2 = get('/api/automatico')['ultimo']
    assert u2['origen'] == 'programado' and u2['ts'] > u['ts'], u2
    c, r = post('/api/automatico/config', {'activo': False, 'hora': '03:15'})
    assert c == 200 and not r['activo'], r
    return 'se enciende (tarea de Windows en ensayo), hace una pasada y deja el resumen (%s); con el programa abierto, la tarea se lo pide a él' % u['carpetas'][0]['estado']


def p_soltar_y_buscar():
    s = get('/api/soltado' + q(nombre='nada'))
    assert s['ruta'] is None
    b = get('/api/buscar-carpeta' + q(nombre='Cliente Smoke'))
    assert isinstance(b['coincidencias'], list)
    return 'puente de arrastre responde; búsqueda de carpeta responde'


def p_actualizacion_offline():
    # con una URL imposible el programa no puede quedarse colgado ni romper nada
    a = get('/api/actualizacion')
    assert a['version']
    c, r = post('/api/config', {'actualizaciones': {'auto': False}})
    assert c == 200 and r['config']['actualizaciones']['auto'] is False, r
    return 'estado legible; el interruptor de arranque se guarda'


def p_clientes_config():
    e = get('/api/estado')
    cl = dict(e['config']['clientes'])
    cl['Cliente Smoke'] = {'tipo': 'contenido', 'alias': ['smoke']}
    c, r = post('/api/clientes', {'clientes': cl})
    assert c == 200
    c, r = post('/api/regla', {'cliente': 'Cliente Smoke', 'origen': 'Sueltos', 'destino': '_Revisar/Sueltos', 'cubeta': 'REVISAR'})
    assert c == 200
    return 'clientes y reglas se guardan'


def p_quitar():
    c, r = post('/api/raiz/quitar', {'raiz': RAIZ, 'datos': True})
    assert c == 200, r
    assert get('/api/estado')['raices'] == []
    return 'desconectada con sus datos'


for nombre, fn in (('Interfaz sin nada externo', p_estatico), ('Registrar carpeta', p_registrar), ('Escanear', p_escanear),
                   ('Estado y lista de acciones', p_carpeta), ('Qué hay (mapa de peso, discos)', p_que_hay),
                   ('Qué pesa por dentro y tipo de trabajo', p_tipo_y_pesa), ('Por nombrar (vocabulario, diccionario)', p_nombrar), ('Calcular el plan', p_plan),
                   ('Aplicar el plan', p_aplicar), ('Deshacer', p_deshacer), ('Volver a aplicar', p_reaplicar),
                   ('Historial y papelera', p_historial), ('Repetidos', p_repetidos), ('Stock viejo', p_stock),
                   ('Carpetas vacías', p_vacias), ('Mapa interactivo', p_mapa), ('Colores en Windows', p_colores),
                   ('Motores de conversión', p_motores), ('Leer clips (ffprobe)', p_analizar),
                   ('Convertir, verificar, reemplazar', p_convertir), ('Comparar original y convertido', p_comparar), ('Resumen de la tanda', p_tanda), ('Convertir un BRAW', p_braw), ('Control de la cola', p_cola_control),
                   ('Respaldar', p_respaldo), ('Asistente / hacer de una vez', p_asistente),
                   ('Pieza nueva', p_pieza), ('Bandeja de entrada y tarjeta', p_entrada), ('Orden automático', p_automatico),
                   ('Arrastrar y buscar carpeta', p_soltar_y_buscar), ('Actualizaciones', p_actualizacion_offline),
                   ('Clientes, reglas', p_clientes_config), ('Quitar de la lista', p_quitar)):
    prueba(nombre, fn)

proc.kill()
proc.wait()
log.close()
print('%-40s %-6s %-8s %s' % ('FUNCIÓN', 'ESTADO', 'TIEMPO', 'DETALLE'))
print('-' * 120)
for n, e, t, d in FILAS:
    print('%-40s %-6s %-8s %s' % (n, e, t, d))
malas = [f for f in FILAS if f[1] != 'OK']
print('-' * 120)
print('%d de %d funciones OK' % (len(FILAS) - len(malas), len(FILAS)))
if malas:
    print()
    print('== últimas líneas del servidor ==')
    print(open(os.path.join(TMP, 'servidor.log'), encoding='utf-8', errors='replace').read()[-1500:])
shutil.rmtree(TMP, ignore_errors=True)
sys.exit(1 if malas else 0)          # la compilación del estudio se detiene si algo falla