#!/usr/bin/env python3 """Compare declared no-op XLSX round trips from WolfXL and headless LibreOffice. The producer writes a receipt bundle for each invocation. It is deliberately a package-fidelity comparison, not a claim about Excel or general spreadsheet parity. LibreOffice is invoked only through isolated headless profiles. """ from __future__ import annotations import argparse import hashlib import importlib.util import json import shutil import sys import tempfile from collections import Counter from datetime import datetime, timezone from pathlib import Path from types import ModuleType from typing import Any from zipfile import BadZipFile, ZipFile SCRIPT_DIR = Path(__file__).resolve().parent ROOT = SCRIPT_DIR.parent def _load_script_module(name: str, path: Path) -> ModuleType: spec = importlib.util.spec_from_file_location(name, path) if spec is None or spec.loader is None: raise ImportError(f"could not load {name} from {path}") module = importlib.util.module_from_spec(spec) sys.modules[name] = module spec.loader.exec_module(module) return module audit_ooxml_fidelity = _load_script_module( "audit_ooxml_fidelity", SCRIPT_DIR / "audit_ooxml_fidelity.py" ) codex_runtime_identity = _load_script_module( "codex_runtime_identity", SCRIPT_DIR / "codex_runtime_identity.py" ) libreoffice_adapter = _load_script_module( "competitor_workload_libreoffice_adapter", SCRIPT_DIR / "competitor_workload_libreoffice_adapter.py", ) from wolfxl import load_workbook # noqa: E402 SCHEMA_VERSION = 1 DEFAULT_TIMEOUT_SECONDS = 120 def _utc_now() -> str: return datetime.now(timezone.utc).isoformat() def _sha256(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as handle: for chunk in iter(lambda: handle.read(1024 * 1024), b""): digest.update(chunk) return digest.hexdigest() def _write_json(path: Path, value: Any) -> None: path.write_text(json.dumps(value, indent=2, sort_keys=True) + "\n", encoding="utf-8") def _write_jsonl(path: Path, values: list[dict[str, Any]]) -> None: with path.open("w", encoding="utf-8") as handle: for value in values: handle.write(json.dumps(value, sort_keys=True) + "\n") def _safe_name(value: str) -> str: return "".join(character if character.isalnum() or character in "._-" else "_" for character in value) def _part_hashes(path: Path) -> dict[str, str]: with ZipFile(path) as archive: return { name: hashlib.sha256(archive.read(name)).hexdigest() for name in sorted(archive.namelist()) } def _package_part_changes(before: Path, after: Path) -> list[dict[str, Any]]: """Supplement the repository audit with byte hashes for its audited package parts.""" before_hashes = _part_hashes(before) after_hashes = _part_hashes(after) changes: list[dict[str, Any]] = [] for part in sorted(before_hashes.keys() | after_hashes.keys()): before_hash = before_hashes.get(part) after_hash = after_hashes.get(part) if before_hash == after_hash: continue if before_hash is None: kind = "added" elif after_hash is None: kind = "removed" else: kind = "changed" changes.append( { "part": part, "kind": kind, "before_sha256": before_hash, "after_sha256": after_hash, } ) return changes def _source_identity() -> dict[str, Any]: return { **codex_runtime_identity.source_metadata(ROOT), "built_hash_or_baked_constant": _sha256(Path(__file__)), "build_mode": "source_python", } def _scenario_id(input_path: Path, source_sha256: str) -> str: return f"{_safe_name(input_path.stem)}-{source_sha256[:12]}" def _terminal_failure(reason: str, *, started_at_utc: str) -> dict[str, Any]: return { "state": "failed", "returncode": None, "stdout": "", "stderr": reason, "command": [], "started_at_utc": started_at_utc, "completed_at_utc": _utc_now(), } def _audit_preservation( input_path: Path, output_path: Path, *, expected_changed_parts: set[str], ) -> dict[str, Any]: """Run the repository's audit before classifying raw package-byte drift.""" audit = audit_ooxml_fidelity.audit(input_path, output_path, phase_timings=True) package_part_changes = _package_part_changes(input_path, output_path) expected_part_changes = [ change for change in package_part_changes if change["part"] in expected_changed_parts ] unrelated_part_drift = [ change for change in package_part_changes if change["part"] not in expected_changed_parts ] return { "declared_expected_changed_parts": sorted(expected_changed_parts), "package_part_changes": package_part_changes, "expected_declared_part_changes": expected_part_changes, "unrelated_part_drift": unrelated_part_drift, "audit": audit, } def _preservation_status(preservation: dict[str, Any]) -> str: if preservation["unrelated_part_drift"] or preservation["audit"]["issue_count"]: return "drifted" return "passed" def _wolfxl_roundtrip( input_path: Path, output_path: Path, *, source_metadata: dict[str, Any], timeout_seconds: int, ) -> dict[str, Any]: started_at_utc = _utc_now() command = ["wolfxl.load_workbook", str(input_path), "workbook.save", str(output_path)] output_path.parent.mkdir(parents=True, exist_ok=True) try: workbook = load_workbook(input_path) workbook.save(output_path) if not output_path.is_file(): raise FileNotFoundError("WolfXL save completed without an XLSX output") preservation = _audit_preservation(input_path, output_path, expected_changed_parts=set()) except Exception as exc: return { "engine": "wolfxl", "status": "failed", "reason": f"{type(exc).__name__}: {exc}", "terminal_status": _terminal_failure( f"{type(exc).__name__}: {exc}", started_at_utc=started_at_utc ), "metadata": { **source_metadata, "scenario": "ooxml_noop_roundtrip", "iteration_count": 1, "timeout_seconds": timeout_seconds, "cache_state": "not_applicable", "artifact_path": str(output_path), "command": command, "started_at_utc": started_at_utc, "completed_at_utc": _utc_now(), }, } completed_at_utc = _utc_now() return { "engine": "wolfxl", "status": _preservation_status(preservation), "terminal_status": { "state": "completed", "returncode": 0, "stdout": "", "stderr": "", "command": command, "started_at_utc": started_at_utc, "completed_at_utc": completed_at_utc, }, "metadata": { **source_metadata, "scenario": "ooxml_noop_roundtrip", "iteration_count": 1, "timeout_seconds": timeout_seconds, "cache_state": "not_applicable", "artifact_path": str(output_path), "command": command, "started_at_utc": started_at_utc, "completed_at_utc": completed_at_utc, }, "output_sha256": _sha256(output_path), "output_size": output_path.stat().st_size, "preservation": preservation, } def _libreoffice_runtime( soffice_path: str | Path | None, *, work_dir: Path, timeout_seconds: int, ) -> tuple[Path | None, dict[str, Any], dict[str, Any] | None]: requested_path, binary_path = libreoffice_adapter._resolve_soffice_path(soffice_path) identity = libreoffice_adapter.runtime_identity(requested_path) if binary_path is None or not binary_path.is_file(): return None, identity, None terminal = libreoffice_adapter._run_soffice( binary_path, ["--version"], profile_path=work_dir / "profile-version", work_dir=work_dir, timeout_seconds=timeout_seconds, ) identity["version_stdout"] = terminal.get("stdout", "").strip() identity["version_terminal_status"] = terminal if terminal["state"] != "completed": return None, identity, terminal return binary_path, identity, terminal def _libreoffice_roundtrip( input_path: Path, output_path: Path, *, binary_path: Path | None, runtime: dict[str, Any], version_terminal: dict[str, Any] | None, source_metadata: dict[str, Any], timeout_seconds: int, ) -> dict[str, Any]: started_at_utc = _utc_now() metadata: dict[str, Any] = { **source_metadata, "scenario": "ooxml_noop_roundtrip", "iteration_count": 1, "timeout_seconds": timeout_seconds, "cache_state": "isolated_profile_no_reuse", "artifact_path": str(output_path), "command": [], "started_at_utc": started_at_utc, "completed_at_utc": None, } if binary_path is None: reason = "LibreOffice soffice executable was not found" if version_terminal is not None: reason = f"LibreOffice version probe {version_terminal['state']}" metadata["completed_at_utc"] = _utc_now() return { "engine": "libreoffice", "status": "unavailable", "reason": reason, "runtime_identity": runtime, **runtime, "terminal_status": {"version": version_terminal, "state": "unavailable"}, "metadata": metadata, } with tempfile.TemporaryDirectory(prefix="wolfxl-libreoffice-preservation-") as directory: work_dir = Path(directory) copied_input = work_dir / "input.xlsx" converted_dir = work_dir / "converted" shutil.copyfile(input_path, copied_input) converted_path, save_terminal = libreoffice_adapter._convert_to_xlsx( binary_path, copied_input, converted_dir, profile_path=work_dir / "profile-save", work_dir=work_dir, timeout_seconds=timeout_seconds, ) metadata["command"] = save_terminal["command"] terminal_status = {"version": version_terminal, "save": save_terminal} if save_terminal["state"] != "completed" or not converted_path.is_file(): reason = ( libreoffice_adapter._failed_terminal_reason("save", save_terminal) if save_terminal["state"] != "completed" else "LibreOffice completed without producing the expected XLSX output" ) metadata["completed_at_utc"] = _utc_now() return { "engine": "libreoffice", "status": "failed", "reason": reason, "runtime_identity": runtime, **runtime, "terminal_status": terminal_status, "metadata": metadata, } output_path.parent.mkdir(parents=True, exist_ok=True) shutil.move(str(converted_path), str(output_path)) try: preservation = _audit_preservation(input_path, output_path, expected_changed_parts=set()) except Exception as exc: metadata["completed_at_utc"] = _utc_now() return { "engine": "libreoffice", "status": "failed", "reason": f"{type(exc).__name__}: {exc}", "runtime_identity": runtime, **runtime, "terminal_status": terminal_status, "metadata": metadata, } metadata["completed_at_utc"] = _utc_now() return { "engine": "libreoffice", "status": _preservation_status(preservation), "runtime_identity": runtime, **runtime, "terminal_status": terminal_status, "metadata": metadata, "output_sha256": _sha256(output_path), "output_size": output_path.stat().st_size, "preservation": preservation, } def _relative_artifact(output_dir: Path, result: dict[str, Any]) -> None: artifact_path = result.get("metadata", {}).get("artifact_path") if isinstance(artifact_path, str): try: result["metadata"]["artifact_path"] = Path(artifact_path).relative_to(output_dir).as_posix() except ValueError: pass def _copy_harness_source(output_dir: Path) -> list[dict[str, str]]: harness_dir = output_dir / "harness" harness_dir.mkdir(parents=True, exist_ok=True) sources = [ Path(__file__), SCRIPT_DIR / "competitor_workload_libreoffice_adapter.py", ] copied: list[dict[str, str]] = [] for source in sources: destination = harness_dir / source.name shutil.copyfile(source, destination) copied.append( { "path": destination.relative_to(output_dir).as_posix(), "sha256": _sha256(destination), } ) return copied def _write_bundle_hashes(output_dir: Path) -> None: entries: dict[str, str] = {} for path in sorted(output_dir.rglob("*")): if path.is_file() and path.name != "bundle.sha256.json": entries[path.relative_to(output_dir).as_posix()] = _sha256(path) _write_json(output_dir / "bundle.sha256.json", {"schema_version": 1, "files": entries}) def run_comparison( input_paths: list[Path], output_dir: Path, *, soffice_path: str | Path | None, timeout_seconds: int, ) -> dict[str, Any]: if timeout_seconds <= 0: raise ValueError("timeout_seconds must be positive") output_dir = output_dir.resolve() if output_dir.exists() and any(output_dir.iterdir()): raise ValueError(f"output directory must be empty: {output_dir}") for input_path in input_paths: if input_path.suffix.lower() != ".xlsx": raise ValueError(f"only .xlsx inputs are supported: {input_path}") if not input_path.is_file(): raise FileNotFoundError(f"missing input workbook: {input_path}") output_dir.mkdir(parents=True, exist_ok=True) source_metadata = _source_identity() results: list[dict[str, Any]] = [] input_records: list[dict[str, str]] = [] with tempfile.TemporaryDirectory(prefix="wolfxl-libreoffice-preservation-runtime-") as directory: binary_path, runtime, version_terminal = _libreoffice_runtime( soffice_path, work_dir=Path(directory), timeout_seconds=timeout_seconds, ) for input_path in input_paths: resolved_input = input_path.resolve() source_sha256 = _sha256(resolved_input) scenario_id = _scenario_id(resolved_input, source_sha256) input_records.append( { "path": str(resolved_input), "sha256": source_sha256, "scenario_id": scenario_id, } ) artifact_root = output_dir / "artifacts" / scenario_id wolfxl_result = _wolfxl_roundtrip( resolved_input, artifact_root / "wolfxl" / "output.xlsx", source_metadata=source_metadata, timeout_seconds=timeout_seconds, ) libreoffice_result = _libreoffice_roundtrip( resolved_input, artifact_root / "libreoffice" / "output.xlsx", binary_path=binary_path, runtime=runtime, version_terminal=version_terminal, source_metadata=source_metadata, timeout_seconds=timeout_seconds, ) for result in (wolfxl_result, libreoffice_result): _relative_artifact(output_dir, result) results.append( { "schema_version": SCHEMA_VERSION, "scenario_id": scenario_id, "input_path": str(resolved_input), "input_sha256": source_sha256, **result, } ) status_counts = Counter(str(result["status"]) for result in results) failed_results = [result for result in results if result["status"] != "passed"] harness_sources = _copy_harness_source(output_dir) manifest = { "schema_version": SCHEMA_VERSION, "producer": Path(__file__).name, "operation": "ooxml_noop_roundtrip", "source_identity": source_metadata, "inputs": input_records, "runtime_identity": runtime, "harness_sources": harness_sources, "receipt_layout": [ "manifest.json", "raw.jsonl", "failures.jsonl", "summary.json", "status.json", "bundle.sha256.json", "harness/", ], "generated_at_utc": _utc_now(), } summary = { "schema_version": SCHEMA_VERSION, "scenario_count": len(input_records), "engine_result_count": len(results), "status_counts": dict(sorted(status_counts.items())), "failure_count": len(failed_results), "scope": "Spreadsheet OOXML no-op round trips for the supplied XLSX scenarios only.", "non_claims": [ "This receipt does not establish byte identity unless the recorded package hashes show no changes.", "This receipt does not establish Excel parity, full spreadsheet parity, or performance superiority.", ], } terminal_state = "completed" if not failed_results else "completed_with_failures" status = { "schema_version": SCHEMA_VERSION, "state": terminal_state, "failure_count": len(failed_results), "completed_at_utc": _utc_now(), } manifest = codex_runtime_identity.public_receipt(manifest, repo_root=ROOT) results = codex_runtime_identity.public_receipt(results, repo_root=ROOT) failed_results = codex_runtime_identity.public_receipt(failed_results, repo_root=ROOT) _write_json(output_dir / "manifest.json", manifest) _write_jsonl(output_dir / "raw.jsonl", results) _write_jsonl(output_dir / "failures.jsonl", failed_results) _write_json(output_dir / "summary.json", summary) _write_json(output_dir / "status.json", status) _write_bundle_hashes(output_dir) return {"manifest": manifest, "summary": summary, "status": status, "results": results} def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("workbook", type=Path, nargs="+", help="one or more XLSX inputs") parser.add_argument("--output-dir", type=Path, required=True) parser.add_argument( "--soffice-path", type=Path, help="explicit soffice executable; otherwise LIBREOFFICE_SOFFICE_PATH then PATH is used", ) parser.add_argument("--timeout-seconds", type=int, default=DEFAULT_TIMEOUT_SECONDS) parser.add_argument( "--allow-incomplete", action="store_true", help="return zero after producing receipts with drift, failures, or unavailable engines", ) args = parser.parse_args(argv) try: report = run_comparison( args.workbook, args.output_dir, soffice_path=args.soffice_path, timeout_seconds=args.timeout_seconds, ) except (OSError, ValueError, BadZipFile) as exc: parser.error(str(exc)) print(json.dumps({"summary": report["summary"], "status": report["status"]}, sort_keys=True)) return 0 if report["status"]["state"] == "completed" or args.allow_incomplete else 1 if __name__ == "__main__": raise SystemExit(main())