e74b8f7ece
- Tras «Aplicar» en una carpeta grande el inventario se ponía al día con la base tomada más de diez minutos y «Buscar clips» se rendía con «database is locked». Ahora las preguntas al disco van antes y en paralelo y lo de dentro de cada carpeta se busca por índice: 1,3 s para 2.493 movimientos, mismas filas de archivos que antes y 183 carpetas que antes se perdían. La lectura de clips reintenta si la base está ocupada. - Las fotos de «Alta» y «Baja» chocaban al juntarse en una carpeta: las subcarpetas con sentido se conservan, el plan avisa antes de dos archivos iguales en una fila, y un número entre dos palabras («Fiesta 6 Años») ya no se quita. - Los fallos de «Aplicar» se agrupan por motivo y dicen qué hacer; el banco de material se explica en llano. - Fechas raras (el reloj de una cámara), escaneo profundo, buscar en todos los clientes, traer lo que el cliente tiene en otros discos, el banco por temas o por fecha, y «Calcular de nuevo» más rápido. - Mover reintenta un momento si el archivo está abierto un instante (lectura de clips, antivirus). - .github/workflows/publicar.yml para compilar y publicar desde un PC del estudio; las pruebas devuelven error si algo falla. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
1170 líneas
75 KiB
Python
1170 líneas
75 KiB
Python
# -*- 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_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_lo_ya_comprimido_no_se_reconvierte(self):
|
||
"""9 clips que ya eran H.265 se reconvirtieron para quedarse en el 95 % de su peso."""
|
||
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"):
|
||
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},
|
||
"C0003.MP4": {"codec": "h264", "perfil": "High", "kbps": 120000},
|
||
"C0004.MP4": {"codec": "av1", "perfil": "Main", "kbps": 30000},
|
||
"C0005.MP4": {"codec": "prores", "perfil": "HQ", "pix_fmt": "yuv422p10le", "kbps": 870000},
|
||
})
|
||
nr = os.path.normpath(raiz)
|
||
cs = {os.path.basename(c["ruta"]): c["excluido"] for c in O.candidatos(nr, 0, "equilibrado", 0)}
|
||
self.assertTrue(cs["A014_06170738_C019.mp4"].startswith("ya viene comprimido (H.264 a 25 Mb/s)"), cs) # por debajo del objetivo (30)
|
||
self.assertTrue(cs["C0002.MP4"].startswith("ya viene comprimido (H.265 a 72 Mb/s)"), cs)
|
||
self.assertTrue(cs["C0004.MP4"].startswith("ya viene comprimido (AV1 a 30 Mb/s)"), cs)
|
||
# lo que guarda cada cuadro entero (XAVC-I, ProRes) y el H.264 muy holgado sí adelgazan de verdad
|
||
self.assertEqual([cs["C0001.MP4"], cs["C0003.MP4"], cs["C0005.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["A014_06170738_C019.mp4"], "")
|
||
# y la lista de acciones cuenta solo 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("3 clips"), conv[0])
|
||
|
||
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_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
|