Munka: «que al cerrar cierre todos los procesos… que al continuar la cola siga por donde iba y vuelva a generar los videos que hagan falta… que identifique bien el material… y convertir antes de mover». Cerrar con trabajo avisa y lo para todo (cola, lectura, respaldo, traer), cierra lo que quede con lo que lanzó y borra lo que quedó a medias. Lo cortado vuelve a esperar en su sitio (no «cancelado») y al continuar se hace entero; también vuelven los fallos por un corte y los convertidos que ya no están. Actualizar con la cola en marcha la detiene y sigue sola al volver. El Convertidor enseña lo grande de la carpeta y qué pasa con cada parte (exportados sin incluir, con un botón). Convertir va antes de ordenar en los pasos y en Inicio; lo que se está convirtiendo no se mueve, y una carpeta con algo abierto ya no se copia entera. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
474 líneas
24 KiB
Python
474 líneas
24 KiB
Python
# -*- coding: utf-8 -*-
|
||
r"""
|
||
Prueba de la ventana, con la ventana de verdad. probar.py y prueba_funcional.py ejercitan el programa
|
||
sin abrirlo; esto abre OrganiLAB sobre una copia temporal (programa y datos aparte, en otro puerto) y
|
||
comprueba desde otro hilo lo que solo se puede comprobar con la ventana delante:
|
||
|
||
py -3.12 prueba_ventana.py el programa desde el código
|
||
py -3.12 prueba_ventana.py --paquete dist\OrganizadorLabstream el paquete, arrancado por su .exe, y su instalación
|
||
… --captura ruta.png deja una imagen de la ventana entera
|
||
|
||
Durante medio minuto se ve una ventana que se abre, se mueve sola y se cierra. No hay que tocar nada.
|
||
Sale con código 0 si todo pasó.
|
||
"""
|
||
import os
|
||
import sys
|
||
import json
|
||
import time
|
||
import zlib
|
||
import shutil
|
||
import socket
|
||
import struct
|
||
import ctypes
|
||
import tempfile
|
||
import threading
|
||
import subprocess
|
||
import urllib.request
|
||
from ctypes import wintypes as W
|
||
|
||
AQUI = os.path.dirname(os.path.abspath(__file__))
|
||
MODULOS = ("organizador.py", "reglas.py", "asistente.py", "actualizador.py", "resolve_bridge.py", "braw.py", "firma.py",
|
||
"ventana.py", "instalar.py", "WebView2Loader.dll")
|
||
RES = [] # (nombre, pasó, detalle)
|
||
U = ctypes.WinDLL("user32")
|
||
G = ctypes.WinDLL("gdi32")
|
||
ENUM = ctypes.WINFUNCTYPE(W.BOOL, W.HWND, W.LPARAM)
|
||
for _n, _a, _r in (("GetWindow", [W.HWND, W.UINT], W.HWND), ("GetClassNameW", [W.HWND, W.LPWSTR, ctypes.c_int], ctypes.c_int),
|
||
("GetWindowTextW", [W.HWND, W.LPWSTR, ctypes.c_int], ctypes.c_int), ("IsWindowVisible", [W.HWND], W.BOOL),
|
||
("EnumWindows", [ENUM, W.LPARAM], W.BOOL), ("EnumChildWindows", [W.HWND, ENUM, W.LPARAM], W.BOOL),
|
||
("SendMessageW", [W.HWND, W.UINT, W.WPARAM, ctypes.c_void_p], ctypes.c_ssize_t),
|
||
("PostMessageW", [W.HWND, W.UINT, W.WPARAM, W.LPARAM], W.BOOL), ("GetDlgItem", [W.HWND, ctypes.c_int], W.HWND),
|
||
("GetWindowDC", [W.HWND], W.HANDLE), ("ReleaseDC", [W.HWND, W.HANDLE], ctypes.c_int),
|
||
("PrintWindow", [W.HWND, W.HANDLE, W.UINT], W.BOOL), ("GetWindowRect", [W.HWND, ctypes.POINTER(W.RECT)], W.BOOL)):
|
||
getattr(U, _n).argtypes, getattr(U, _n).restype = _a, _r
|
||
for _n, _a, _r in (("CreateCompatibleDC", [W.HANDLE], W.HANDLE), ("CreateCompatibleBitmap", [W.HANDLE, ctypes.c_int, ctypes.c_int], W.HANDLE),
|
||
("SelectObject", [W.HANDLE, W.HANDLE], W.HANDLE), ("DeleteObject", [W.HANDLE], W.BOOL), ("DeleteDC", [W.HANDLE], W.BOOL),
|
||
("GetDIBits", [W.HANDLE, W.HANDLE, W.UINT, W.UINT, ctypes.c_void_p, ctypes.c_void_p, W.UINT], ctypes.c_int)):
|
||
getattr(G, _n).argtypes, getattr(G, _n).restype = _a, _r
|
||
WM_CLOSE, WM_COMMAND, WM_SETTEXT, WM_GETICON = 0x0010, 0x0111, 0x000C, 0x007F
|
||
|
||
|
||
def anota(nombre, ok, detalle=""):
|
||
RES.append((nombre, bool(ok), str(detalle)))
|
||
print((" bien " if ok else " MAL ") + nombre + (f" · {detalle}" if detalle else ""), flush=True)
|
||
|
||
|
||
def espera(condicion, plazo=15.0, paso=0.1):
|
||
fin = time.time() + plazo
|
||
while time.time() < fin:
|
||
r = condicion()
|
||
if r:
|
||
return r
|
||
time.sleep(paso)
|
||
return condicion()
|
||
|
||
|
||
def puerto_libre():
|
||
s = socket.socket()
|
||
s.bind(("127.0.0.1", 0))
|
||
p = s.getsockname()[1]
|
||
s.close()
|
||
return p
|
||
|
||
|
||
def clase(h):
|
||
b = ctypes.create_unicode_buffer(200)
|
||
U.GetClassNameW(h, b, 200)
|
||
return b.value
|
||
|
||
|
||
def texto(h):
|
||
b = ctypes.create_unicode_buffer(2000)
|
||
U.GetWindowTextW(h, b, 2000)
|
||
return b.value
|
||
|
||
|
||
def hijos(h):
|
||
out = []
|
||
U.EnumChildWindows(h, ENUM(lambda x, _: out.append(x) or True), 0)
|
||
return out
|
||
|
||
|
||
def dialogo_de(hwnd):
|
||
"""El cuadro de diálogo de Windows abierto sobre esa ventana, si hay alguno."""
|
||
out = []
|
||
U.EnumWindows(ENUM(lambda x, _: (out.append(x) if clase(x) == "#32770" and U.GetWindow(x, 4) == hwnd and U.IsWindowVisible(x) else None) or True), 0)
|
||
return out[0] if out else None
|
||
|
||
|
||
def marco(hwnd, ruta):
|
||
"""La ventana entera (barra de título incluida), tal como la pinta Windows, a un PNG."""
|
||
r = W.RECT()
|
||
U.GetWindowRect(hwnd, ctypes.byref(r))
|
||
ancho, alto = r.right - r.left, r.bottom - r.top
|
||
dc = U.GetWindowDC(hwnd)
|
||
mem = G.CreateCompatibleDC(dc)
|
||
bmp = G.CreateCompatibleBitmap(dc, ancho, alto)
|
||
G.SelectObject(mem, bmp)
|
||
U.PrintWindow(hwnd, mem, 2)
|
||
cab = ctypes.create_string_buffer(struct.pack("<IiiHHIIiiII", 40, ancho, -alto, 1, 32, 0, 0, 0, 0, 0, 0), 64)
|
||
buf = ctypes.create_string_buffer(ancho * alto * 4)
|
||
G.GetDIBits(mem, bmp, 0, alto, buf, cab, 0)
|
||
G.DeleteObject(bmp), G.DeleteDC(mem), U.ReleaseDC(hwnd, dc)
|
||
filas, paso, bgra = bytearray(), ancho * 4, buf.raw
|
||
for y in range(alto):
|
||
fila, rgb = bgra[y * paso:(y + 1) * paso], bytearray(ancho * 3)
|
||
rgb[0::3], rgb[1::3], rgb[2::3] = fila[2::4], fila[1::4], fila[0::4]
|
||
filas += b"\x00" + rgb
|
||
|
||
def trozo(t, d):
|
||
return struct.pack(">I", len(d)) + t + d + struct.pack(">I", zlib.crc32(t + d) & 0xffffffff)
|
||
with open(ruta, "wb") as fh:
|
||
fh.write(b"\x89PNG\r\n\x1a\n" + trozo(b"IHDR", struct.pack(">IIBBBBB", ancho, alto, 8, 2, 0, 0, 0))
|
||
+ trozo(b"IDAT", zlib.compress(bytes(filas), 6)) + trozo(b"IEND", b""))
|
||
return ancho, alto
|
||
|
||
|
||
def cliente_de_ensayo(tmp):
|
||
raiz = os.path.join(tmp, "Cliente de ensayo")
|
||
for rel in ("Rodaje/C0001.MP4", "Entregas/Video final.mp4", "notas.txt"):
|
||
p = os.path.join(raiz, rel)
|
||
os.makedirs(os.path.dirname(p), exist_ok=True)
|
||
with open(p, "wb") as fh:
|
||
fh.write(os.urandom(2048))
|
||
return raiz
|
||
|
||
|
||
def marca_de(datos):
|
||
return (zlib.crc32(os.path.normcase(os.path.abspath(datos)).encode("utf-8")) & 0x7FFFFFFF) or 1
|
||
|
||
|
||
def cerrar_las_de(V, *marcas):
|
||
"""Cierra las copias de ensayo que sigan abiertas (con sus cuadros de diálogo), pase lo que pase en la
|
||
prueba: una copia olvidada se queda en la pantalla y, peor, atendiendo."""
|
||
for _ in range(3):
|
||
abiertas = [h for m in marcas for h in V._ventanas(m)]
|
||
if not abiertas:
|
||
return
|
||
for h in abiertas:
|
||
d = dialogo_de(h)
|
||
if d:
|
||
U.PostMessageW(d, WM_CLOSE, 0, 0)
|
||
U.PostMessageW(h, WM_CLOSE, 0, 0)
|
||
time.sleep(3)
|
||
|
||
|
||
def captura_pedida():
|
||
return sys.argv[sys.argv.index("--captura") + 1] if "--captura" in sys.argv else ""
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# desde el código: el programa corre en este mismo proceso y se le puede preguntar todo
|
||
# --------------------------------------------------------------------------
|
||
|
||
def desde_el_codigo(tmp):
|
||
app = os.path.join(tmp, "app")
|
||
os.makedirs(app)
|
||
for f in MODULOS:
|
||
shutil.copy(os.path.join(AQUI, f), app)
|
||
shutil.copytree(os.path.join(AQUI, "static"), os.path.join(app, "static"))
|
||
carpeta = cliente_de_ensayo(tmp)
|
||
puerto = puerto_libre()
|
||
os.environ["ORGANIZADOR_PUERTO"] = str(puerto)
|
||
sys.path.insert(0, app)
|
||
sys.argv = [os.path.join(app, "organizador.py")]
|
||
import organizador as O
|
||
import ventana as V
|
||
base = f"http://127.0.0.1:{puerto}"
|
||
|
||
def pedir(ruta, datos=None):
|
||
rq = urllib.request.Request(base + ruta, None if datos is None else json.dumps(datos).encode(),
|
||
{"Content-Type": "application/json", "Origin": base})
|
||
return json.loads(urllib.request.urlopen(rq, timeout=90).read())
|
||
|
||
def otra_copia():
|
||
return subprocess.run([sys.executable, os.path.join(app, "organizador.py")], cwd=app, timeout=40,
|
||
capture_output=True, text=True, encoding="utf-8", errors="replace")
|
||
|
||
def pruebas():
|
||
v = espera(lambda: O.VENTANA if O.VENTANA is not None and O.VENTANA.lista else None, 60)
|
||
anota("la ventana abre y carga la interfaz", v is not None, "" if v else "no cargó en 60 s")
|
||
if v is None:
|
||
os._exit(1)
|
||
time.sleep(1.5)
|
||
h = v.hwnd
|
||
anota("es una ventana de Windows propia, con su título y su icono",
|
||
clase(h) == V.CLASE and texto(h) == "OrganiLAB" and U.SendMessageW(h, WM_GETICON, 1, None) != 0,
|
||
f"clase {clase(h)} · «{texto(h)}»")
|
||
anota("dentro va la interfaz del programa", v.evaluar("ESTADO.version") == O.VERSION, v.evaluar("ESTADO.version"))
|
||
|
||
def identidad():
|
||
almacen, out = ctypes.c_void_p(), {}
|
||
V._dll().s.SHGetPropertyStoreForWindow(h, ctypes.byref(V._guid(V.IID_PROPIEDADES)), ctypes.byref(almacen))
|
||
for nombre, pid in (("id", 5), ("orden", 2), ("nombre", 4), ("icono", 3)):
|
||
clave, pv = V.PROPERTYKEY(V._guid(V.FMT_APPUSERMODEL), pid), V.PROPVARIANT()
|
||
V._m(almacen.value, 5, ctypes.byref(clave), ctypes.byref(pv), tipos=(ctypes.POINTER(V.PROPERTYKEY), ctypes.POINTER(V.PROPVARIANT)))
|
||
out[nombre] = ctypes.wstring_at(pv.valor) if pv.vt == 31 and pv.valor else ""
|
||
V._soltar(almacen.value)
|
||
return out
|
||
quien = v.en_ui(identidad)
|
||
anota("la barra de tareas la conoce como OrganiLAB y sabe cómo volver a abrirla",
|
||
quien["id"] == V.IDENTIDAD and "organizador.py" in quien["orden"] and quien["nombre"] == "OrganiLAB" and quien["icono"].endswith("organilab.ico"),
|
||
f"{quien['id']} · {quien['orden']}")
|
||
|
||
def ajustes():
|
||
aj = ctypes.c_void_p()
|
||
V._m(v._wv, 3, ctypes.byref(aj), tipos=(V.PPVOID,))
|
||
a3 = V._qi(aj.value, V.IID_AJUSTES3)
|
||
r = {"barra de estado": V._entero(aj.value, 9), "herramientas de desarrollo": V._entero(aj.value, 11),
|
||
"zoom": V._entero(aj.value, 17), "página de error": V._entero(aj.value, 19), "teclas de navegador": V._entero(a3, 23)}
|
||
V._soltar(a3), V._soltar(aj.value)
|
||
return r
|
||
aj = v.en_ui(ajustes)
|
||
anota("lo que es de navegador está apagado", not any(aj.values()), ", ".join(k for k, x in aj.items() if x) or "todo apagado")
|
||
v.evaluar("irA('manual')")
|
||
entradas = v.evaluar("document.querySelector('#t-manual').classList.contains('on') ? document.querySelectorAll('#t-manual dt').length : 0")
|
||
anota("el manual está dentro del programa", (entradas or 0) > 35, f"{entradas} entradas")
|
||
v.evaluar("irA('carpeta')")
|
||
anota("la interfaz sabe que está en la ventana", v.evaluar("document.documentElement.classList.contains('nativo')") is True)
|
||
|
||
# el avance en la barra de tareas
|
||
v.progreso(0.5)
|
||
anota("pone avance en el botón de la barra de tareas", espera(lambda: v.en_ui(lambda: bool(v._barra)), 5))
|
||
v.progreso(None)
|
||
if captura_pedida():
|
||
anota("captura de la ventana", True, "%d×%d px en %s" % (*marco(h, captura_pedida()), captura_pedida()))
|
||
|
||
# el selector de carpetas de Windows, pedido como lo pide la interfaz
|
||
caja = {}
|
||
threading.Thread(target=lambda: caja.update(r=pedir("/api/elegir-carpeta", {})), daemon=True).start()
|
||
d = espera(lambda: dialogo_de(h), 15)
|
||
if d:
|
||
time.sleep(1.0)
|
||
campo = U.GetDlgItem(d, 1152) # el campo «Carpeta:»
|
||
if campo:
|
||
U.SendMessageW(campo, WM_SETTEXT, 0, ctypes.c_wchar_p(carpeta))
|
||
time.sleep(0.3)
|
||
U.SendMessageW(d, WM_COMMAND, 1, U.GetDlgItem(d, 1))
|
||
# con una ruta escrita, el primer «Elegir carpeta» entra en ella y el segundo la elige
|
||
if not espera(lambda: "r" in caja, 3):
|
||
U.SendMessageW(d, WM_COMMAND, 1, U.GetDlgItem(d, 1))
|
||
if not espera(lambda: "r" in caja, 8):
|
||
U.PostMessageW(d, WM_CLOSE, 0, 0)
|
||
espera(lambda: "r" in caja, 8)
|
||
elegida = (caja.get("r") or {}).get("ruta")
|
||
anota("el selector de carpetas de Windows se abre sobre la ventana y devuelve la carpeta",
|
||
bool(d) and elegida and os.path.normcase(elegida) == os.path.normcase(carpeta),
|
||
elegida or ("se abrió pero no devolvió carpeta" if d else "no se abrió"))
|
||
|
||
# soltar una carpeta: el motor entrega los archivos y la ventana lee su ruta completa
|
||
def cdp(metodo, parametros):
|
||
hecho = threading.Event()
|
||
p, sosten = V._objeto("{5C4889F0-5EF6-4C5A-952C-D8F1B92D0574}", (ctypes.c_long, W.LPCWSTR), lambda hr, r: hecho.set(), fijo=False)
|
||
v.en_ui(lambda: V._m(v._wv, 36, metodo, json.dumps(parametros), p, tipos=(W.LPCWSTR, W.LPCWSTR, V.PVOID)))
|
||
hecho.wait(5)
|
||
return sosten
|
||
x, y = v.evaluar("[Math.round(innerWidth * 0.6), Math.round(innerHeight * 0.4)]")
|
||
carga = {"items": [], "files": [carpeta], "dragOperationsMask": 1}
|
||
for tipo in ("dragEnter", "dragOver", "drop"):
|
||
cdp("Input.dispatchDragEvent", {"type": tipo, "x": x, "y": y, "data": carga})
|
||
llego = espera(lambda: [r for r in O.SOLTADOS["rutas"] if os.path.normcase(r) == os.path.normcase(carpeta)], 8)
|
||
anota("una carpeta soltada llega con su ruta completa", bool(llego), (llego or ["no llegó"])[0])
|
||
propuesta = espera(lambda: v.evaluar("document.querySelector('#modal').classList.contains('on') && !!document.querySelector('#nuevoCliente')"), 8)
|
||
anota("y la interfaz propone añadirla", propuesta is True, v.evaluar("document.querySelector('#modalTitulo').textContent"))
|
||
v.evaluar("document.querySelector('#modal').classList.remove('on')")
|
||
|
||
# una sola copia abierta
|
||
p = otra_copia()
|
||
anota("abrirlo otra vez no abre una segunda copia", p.returncode == 0 and len(V._ventanas(O._marca())) == 1,
|
||
f"código {p.returncode} · {len(V._ventanas(O._marca()))} ventana(s)")
|
||
|
||
# cerrar con trabajo en marcha: avisa, se quita de la vista y lo para todo (un movimiento de archivos se deja
|
||
# acabar); si se abre otra vez mientras tanto, vuelve; y al acabar de parar se cierra sola
|
||
O.APLICANDO.update(total=10, hechas=3, raiz="ensayo")
|
||
U.PostMessageW(h, WM_CLOSE, 0, 0)
|
||
d = espera(lambda: dialogo_de(h), 8)
|
||
aviso = " ".join(texto(x) for x in hijos(d)) if d else ""
|
||
anota("al cerrar con trabajo en marcha, avisa de qué está haciendo", "moviendo archivos" in aviso, aviso[:90])
|
||
if d:
|
||
U.PostMessageW(d, WM_COMMAND, 1, 0)
|
||
anota("la ventana se quita de la vista mientras se para todo", espera(lambda: v.oculta and not U.IsWindowVisible(h), 8))
|
||
otra_copia()
|
||
anota("abrirlo otra vez devuelve esa misma ventana", espera(lambda: not v.oculta and U.IsWindowVisible(h), 8))
|
||
U.PostMessageW(h, WM_CLOSE, 0, 0)
|
||
d = espera(lambda: dialogo_de(h), 8)
|
||
if d:
|
||
U.PostMessageW(d, WM_COMMAND, 1, 0)
|
||
espera(lambda: v.oculta, 8)
|
||
O.APLICANDO.update(hechas=10) # el movimiento termina: con todo parado, el programa se cierra solo
|
||
|
||
fin, forzado = threading.Event(), []
|
||
|
||
def con_red():
|
||
try:
|
||
pruebas()
|
||
except Exception as ex: # noqa
|
||
anota("la prueba llegó hasta el final", False, repr(ex))
|
||
O.APLICANDO.update(hechas=O.APLICANDO.get("total", 0))
|
||
if not fin.wait(10) and O.VENTANA is not None: # no se cerró solo: se cierra a la fuerza para no quedarse aquí
|
||
forzado.append(1)
|
||
O.VENTANA.cerrar()
|
||
|
||
threading.Thread(target=con_red, daemon=True).start()
|
||
t0 = time.time()
|
||
O.main()
|
||
fin.set()
|
||
anota("al terminar el trabajo, el programa se cierra solo", not forzado, f"{time.time() - t0:.0f} s en total")
|
||
datos = os.path.join(app, "datos")
|
||
try:
|
||
sitio = json.load(open(os.path.join(datos, "ventana.json"), encoding="utf-8"))
|
||
except Exception: # noqa
|
||
sitio = {}
|
||
anota("recuerda dónde y de qué tamaño quedó", sitio.get("ancho", 0) > 200, json.dumps(sitio))
|
||
anota("no deja rastro de la sesión", not os.path.exists(os.path.join(datos, "sesion.json")))
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# el paquete: se arranca por su .exe, como lo arrancaría una persona, y se mira desde fuera
|
||
# --------------------------------------------------------------------------
|
||
|
||
def el_paquete(tmp, origen):
|
||
sys.path.insert(0, AQUI)
|
||
import ventana as V
|
||
app = os.path.join(tmp, "OrganizadorLabstream")
|
||
shutil.copytree(origen, app, ignore=shutil.ignore_patterns("ffmpeg", "datos", "actualizacion", "__pycache__"))
|
||
exe = os.path.join(app, "OrganizadorLabstream.exe")
|
||
puerto = puerto_libre()
|
||
entorno = dict(os.environ, ORGANIZADOR_PUERTO=str(puerto))
|
||
base = f"http://127.0.0.1:{puerto}"
|
||
p = subprocess.run([exe, "-B", "-c", "import sys; print(sys.version_info[:2])"], capture_output=True, text=True, timeout=60, stdin=subprocess.DEVNULL)
|
||
anota("el .exe del paquete arranca en este Windows", p.returncode == 0, (p.stdout or p.stderr).strip()[:80])
|
||
proc = subprocess.Popen([exe], cwd=tmp, env=entorno)
|
||
|
||
def estado():
|
||
try:
|
||
return json.loads(urllib.request.urlopen(base + "/api/estado", timeout=2).read())
|
||
except Exception: # noqa
|
||
return None
|
||
e = espera(estado, 40, 0.3)
|
||
anota("doble clic en el .exe abre el programa", bool(e), f"versión {e['version']}" if e else "no respondió en 40 s")
|
||
if not e:
|
||
proc.kill()
|
||
return
|
||
anota("corre con el motor de su propia carpeta", e.get("congelado") is True)
|
||
marca = marca_de(os.path.join(app, "datos"))
|
||
hs = espera(lambda: V._ventanas(marca), 30)
|
||
h = hs[0] if hs else None
|
||
anota("con su ventana de Windows", bool(h) and texto(h) == "OrganiLAB" and bool(U.IsWindowVisible(h)), f"clase {clase(h)}" if h else "sin ventana")
|
||
if h:
|
||
time.sleep(4)
|
||
if captura_pedida():
|
||
anota("captura de la ventana", True, "%d×%d px en %s" % (*marco(h, captura_pedida()), captura_pedida()))
|
||
subprocess.run([exe], cwd=tmp, env=entorno, timeout=40)
|
||
anota("abrirlo otra vez no abre una segunda copia", len(V._ventanas(marca)) == 1, f"{len(V._ventanas(marca))} ventana(s)")
|
||
U.PostMessageW(h, WM_CLOSE, 0, 0)
|
||
try:
|
||
proc.wait(25)
|
||
except subprocess.TimeoutExpired:
|
||
proc.kill()
|
||
anota("al cerrar la ventana se cierra el programa", proc.returncode == 0, f"código {proc.returncode}")
|
||
anota("y deja de atender", estado() is None)
|
||
cerrar_las_de(V, marca)
|
||
p = subprocess.run([exe, "ayuda"], capture_output=True, timeout=60, stdin=subprocess.DEVNULL, env=entorno)
|
||
anota("por consola responde a sus comandos", p.returncode == 0 and b"asistente" in p.stdout)
|
||
|
||
|
||
def instalar_el_paquete(tmp, origen):
|
||
"""La instalación de punta a punta, sin tocar el Windows de verdad: con ORGANILAB_ENSAYO el «equipo» es una
|
||
carpeta temporal. El paquete se abre desde unas «Descargas», ofrece instalarse, se copia a su sitio, se da
|
||
de alta, se cierra y vuelve a abrirse desde allí; y abrir después la copia de Descargas abre la instalada."""
|
||
import winreg
|
||
equipo = os.path.join(tmp, "equipo")
|
||
os.environ["ORGANILAB_ENSAYO"] = equipo
|
||
os.environ.pop("ORGANIZADOR_PUERTO", None)
|
||
sys.path.insert(0, AQUI)
|
||
import ventana as V
|
||
import instalar
|
||
descarga = os.path.join(tmp, "Descargas", "OrganizadorLabstream")
|
||
shutil.copytree(origen, descarga, ignore=shutil.ignore_patterns("ffmpeg", "actualizacion", "__pycache__"))
|
||
os.makedirs(os.path.join(descarga, "ffmpeg"))
|
||
with open(os.path.join(descarga, "ffmpeg", "LICENSE.txt"), "w") as fh:
|
||
fh.write("en el paquete completo aquí va ffmpeg")
|
||
destino = instalar.sitio_definitivo()
|
||
exe_d, exe_i = os.path.join(descarga, "OrganizadorLabstream.exe"), os.path.join(destino, "OrganizadorLabstream.exe")
|
||
marca_d, marca_i = marca_de(os.path.join(descarga, "datos")), marca_de(os.path.join(destino, "datos"))
|
||
try:
|
||
proc = subprocess.Popen([exe_d], cwd=tmp)
|
||
hs = espera(lambda: V._ventanas(marca_d), 40)
|
||
if not hs:
|
||
anota("el paquete abre desde Descargas", False)
|
||
proc.kill()
|
||
return
|
||
h = hs[0]
|
||
d = espera(lambda: dialogo_de(h), 25)
|
||
pregunta = " ".join(texto(x) for x in hijos(d)) if d else ""
|
||
anota("la primera vez ofrece instalarlo en el equipo", "Instalar OrganiLAB en este equipo" in pregunta, pregunta[8:70])
|
||
if d:
|
||
U.PostMessageW(d, WM_COMMAND, 6, 0) # «Sí»
|
||
|
||
def aviso_final():
|
||
x = dialogo_de(h)
|
||
return x if x and "quedó instalado" in " ".join(texto(y) for y in hijos(x)) else None
|
||
d = espera(aviso_final, 90)
|
||
anota("se copia a su sitio y lo dice", bool(d))
|
||
if d:
|
||
U.PostMessageW(d, WM_CLOSE, 0, 0) # solo tiene «Aceptar»
|
||
hi = espera(lambda: V._ventanas(marca_i), 60)
|
||
anota("se cierra y vuelve a abrirse solo desde su sitio", bool(hi) and espera(lambda: not V._ventanas(marca_d), 30) is True)
|
||
try:
|
||
proc.wait(30)
|
||
except subprocess.TimeoutExpired:
|
||
proc.kill()
|
||
anota("en su sitio están el programa, sus datos y su ffmpeg",
|
||
all(os.path.exists(os.path.join(destino, x)) for x in ("OrganizadorLabstream.exe", "runtime", "static", "datos\\config.json", "ffmpeg\\LICENSE.txt")))
|
||
a = instalar.leer_acceso(instalar._accesos()[0]) if os.path.exists(instalar._accesos()[0]) else None
|
||
anota("queda dado de alta desde su sitio (Inicio, Escritorio y Aplicaciones)",
|
||
instalar.instalado_aqui(destino) and bool(a) and instalar._misma(a["exe"], exe_i) and os.path.exists(instalar._accesos()[1]),
|
||
a["exe"] if a else "sin acceso directo")
|
||
anota("del origen no se borra nada, y queda la nota de la mudanza", os.path.exists(exe_d) and bool(instalar.mudado(descarga)))
|
||
if hi:
|
||
try:
|
||
with open(os.path.join(destino, "datos", "sesion.json"), encoding="utf-8") as fh:
|
||
puerto = json.load(fh)["puerto"]
|
||
e = json.loads(urllib.request.urlopen(f"http://127.0.0.1:{puerto}/api/estado", timeout=5).read())
|
||
except Exception: # noqa
|
||
e = {}
|
||
anota("el instalado atiende con sus datos", e.get("congelado") is True and (e.get("config") or {}).get("instalacion", {}).get("ofrecida") is True,
|
||
f"versión {e.get('version')}")
|
||
subprocess.run([exe_d], cwd=tmp, timeout=40)
|
||
time.sleep(2.5)
|
||
anota("abrir la copia de Descargas abre la instalada", not V._ventanas(marca_d) and len(V._ventanas(marca_i)) == 1)
|
||
U.PostMessageW(hi[0], WM_CLOSE, 0, 0)
|
||
anota("y se cierra como siempre", espera(lambda: not V._ventanas(marca_i), 25) is True)
|
||
time.sleep(2)
|
||
finally:
|
||
cerrar_las_de(V, marca_d, marca_i)
|
||
for clave in (instalar.CLAVE, "Software\\OrganiLAB-ensayo"):
|
||
try:
|
||
winreg.DeleteKey(winreg.HKEY_CURRENT_USER, clave)
|
||
except OSError:
|
||
pass
|
||
os.environ.pop("ORGANILAB_ENSAYO", None)
|
||
|
||
|
||
def main():
|
||
guardia = threading.Timer(300, lambda: os._exit(3)) # pase lo que pase, no se queda colgada
|
||
guardia.daemon = True
|
||
guardia.start()
|
||
tmp = tempfile.mkdtemp(prefix="organilab-ventana-")
|
||
try:
|
||
if "--paquete" in sys.argv:
|
||
el_paquete(tmp, os.path.abspath(sys.argv[sys.argv.index("--paquete") + 1]))
|
||
instalar_el_paquete(tmp, os.path.abspath(sys.argv[sys.argv.index("--paquete") + 1]))
|
||
else:
|
||
desde_el_codigo(tmp)
|
||
finally:
|
||
shutil.rmtree(tmp, ignore_errors=True)
|
||
mal = [n for n, ok, _ in RES if not ok]
|
||
print(f"\n{len(RES) - len(mal)} de {len(RES)} comprobaciones pasan" + (": fallan " + "; ".join(mal) if mal else ""))
|
||
return 1 if mal else 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|