maestro/runtime/excel-recalculate.py

#!/usr/bin/env python3
"""Recalculate one trusted local XLSX through a dedicated LibreOffice UNO pipe.

The helper accepts exactly one JSON line on stdin and writes exactly one JSON
line on stdout. It never emits user-provided cell values or formulas.
"""

import json
import os
import sys
import time

import uno


CANARY_PREFIX = "_MAESTRO_RECALC_"
CONNECT_ATTEMPTS = 300
CONNECT_DELAY_SECONDS = 0.1


class RecalculationError(Exception):
    """Known failure with a stable, non-sensitive machine code."""


def property_value(name, value):
    prop = uno.createUnoStruct("com.sun.star.beans.PropertyValue")
    prop.Name = name
    prop.Value = value
    return prop


def connect(pipe_name):
    local_context = uno.getComponentContext()
    resolver = local_context.ServiceManager.createInstanceWithContext(
        "com.sun.star.bridge.UnoUrlResolver", local_context
    )
    url = f"uno:pipe,name={pipe_name};urp;StarOffice.ComponentContext"
    last_error = None
    for _ in range(CONNECT_ATTEMPTS):
        try:
            return resolver.resolve(url)
        except Exception as error:  # UNO raises several bridge-specific types.
            last_error = error
            time.sleep(CONNECT_DELAY_SECONDS)
    raise RecalculationError("uno_pipe_connection_failed") from last_error


def column_name(index):
    value = index + 1
    result = ""
    while value:
        value, remainder = divmod(value - 1, 26)
        result = chr(65 + remainder) + result
    return result


def collect_formula_errors(document):
    errors = {}
    formula_flag = uno.getConstantByName(
        "com.sun.star.sheet.CellFlags.FORMULA"
    )
    sheets = document.getSheets()
    for sheet_name in sheets.getElementNames():
        if sheet_name.startswith(CANARY_PREFIX):
            continue
        sheet = sheets.getByName(sheet_name)
        formula_ranges = sheet.queryContentCells(formula_flag)
        for area in formula_ranges.getRangeAddresses():
            for row in range(area.StartRow, area.EndRow + 1):
                for column in range(area.StartColumn, area.EndColumn + 1):
                    cell = sheet.getCellByPosition(column, row)
                    error_code = int(cell.getError())
                    if error_code:
                        address = f"{column_name(column)}{row + 1}"
                        errors[f"{sheet_name}!{address}"] = error_code
    return errors


def find_canary_value(document):
    sheets = document.getSheets()
    canaries = [
        name for name in sheets.getElementNames() if name.startswith(CANARY_PREFIX)
    ]
    if len(canaries) != 1:
        raise RecalculationError("recalculation_canary_missing_or_ambiguous")
    return sheets.getByName(canaries[0]).getCellRangeByName("A1").getValue()


def recalculate(request):
    input_path = os.path.realpath(os.path.abspath(request["input_path"]))
    output_path = os.path.realpath(os.path.abspath(request["output_path"]))
    if input_path == output_path:
        raise RecalculationError("input_and_output_paths_must_differ")
    context = connect(request["pipe"])
    desktop = context.ServiceManager.createInstanceWithContext(
        "com.sun.star.frame.Desktop", context
    )
    input_url = uno.systemPathToFileUrl(input_path)
    output_url = uno.systemPathToFileUrl(output_path)
    load_properties = (
        property_value("Hidden", True),
        property_value(
            "UpdateDocMode",
            uno.getConstantByName("com.sun.star.document.UpdateDocMode.NO_UPDATE"),
        ),
        property_value(
            "MacroExecutionMode",
            uno.getConstantByName(
                "com.sun.star.document.MacroExecMode.NEVER_EXECUTE"
            ),
        ),
    )
    document = desktop.loadComponentFromURL(input_url, "_blank", 0, load_properties)
    if document is None:
        raise RecalculationError("spreadsheet_load_failed")
    try:
        document.calculateAll()
        formula_errors = collect_formula_errors(document)
        canary_value = find_canary_value(document)
        document.storeToURL(
            output_url,
            (
                property_value("FilterName", "Calc MS Excel 2007 XML"),
                property_value("Overwrite", True),
            ),
        )
        return {
            "ok": True,
            "formula_error_codes": formula_errors,
            "canary_value": canary_value,
        }
    finally:
        try:
            document.close(True)
        except Exception:
            document.dispose()


def main():
    try:
        request = json.loads(sys.stdin.readline())
        # profile_uri is applied by the Node-owned soffice process and retained
        # in this boundary contract so the caller must identify that isolated profile.
        required = ("input_path", "output_path", "pipe", "profile_uri")
        if not isinstance(request, dict) or any(
            not isinstance(request.get(key), str) or not request[key] for key in required
        ):
            raise ValueError("invalid_request")
        response = recalculate(request)
    except RecalculationError as error:
        response = {"ok": False, "error": str(error)}
    except Exception as error:
        response = {"ok": False, "error": type(error).__name__}
    sys.stdout.write(json.dumps(response, separators=(",", ":")) + "\n")
    sys.stdout.flush()
    return 0 if response.get("ok") else 1


if __name__ == "__main__":
    raise SystemExit(main())