Labstream/labstream-organizador
1
0
Fork 0
Código Incidencias Pull Requests Paquetes Proyectos Lanzamientos Wiki Actividad
Files
6c21145b1c0ec3a271d8f1ee84c68d6e434588ce
labstream-organizador/prueba_ventana.py
Labstream 252b1fd00e 3.0.0: OrganiLAB pasa a ser un programa de Windows 
- ventana.py: la ventana es un marco Win32 propio con WebView2 dentro,
  manejado por COM con ctypes. Fuera pywebview, pythonnet y clr_loader y
  sus binarios sin firmar. Identidad en la barra de tareas, una sola copia
  abierta por carpeta de datos, selector de carpetas de Windows, rutas
  completas al soltar, avance en la barra de tareas, memoria de posición,
  aviso al cerrar con trabajo en marcha, y nada de navegador (menú, F5,
  zoom, permisos). Si el motor se cae, se monta de nuevo solo.
- arranque.py y empaquetar.py: el paquete ya no lleva binario propio.
  OrganizadorLabstream.exe es pythonw.exe (firmado por la PSF) con otro
  nombre y arranca el programa desde sitecustomize; abre con Smart App
  Control encendido. Fuera PyInstaller. Al empaquetar se comprueba la firma
  de cada binario con Windows y se arranca el motor del paquete.
- instalar.py: accesos directos en Inicio y Escritorio con la identidad de
  la ventana, y la línea en Configuración › Aplicaciones. Se ofrece una vez,
  la primera que se abre un paquete en un equipo.
- Dos copias sobre los mismos datos ya no se pisan: la segunda trae la
  ventana al frente (antes abría el navegador y daba por huérfanas las
  conversiones en marcha de la primera). Si el puerto 8765 está ocupado
  por otro programa, se toma uno libre.
- Por consola, el paquete ya no consulta si hay versión nueva con las
  actualizaciones automáticas apagadas.
- Pruebas: probar.py 17, prueba_funcional.py 26 de 26 y prueba_ventana.py
  (nueva), que abre la ventana de verdad: 17 comprobaciones desde el código
  y 8 arrancando el paquete por su .exe.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
2026-10-04 09:54:31 -05:00

374 líneas
18 KiB
Python
Original Blame Histórico

Este archivo contiene caracteres Unicode ambiguos

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

# -*- coding: utf-8 -*-
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
… --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", "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 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")
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 oculta, vuelve si se abre otra vez y se cierra sola al acabar
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 y el programa sigue", 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 trabajo termina: el programa tiene que cerrarse 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)
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 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]))
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())
Referencia en una nueva incidencia Ver la culpa de Git Copiar Permalink