"""PyScript glue for VMAS XML Compare — Web Studio. All UI chrome (themes, views, drag & drop) lives in ui.js; this module owns the Python side: loading files into the Pyodide FS, running the comparison engine (core.py — auto-synced from the desktop tool), progress reporting and building Blob downloads for the HTML / Excel reports. """ import asyncio import logging import os import traceback from js import document, window, Blob, URL, Object, Uint8Array from pyodide.ffi import create_proxy, to_js ui = window.vmasUI WORK_SRC = "/work/src" WORK_TGT = "/work/tgt" # --------------------------------------------------------------------------- # Logging bridge: python logging -> ui.js log console # --------------------------------------------------------------------------- class UILogHandler(logging.Handler): def emit(self, record): kind = "" if record.levelno >= logging.ERROR: kind = "err" elif record.levelno >= logging.WARNING: kind = "warn" ui.logLine(self.format(record), kind) def setup_logger() -> logging.Logger: fmt = logging.Formatter("%(asctime)s - %(levelname)s - %(message)s", datefmt="%H:%M:%S") handler = UILogHandler() handler.setFormatter(fmt) logger = logging.getLogger("WebLogger") logger.setLevel(logging.INFO) logger.propagate = False logger.addHandler(handler) vmas_logger = logging.getLogger("VMASCompareTool") vmas_logger.setLevel(logging.INFO) vmas_logger.propagate = False vmas_logger.handlers = [handler] return logger log = setup_logger() async def yield_to_ui(): """Give the browser a chance to repaint between heavy steps.""" await asyncio.sleep(0.02) # --------------------------------------------------------------------------- # Blob helpers # --------------------------------------------------------------------------- _active_urls = {} def make_blob_url(key: str, data, mime: str) -> str: """Create an object URL for data (str or bytes), revoking the previous one.""" if key in _active_urls: try: URL.revokeObjectURL(_active_urls[key]) except Exception: pass opts = to_js({"type": mime}, dict_converter=Object.fromEntries) blob = Blob.new(to_js([data]), opts) url = URL.createObjectURL(blob) _active_urls[key] = url return url # --------------------------------------------------------------------------- # Boot: import the engine, optionally fetch xmldiff, unlock the UI # --------------------------------------------------------------------------- CORE_VERSION = "?" try: ui.setBootStatus("Importing comparison engine…") from core import VMASXMLComparator, VERSION as CORE_VERSION # noqa: E402 ui.engineReady(CORE_VERSION) log.info(f"Python engine ready (engine v{CORE_VERSION}, lxml active).") log.info("Select or drop two files (XML or text), then press Run Comparison.") except Exception as boot_err: # pragma: no cover - surfaced in the UI ui.engineError(str(boot_err)) raise # Structural verification note: the web build does not use xmldiff (O(n^2), # not in the Pyodide distribution). core.py ships a fast O(n) structural scan # instead, so no runtime package fetch is needed — fully static, Vercel-safe. # --------------------------------------------------------------------------- # Progress mapping (engine percent -> stage label) # --------------------------------------------------------------------------- STAGES = [ (0, "Preparing…"), (10, "Parsing source XML…"), (25, "Parsing target XML…"), (40, "Comparing modules…"), (85, "Cross-verification & audit…"), (92, "Generating report…"), (100, "Done"), ] def stage_for(pct: int) -> str: label = STAGES[0][1] for threshold, name in STAGES: if pct >= threshold: label = name return label # --------------------------------------------------------------------------- # Comparison flow # --------------------------------------------------------------------------- run_btn = document.getElementById("run-btn") def set_running(running: bool): run_btn.disabled = running run_btn.querySelector(".btn-text").innerText = "Processing…" if running else "Run Comparison" ui.setLed("busy" if running else "ready") async def save_input_file(js_file, directory: str) -> str: """Copy a browser File object into the Pyodide filesystem.""" os.makedirs(directory, exist_ok=True) # src/tgt live in separate directories so same-named files never collide # while the report still shows the real file names. path = os.path.join(directory, js_file.name) buf = await js_file.arrayBuffer() with open(path, "wb") as fh: fh.write(Uint8Array.new(buf).to_bytes()) return path async def run_comparison(event): file1_input = document.getElementById("file1") file2_input = document.getElementById("file2") if file1_input.files.length == 0 or file2_input.files.length == 0: log.error("Please select BOTH Source (File 1) and Target (File 2) files.") return set_running(True) ui.resetProgress() ui.setProgress(2, "Reading files…") await yield_to_ui() f1 = file1_input.files.item(0) f2 = file2_input.files.item(0) try: log.info(f"Reading {f1.name} ({f1.size} bytes) into the virtual filesystem…") p1 = await save_input_file(f1, WORK_SRC) log.info(f"Reading {f2.name} ({f2.size} bytes) into the virtual filesystem…") p2 = await save_input_file(f2, WORK_TGT) await yield_to_ui() comparator = VMASXMLComparator(p1, p2, logger=log) def on_progress(pct): ui.setProgress(pct, stage_for(pct)) comparator.set_progress = on_progress log.info("Starting comparison… large files may take a moment.") ui.setProgress(8, "Comparing…") await yield_to_ui() report = comparator.compare() log.info("Comparison complete. Building interactive HTML report…") ui.setProgress(94, "Rendering HTML report…") await yield_to_ui() report_path = "/work/report.html" comparator.generate_html_report(report, report_path) with open(report_path, "r", encoding="utf-8") as fh: report_html = fh.read() html_url = make_blob_url("html", report_html, "text/html;charset=utf-8") document.getElementById("report-frame").src = html_url document.getElementById("download-report").href = html_url document.getElementById("open-report-tab").href = html_url ui.setProgress(100, "Done") # Summary line for the log if report.get("is_text_mode"): lines1 = report.get("lines_file1", 0) lines2 = report.get("lines_file2", 0) n_moved = len(report.get("moved_lines", [])) summary = f"Text comparison complete ({lines1} vs {lines2} lines)." if n_moved: summary += f" {n_moved} moved line(s) detected." log.info(summary + " Opening report…") else: n_add = len(report.get("added_modules", [])) n_del = len(report.get("deleted_modules", [])) n_mod = len(report.get("modified_modules", {})) if n_add or n_del or n_mod: log.info(f"Differences found: +{n_add} added / -{n_del} deleted / " f"~{n_mod} modified modules. Opening report…") else: log.info("No module-level differences found. Opening report…") files_label = document.getElementById("report-files") files_label.innerHTML = "" files_label.append(f1.name) arrow = document.createElement("b") arrow.textContent = " → " files_label.appendChild(arrow) files_label.append(f2.name) files_label.title = f"{f1.name} → {f2.name}" await yield_to_ui() ui.showView("report") except Exception as exc: log.error(f"Error during comparison: {exc}") log.error(traceback.format_exc()) ui.setLed("error") ui.resetProgress() finally: set_running(False) run_proxy = create_proxy(run_comparison) run_btn.addEventListener("click", run_proxy)