Labstream/labstream-organizador
1
0
Fork 0
Código Incidencias Pull Requests Paquetes Proyectos Lanzamientos Wiki Actividad
Files
c91473aea422bf073cd9898da91e8435d51da789
labstream-organizador/prueba_orden.py
Labstream 8ef3af6ad7 Inventario fiel al disco: mayúsculas de las carpetas, mover durante el escaneo, y no convertir dos veces 
Cotejado archivo por archivo cada disco con el inventario: el escaneo estaba completo, pero el inventario
apuntaba la carpeta con las mayúsculas del plan (1.039 fotos en Vía Sumapaz, tres carpetas donde hay una), un
proxy apartado en pleno escaneo quedó dos veces, y los 11 verificados de Línea Gene se habían vuelto a encolar
con la 3.7. Ahora la carpeta se escribe como en el disco y se recuenta donde quedó al final, el escaneo vuelve a
recorrer si la app movió algo mientras, y al reenganchar se quita la conversión que sobra.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-14 14:48:55 -05:00

1394 líneas
90 KiB
Python
Original Blame Histórico

Este archivo contiene caracteres Unicode ambiguos

Este archivo contiene caracteres Unicode que pueden confundirse con otros caracteres. Si crees que esto es intencional, puedes ignorar esta advertencia. Usa el botón de Escape para revelarlos.

# -*- coding: utf-8 -*-
"""Pruebas del motor de orden. Cada caso reproduce en pequeño algo medido en una carpeta de verdad (la
de Andrea Galeano, el 4-oct-2026): monta un cliente de ensayo en una carpeta temporal, lo escanea, pide
el plan y, donde importa, lo aplica y mira cómo queda el disco.
py -3.12 prueba_orden.py
"""
import os
import sys
import json
import time
import shutil
import tempfile
import threading
import unittest
AQUI = os.path.dirname(os.path.abspath(__file__))
TMP = tempfile.mkdtemp(prefix="organilab-motor-")
APP = os.path.join(TMP, "app")
os.makedirs(os.path.join(APP, "datos"))
for _f in ("organizador.py", "reglas.py", "asistente.py", "actualizador.py", "firma.py", "instalar.py", "ventana.py",
"resolve_bridge.py", "braw.py"):
shutil.copy(os.path.join(AQUI, _f), APP)
# un diccionario como el que dejaba la 3.1.1: con una carpeta de la propia app aprendida como grabación
with open(os.path.join(APP, "datos", "diccionario.json"), "w", encoding="utf-8") as _fh:
json.dump({"carpetas": {"basura": "GRABACION", "mi carpeta rara": "FOTOS"}, "reglas": [], "aprendido": []}, _fh)
sys.path.insert(0, APP)
import organizador as O # noqa: E402 (tiene que importarse desde la copia: sus datos van junto a él)
import asistente # noqa: E402
R = O.R
S = os.sep
def cliente(nombre, tipo="contenido"):
raiz = os.path.join(TMP, nombre)
os.makedirs(raiz)
O.CONFIG["raices"].append({"ruta": raiz, "cliente": nombre, "tipo": tipo})
return raiz
def mk(raiz, rel, y, m, d, tam=2048):
"""Crea un archivo con esa fecha de modificación (a mediodía)."""
p = os.path.join(raiz, rel.replace("/", S))
os.makedirs(os.path.dirname(p), exist_ok=True)
with open(p, "wb") as fh:
fh.write(os.urandom(tam))
t = time.mktime(time.struct_time((y, m, d, 12, 0, 0, 0, 0, -1)))
os.utime(p, (t, t))
return p
def plan(raiz):
O.escanear(raiz)
prog = O.PROGRESO[os.path.normpath(raiz)]
assert prog["estado"] == "listo", prog
return O.Plan(O.raiz_cfg(raiz)).correr()
def aplicar(raiz, p):
r = O.aplicar(raiz, [f for f in p["filas"] if f["marcado"]])
assert not r["errores"], r["errores"]
return r
def arbol(raiz):
return sorted(os.path.relpath(os.path.join(b, f), raiz) for b, _, fs in os.walk(raiz) for f in fs)
def nombres(raiz):
return sorted(os.path.basename(x) for x in arbol(raiz))
def rel(*partes):
return S.join(partes)
class Reglas(unittest.TestCase):
def test_un_clip_convertido_sigue_siendo_de_su_camara(self):
"""166 clips ya convertidos habían perdido su cámara: pasaban por exportados."""
self.assertEqual(R.firma_archivo("C0001_10bit.mp4"), "SONY")
self.assertEqual(R.firma_archivo("20251111_Labstream5320_10bit.mp4"), "SONY")
self.assertEqual(R.firma_archivo("A002_03110416_C042_10bit_10bit.mp4"), "BM_PROXY")
self.assertEqual(R.firma_archivo("A006_06010242_C114 (2)_10bit.mp4"), "BM_PROXY")
self.assertIsNone(R.firma_archivo("INMODE_10bit.mp4")) # no era de cámara: sigue sin serlo
self.assertEqual(R.rol_archivo("C0001_10bit.mp4", ".mp4", R.firma_archivo("C0001_10bit.mp4")), "camara_video")
self.assertTrue(R.es_conversion("C0001_10bit.mp4"))
self.assertTrue(R.es_conversion("C0001_10bit_10bit.mp4"))
self.assertFalse(R.es_conversion("C0001.MP4"))
self.assertEqual(R.sin_conversion("C0001_10bit_10bit.mp4"), "C0001.mp4")
self.assertEqual(R.raiz_derivado("C0001_10bit_10bit.mp4"), "C0001")
self.assertIsNone(R.raiz_derivado("C0001.MP4"))
def test_lo_aprendido_vale_por_el_nombre_entero(self):
"""Confirmar que «Videos» era una grabación convertía en grabación a «videos ia» y a «01- Orden»."""
d = dict(R.DICCIONARIO_SEMILLA, videos="GRABACION", orden="GRABACION")
self.assertEqual(R.cubeta_carpeta("Videos", d)[0], "GRABACION")
self.assertEqual(R.cubeta_carpeta("videos ia", d)[0], "IA") # la palabra de fábrica sí vale
self.assertIsNone(R.cubeta_carpeta("01- Orden", d)[0])
self.assertIsNone(R.cubeta_carpeta("Orden nuevo de cosas", d)[0])
self.assertEqual(R.cubeta_carpeta("04_Revisar-Borrar", R.DICCIONARIO_SEMILLA)[0], "REVISAR")
def test_fecha_de_blackmagic(self):
"""El nombre trae mes y día. Si el archivo no se ha tocado desde el rodaje, el año es seguro."""
intacto = time.mktime(time.struct_time((2025, 12, 9, 18, 30, 0, 0, 0, -1)))
reescrito = time.mktime(time.struct_time((2026, 7, 1, 10, 0, 0, 0, 0, -1)))
self.assertEqual(R.fecha_de_nombre("A014_12091814_C001.braw", intacto), ("2025-12-09", "nombre"))
self.assertEqual(R.fecha_de_nombre("A014_12091814_C001.mp4", reescrito), ("2025-12-09", "nombre+mtime"))
self.assertEqual(R.clave_rodaje("A014_12091814_C001_10bit.mp4"), "A014_1209")
self.assertIsNone(R.clave_rodaje("C0001.MP4"))
def test_carpetas_de_tarjeta(self):
for n in ("DCIM", "100MEDIA", "PRIVATE", "M4ROOT", "CLIP", "A001_0305XY", "100MSDCF"):
self.assertTrue(R.es_carpeta_de_tarjeta(n), n)
for n in ("Tomas consultorio", "Entrevista", "Maquinas"):
self.assertFalse(R.es_carpeta_de_tarjeta(n), n)
def test_rodaje_de_una_misma_tarjeta(self):
"""Los clips de la misma tarjeta y día comparten año, y la madrugada cuenta como la noche anterior."""
def fila(nombre, fecha, origen):
return ("x" + S + nombre, "x", "x", nombre, ".mp4", 1, 0.0, "camara_video", "BM_PROXY", "", None, None,
fecha, origen, 0, 0, 0)
filas = [fila("A002_03020727_C001.mp4", "2025-03-02", "nombre+mtime"), # reescrito en dic-2025
fila("A002_03020727_C002.mp4", "2026-03-02", "nombre+mtime"), # reescrito en jul-2026
fila("A014_12091814_C001.braw", "2025-12-09", "nombre"), # intacto
fila("A014_12091814_C001.mp4", "2025-12-09", "nombre+mtime"),
fila("A006_05312107_C001.mp4", "2025-05-31", "nombre+mtime"), # empieza de noche…
fila("A006_06010000_C002.mp4", "2025-06-01", "nombre+mtime"), # …y sigue de madrugada
fila("A003_06010400_C019.mp4", "2025-06-01", "nombre+mtime"), # otra tarjeta, misma madrugada
fila("A009_06011500_C001.mp4", "2025-06-01", "nombre+mtime")] # esa tarde: otro rodaje
O._fechas_de_rodaje(filas)
por = {f[3]: (f[12], f[13]) for f in filas}
self.assertEqual(por["A002_03020727_C002.mp4"], ("2025-03-02", "nombre+mtime")) # el año más antiguo del grupo
self.assertEqual(por["A014_12091814_C001.mp4"], ("2025-12-09", "nombre")) # el año seguro se contagia
self.assertEqual(por["A006_06010000_C002.mp4"][0], "2025-05-31")
self.assertEqual(por["A003_06010400_C019.mp4"][0], "2025-05-31")
self.assertEqual(por["A009_06011500_C001.mp4"][0], "2025-06-01")
class Arranque(unittest.TestCase):
def test_se_olvida_lo_que_era_de_la_app(self):
"""«basura» se aprendió como grabación porque «Por nombrar» preguntaba por _Para borrar\\basura."""
self.assertNotIn("basura", O.DICC["carpetas"])
self.assertEqual(O.DICC["carpetas"].get("mi carpeta rara"), "FOTOS") # lo del cliente se conserva
with open(O.DICC_PATH, encoding="utf-8") as fh:
self.assertNotIn("basura", json.load(fh)["carpetas"])
class Motor(unittest.TestCase):
def aprende(self, token, cubeta):
O.DICC["carpetas"][token] = cubeta
self.addCleanup(lambda: O.DICC["carpetas"].pop(token, None))
def test_un_cajon_no_se_desarma(self):
"""«Orden» (una organización anterior entera) se aprendió como grabación y el plan la iba a repartir
por cámara y por mes, proyectos incluidos. Ahora se abre, y cada cosa va a su sitio."""
raiz = cliente("Cajon")
mk(raiz, "2026/05- Mayo/03- El cuello/portada vieja.png", 2026, 5, 10)
mk(raiz, "Orden/01_Grabaciones/Tomas consultorio/20251111_Labstream5320.MP4", 2025, 11, 11)
mk(raiz, "Orden/01_Grabaciones/Tomas consultorio/20251111_Labstream5321_10bit.mp4", 2026, 1, 10)
mk(raiz, "Orden/01_Grabaciones/Tomas consultorio/Duplicados/20251111_Labstream5322.MP4", 2025, 11, 11)
mk(raiz, "Orden/01_Grabaciones/Videos Maquinas/A006_05312107_C001.mp4", 2026, 1, 10)
mk(raiz, "Orden/01_Grabaciones/Videos Maquinas/A006_06010000_C002.mp4", 2026, 1, 10)
mk(raiz, "Orden/01_Grabaciones/Videos Maquinas/A006_06010030_C003.braw", 2025, 6, 1)
mk(raiz, "Orden/02_Proyectos/Mayo/03 - El cuello/Exportado/final.mp4", 2026, 5, 20)
mk(raiz, "Orden/02_Proyectos/Mayo/03 - El cuello/Portada/p.png", 2026, 5, 20)
mk(raiz, "Orden/02_Proyectos/Mayo/03 - El cuello/proyecto.prproj", 2026, 5, 20)
mk(raiz, "Orden/03_Recursos/logo.png", 2026, 5, 20)
self.aprende("orden", "GRABACION")
antes = nombres(raiz)
p = plan(raiz)
self.assertFalse([f for f in p["filas"] if f["origen_rel"] == "Orden"], "sigue queriendo mover «Orden» entera")
self.assertTrue(any("guarda dentro carpetas de otras fechas" in a for a in p["avisos"]), p["avisos"])
destinos = {f["destino_rel"] for f in p["filas"]}
self.assertIn(rel("2025", "11- Noviembre", "00_Grabacion", "11- Martes", "Tomas consultorio", "SONY"), destinos)
self.assertIn(rel("2025", "11- Noviembre", "00_Grabacion", "11- Martes", "Tomas consultorio", "SONY", "Duplicados"), destinos)
# el rodaje del 31 de mayo siguió de madrugada: todo a mayo de 2025, el año que da el .braw intacto
self.assertIn(rel("2025", "05- Mayo", "00_Grabacion", "31- Sábado", "Maquinas", "BLACKMAGIC_MP4"), destinos)
self.assertIn(rel("2025", "05- Mayo", "00_Grabacion", "31- Sábado", "Maquinas", "URSA_BRAW"), destinos)
self.assertFalse([d for d in destinos if "06- Junio" in d], destinos)
junta = [f for f in p["filas"] if f["accion"] == "renombrar" and f["origen_rel"].endswith("03 - El cuello")]
self.assertTrue(junta and junta[0]["marcado"] and "se junta" in junta[0]["motivo"], junta)
aplicar(raiz, p)
self.assertEqual(nombres(raiz), antes, "se perdió o se duplicó algún archivo")
for esperado in (rel("2025", "11- Noviembre", "00_Grabacion", "11- Martes", "Tomas consultorio", "SONY", "20251111_Labstream5320.MP4"),
rel("2025", "11- Noviembre", "00_Grabacion", "11- Martes", "Tomas consultorio", "SONY", "20251111_Labstream5321_10bit.mp4"),
rel("2025", "05- Mayo", "00_Grabacion", "31- Sábado", "Maquinas", "BLACKMAGIC_MP4", "A006_06010000_C002.mp4"),
rel("2025", "05- Mayo", "00_Grabacion", "31- Sábado", "Maquinas", "URSA_BRAW", "A006_06010030_C003.braw"),
rel("2026", "05- Mayo", "03- El cuello", "Exportado", "final.mp4"),
rel("2026", "05- Mayo", "03- El cuello", "proyecto.prproj"),
rel("2026", "05- Mayo", "03- El cuello", "portada vieja.png")):
self.assertTrue(os.path.isfile(os.path.join(raiz, esperado)), "falta " + esperado + "\n" + "\n".join(arbol(raiz)))
self.assertFalse(os.path.isdir(os.path.join(raiz, "2026", "05- Mayo", "03 - El cuello")), "quedó la pieza gemela")
# y con todo en su sitio, no queda nada que proponer sobre lo mismo
self.assertFalse([f for f in O.Plan(O.raiz_cfg(raiz)).correr()["filas"] if f["marcado"]])
def test_una_pieza_no_es_una_grabacion(self):
"""Una pieza numerada cuyo título alguien confirmó como «grabación» se quedaba sin sus archivos."""
raiz = cliente("Pieza")
for n in ("final VR1.mp4", "toma 2.mp4", "Musica/tema.wav", "Portada/p.png"):
mk(raiz, "2026/02- Febrero/03- Se me cayó el parpado/" + n, 2026, 2, 10)
mk(raiz, "2026/08- Agosto/03- Grabacion Agosto/C0001.MP4", 2026, 8, 5)
self.aprende("se me cayo el parpado", "GRABACION")
p = plan(raiz)
self.assertFalse([f for f in p["filas"] if "parpado" in f["origen_rel"]], "desarma la pieza")
# la sesión del mes, con nombre de fábrica y clips de cámara, sí se reparte
self.assertIn(rel("2026", "08- Agosto", "00_Grabacion", "05- Miércoles", "SONY"), {f["destino_rel"] for f in p["filas"]})
def test_lo_que_no_es_rodaje_no_va_a_otros(self):
"""Dentro de una carpeta de grabación había proyectos e imágenes de IA que acababan en 00_Grabacion\\OTROS."""
raiz = cliente("Mezcla")
for n in ("C0001.MP4", "C0002.MP4", "C0003.MP4"):
mk(raiz, "Grabación evento/" + n, 2026, 4, 10)
mk(raiz, "Grabación evento/guion.docx", 2026, 4, 10)
mk(raiz, "Grabación evento/proyecto.drp", 2026, 4, 10)
mk(raiz, "Grabación evento/Export/final.mp4", 2026, 4, 12)
p = plan(raiz)
por_archivo = {os.path.basename(a): f for f in p["filas"] for a in (f["archivos"] or [])}
self.assertEqual(por_archivo["C0001.MP4"]["destino_rel"], rel("2026", "04- Abril", "00_Grabacion", "10- Viernes", "evento", "SONY"))
self.assertTrue(por_archivo["C0001.MP4"]["marcado"], "un original de cámara fechado por su archivo debe ir marcado")
self.assertEqual(por_archivo["proyecto.drp"]["destino_rel"], rel("01_Proyectos", "DaVinci Resolve"))
self.assertNotIn("00_Grabacion", por_archivo["guion.docx"]["destino_rel"])
self.assertNotIn("final.mp4", por_archivo) # va con su carpeta, no suelto
export = [f for f in p["filas"] if f["origen_rel"] == rel("Grabación evento", "Export")]
self.assertTrue(export and export[0]["destino_rel"] == rel("2026", "04- Abril", "_Exportados"), export)
self.assertFalse([f for f in p["filas"] if f["destino_rel"].endswith("OTROS")])
def test_repaso_de_lo_ya_ordenado(self):
"""160 clips de diciembre de 2025 vivían en la 00_Grabacion de marzo, y nada los volvía a mirar."""
raiz = cliente("Repaso")
g = "2025/03- Marzo/00_Grabacion/"
mk(raiz, g + "URSA_BRAW/A014_12091814_C001.braw", 2025, 12, 9)
mk(raiz, g + "URSA_BRAW/A014_12100130_C002.braw", 2025, 12, 10) # la misma noche, ya de madrugada
mk(raiz, g + "URSA_BRAW/Duplicados/A014_12091814_C001.mp4", 2026, 7, 1) # reescrito después
mk(raiz, g + "SONY/C0001.MP4", 2025, 12, 9) # sin fecha en el nombre, mismo día
mk(raiz, g + "SONY/C0002.MP4", 2025, 3, 5) # este sí es de marzo
mk(raiz, g + "URSA_BRAW/A002_03110342_C036.braw", 2025, 3, 11) # y este
antes = nombres(raiz)
p = plan(raiz)
filas = [f for f in p["filas"] if "ya estaba ordenado" in f["motivo"]]
movidos = sorted(os.path.basename(a) for f in filas for a in f["archivos"])
self.assertEqual(movidos, ["A014_12091814_C001.braw", "A014_12091814_C001.mp4", "A014_12100130_C002.braw", "C0001.MP4"])
self.assertTrue(all(f["marcado"] for f in filas), filas)
self.assertTrue(any("en un mes que no es el de su rodaje" in a for a in p["avisos"]), p["avisos"])
aplicar(raiz, p)
self.assertEqual(nombres(raiz), antes)
dic = os.path.join(raiz, "2025", "12- Diciembre", "00_Grabacion")
for esperado in ("09- Martes" + S + "URSA_BRAW" + S + "A014_12091814_C001.braw", "09- Martes" + S + "URSA_BRAW" + S + "A014_12100130_C002.braw", # la madrugada sigue siendo la noche del 9
"09- Martes" + S + "URSA_BRAW" + S + "Duplicados" + S + "A014_12091814_C001.mp4", "09- Martes" + S + "SONY" + S + "C0001.MP4"):
self.assertTrue(os.path.isfile(os.path.join(dic, esperado)), "falta " + esperado + "\n" + "\n".join(arbol(raiz)))
# y lo que sí era de marzo, pero estaba sin carpeta de día, entra en la suya (3.6.0: el día va siempre)
mar = os.path.join(raiz, "2025", "03- Marzo", "00_Grabacion")
self.assertTrue(os.path.isfile(os.path.join(mar, "05- Miércoles", "SONY", "C0002.MP4")), "\n".join(arbol(raiz)))
self.assertTrue(os.path.isfile(os.path.join(mar, "11- Martes", "URSA_BRAW", "A002_03110342_C036.braw")), "\n".join(arbol(raiz)))
self.assertFalse([f for f in O.Plan(O.raiz_cfg(raiz)).correr()["filas"] if f["marcado"]], "el repaso no termina")
def test_carpetas_gemelas_y_papelera_vieja(self):
"""«Exportados» y «_Exportados» convivían en el mismo mes, y lo apartado, en dos papeleras."""
raiz = cliente("Gemelas")
mk(raiz, "2026/03- Marzo/Exportados/a.mp4", 2026, 3, 10)
mk(raiz, "2026/03- Marzo/_Exportados/b.mp4", 2026, 3, 11)
mk(raiz, "2026/05- Mayo/EXPORTADOS/c.mp4", 2026, 5, 11)
mk(raiz, "_Papelera/basura/x.tmp", 2026, 1, 1)
mk(raiz, "_Para borrar/basura/y.tmp", 2026, 1, 1)
p = plan(raiz)
gemela = [f for f in p["filas"] if f["origen_rel"] == rel("2026", "03- Marzo", "Exportados")]
self.assertTrue(gemela and gemela[0]["destino_rel"] == rel("2026", "03- Marzo", "_Exportados") and gemela[0]["marcado"], gemela)
vieja = [f for f in p["filas"] if f["origen_rel"] == "_Papelera"]
self.assertTrue(vieja and vieja[0]["destino_rel"] == "_Para borrar" and vieja[0]["paso"] == "limpiar", vieja)
self.assertEqual([a["titulo"] for a in O.acciones(raiz)["acciones"] if a["clave"] == "limpiar"], ["Juntar las dos papeleras"])
aplicar(raiz, p)
self.assertEqual(arbol(raiz), sorted([rel("2026", "03- Marzo", "_Exportados", "a.mp4"), rel("2026", "03- Marzo", "_Exportados", "b.mp4"),
rel("2026", "05- Mayo", "_Exportados", "c.mp4"),
rel("_Para borrar", "basura", "x.tmp"), rel("_Para borrar", "basura", "y.tmp")]))
def test_por_nombrar_solo_pregunta_lo_que_importa(self):
"""De 6 preguntas, 6 sobraban: carpetas de la app, de dentro de la papelera o de dentro de una pieza."""
raiz = cliente("Nombrar")
mk(raiz, "_Para borrar/basura/Videos/c.mp4", 2026, 1, 1) # dentro de la papelera
mk(raiz, "2026/01- Enero/00_Grabacion/Orden/SONY/C0001.MP4", 2026, 1, 5) # una etiqueta de la propia app
mk(raiz, "2026/01- Enero/01- Pieza/Rara/x.png", 2026, 1, 5) # dentro de una pieza
mk(raiz, "2026/01- Enero/Sin numero/x.png", 2026, 1, 6) # en un mes: será una pieza
for n in ("Exportado/f.mp4", "Portada/p.png", "Musica/m.wav", "r.prproj"):
mk(raiz, "Botox labios/" + n, 2026, 2, 2) # una pieza montada, suelta en la raíz
for n in ("a.xyz", "b.xyz", "c.xyz"):
mk(raiz, "Cosas/" + n, 2026, 2, 2)
for i in range(5):
mk(raiz, "Tarjeta/C%04d.MP4" % i, 2026, 2, 3) # llena de clips: se reparte sola
os.makedirs(os.path.join(raiz, "Vacia", "dentro"))
plan(raiz)
voc = {t["token"]: t for t in O.vocabulario(raiz)}
self.assertEqual(sorted(k for k, t in voc.items() if not t["cubeta"]), ["botox labios", "cosas"], sorted(voc))
for sobra in ("videos", "orden", "rara", "sin numero", "tarjeta", "vacia", "basura", "sony"):
self.assertNotIn(sobra, voc)
t = voc["botox labios"]
self.assertEqual((t["sugerencia"], t["sug_conf"]), ("PIEZA", "alta"), t)
self.assertNotEqual(voc["cosas"].get("sugerencia"), "GRABACION")
nombrar = [a for a in O.acciones(raiz)["acciones"] if a["clave"] == "nombrar"][0]
self.assertIn("2 carpetas", nombrar["cifra"])
def test_piezas_gemelas_en_el_mismo_mes(self):
"""La misma pieza escrita de dos formas en el mismo mes: se juntan, y solo se mueve una de las dos."""
raiz = cliente("Gemelas de pieza")
mk(raiz, "2026/05- Mayo/03 - El cuello/a.mp4", 2026, 5, 10)
mk(raiz, "2026/05- Mayo/03- El cuello/b.mp4", 2026, 5, 10)
mk(raiz, "2026/06- Junio/Botox/c.mp4", 2026, 6, 10)
mk(raiz, "2026/06- Junio/02- Botox/d.mp4", 2026, 6, 10)
p = plan(raiz)
self.assertEqual(sorted((f["origen_rel"], f["destino_rel"], f["marcado"]) for f in p["filas"]),
[(rel("2026", "05- Mayo", "03 - El cuello"), rel("2026", "05- Mayo", "03- El cuello"), True),
(rel("2026", "06- Junio", "Botox"), rel("2026", "06- Junio", "02- Botox"), True)])
aplicar(raiz, p)
self.assertEqual(arbol(raiz), [rel("2026", "05- Mayo", "03- El cuello", "a.mp4"), rel("2026", "05- Mayo", "03- El cuello", "b.mp4"),
rel("2026", "06- Junio", "02- Botox", "c.mp4"), rel("2026", "06- Junio", "02- Botox", "d.mp4")])
def test_una_campana_con_fechas_no_es_un_cajon(self):
"""Abrir lo que guarda meses dentro vale donde las piezas van por mes; una campaña se mueve entera."""
raiz = cliente("Anuncios", "pauta")
mk(raiz, "Campaña Navidad/2025/Noviembre/anuncio.mp4", 2025, 11, 20)
mk(raiz, "Campaña Navidad/2025/Noviembre/guion.docx", 2025, 11, 20)
p = plan(raiz)
self.assertFalse([a for a in p["avisos"] if "guarda dentro" in a], p["avisos"])
self.assertEqual([(f["origen_rel"], f["accion"]) for f in p["filas"]], [("Campaña Navidad", "mover carpeta")])
def test_lo_dudoso_no_cuenta_como_pendiente(self):
"""Con lo seguro hecho, la carpeta está al día aunque quede algo que la app no mueve sola."""
raiz = cliente("Dudas")
mk(raiz, "2026/03- Marzo/01 - Botox/final.mp4", 2026, 3, 10) # solo falta un retoque de nombre, sin marcar
p = plan(raiz)
self.assertEqual([(f["accion"], f["marcado"]) for f in p["filas"]], [("renombrar", False)])
O.PLANES[os.path.normpath(raiz)] = p
a = O.acciones(raiz)
fechas = [x for x in a["acciones"] if x["clave"] == "fechas"][0]
self.assertEqual((fechas["estado"], fechas["cifra"]), ("espera", "1 dudoso"))
self.assertNotIn("fechas", a["automatico"])
def test_hacer_lo_seguro_no_mueve_lo_dudoso(self):
filas = [{"cubeta": "GRABACION", "marcado": False, "confianza": "baja"},
{"cubeta": "GRABACION", "marcado": True, "confianza": "media"},
{"cubeta": "BASURA", "marcado": True, "confianza": "alta"},
{"cubeta": "PROXY", "marcado": False, "confianza": "baja"}, # sin su original, o dentro de «No tocar»
{"cubeta": "EXPORTADOS", "marcado": False, "confianza": "baja"}]
elegidas = asistente.filas_elegidas(filas, {"carpetas": "seguras"})
self.assertEqual([(f["cubeta"], f["marcado"]) for f in elegidas], [("GRABACION", True), ("BASURA", True)])
def test_banco_de_material(self):
"""Vía Sumapaz: 3,6 TB en «Material Vía (No tocar)», por temas, que el plan dejaba fuera por la palabra
«material». Los temas se quedan; dentro, las grabaciones por fecha y cámara; todo sin marcar."""
raiz = cliente("Banco")
b = "Material Vía (No tocar)/"
mk(raiz, b + "Lugares/Peajes/Drone/DJI_20240312101500_0001_D.MP4", 2024, 3, 12)
mk(raiz, b + "Lugares/Peajes/URSA/A001_03121015_C001.mp4", 2024, 3, 12)
mk(raiz, b + "Lugares/Peajes/URSA/Proxy/A001_03121015_C001.mov", 2024, 3, 12) # proxy con su original
mk(raiz, b + "Lugares/Peajes/Final peajes.mp4", 2024, 4, 2) # montado: no se toca
mk(raiz, b + "Material Importante/Social/Mercados/C0001.MP4", 2025, 5, 2)
mk(raiz, b + "Material Importante/UF9/Drone/DJI_0001.MP4", 2009, 1, 19) # reloj sin poner
mk(raiz, "2026/03- Marzo/Grabacion/C0010.MP4", 2026, 3, 12)
mk(raiz, "2026/03- Marzo/Grabacion/Proxy/C0010.MOV", 2026, 3, 12) # proxy con original
mk(raiz, "2026/04- Abril/01- Viejo/Proxy/C0099.MOV", 2026, 4, 2) # el único que queda
dicc = O.DICC["carpetas"]
self.assertEqual(O.R.cubeta_carpeta("Material Vía (No tocar)", dicc)[0], "BANCO")
self.assertIsNone(O.R.cubeta_carpeta("Material juan", dicc)[0], "«material» suelto ya no ignora la carpeta")
self.assertEqual(O.R.cubeta_carpeta("Material", dicc)[0], "IGNORAR")
antes = nombres(raiz)
p = plan(raiz)
banco = [f for f in p["filas"] if f["paso"] == "banco"]
destinos = {f["destino_rel"] for f in banco}
B = "Material Vía (No tocar)"
self.assertIn(rel(B, "Lugares", "Peajes", "2024-03-12", "OSMO"), destinos)
self.assertIn(rel(B, "Lugares", "Peajes", "2024-03-12", "BLACKMAGIC_MP4"), destinos)
self.assertIn(rel(B, "Material Importante", "Social", "Mercados", "2025-05-02", "SONY"), destinos)
self.assertFalse([f for f in banco if f["marcado"]], "en «No tocar» nada va marcado")
self.assertFalse([f for f in p["filas"] if any("Final peajes" in a for a in (f["archivos"] or []))], "movió lo montado")
reloj = [f for f in banco if "reloj sin poner" in f["motivo"]]
self.assertTrue(reloj and reloj[0]["confianza"] == "baja", [f["motivo"] for f in banco])
prox = [f for f in p["filas"] if f["cubeta"] == "PROXY"]
dentro = [f for f in prox if "Peajes" in f["origen_rel"]]
self.assertTrue(dentro and not dentro[0]["marcado"] and "No tocar" in dentro[0]["motivo"], dentro)
sin = [f for f in prox if "01- Viejo" in f["origen_rel"]]
self.assertTrue(sin and not sin[0]["marcado"] and "OJO" in sin[0]["motivo"], sin)
con = [f for f in prox if "03- Marzo" in f["origen_rel"]]
self.assertTrue(con and con[0]["marcado"], con)
# «hacer de una vez» no toca nada de lo que va sin marcar
self.assertFalse([f for f in asistente.filas_elegidas(p["filas"], {"carpetas": "seguras"}) if not f["marcado"]])
dentro_antes = [x for x in arbol(raiz) if x.startswith(B)]
aplicar(raiz, p)
self.assertEqual([x for x in arbol(raiz) if x.startswith(B)], dentro_antes, "tocó la carpeta «No tocar» sin que nadie lo marcara")
self.assertTrue(os.path.isfile(os.path.join(raiz, "2026", "04- Abril", "01- Viejo", "Proxy", "C0099.MOV")), "apartó el proxy sin original")
# si alguien lo marca, queda donde dice
p2 = O.Plan(O.raiz_cfg(raiz)).correr()
for f in p2["filas"]:
f["marcado"] = f["paso"] == "banco"
aplicar(raiz, p2)
self.assertEqual(nombres(raiz), antes, "se perdió o se duplicó algún archivo")
for esperado in (rel(B, "Lugares", "Peajes", "2024-03-12", "OSMO", "DJI_20240312101500_0001_D.MP4"),
rel(B, "Lugares", "Peajes", "2024-03-12", "BLACKMAGIC_MP4", "A001_03121015_C001.mp4"),
rel(B, "Lugares", "Peajes", "Final peajes.mp4"),
rel(B, "Material Importante", "Social", "Mercados", "2025-05-02", "SONY", "C0001.MP4")):
self.assertTrue(os.path.isfile(os.path.join(raiz, esperado)), "falta " + esperado + "\n" + "\n".join(arbol(raiz)))
# y «Qué pesa» entra hasta el fondo y dice cómo la trata la app
d = O.carpeta_detalle(raiz, os.path.join(raiz, B))
self.assertEqual(d["trato"]["cubeta"], "BANCO")
self.assertTrue(d["niveles"] >= 3 and "Lugares" in [h["nombre"] for h in d["hijos"]], d)
self.assertIn("banco", [a["clave"] for a in O.acciones(raiz)["acciones"]])
def test_alta_y_baja_no_chocan(self):
"""Vía Sumapaz: cada evento guardaba la misma foto en «Alta» y en «Baja»; el plan las juntaba en una
sola carpeta y 753 no se movieron. Las subcarpetas con sentido se conservan."""
raiz = cliente("AltaBaja")
for sub, tam in (("Alta", 8192), ("Baja", 2048)):
for n in ("IMG_5154.jpg", "IMG_5155.jpg"):
mk(raiz, f"Fotos varias/2022/Enero/Evento recaudadoras/{sub}/{n}", 2022, 1, 2, tam)
mk(raiz, "Fotos varias/2022/Diciembre/Fiesta 6 Años/IMG_3135.jpg", 2022, 12, 14)
antes = nombres(raiz)
p = plan(raiz)
self.assertFalse([f for f in p["filas"] if "OJO" in f["motivo"]], [f["motivo"] for f in p["filas"]])
por = {}
for f in p["filas"]:
for a in f["archivos"] or []:
por[a] = f["destino_rel"]
alta = [d for a, d in por.items() if os.sep + "Alta" + os.sep in a]
baja = [d for a, d in por.items() if os.sep + "Baja" + os.sep in a]
self.assertTrue(alta and baja and all(d.endswith("Alta") for d in alta) and all(d.endswith("Baja") for d in baja), por)
self.assertTrue(any("Fiesta 6 Años" in d for d in por.values()), por)
aplicar(raiz, p) # sin errores: ninguna foto choca con su gemela
self.assertEqual(nombres(raiz), antes)
def test_reloj_de_una_camara(self):
"""Una cámara con el reloj sin poner: se corrige con la fecha real del primer clip, sigue corregida al
volver a escanear, el plan la lleva a su mes de verdad, y se deshace."""
raiz = cliente("Reloj")
mk(raiz, "Vía Mujeres/Drone/DJI_0001.MP4", 2009, 1, 19, 4096)
mk(raiz, "Vía Mujeres/Drone/DJI_0002.MP4", 2009, 1, 20, 4096)
plan(raiz) # escanea
r = O.relojes(raiz)
self.assertEqual([(g["fuente"], g["n"], g["desde"]) for g in r["grupos"]], [("DJI", 2, "2009-01-19")], r)
c = O.corregir_reloj(raiz, r["grupos"][0]["dir"], "DJI", "2024-03-12")
self.assertTrue(c.get("ok") and c["n"] == 2, c)
con = O.db()
fechas = sorted((x["fecha"][:10], x["fecha_origen"]) for x in con.execute("SELECT fecha, fecha_origen FROM archivos WHERE raiz = ? AND nombre LIKE 'DJI_%'", (raiz,)))
con.close()
self.assertEqual(fechas, [("2024-03-12", "corregida"), ("2024-03-13", "corregida")])
p = plan(raiz) # vuelve a escanear: la corrección se mantiene
self.assertFalse(O.relojes(raiz)["grupos"])
self.assertTrue(any("2024" in f["destino_rel"] and "03- Marzo" in f["destino_rel"] for f in p["filas"]), [f["destino_rel"] for f in p["filas"]])
d = O.deshacer_reloj(raiz, O.relojes(raiz)["corregidos"][0]["grupo"])
self.assertTrue(d.get("ok"), d)
self.assertEqual(O.relojes(raiz)["grupos"][0]["desde"], "2009-01-19")
# y buscar la encuentra, con tilde o sin ella
b = O.buscar("via mujeres dji_0002")
self.assertEqual([x["nombre"] for x in b["resultados"]], ["DJI_0002.MP4"], b)
self.assertEqual(O.buscar("x").get("error"), "escribe al menos dos letras")
def test_banco_por_fecha(self):
"""El banco también se puede ordenar por fecha: a la 00_Grabacion de su mes, con el tema de etiqueta."""
raiz = cliente("BancoFecha")
mk(raiz, "Archivo (No tocar)/Lugares/Peajes/Drone/DJI_20240312101500_0001_D.MP4", 2024, 3, 12)
mk(raiz, "Archivo (No tocar)/Lugares/Peajes/Final.mp4", 2024, 4, 2)
O.CONFIG.setdefault("bancos", {})[os.path.normpath(raiz).lower()] = {O.R.norm("Archivo (No tocar)"): "fecha"}
self.addCleanup(lambda: O.CONFIG["bancos"].pop(os.path.normpath(raiz).lower(), None))
p = plan(raiz)
self.assertEqual([b["modo"] for b in p["bancos"]], ["fecha"])
banco = [f for f in p["filas"] if f["paso"] == "banco"]
self.assertEqual([f["destino_rel"] for f in banco], [rel("2024", "03- Marzo", "00_Grabacion", "12- Martes", "Peajes", "OSMO")])
self.assertFalse(banco[0]["marcado"])
def test_otros_discos(self):
"""El mismo cliente en otros discos: se ofrece traerlo, sin contar las carpetas ya registradas."""
raiz = cliente("Repartido")
previo = dict(O._DISCOS_CACHE)
self.addCleanup(lambda: (O._DISCOS_CACHE.clear(), O._DISCOS_CACHE.update(previo)))
O._DISCOS_CACHE.update(ts=time.time(), datos={"discos": [{"letra": "K:", "unc": r"\\nas\Entregas", "nombre": ""}],
"clientes": {"Repartido": [r"K:\Repartido viejo", raiz]}})
s = O.otros_sitios(raiz)
self.assertEqual([(x["ruta"], x["unc"]) for x in s], [(r"K:\Repartido viejo", r"\\nas\Entregas")])
def test_el_convertidor_no_repite(self):
"""22 clips se habían convertido dos veces y algunos tres («…_10bit_10bit_10bit.mp4»)."""
raiz = cliente("Convertidos")
mk(raiz, "2026/04- Abril/00_Grabacion/SONY/C0001.MP4", 2025, 4, 10)
mk(raiz, "2026/04- Abril/00_Grabacion/SONY/C0002_10bit.mp4", 2025, 4, 10)
plan(raiz)
con = O.db()
with con:
con.execute("UPDATE archivos SET analizado = 1, codec = 'hevc', ancho = 1920, alto = 1080, fps = 25, dur = 10, "
"kbps = 90000 WHERE raiz = ?", (os.path.normpath(raiz),))
roles = {r["nombre"]: r["rol"] for r in con.execute("SELECT nombre, rol FROM archivos WHERE raiz = ?", (os.path.normpath(raiz),))}
con.close()
self.assertEqual(roles, {"C0001.MP4": "camara_video", "C0002_10bit.mp4": "camara_video"})
cs = {os.path.basename(c["ruta"]): c["excluido"] for c in O.candidatos(os.path.normpath(raiz), 0, "equilibrado", 0)}
self.assertEqual(cs["C0001.MP4"], "")
self.assertIn("ya es una conversión", cs["C0002_10bit.mp4"])
def test_el_sufijo_del_convertidor_se_puede_cambiar(self):
"""El sufijo que pone el convertidor es un ajuste: si se cambia, las reglas tienen que enterarse."""
antes = O.CONFIG["optimizar"].get("sufijo")
self.addCleanup(lambda: O.CONFIG["optimizar"].__setitem__("sufijo", antes))
self.assertFalse(R.es_conversion("C0001_arch.mp4"))
O.CONFIG["optimizar"]["sufijo"] = "_arch"
O._sufijos_conversion()
self.assertTrue(R.es_conversion("C0001_arch.mp4"))
self.assertEqual(R.firma_archivo("C0001_arch.mp4"), "SONY")
self.assertTrue(R.es_conversion("C0001_10bit.mp4")) # y los de siempre siguen valiendo
def test_el_inventario_escribe_la_carpeta_como_el_disco(self):
"""Vía Sumapaz, 14-oct-2026: el plan movió fotos a «Altas Print» y «Altas print», que en el disco son la misma
«ALTAS PRINT»; el inventario apuntó tres carpetas distintas con 1.039 fotos mal escritas."""
raiz = cliente("Mayusculas")
nr = os.path.normpath(raiz)
a = mk(raiz, "Sueltas/foto1.jpg", 2024, 5, 3)
b = mk(raiz, "Sueltas/foto2.jpg", 2024, 5, 3)
mk(raiz, "2024/05- Mayo/_Fotos/ALTAS PRINT/foto0.jpg", 2024, 5, 3)
O.escanear(raiz)
movidos = []
O.mover(a, os.path.join(nr, "2024", "05- Mayo", "_Fotos", "Altas Print", "foto1.jpg"), "prueba", movidos)
O.mover(b, os.path.join(nr, "2024", "05- mayo", "_fotos", "altas print", "foto2.jpg"), "prueba", movidos)
bien = os.path.join(nr, "2024", "05- Mayo", "_Fotos", "ALTAS PRINT")
self.assertEqual([d for _, d in movidos], [os.path.join(bien, "foto1.jpg"), os.path.join(bien, "foto2.jpg")])
O.actualizar_inventario(movidos)
con = O.db()
rutas = {r[0] for r in con.execute("SELECT ruta FROM archivos WHERE raiz = ?", (nr,))}
carpetas = [r[0] for r in con.execute("SELECT ruta FROM carpetas WHERE raiz = ? AND lower(ruta) = lower(?)", (nr, bien))]
con.close()
self.assertTrue({os.path.join(bien, n) for n in ("foto0.jpg", "foto1.jpg", "foto2.jpg")} <= rutas, rutas)
self.assertEqual(carpetas, [bien])
# una carpeta que no existía se crea como la escribe el plan, y las siguientes la siguen igual
c = mk(raiz, "Sueltas/foto3.jpg", 2024, 5, 3)
d = mk(raiz, "Sueltas/foto4.jpg", 2024, 5, 3)
movidos = []
O.mover(c, os.path.join(nr, "Nueva Carpeta", "foto3.jpg"), "prueba", movidos)
O.mover(d, os.path.join(nr, "NUEVA CARPETA", "foto4.jpg"), "prueba", movidos)
self.assertEqual({os.path.dirname(x) for _, x in movidos}, {os.path.join(nr, "Nueva Carpeta")})
def test_escaneo_con_movimientos_a_la_vez(self):
"""Vía Sumapaz, 14-oct-2026: un proxy apartado a _Para borrar en pleno escaneo quedó dos veces en el
inventario (visto en su sitio y luego en la papelera). Si la app mueve algo mientras se recorre, se recorre otra vez."""
raiz = cliente("EnMarcha")
nr = os.path.normpath(raiz)
p = mk(raiz, "2024/09- Septiembre/03- Botellitas/Proxy/Botellas Vr2.mov", 2024, 9, 3)
mk(raiz, "2024/09- Septiembre/03- Botellitas/Botellas Vr2.mov", 2024, 9, 3, tam=4096)
real = O._fechas_de_rodaje
llamadas = []
def a_la_vez(filas):
llamadas.append(len(filas))
if len(llamadas) == 1: # después de recorrer, antes de guardar: la app aparta el proxy
O.mover(p, os.path.join(nr, R.PAPELERA, "proxies", os.path.relpath(p, nr)), "aparte", [])
return real(filas)
O._fechas_de_rodaje = a_la_vez
self.addCleanup(setattr, O, "_fechas_de_rodaje", real)
O.escanear(raiz)
self.assertEqual(len(llamadas), 2)
con = O.db()
proxies = [r[0] for r in con.execute("SELECT ruta FROM archivos WHERE raiz = ? AND nombre = 'Botellas Vr2.mov' AND bytes = 2048", (nr,))]
con.close()
self.assertEqual(proxies, [os.path.join(nr, R.PAPELERA, "proxies", os.path.relpath(p, nr))])
self.assertEqual(O.PROGRESO[nr]["estado"], "listo")
def test_inventario_de_otra_version(self):
"""Lo que apunta el escaneo cambió: un inventario viejo se relee antes de planear nada."""
raiz = cliente("Viejo")
mk(raiz, "Grabacion/C0001.MP4", 2026, 4, 10)
plan(raiz)
self.assertNotEqual(O.acciones(raiz)["acciones"][0]["clave"], "escanear")
con = O.db()
with con:
con.execute("UPDATE escaneos SET reglas = 0 WHERE raiz = ?", (os.path.normpath(raiz),))
con.close()
a = O.acciones(raiz)
self.assertEqual((a["acciones"][0]["clave"], a["acciones"][0]["estado"], a["automatico"]), ("escanear", "toca", []))
self.assertTrue(any("versión anterior" in x for x in O.Plan(O.raiz_cfg(raiz)).correr()["avisos"]))
class Nuevo(unittest.TestCase):
"""Lo que entra nuevo: una pieza, el material de una tarjeta, y la pasada automática."""
def test_pieza_nueva(self):
"""Las piezas se creaban a mano: dos «01» en el mismo mes, «03 - X» junto a «03- X», «portasda»."""
raiz = cliente("Piezas")
mk(raiz, "2026/10- Octubre/01- Uno/a.mp4", 2026, 10, 1)
mk(raiz, "2026/10- Octubre/02- Dos/b.mp4", 2026, 10, 2)
plan(raiz)
previa = O.nueva_pieza(raiz, "Ozempic face", 2026, 10, crear=False)
self.assertEqual((previa["rel"], previa["existe"]), (rel("2026", "10- Octubre", "03- Ozempic face"), ""))
self.assertFalse(os.path.exists(previa["ruta"]), "la vista previa no debe crear nada")
r = O.nueva_pieza(raiz, "Ozempic face", 2026, 10)
self.assertTrue(r.get("ok"), r)
subs = sorted(os.listdir(r["ruta"]))
self.assertEqual(subs, sorted(O.CONFIG_DEFECTO["pieza"]["subcarpetas"]))
self.assertIn("ya existe", O.nueva_pieza(raiz, "ozempic FACE", 2026, 10).get("error", "")) # la misma, escrita distinto
self.assertEqual(O.nueva_pieza(raiz, "07- Otra cosa", 2026, 10, crear=False)["nombre"], "04- Otra cosa") # el número lo pone la app
for malo in ("", "a/b", "x" * 90):
self.assertTrue(O.nueva_pieza(raiz, malo, 2026, 10).get("error"), malo)
mes_nuevo = O.nueva_pieza(raiz, "Primera", 2026, 11)
self.assertEqual(mes_nuevo["rel"], rel("2026", "11- Noviembre", "01- Primera"))
# una pieza recién creada no es un hueco que borrar, ni el mes que se creó para ella
self.assertEqual(O.carpetas_vacias(raiz)["carpetas"], [])
self.assertFalse([f for f in O.Plan(O.raiz_cfg(raiz)).correr()["filas"]], "el plan quiere mover una pieza recién creada")
# deshacer la quita si sigue vacía; si ya tiene algo dentro, la deja
mk(raiz, "2026/11- Noviembre/01- Primera/Guion/guion.docx", 2026, 11, 2)
d = O.deshacer(mes_nuevo["lote"])
self.assertTrue(os.path.isfile(os.path.join(mes_nuevo["ruta"], "Guion", "guion.docx")) and d["errores"], d)
d = O.deshacer(r["lote"])
self.assertFalse(os.path.exists(r["ruta"]) or d["errores"], d)
# donde las piezas no van por mes (campañas), sin número
anuncios = cliente("Piezas de pauta", "pauta")
self.assertEqual(O.nueva_pieza(anuncios, "Navidad", 2026, 12)["rel"], rel("Campañas", "2026_Navidad"))
def test_bandeja_de_entrada(self):
"""Lo que se deja en _Entrada se reparte solo por rodaje y cámara, y la bandeja queda vacía."""
raiz = cliente("Bandeja")
mk(raiz, "2026/09- Septiembre/01- Pieza/a.png", 2026, 9, 1)
t = "_Entrada/Tarjeta A/PRIVATE/M4ROOT/"
mk(raiz, t + "CLIP/C0001.MP4", 2026, 9, 20)
mk(raiz, t + "CLIP/C0001M01.XML", 2026, 9, 20)
mk(raiz, t + "MEDIAPRO.XML", 2026, 9, 20)
mk(raiz, t + "THMBNL/C0001T01.JPG", 2026, 9, 20)
mk(raiz, t + "SUB/C0001S03.MP4", 2026, 9, 20)
mk(raiz, "_Entrada/A001_09211200_C001.braw", 2026, 9, 21)
mk(raiz, "_Entrada/Entrevista Dra/C0005.MP4", 2026, 9, 22)
antes = nombres(raiz)
plan(raiz)
self.assertEqual([a["clave"] for a in O.acciones(raiz)["acciones"] if a["clave"] == "entrada"], ["entrada"])
self.assertGreaterEqual(O.entrada(raiz)["n"], 4)
r = O.ordenar_entrada(raiz)
self.assertTrue(r.get("ok") and r["movimientos"] and not r["errores"], r)
self.assertEqual(nombres(raiz), antes)
g = os.path.join(raiz, "2026", "09- Septiembre", "00_Grabacion")
for esperado in (os.path.join(g, R.nombre_dia("2026-09-20"), "SONY", "C0001.MP4"),
os.path.join(g, R.nombre_dia("2026-09-20"), "SONY", "C0001M01.XML"),
os.path.join(g, R.nombre_dia("2026-09-21"), "URSA_BRAW", "A001_09211200_C001.braw"),
os.path.join(g, R.nombre_dia("2026-09-22"), "Entrevista Dra", "SONY", "C0005.MP4"),
os.path.join(raiz, "_Para borrar", "proxies", "_Entrada", "Tarjeta A", "PRIVATE", "M4ROOT", "SUB", "C0001S03.MP4"),
os.path.join(raiz, "_Para borrar", "basura", "_Entrada", "Tarjeta A", "PRIVATE", "M4ROOT", "THMBNL", "C0001T01.JPG"),
os.path.join(raiz, "_Para borrar", "basura", "_Entrada", "Tarjeta A", "PRIVATE", "M4ROOT", "MEDIAPRO.XML")):
self.assertTrue(os.path.isfile(esperado), "falta " + os.path.relpath(esperado, raiz) + "\n" + "\n".join(arbol(raiz)))
# la bandeja se queda, vacía y sin el esqueleto de la tarjeta, y no sale como hueco que borrar
self.assertEqual(os.listdir(os.path.join(raiz, "_Entrada")), [])
self.assertEqual((r["quedan"], O.entrada(raiz)["n"]), (0, 0))
self.assertFalse([a for a in O.acciones(raiz)["acciones"] if a["clave"] == "entrada"])
self.assertNotIn("_Entrada", [c["rel"] for c in O.carpetas_vacias(raiz)["carpetas"]])
# y se deshace: cada cosa vuelve a su sitio dentro de la tarjeta
d = O.deshacer(r["lote"])
self.assertFalse(d["errores"], d)
self.assertTrue(os.path.isfile(os.path.join(raiz, "_Entrada", "Tarjeta A", "PRIVATE", "M4ROOT", "CLIP", "C0001.MP4")))
self.assertTrue(os.path.isfile(os.path.join(raiz, "_Entrada", "Entrevista Dra", "C0005.MP4")))
def test_traer_de_una_tarjeta(self):
"""La copia desde la tarjeta se comprueba, conserva la fecha del rodaje y no toca el origen."""
raiz = cliente("Traer")
mk(raiz, "2026/08- Agosto/01- Pieza/a.png", 2026, 8, 1)
tarjeta = os.path.join(TMP, "tarjeta-sd")
clip = mk(tarjeta, "DCIM/100MEDIA/DJI_0001.MP4", 2026, 8, 10, tam=300_000)
mk(tarjeta, "System Volume Information/indice.dat", 2026, 8, 10)
with open(clip, "rb") as fh:
contenido = fh.read()
self.assertTrue(O.plan_traer(raiz, os.path.join(raiz, "2026")).get("error")) # de dentro no se trae
self.assertTrue(O.plan_traer(raiz, os.path.join(TMP, "no-existe")).get("error"))
p = O.plan_traer(raiz, tarjeta)
self.assertEqual((p["n"], p["bytes"]), (1, 300_000), p)
self.assertTrue(p["carpeta"].startswith("Tarjeta "))
r = O.traer(raiz, tarjeta)
self.assertEqual((r["estado"], r["copiados"], r["errores"]), ("listo", 1, []), r)
self.assertTrue(r["ordenado"]["ok"] and not r["ordenado"]["errores"], r["ordenado"])
final = os.path.join(raiz, "2026", "08- Agosto", "00_Grabacion", "10- Lunes", "DJI", "DJI_0001.MP4")
self.assertTrue(os.path.isfile(final), "\n".join(arbol(raiz)))
with open(final, "rb") as fh:
self.assertEqual(fh.read(), contenido)
self.assertLess(abs(os.path.getmtime(final) - os.path.getmtime(clip)), 2, "no conservó la fecha del rodaje")
with open(clip, "rb") as fh:
self.assertEqual(fh.read(), contenido, "tocó el origen")
self.assertEqual(arbol(tarjeta), [rel("DCIM", "100MEDIA", "DJI_0001.MP4"), rel("System Volume Information", "indice.dat")])
actas = os.listdir(os.path.join(raiz, ".organilab", "entradas"))
self.assertEqual(len(actas), 1)
with open(os.path.join(raiz, ".organilab", "entradas", actas[0]), encoding="utf-8-sig") as fh:
self.assertIn(rel("DCIM", "100MEDIA", "DJI_0001.MP4") + ";300000;", fh.read())
# una copia que no queda igual que el original no se da por buena
malo = mk(tarjeta, "DCIM/100MEDIA/DJI_0002.MP4", 2026, 8, 10, tam=50_000)
real = O.hashlib.blake2b
vez = []
class Trucado: # la segunda lectura (la de comprobación) sale distinta
def __init__(self, **kw):
self.h = real(**kw)
vez.append(self)
self.malo = len(vez) % 2 == 0
def update(self, b):
self.h.update(b[::-1] if self.malo else b)
def digest(self):
return self.h.digest()
def hexdigest(self):
return self.h.hexdigest()
O.hashlib.blake2b = Trucado
try:
estado = {"bytes": 0}
destino = os.path.join(raiz, "_Entrada", "prueba", "DJI_0002.MP4")
with self.assertRaises(OSError):
O._copiar_comprobando(malo, destino, estado)
finally:
O.hashlib.blake2b = real
self.assertFalse(os.path.exists(destino) or os.path.exists(destino + ".copiando"), "dejó una copia sin comprobar")
self.assertEqual(estado["bytes"], 0)
def test_versiones_del_mismo_clip(self):
"""39 clips estaban dos veces con el mismo nombre y distinta calidad, y no cabían juntos en su mes."""
raiz = cliente("Versiones")
oct_, may = "2025/10- Octubre/00_Grabacion/BLACKMAGIC_MP4/", "2026/05- Mayo/00_Grabacion/BLACKMAGIC_MP4/"
buena = mk(raiz, oct_ + "A002_03020727_C001.mp4", 2025, 12, 1, tam=9000)
peor = mk(raiz, may + "A002_03020727_C001.mp4", 2026, 7, 1, tam=3000)
mk(raiz, oct_ + "C0001.MP4", 2025, 10, 5, tam=5000) # se llaman igual…
mk(raiz, may + "C0001.MP4", 2026, 5, 5, tam=4000) # …pero son clips distintos: duran distinto
# dos exportados con el mismo nombre no son un duplicado: son dos entregas de la pieza (la vertical y
# la horizontal). No son material de cámara, así que ni se leen: `sonda` fallaría si se los pasaran.
mk(raiz, "2026/02- Febrero/02- Liftera/Exportados/Liftera VR3 Final.mp4", 2026, 2, 20, tam=7000)
mk(raiz, "2026/02- Febrero/01- Otra/Exportados/Liftera VR3 Final.mp4", 2026, 2, 21, tam=6000)
p = plan(raiz)
self.assertTrue([f for f in p["filas"] if "OJO" in f["motivo"]], "sin comparar, las dos versiones chocan en su mes")
medidas = {buena: (12.0, 3840, 2160), peor: (12.02, 1920, 1080),
os.path.join(raiz, oct_.replace("/", S), "C0001.MP4"): (10.0, 1920, 1080),
os.path.join(raiz, may.replace("/", S), "C0001.MP4"): (25.0, 1920, 1080)}
def sonda(ruta):
d, an, al = medidas[ruta]
return {"codec": "h264", "perfil": "", "pix_fmt": "yuv420p", "ancho": an, "alto": al, "fps": 25.0, "dur": d,
"kbps": 1000, "alfa": 0, "acodec": "aac", "pistas_audio": 1}
viejos = (O.ffprobe, O.herramientas)
O.ffprobe, O.herramientas = sonda, lambda: {"ffprobe": {"ok": True}}
try:
self.assertEqual(O.duplicados(raiz, minimo=0)["por_comparar"], 2)
d = O.duplicados(raiz, minimo=0, calidades=True)
finally:
O.ffprobe, O.herramientas = viejos
self.assertEqual([(g["tipo"], g["queda"], g["mueven"]) for g in d["grupos"]], [("calidad", buena, [peor])], d["grupos"])
r = O.apartar_duplicados(raiz, d["grupos"])
self.assertEqual((r["movimientos"], r["errores"]), (1, []), r)
self.assertTrue(os.path.isfile(os.path.join(os.path.dirname(buena), "Duplicados", "A002_03020727_C001.mp4")))
self.assertTrue(os.path.isfile(os.path.join(raiz, may.replace("/", S), "C0001.MP4")), "apartó un clip que solo se llamaba igual")
# ya no chocan: las dos van a su mes, la peor dentro de Duplicados
p = O.Plan(O.raiz_cfg(raiz)).correr()
self.assertFalse([f for f in p["filas"] if "OJO" in f["motivo"]], [f["motivo"] for f in p["filas"]])
aplicar(raiz, p)
mar = os.path.join(raiz, "2025", "03- Marzo", "00_Grabacion", "02- Domingo", "BLACKMAGIC_MP4")
self.assertTrue(os.path.isfile(os.path.join(mar, "A002_03020727_C001.mp4")), "\n".join(arbol(raiz)))
self.assertTrue(os.path.isfile(os.path.join(mar, "Duplicados", "A002_03020727_C001.mp4")), "\n".join(arbol(raiz)))
def test_orden_automatico(self):
"""La pasada automática hace lo seguro en cada carpeta, y se frena sola cuando no debe seguir."""
normal = cliente("Auto normal")
for n in ("C0001.MP4", "C0002.MP4"):
mk(normal, "2026/04- Abril/Grabacion/" + n, 2026, 4, 10)
mk(normal, "2026/04- Abril/01 - Botox/final.mp4", 2026, 4, 12) # un retoque de nombre: dudoso, no se toca
mucho = cliente("Auto demasiado")
for i in range(6):
mk(mucho, "2026/04- Abril/Grabacion/C%04d.MP4" % i, 2026, 4, 10)
todas, ajuste = O.CONFIG["raices"], dict(O.CONFIG["automatico"])
desconectada = {"ruta": os.path.join(TMP, "no-conectada"), "cliente": "Sin disco", "tipo": "contenido"}
O.CONFIG["raices"] = [r for r in todas if r["cliente"] == "Auto normal"] + [desconectada]
ensayo = os.path.join(TMP, "ensayo-tarea")
os.makedirs(ensayo, exist_ok=True)
os.environ["ORGANILAB_ENSAYO"] = ensayo
try:
res = O.automatico(origen="prueba")
self.assertEqual([(c["cliente"], c["estado"]) for c in res["carpetas"]], [("Auto normal", "con dudas"), ("Sin disco", "sin conexion")])
c = res["carpetas"][0]
self.assertEqual((c["movimientos"], c["dudosos"], c["errores"]), (2, 1, []), c)
self.assertTrue(os.path.isfile(os.path.join(normal, "2026", "04- Abril", "00_Grabacion", "10- Viernes", "SONY", "C0001.MP4")))
self.assertTrue(os.path.isdir(os.path.join(normal, "2026", "04- Abril", "01 - Botox")), "tocó lo dudoso")
self.assertEqual(O.estado_automatico()["ultimo"]["origen"], "prueba")
self.assertFalse(O.AUTO["activo"])
# con más movimientos que el límite, la carpeta no se toca y se dice por qué
O.CONFIG["raices"] = [r for r in todas if r["cliente"] == "Auto demasiado"]
O.CONFIG["automatico"]["limite"] = 5
c = O.automatico(origen="prueba")["carpetas"][0]
self.assertEqual(c["estado"], "demasiado", c)
self.assertTrue(os.path.isfile(os.path.join(mucho, "2026", "04- Abril", "Grabacion", "C0000.MP4")), "movió pese al límite")
# encenderlo da de alta la tarea de Windows (aquí, en una carpeta de ensayo) con la orden «automatico»
self.assertIn("error", O.configurar_automatico(True, "25:00"))
r = O.configurar_automatico(True, "03:30")
self.assertTrue(r.get("ok") and r["activo"] and r["hora"] == "03:30", r)
with open(os.path.join(ensayo, "tarea.json"), encoding="utf-8") as fh:
tarea = json.load(fh)
self.assertTrue(tarea["activo"] and tarea["hora"] == "03:30" and tarea["orden"].endswith(" automatico"), tarea)
self.assertFalse(O.configurar_automatico(False, "03:30")["activo"])
finally:
os.environ.pop("ORGANILAB_ENSAYO", None)
O.CONFIG["raices"] = todas
O.CONFIG["automatico"] = ajuste
# el lector de BRAW de mentira: habla como braw.py (una línea JSON por evento) sin decodificar nada
LECTOR_FALSO = """# -*- coding: utf-8 -*-
import os, sys, json, time
t = json.load(open(sys.argv[2], encoding="utf-8"))
modo = os.environ.get("BRAW_FALSO", "bien")
def di(**kw):
print(json.dumps(kw), flush=True)
di(msg="2160x4096 · 30 fps · 60 cuadros · curva de prueba")
if modo == "falla":
di(estado="error", log="el decodificador no pudo abrir el clip (prueba)")
sys.exit(1)
with open(t["tmp"], "wb") as fh:
fh.write(b"x" * 200000)
if modo == "lento":
for i in range(300):
di(pct=i / 3.0)
if os.path.exists(t["cancelar"]):
os.remove(t["tmp"])
di(estado="cancelado")
sys.exit(1)
time.sleep(0.1)
di(pct=50.0)
di(estado="listo", pct=100, segundos=0.1)
"""
class Convertidor(unittest.TestCase):
"""El convertidor, medido el 5-oct-2026 sobre la carpeta de Andrea Galeano: los BRAW no habían salido
nunca (87 clips, 462 GB, tres tandas fallidas por la vía de Resolve) y lo que ya venía bien comprimido
se reconvertía para ahorrar un 5 %, con una generación de pérdida encima."""
MOTORES = {"ffmpeg": True, "gpu": {"nombre": "", "hevc": False, "h264": False, "chroma422": False, "uhq": False},
"cpu": {"libx265": True, "libx264": True, "prores_ks": True}, "formatos": {k: "cpu" for k in O.FORMATOS}, "chroma422": True}
def setUp(self):
antes = (O.decodificador_braw, O.motores, O.medir_braw, O.ffprobe)
def devolver():
O.decodificador_braw, O.motores, O.medir_braw, O.ffprobe = antes
O.COLA["cancelar"] = False
os.environ.pop("BRAW_FALSO", None)
self.de_verdad = antes
self.addCleanup(devolver)
O.decodificador_braw = lambda: {"ok": True, "carpeta": "X:\\decodificador"}
O.motores = lambda: self.MOTORES
def leidos(self, raiz, datos):
"""Apunta en el inventario lo que ffprobe habría leído de cada clip."""
con = O.db()
with con:
for nombre, d in datos.items():
d = dict({"perfil": "", "pix_fmt": "yuv420p", "ancho": 3840, "alto": 2160, "fps": 30, "dur": 60, "acodec": "aac"}, **d)
n = con.execute("UPDATE archivos SET analizado = 1, codec = ?, perfil = ?, pix_fmt = ?, ancho = ?, alto = ?, fps = ?, dur = ?, "
"kbps = ?, acodec = ? WHERE raiz = ? AND nombre = ?",
(d["codec"], d["perfil"], d["pix_fmt"], d["ancho"], d["alto"], d["fps"], d["dur"], d["kbps"], d["acodec"],
os.path.normpath(raiz), nombre)).rowcount
assert n == 1, nombre
con.close()
def trabajo(self, id_):
con = O.db()
t = dict(con.execute("SELECT * FROM trabajos WHERE id = ?", (id_,)).fetchone())
con.close()
return t
def test_se_convierte_lo_que_ahorra_un_30(self):
"""Munka, 14-oct-2026: «si un archivo se puede optimizar así sea un 30 % se convierta». Medido sobre las 286
conversiones hechas: un H.265 de cámara se queda en el 85-99 % (6 se reconvirtieron para nada), y en cambio
los AV1 de 85 Mb/s que no se ofrecían quedan en la décima parte."""
raiz = cliente("Comprimidos")
g = "2025/06- Junio/00_Grabacion/"
for n in ("A014_06170738_C019.mp4", "C0001.MP4", "C0002.MP4", "C0003.MP4", "C0004.MP4", "C0005.MP4", "C0006.MP4",
"A015_05250930_C004.mp4", "Duplicados/C0003.MP4"):
mk(raiz, g + n, 2025, 6, 17)
O.escanear(raiz)
self.leidos(raiz, {
"A014_06170738_C019.mp4": {"codec": "h264", "perfil": "Main", "ancho": 4608, "alto": 2592, "kbps": 25000},
"C0001.MP4": {"codec": "h264", "perfil": "High 4:2:2 Intra", "pix_fmt": "yuv422p10le", "fps": 59.94, "kbps": 596000},
"C0002.MP4": {"codec": "hevc", "perfil": "Main 10", "pix_fmt": "yuv420p10le", "fps": 59.94, "kbps": 72000},
"C0004.MP4": {"codec": "av1", "perfil": "Main", "kbps": 22000},
"C0005.MP4": {"codec": "prores", "perfil": "HQ", "pix_fmt": "yuv422p10le", "kbps": 870000},
"C0006.MP4": {"codec": "hevc", "perfil": "Main 10", "pix_fmt": "yuv420p10le", "kbps": 200000},
"A015_05250930_C004.mp4": {"codec": "av1", "perfil": "Main", "ancho": 2160, "alto": 3840, "fps": 59.94, "kbps": 83000},
})
con = O.db()
with con:
con.execute("UPDATE archivos SET analizado = 1, codec = 'h264', perfil = 'High', pix_fmt = 'yuv420p', ancho = 3840, alto = 2160, "
"fps = 30, dur = 60, kbps = 120000 WHERE nombre = 'C0003.MP4'") # el clip y su copia en Duplicados
con.close()
nr = os.path.normpath(raiz)
cands = {os.path.relpath(c["ruta"], os.path.join(nr, g.replace("/", S))): c for c in O.candidatos(nr, 0, "equilibrado", 0)}
cs = {k: c["excluido"] for k, c in cands.items()}
# un H.265 de cámara casi no adelgaza: no se ofrece; uno muy holgado (el techo es 1,8 veces el objetivo), sí
self.assertTrue(cs["C0002.MP4"].startswith("convertido ahorraría solo un 7 % (H.265 a 72 Mb/s): se convierte desde un 30 %"), cs)
self.assertEqual(cs["C0006.MP4"], "")
# el H.264 por debajo del objetivo y los AV1 (el de móvil a 83 Mb/s, uno a 22), que antes no se ofrecían, ahora sí
self.assertEqual([cs["A014_06170738_C019.mp4"], cs["A015_05250930_C004.mp4"], cs["C0004.MP4"]], ["", "", ""])
# lo que guarda cada cuadro entero (XAVC-I, ProRes) y el H.264 muy holgado, como siempre
self.assertEqual([cs["C0001.MP4"], cs["C0003.MP4"], cs["C0005.MP4"]], ["", "", ""])
# lo apartado como repetido no se convierte: se borra
self.assertTrue(cs[S.join(("Duplicados", "C0003.MP4"))].startswith("está apartado como repetido"), cs)
# lo que se estima: un AV1 a 83 Mb/s, como mucho el objetivo (49 % menos); un H.264 a 25, como mucho 0,6 veces
a = cands["A015_05250930_C004.mp4"]
self.assertAlmostEqual(a["estimado"] / (83000 * 1000 / 8 * 60), 1 - 0.49, delta=0.01)
# el mínimo se cambia: pidiendo un 45 %, el H.264 a 25 Mb/s (39 %) ya no entra
cs45 = {os.path.basename(c["ruta"]): c["excluido"] for c in O.candidatos(nr, 0, "equilibrado", 0, ahorro=45)}
self.assertTrue(cs45["A014_06170738_C019.mp4"].startswith("convertido ahorraría solo un 39 %"), cs45)
self.assertEqual(cs45["A015_05250930_C004.mp4"], "")
# a un formato de edición no se le aplica: ahí no se busca que pese menos
cs = {os.path.basename(c["ruta"]): c["excluido"] for c in O.candidatos(nr, 0, "equilibrado", 0, False, "prores_lt")}
self.assertEqual(cs["C0002.MP4"], "")
# la lista de acciones cuenta lo que se puede aligerar
ajustes = dict(O.CONFIG["optimizar"])
self.addCleanup(lambda: O.CONFIG["optimizar"].update(ajustes))
O.CONFIG["optimizar"].update(min_mb=0, meses_minimos=0)
conv = [a for a in O.acciones(raiz)["acciones"] if a["clave"] == "convertir"]
self.assertEqual(len(conv), 1)
self.assertTrue(conv[0]["cifra"].startswith("7 clips"), conv[0])
# y después de convertir se descarta lo que ahorre menos del mínimo con 10 puntos de margen (con 30, del 20 %)
t = {"formato": "h265_10bit", "ruta": "C0002.MP4", "bytes_in": 100, "bytes_out": 81}
self.assertTrue(O._no_compensa(t).startswith("no compensa: pesa el 81 %"))
self.assertEqual(O._no_compensa(dict(t, bytes_out=79)), "")
O.CONFIG["optimizar"]["ahorro_minimo"] = 50
self.assertTrue(O._no_compensa(dict(t, bytes_out=61)).startswith("no compensa"))
def test_un_nombre_parecido_no_es_una_conversion(self):
"""El sufijo solo no basta: «Entrevista_HQ.mp4» en H.264 no lo hizo este programa."""
raiz = cliente("Parecidos")
g = "2025/06- Junio/00_Grabacion/"
for n in ("Entrevista_HQ.mp4", "C0001_10bit.mp4", "C0002_HQ.mov", "C0003_h264.mp4"):
mk(raiz, g + n, 2025, 6, 17)
O.escanear(raiz)
self.leidos(raiz, {"Entrevista_HQ.mp4": {"codec": "h264", "perfil": "High", "kbps": 400000},
"C0001_10bit.mp4": {"codec": "hevc", "perfil": "Main 10", "kbps": 400000},
"C0002_HQ.mov": {"codec": "prores", "perfil": "HQ", "kbps": 870000},
"C0003_h264.mp4": {"codec": "hevc", "perfil": "Main", "kbps": 400000}})
cs = {os.path.basename(c["ruta"]): c["excluido"] for c in O.candidatos(os.path.normpath(raiz), 0, "equilibrado", 0, True)}
self.assertEqual(cs["Entrevista_HQ.mp4"], "")
self.assertEqual(cs["C0003_h264.mp4"], "") # dice _h264 y es H.265: tampoco es nuestro
self.assertIn("ya es una conversión", cs["C0001_10bit.mp4"])
self.assertIn("ya es una conversión", cs["C0002_HQ.mov"])
def test_los_braw_van_por_el_lector(self):
raiz = cliente("Braws")
g = "2025/12- Diciembre/00_Grabacion/URSA_BRAW/"
a = mk(raiz, g + "A014_12091814_C001.braw", 2025, 12, 9)
b = mk(raiz, g + "A014_12091819_C002.braw", 2025, 12, 9)
O.escanear(raiz)
como_ffprobe = {"codec": "braw", "pix_fmt": "", "ancho": 4640, "alto": 2624, "kbps": 560000, "acodec": "pcm_s24le"}
self.leidos(raiz, {"A014_12091814_C001.braw": como_ffprobe, "A014_12091819_C002.braw": como_ffprobe})
nr = os.path.normpath(raiz)
cs = O.candidatos(nr, 0, "equilibrado", 0)
self.assertEqual([(c["motor"], c["excluido"]) for c in cs], [("braw", ""), ("braw", "")])
# con el lector, un BRAW sale a cualquier formato
self.assertEqual([c["excluido"] for c in O.candidatos(nr, 0, "equilibrado", 0, False, "prores_lt")], ["", ""])
# sin decodificador se intenta por Resolve, y por ahí solo a H.265 de 10 bits
O.decodificador_braw = lambda: {"ok": False, "motivo": "no está"}
self.assertEqual({c["motor"] for c in O.candidatos(nr, 0, "equilibrado", 0)}, {"resolve"})
self.assertIn("Resolve", O.candidatos(nr, 0, "equilibrado", 0, False, "prores_lt")[0]["excluido"])
self.assertEqual(O.encolar([a], "equilibrado", "420", "rec709", nr, "prores_lt"), [])
viejo = O.encolar([a], "equilibrado", "420", "rec709", nr)
self.assertEqual(self.trabajo(viejo[0])["motor"], "resolve")
# así quedó la cola de verdad: cada clip fallido dos veces, y uno apuntando a donde ya no está
con = O.db()
with con:
con.execute("UPDATE trabajos SET estado = 'error', log = 'Resolve está en la ventana del Gestor de proyectos' WHERE id = ?", (viejo[0],))
for ruta in (a, os.path.join(nr, "Orden", "A014_12091814_C001.braw")):
con.execute("INSERT INTO trabajos (ruta, salida, motor, perfil, chroma, color, kbps, estado, bytes_in, creado, raiz, formato) "
"VALUES (?,?,?,?,?,?,?,?,?,?,?,?)", (ruta, ruta[:-5] + "_10bit.mp4", "resolve", "equilibrado", "420", "rec709", 31047,
"error", 1, time.time(), nr, "h265_10bit"))
con.close()
self.assertEqual(O.braws_al_lector(), 0) # sin decodificador no se toca nada
O.decodificador_braw = lambda: {"ok": True, "carpeta": "X:\\decodificador"}
self.assertEqual(O.braws_al_lector(), 1)
con = O.db()
filas = [tuple(r) for r in con.execute("SELECT ruta, motor, estado FROM trabajos WHERE raiz = ?", (nr,))]
con.close()
self.assertEqual(filas, [(a, "braw", "error")])
# volver a encolar el clip sustituye el intento fallido en vez de dejar otro al lado
nuevos = O.encolar([a, b], "equilibrado", "420", "rec709", nr)
self.assertEqual(len(nuevos), 2)
con = O.db()
filas = sorted(tuple(r) for r in con.execute("SELECT ruta, motor, estado FROM trabajos WHERE raiz = ?", (nr,)))
con.execute("DELETE FROM trabajos WHERE raiz = ?", (nr,))
con.commit()
con.close()
self.assertEqual(filas, [(a, "braw", "pendiente"), (b, "braw", "pendiente")])
def test_un_braw_pasa_por_la_cola(self):
"""La cola con un lector de mentira: lo que importa aquí es lo que hace el programa con lo que el lector dice."""
raiz = cliente("ColaBraw")
g = "2025/03- Marzo/00_Grabacion/URSA_BRAW/"
clips = [mk(raiz, g + f"A002_0311042{i}_C04{i}.braw", 2025, 3, 11, tam=1_000_000) for i in (4, 5, 6)]
mk(raiz, g + "A002_03110424_C044.sidecar", 2025, 3, 11)
O.escanear(raiz)
self.leidos(raiz, {os.path.basename(c): {"codec": "braw", "pix_fmt": "", "ancho": 4128, "alto": 2176, "kbps": 407000, "dur": 2.0,
"acodec": "pcm_s24le"} for c in clips})
nr = os.path.normpath(raiz)
# el decodificador dice el tamaño de verdad: el clip es vertical, y ffprobe lo leyó tumbado y con relleno
O.medir_braw = lambda ruta: {"ok": True, "ancho": 2160, "alto": 4096, "fps": 30.0, "cuadros": 60, "dur": 2.0, "tc": "04:27:26:19",
"camara": "Blackmagic Pocket Cinema Camera 4K", "gamma": "Blackmagic Design Film", "gamut": "Blackmagic Design", "gen": 5}
lector = os.path.join(APP, "braw.py")
shutil.copy(lector, lector + ".de-verdad")
self.addCleanup(lambda: os.replace(lector + ".de-verdad", lector))
with open(lector, "w", encoding="utf-8") as fh:
fh.write(LECTOR_FALSO)
ids = O.encolar(clips, "equilibrado", "420", "rec709", nr)
self.assertEqual(len(ids), 3)
# 1) sale bien
O.correr_braw(self.trabajo(ids[0]))
t = self.trabajo(ids[0])
self.assertEqual(t["estado"], "listo", t["log"])
self.assertEqual(os.path.getsize(t["salida"]), 200000)
self.assertIn("curva de prueba", t["log"])
self.assertEqual(t["kbps"], O.kbps_objetivo(2160, 4096, 30.0, "equilibrado")) # la tasa, con el tamaño de verdad
con = O.db()
f = con.execute("SELECT ancho, alto, perfil FROM archivos WHERE ruta = ?", (clips[0],)).fetchone()
con.close()
self.assertEqual((f["ancho"], f["alto"], f["perfil"]), (2160, 4096, "Film"))
self.assertEqual([x for x in os.listdir(O.DATOS) if x.startswith("braw_")], [])
# la verificación compara con ese tamaño: una salida tumbada no pasa
bien = {"codec": "hevc", "perfil": "Main 10", "pix_fmt": "yuv420p10le", "ancho": 2160, "alto": 4096, "fps": 30.0, "dur": 2.0, "kbps": 20000,
"alfa": 0, "acodec": "aac", "pistas_audio": 1}
O.ffprobe = lambda ruta: dict(bien, ancho=4128, alto=2176)
v = O.verificar([ids[0]])[0]
self.assertFalse(v["ok"])
self.assertIn("2160×4096", v["motivo"])
O.set_trabajo(ids[0], estado="listo")
O.ffprobe = lambda ruta: dict(bien)
self.assertTrue(O.verificar([ids[0]])[0]["ok"])
# al reemplazar, el .sidecar del BRAW se va con él
self.assertTrue(O.reemplazar([ids[0]])[0]["ok"])
apartados = os.path.join(nr, R.PAPELERA, "originales", "2025", "03- Marzo", "00_Grabacion", "URSA_BRAW")
self.assertEqual(sorted(os.listdir(apartados)), ["A002_03110424_C044.braw", "A002_03110424_C044.sidecar"])
self.assertTrue(os.path.isfile(t["salida"]))
# 2) el lector dice que no puede: el motivo llega tal cual y no queda nada a medias
os.environ["BRAW_FALSO"] = "falla"
O.correr_braw(self.trabajo(ids[1]))
t = self.trabajo(ids[1])
self.assertEqual((t["estado"], t["log"]), ("error", "el decodificador no pudo abrir el clip (prueba)"))
self.assertFalse(os.path.exists(t["salida"]) or os.path.exists(t["salida"] + ".parcial.mp4"))
# 3) Detener a mitad de un clip
os.environ["BRAW_FALSO"] = "lento"
hilo = threading.Thread(target=O.correr_braw, args=(self.trabajo(ids[2]),))
hilo.start()
fin = time.time() + 20
while time.time() < fin and not (self.trabajo(ids[2])["pct"] or 0) > 0:
time.sleep(0.1)
O.COLA["cancelar"] = True
hilo.join(40)
self.assertFalse(hilo.is_alive())
t = self.trabajo(ids[2])
self.assertEqual(t["estado"], "cancelado")
self.assertFalse(os.path.exists(t["salida"]) or os.path.exists(t["salida"] + ".parcial.mp4"))
self.assertEqual([x for x in os.listdir(O.DATOS) if x.startswith("braw_")], [])
# 4) si ya hay un archivo con el nombre de la salida, no se pisa
O.COLA["cancelar"] = False
os.environ["BRAW_FALSO"] = "bien"
O.set_trabajo(ids[1], estado="pendiente")
with open(self.trabajo(ids[1])["salida"], "wb") as fh:
fh.write(b"de otro")
O.correr_braw(self.trabajo(ids[1]))
t = self.trabajo(ids[1])
self.assertEqual(t["estado"], "error")
self.assertIn("no se pisa", t["log"])
with open(t["salida"], "rb") as fh:
self.assertEqual(fh.read(), b"de otro")
def test_lo_que_no_compensa_no_reemplaza(self):
"""8 conversiones esperaban con «Reemplazar 8 verificados»: seis eran copias de una copia, al 85-99 % de su peso."""
raiz = cliente("NoCompensa")
g = "2025/11- Noviembre/00_Grabacion/SONY/"
clips = [mk(raiz, g + n, 2025, 11, 11, tam=300000) for n in ("C0001_10bit.mp4", "C0002.MP4", "C0003.MP4")]
O.escanear(raiz)
nr = os.path.normpath(raiz)
self.leidos(raiz, {"C0001_10bit.mp4": {"codec": "hevc", "perfil": "Main 10", "kbps": 72000},
"C0002.MP4": {"codec": "h264", "perfil": "High", "kbps": 120000},
"C0003.MP4": {"codec": "h264", "perfil": "High", "kbps": 120000}})
ids = []
con = O.db()
with con:
# una reconversión (93 %), un clip que apenas aligera (95 %) y uno que sí (40 %)
for ruta, pesa in zip(clips, (280000, 285000, 120000)):
salida = os.path.splitext(ruta)[0] + "_10bit.mp4"
with open(salida, "wb") as fh:
fh.write(b"x" * pesa)
ids.append(con.execute("INSERT INTO trabajos (ruta, salida, motor, perfil, chroma, color, kbps, estado, bytes_in, creado, raiz, formato) "
"VALUES (?,?,?,?,?,?,?,?,?,?,?,?)", (ruta, salida, "ffmpeg", "equilibrado", "420", "rec709", 21150, "listo", 300000,
time.time(), nr, "h265_10bit")).lastrowid)
con.close()
O.ffprobe = lambda ruta: {"codec": "hevc", "perfil": "Main 10", "pix_fmt": "yuv420p10le", "ancho": 3840, "alto": 2160, "fps": 30.0, "dur": 60.0,
"kbps": 20000, "alfa": 0, "acodec": "aac", "pistas_audio": 1}
v = O.verificar(ids)
self.assertEqual([x["ok"] for x in v], [False, False, True])
self.assertIn("93 %", v[0]["motivo"])
self.assertIn("ya era una conversión", v[0]["motivo"])
self.assertIn("95 %", v[1]["motivo"])
self.assertEqual([self.trabajo(i)["estado"] for i in ids], ["descartado", "descartado", "verificado"])
# lo descartado ni se reemplaza…
r = O.reemplazar(ids)
self.assertEqual([x["ok"] for x in r], [False, False, True])
self.assertEqual([os.path.isfile(c) for c in clips], [True, True, False])
# …ni aunque viniera dado por bueno de una versión anterior: se repasa al arrancar y al reemplazar
O.set_trabajo(ids[0], estado="verificado")
O.set_trabajo(ids[1], estado="verificado")
r = O.reemplazar([ids[1]])[0]
self.assertFalse(r["ok"])
self.assertTrue(r["motivo"].startswith("no compensa"), r)
self.assertEqual(O.repasar_verificados(), 1)
self.assertEqual([self.trabajo(i)["estado"] for i in ids[:2]], ["descartado", "descartado"])
self.assertTrue(os.path.isfile(clips[0]) and os.path.isfile(clips[1]))
# y el clip no se vuelve a ofrecer con la misma calidad; con otra, sí
cs = {os.path.basename(c["ruta"]): c["excluido"] for c in O.candidatos(nr, 0, "equilibrado", 0)}
self.assertTrue(cs["C0002.MP4"].startswith("ya se probó con esta calidad: pesa el 95 %"), cs)
cs = {os.path.basename(c["ruta"]): c["excluido"] for c in O.candidatos(nr, 0, "ahorro", 0)}
self.assertEqual(cs["C0002.MP4"], "")
# a un formato de edición no se le pide que pese menos
self.assertEqual(O._no_compensa({"formato": "prores_lt", "ruta": clips[1], "bytes_in": 100, "bytes_out": 900}), "")
def test_lo_que_no_compensa_se_aparta(self):
"""Lo que salió de una conversión que no compensa no se queda junto al original ocupando su nombre, y un
intento que no pasó la verificación se puede rehacer."""
raiz = cliente("Apartar")
g = "2025/11- Noviembre/00_Grabacion/SONY/"
a, b, c = (mk(raiz, g + n, 2025, 11, 11, tam=300000) for n in ("C0001.MP4", "C0002.MP4", "C0003.MP4"))
O.escanear(raiz)
nr = os.path.normpath(raiz)
self.leidos(raiz, {n: {"codec": "h264", "perfil": "High", "kbps": 120000} for n in ("C0001.MP4", "C0002.MP4", "C0003.MP4")})
O.ffprobe = lambda ruta: {"codec": "hevc", "perfil": "Main 10", "pix_fmt": "yuv420p10le", "ancho": 3840, "alto": 2160, "fps": 30.0, "dur": 60.0,
"kbps": 20000, "alfa": 0, "acodec": "aac", "pistas_audio": 1}
ids = O.encolar([a, b, c], "equilibrado", "420", "rec709", nr)
self.assertEqual(len(ids), 3)
for i in ids: # tres conversiones que salieron casi del mismo peso
with open(self.trabajo(i)["salida"], "wb") as fh:
fh.write(b"x" * 290000)
O.set_trabajo(i, estado="listo", bytes_out=290000)
apartado = lambda t: os.path.join(nr, R.PAPELERA, "descartados", os.path.relpath(t["salida"], nr))
# 1) recién convertida y no compensa: lo que salió se aparta en el acto, con su lote en la bitácora
v = O.verificar([ids[0]])[0]
self.assertFalse(v["ok"])
self.assertIn("descartados", v["motivo"])
t0 = self.trabajo(ids[0])
self.assertFalse(os.path.exists(t0["salida"]))
self.assertTrue(os.path.isfile(apartado(t0)))
self.assertTrue(os.path.isfile(a)) # el original, intacto
lotes = [e["lote"] for e in O.leer_bitacora() if e.get("op") == "fin" and e.get("raiz") == nr and e["lote"].startswith("descartados-")]
self.assertEqual(len(lotes), 1)
O.deshacer(lotes[0]) # y «Deshacer» lo devuelve
self.assertTrue(os.path.isfile(t0["salida"]))
# 2) las que venían «verificadas» de antes: al arrancar solo cambian de estado, sin mover nada…
O.set_trabajo(ids[1], estado="verificado")
O.set_trabajo(ids[2], estado="verificado")
self.assertEqual(O.repasar_verificados(), 2)
t1, t2 = self.trabajo(ids[1]), self.trabajo(ids[2])
self.assertEqual((t1["estado"], t2["estado"]), ("descartado", "descartado"))
self.assertTrue(os.path.isfile(t1["salida"]) and os.path.isfile(t2["salida"]))
# …volver a convertir el clip con otra calidad aparta lo que salió de aquel y deja libre la ruta…
nuevos = O.encolar([b], "ahorro", "420", "rec709", nr)
self.assertEqual(len(nuevos), 1)
self.assertFalse(os.path.exists(t1["salida"]))
self.assertTrue(os.path.isfile(apartado(t1)))
# …y «Limpiar terminados» aparta lo de los que quedaran
r = O.apartar_descartados([ids[2]])
self.assertEqual((r["movidos"], r["errores"], r["quedan"]), (1, [], []))
self.assertFalse(os.path.exists(t2["salida"]))
self.assertTrue(os.path.isfile(apartado(t2)))
# 3) un intento que no pasó la verificación se rehace: su propia salida no bloquea la ruta; otra sí
tn = self.trabajo(nuevos[0])
with open(tn["salida"], "wb") as fh:
fh.write(b"y" * 1000)
O.set_trabajo(nuevos[0], estado="error", bytes_out=1000, log="verificación: duración 3.00s vs original 60.00s")
self.assertFalse(O._salida_ajena(self.trabajo(nuevos[0])))
with open(tn["salida"], "wb") as fh:
fh.write(b"z" * 999) # alguien dejó otro archivo con ese nombre
self.assertTrue(O._salida_ajena(self.trabajo(nuevos[0])))
def test_la_cola_sigue_al_clip(self):
"""Línea Gene, 14-oct-2026: el plan llevó «Material Sony» a 00_Grabacion; 138 conversiones en espera
fallaron buscando cada clip donde ya no estaba, los 11 verificados no se podían reemplazar (y su «_10bit»
se quedó en la carpeta vieja), y volver a escanear no los reenganchaba."""
raiz = cliente("ColaSigue")
nr = os.path.normpath(raiz)
g = "Material Sony/"
a = mk(raiz, g + "20250722_Labstream3796.MP4", 2025, 7, 22, tam=300000)
b = mk(raiz, g + "20250722_Labstream3797.MP4", 2025, 7, 22, tam=300001)
c = mk(raiz, g + "20250731_Labstream4033.MP4", 2025, 7, 31, tam=300002)
d = mk(raiz, g + "20250731_Labstream4034.MP4", 2025, 7, 31, tam=300003)
O.escanear(raiz)
self.leidos(raiz, {os.path.basename(x): {"codec": "h264", "perfil": "High", "kbps": 120000} for x in (a, b, c, d)})
ia, ib, ic = O.encolar([a, b, c], "equilibrado", "420", "rec709", nr)
O.set_trabajo(ib, estado="error", log="falló antes por otra cosa")
# el plan se calcula antes de que c termine de convertirse: su «_10bit» aún no está en el inventario
p = O.Plan(O.raiz_cfg(raiz)).correr()
sc = self.trabajo(ic)["salida"]
with open(sc, "wb") as fh:
fh.write(b"x" * 120000)
O.set_trabajo(ic, estado="verificado", bytes_out=120000)
r = aplicar(raiz, p)
self.assertFalse(os.path.exists(a) or os.path.exists(c))
# 1) los trabajos se fueron con sus clips; el que espera sale junto al clip en su sitio nuevo
ta, tb, tc = (self.trabajo(i) for i in (ia, ib, ic))
for t in (ta, tb, tc):
self.assertTrue(os.path.isfile(t["ruta"]), t)
self.assertIn("00_Grabacion", t["ruta"])
self.assertEqual(ta["salida"], os.path.splitext(ta["ruta"])[0] + "_10bit.mp4")
self.assertEqual((ta["estado"], tb["estado"]), ("pendiente", "error"))
# …y el convertido acompañó a su original, aunque el plan no lo conocía
self.assertEqual(os.path.dirname(tc["salida"]), os.path.dirname(tc["ruta"]))
self.assertTrue(os.path.isfile(tc["salida"]))
self.assertFalse(os.path.exists(sc))
# «Buscar clips» no vuelve a ofrecer lo que ya está en la cola o convertido
cs = {os.path.basename(x["ruta"]): x["en_cola"] for x in O.candidatos(nr, 0, "equilibrado", 0)}
self.assertEqual((cs["20250722_Labstream3796.MP4"], cs["20250731_Labstream4033.MP4"], cs["20250731_Labstream4034.MP4"]),
("pendiente", "verificado", ""))
# 2) Deshacer los devuelve, y la cola vuelve con ellos
O.deshacer(r["lote"])
ta, tc = self.trabajo(ia), self.trabajo(ic)
self.assertEqual((ta["ruta"], tc["ruta"], tc["salida"]), (a, c, sc))
self.assertTrue(os.path.isfile(sc))
self.assertEqual(ta["salida"], os.path.splitext(a)[0] + "_10bit.mp4")
# 3) movidos con el Explorador: justo antes de convertir, el que no está falla diciendo qué hacer…
otra = os.path.join(nr, "Otra carpeta")
os.makedirs(otra)
a2, c2 = os.path.join(otra, os.path.basename(a)), os.path.join(otra, os.path.basename(c))
os.rename(a, a2)
os.rename(c, c2) # el original verificado se va; su convertido, no
self.assertIsNone(O._clip_presente(self.trabajo(ia)))
ta = self.trabajo(ia)
self.assertEqual(ta["estado"], "error")
self.assertIn("Vuelve a escanear", ta["log"])
# …y al volver a escanear se reengancha: vuelve a la cola en su sitio nuevo, sin volver a leer el clip
O.escanear(raiz)
ta, tc = self.trabajo(ia), self.trabajo(ic)
self.assertEqual((ta["ruta"], ta["estado"]), (a2, "pendiente"))
self.assertEqual(ta["salida"], os.path.splitext(a2)[0] + "_10bit.mp4")
self.assertTrue(ta["log"].startswith("el clip se había movido"), ta["log"])
self.assertEqual((tc["ruta"], tc["salida"]), (c2, sc))
con = O.db()
leido = con.execute("SELECT analizado, codec FROM archivos WHERE ruta = ?", (a2,)).fetchone()
con.close()
self.assertEqual(tuple(leido), (1, "h264"))
self.assertEqual(O._clip_presente(self.trabajo(ia))["ruta"], a2)
# 4) un intento fallido que ya se volvió a encolar en el sitio nuevo sobra: se quita
bn = os.path.join(otra, os.path.basename(b))
os.rename(b, bn)
O.escanear(raiz)
self.assertEqual(self.trabajo(ib)["ruta"], bn) # el de b (error) se reenganchó y vuelve a esperar
self.assertEqual(self.trabajo(ib)["estado"], "pendiente")
con = O.db()
with con:
viejo = con.execute("INSERT INTO trabajos (ruta, salida, motor, perfil, chroma, color, kbps, estado, bytes_in, creado, raiz, formato) "
"VALUES (?,?,?,?,?,?,?,?,?,?,?,?)", (b, b + "_10bit.mp4", "ffmpeg", "equilibrado", "420", "rec709", 21150, "error",
300001, time.time(), nr, "h265_10bit")).lastrowid
con.close()
O.reubicar_trabajos(nr)
con = O.db()
self.assertIsNone(con.execute("SELECT id FROM trabajos WHERE id = ?", (viejo,)).fetchone())
con.close()
# 5) reemplazar con el convertido en otra carpeta: primero se lleva junto al original
r = O.reemplazar([ic])[0]
self.assertTrue(r["ok"], r)
junto = os.path.join(otra, os.path.basename(sc))
self.assertTrue(os.path.isfile(junto))
self.assertFalse(os.path.exists(sc) or os.path.exists(c2))
self.assertTrue(os.path.isfile(os.path.join(nr, R.PAPELERA, "originales", "Otra carpeta", os.path.basename(c2))))
self.assertEqual(self.trabajo(ic)["salida"], junto)
def test_lo_ya_convertido_no_se_convierte_dos_veces(self):
"""Línea Gene, 14-oct-2026: con la 3.7 los 11 verificados se volvieron a encolar en su sitio nuevo, y uno ya
se había convertido otra vez. Al reengancharlos: la conversión que aún espera sobra; la ya hecha deja de más
la vieja."""
raiz = cliente("DosVeces")
nr = os.path.normpath(raiz)
a = mk(raiz, "Material Sony/20250722_Labstream3805.MP4", 2025, 7, 22, tam=300000)
b = mk(raiz, "Material Sony/20250722_Labstream3766.MP4", 2025, 7, 22, tam=300001)
O.escanear(raiz)
self.leidos(raiz, {os.path.basename(x): {"codec": "h264", "perfil": "High", "kbps": 120000} for x in (a, b)})
ia, ib = O.encolar([a, b], "equilibrado", "420", "rec709", nr)
for i in (ia, ib):
with open(self.trabajo(i)["salida"], "wb") as fh:
fh.write(b"x" * 90000)
O.set_trabajo(i, estado="verificado", bytes_out=90000)
# la 3.7: los clips se movieron, la cola no se enteró, y se volvieron a encolar en el sitio nuevo
nueva = os.path.join(nr, "2025", "07- Julio", "00_Grabacion", "22- Martes", "SONY")
os.makedirs(nueva)
a2, b2 = os.path.join(nueva, os.path.basename(a)), os.path.join(nueva, os.path.basename(b))
os.rename(a, a2)
os.rename(b, b2)
with open(os.path.splitext(a2)[0] + "_10bit.mp4", "wb") as fh: # el de a ya se convirtió otra vez allí
fh.write(b"y" * 90000)
con = O.db()
with con:
ids = []
for ruta, estado in ((a2, "verificado"), (b2, "pendiente")):
ids.append(con.execute("INSERT INTO trabajos (ruta, salida, motor, perfil, chroma, color, kbps, estado, bytes_in, bytes_out, creado, raiz, formato) "
"VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?)", (ruta, os.path.splitext(ruta)[0] + "_10bit.mp4", "ffmpeg", "equilibrado", "420",
"rec709", 21150, estado, os.path.getsize(ruta), 90000, time.time(), nr,
"h265_10bit")).lastrowid)
con.close()
O.escanear(raiz)
ta, tb = self.trabajo(ia), self.trabajo(ib)
# el de a ya se había vuelto a convertir: sobra el viejo, que queda «no compensa» con su porqué
self.assertEqual(ta["estado"], "descartado")
self.assertIn("se volvió a convertir en su sitio nuevo", ta["log"])
self.assertEqual(self.trabajo(ids[0])["estado"], "verificado")
# el de b aún esperaba: se quita de la cola, y el viejo se reengancha (se reemplaza junto a su original)
con = O.db()
self.assertIsNone(con.execute("SELECT id FROM trabajos WHERE id = ?", (ids[1],)).fetchone())
con.close()
self.assertEqual((tb["estado"], tb["ruta"]), ("verificado", b2))
r = O.reemplazar([ib])[0]
self.assertTrue(r["ok"], r)
self.assertTrue(os.path.isfile(os.path.join(nueva, os.path.basename(tb["salida"]))))
def test_el_codificador_que_toca(self):
fm = O.FORMATOS["h265_10bit"]
t = {"chroma": "420", "perfil": "equilibrado", "kbps": 30000}
con_uhq = dict(self.MOTORES, gpu=dict(self.MOTORES["gpu"], hevc=True, h264=True, uhq=True), formatos={k: "gpu" for k in O.FORMATOS})
O.motores = lambda: con_uhq
args, pix = O._args_video(t, fm, "gpu", 30)
self.assertEqual((pix, args[args.index("-tune") + 1], args[args.index("-profile:v") + 1]), ("p010le", "uhq", "main10"))
self.assertNotIn("-bf", args) # «uhq» no admite que se le fijen los cuadros B
self.assertEqual(O._args_video(dict(t, chroma="422"), fm, "gpu", 30)[1], "p210le")
self.assertIn("hq", O._args_video(dict(t, chroma="422"), fm, "gpu", 30)[0]) # solo donde se midió: 4:2:0
self.assertIn("hq", O._args_video(t, O.FORMATOS["h264"], "gpu", 30)[0])
O.motores = lambda: dict(con_uhq, gpu=dict(con_uhq["gpu"], uhq=False))
args, _ = O._args_video(t, fm, "gpu", 30)
self.assertEqual(args[args.index("-tune") + 1], "hq")
self.assertIn("-bf", args)
# la tarjeta tiene un tamaño máximo de imagen por códec: un 4.6K en H.264 falla («No capable devices found»)
self.assertEqual(O._motor_de("h264", "420", 4608, 2592)[0], "cpu")
self.assertEqual(O._motor_de("h264", "420", 3840, 2160)[0], "gpu")
self.assertEqual(O._motor_de("h265_10bit", "420", 4608, 2592)[0], "gpu")
self.assertEqual(O._motor_de("h265_10bit", "420", 12288, 6480)[0], "cpu")
# el contenedor guarda las etiquetas de la cámara, sin arrastrar las marcas de formato del original
cont, audio = O._args_salida(fm, None)
self.assertIn("+faststart+use_metadata_tags", cont)
self.assertIn("major_brand=", cont)
self.assertEqual(audio, ["-c:a", "aac", "-b:a", "320k"])
def test_el_lector_sin_decodificador(self):
"""Lo que braw.py hace sin tocar el decodificador, y cómo responde si lo que encuentra no sirve."""
import braw
self.assertEqual([braw.fraccion(x) for x in (23.976, 24.0, 25.0, 29.97, 30.0, 59.94, 60.0)],
["24000/1001", "24", "25", "30000/1001", "30", "60000/1001", "60"])
falso = os.path.join(TMP, "decodificador-falso")
os.makedirs(falso)
self.assertNotEqual(braw.buscar(falso), falso)
with open(os.path.join(falso, braw.NOMBRE), "wb") as fh:
fh.write(b"esto no es una DLL")
self.assertEqual(braw.buscar(falso), falso)
r = braw.sonda(falso)
self.assertFalse(r["ok"])
self.assertTrue(r["motivo"])
r = braw.info(os.path.join(TMP, "no-existe.braw"), falso)
self.assertFalse(r["ok"])
self.assertTrue(r["error"].startswith("el clip ya no está en esa ruta"), r)
@unittest.skipUnless(os.environ.get("ORGANILAB_BRAW_PRUEBA"), "sin un .braw de prueba (variable ORGANILAB_BRAW_PRUEBA)")
def test_un_braw_de_verdad(self):
"""Con el decodificador instalado y un clip real: se mide, se convierte, se verifica."""
O.decodificador_braw, O.motores, O.medir_braw, O.ffprobe = self.de_verdad
self.assertTrue(O.decodificador_braw()["ok"], O.decodificador_braw())
raiz = cliente("BrawReal")
destino = os.path.join(raiz, "2025", "03- Marzo", "00_Grabacion", "URSA_BRAW", "A002_03110347_C038.braw")
os.makedirs(os.path.dirname(destino))
shutil.copy(os.environ["ORGANILAB_BRAW_PRUEBA"], destino)
O.escanear(raiz)
nr = os.path.normpath(raiz)
O.analizar(nr, 0)
real = O.medir_braw(destino)
self.assertTrue(real["ok"], real)
c = O.candidatos(nr, 0, "equilibrado", 0)[0]
self.assertEqual((c["motor"], c["excluido"], c["res"]), ("braw", "", f"{real['ancho']}x{real['alto']}"))
ids = O.encolar([destino], "equilibrado", "420", "rec709", nr)
O.correr_braw(self.trabajo(ids[0]))
t = self.trabajo(ids[0])
self.assertEqual(t["estado"], "listo", t["log"])
v = O.verificar(ids)[0]
self.assertTrue(v["ok"], v)
i = O.ffprobe(t["salida"])
self.assertEqual((i["codec"], i["pix_fmt"], i["ancho"], i["alto"]), ("hevc", "yuv420p10le", real["ancho"], real["alto"]))
if __name__ == "__main__":
try:
r = unittest.main(verbosity=2, exit=False)
finally:
shutil.rmtree(TMP, ignore_errors=True)
sys.exit(0 if r.result.wasSuccessful() else 1) # la compilación del estudio se detiene si algo falla
Referencia en una nueva incidencia Ver la culpa de Git Copiar Permalink