| 1 | from __future__ import annotations |
| 2 | |
| 3 | import os |
| 4 | import shutil |
| 5 | import subprocess |
| 6 | import tempfile |
| 7 | import zipfile |
| 8 | from pathlib import Path |
| 9 | from typing import Any |
| 10 | |
| 11 | |
| 12 | SOFFICE_BINARIES = ("soffice", "libreoffice") |
| 13 | CONVERT_TIMEOUT_SECONDS = 45 |
| 14 | ODF_MIMETYPES = { |
| 15 | "odt": "application/vnd.oasis.opendocument.text", |
| 16 | "ods": "application/vnd.oasis.opendocument.spreadsheet", |
| 17 | "odp": "application/vnd.oasis.opendocument.presentation", |
| 18 | } |
| 19 | |
| 20 | |
| 21 | def find_soffice() -> str: |
| 22 | for name in SOFFICE_BINARIES: |
| 23 | path = shutil.which(name) |
| 24 | if path: |
| 25 | return path |
| 26 | return "" |
| 27 | |
| 28 | |
| 29 | def collect_status() -> dict[str, Any]: |
| 30 | soffice = find_soffice() |
| 31 | status = { |
| 32 | "ok": True, |
| 33 | "state": "healthy" if soffice else "missing", |
| 34 | "healthy": bool(soffice), |
| 35 | "soffice": soffice, |
| 36 | "message": "LibreOffice is available." if soffice else "LibreOffice is not installed in this runtime.", |
| 37 | } |
| 38 | try: |
| 39 | from plugins._desktop.helpers import desktop_session |
| 40 | |
| 41 | status["desktop"] = desktop_session.collect_desktop_status() |
| 42 | except Exception as exc: |
| 43 | status["desktop"] = {"ok": False, "healthy": False, "error": str(exc)} |
| 44 | return status |
| 45 | |
| 46 | |
| 47 | def validate_docx(path: str | Path) -> dict[str, Any]: |
| 48 | source = Path(path) |
| 49 | if not source.exists(): |
| 50 | return {"ok": False, "error": f"File not found: {source}"} |
| 51 | try: |
| 52 | with zipfile.ZipFile(source) as archive: |
| 53 | archive.getinfo("[Content_Types].xml") |
| 54 | archive.getinfo("word/document.xml") |
| 55 | except Exception as exc: |
| 56 | return {"ok": False, "error": f"DOCX package validation failed: {exc}"} |
| 57 | |
| 58 | soffice = find_soffice() |
| 59 | if not soffice: |
| 60 | return {"ok": True, "warning": "LibreOffice binary was not available; package validation only."} |
| 61 | |
| 62 | with tempfile.TemporaryDirectory(prefix="a0-office-validate-") as temp_dir: |
| 63 | result = _run_soffice( |
| 64 | soffice, |
| 65 | [ |
| 66 | "--headless", |
| 67 | "--safe-mode", |
| 68 | "--convert-to", |
| 69 | "pdf", |
| 70 | "--outdir", |
| 71 | temp_dir, |
| 72 | str(source), |
| 73 | ], |
| 74 | timeout=CONVERT_TIMEOUT_SECONDS, |
| 75 | ) |
| 76 | if result.returncode != 0: |
| 77 | return {"ok": False, "error": _format_process_error(result)} |
| 78 | return {"ok": True} |
| 79 | |
| 80 | |
| 81 | def validate_odf(path: str | Path) -> dict[str, Any]: |
| 82 | source = Path(path) |
| 83 | if not source.exists(): |
| 84 | return {"ok": False, "error": f"File not found: {source}"} |
| 85 | ext = source.suffix.lower().lstrip(".") |
| 86 | expected_mimetype = ODF_MIMETYPES.get(ext) |
| 87 | if not expected_mimetype: |
| 88 | return {"ok": False, "error": f"Unsupported ODF extension: {ext}"} |
| 89 | try: |
| 90 | with zipfile.ZipFile(source) as archive: |
| 91 | first = archive.infolist()[0] |
| 92 | mimetype = archive.read("mimetype").decode("utf-8") |
| 93 | archive.getinfo("content.xml") |
| 94 | archive.getinfo("META-INF/manifest.xml") |
| 95 | except Exception as exc: |
| 96 | return {"ok": False, "error": f"ODF package validation failed: {exc}"} |
| 97 | if first.filename != "mimetype" or first.compress_type != zipfile.ZIP_STORED: |
| 98 | return {"ok": False, "error": "ODF mimetype must be the first uncompressed package entry."} |
| 99 | if mimetype != expected_mimetype: |
| 100 | return {"ok": False, "error": f"ODF mimetype mismatch: expected {expected_mimetype}, got {mimetype}"} |
| 101 | return {"ok": True} |
| 102 | |
| 103 | |
| 104 | def convert_document(path: str | Path, target_format: str, output_dir: str | Path | None = None) -> dict[str, Any]: |
| 105 | source = Path(path) |
| 106 | if not source.exists(): |
| 107 | return {"ok": False, "error": f"File not found: {source}"} |
| 108 | soffice = find_soffice() |
| 109 | if not soffice: |
| 110 | return {"ok": False, "error": "LibreOffice is not installed in this runtime."} |
| 111 | |
| 112 | target_format = str(target_format or "").lower().strip().lstrip(".") |
| 113 | if not target_format: |
| 114 | return {"ok": False, "error": "target_format is required."} |
| 115 | |
| 116 | destination_dir = Path(output_dir) if output_dir else source.parent |
| 117 | destination_dir.mkdir(parents=True, exist_ok=True) |
| 118 | before = {item.name for item in destination_dir.iterdir()} if destination_dir.exists() else set() |
| 119 | result = _run_soffice( |
| 120 | soffice, |
| 121 | [ |
| 122 | "--headless", |
| 123 | "--safe-mode", |
| 124 | "--convert-to", |
| 125 | target_format, |
| 126 | "--outdir", |
| 127 | str(destination_dir), |
| 128 | str(source), |
| 129 | ], |
| 130 | timeout=CONVERT_TIMEOUT_SECONDS, |
| 131 | ) |
| 132 | if result.returncode != 0: |
| 133 | return {"ok": False, "error": _format_process_error(result)} |
| 134 | |
| 135 | expected = destination_dir / f"{source.stem}.{target_format}" |
| 136 | if expected.exists(): |
| 137 | return {"ok": True, "path": str(expected)} |
| 138 | |
| 139 | created = [item for item in destination_dir.iterdir() if item.name not in before] |
| 140 | if created: |
| 141 | return {"ok": True, "path": str(created[0])} |
| 142 | return {"ok": False, "error": "LibreOffice completed without producing an output file."} |
| 143 | |
| 144 | |
| 145 | def _run_soffice(soffice: str, args: list[str], timeout: int) -> subprocess.CompletedProcess[str]: |
| 146 | env = { |
| 147 | **os.environ, |
| 148 | "HOME": os.environ.get("HOME") or "/tmp", |
| 149 | "SAL_USE_VCLPLUGIN": os.environ.get("SAL_USE_VCLPLUGIN") or "gen", |
| 150 | } |
| 151 | return subprocess.run( |
| 152 | [soffice, *args], |
| 153 | check=False, |
| 154 | text=True, |
| 155 | capture_output=True, |
| 156 | timeout=timeout, |
| 157 | env=env, |
| 158 | ) |
| 159 | |
| 160 | |
| 161 | def _format_process_error(result: subprocess.CompletedProcess[str]) -> str: |
| 162 | details = (result.stderr or result.stdout or "").strip() |
| 163 | if details: |
| 164 | return f"LibreOffice exited with {result.returncode}: {details}" |
| 165 | return f"LibreOffice exited with {result.returncode}." |