Coverage for src/rtflite/convert.py: 92%
115 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-07 03:44 +0000
1import os
2import platform
3import re
4import shutil
5import subprocess
6import tempfile
7from collections.abc import Sequence
8from math import isfinite
9from pathlib import Path
11from .dictionary.libreoffice import DEFAULT_PATHS, MIN_VERSION
14class LibreOfficeConverter:
15 """Convert RTF documents to other formats using LibreOffice.
17 Convert RTF files to various formats including PDF, DOCX, HTML, and others
18 using LibreOffice in headless mode.
20 Requirements:
21 - LibreOffice 7.1 or later must be installed.
22 - Automatically finds LibreOffice in standard installation paths.
23 - For custom installations, provide `executable_path` parameter.
25 Note:
26 The converter runs LibreOffice in headless mode, so no GUI is required.
27 This makes it suitable for server environments and automated workflows.
28 """
30 def __init__(
31 self,
32 executable_path: str | Path | None = None,
33 *,
34 timeout: float | None = 120,
35 ) -> None:
36 """Initialize converter with optional executable path.
38 Args:
39 executable_path: Path (or executable name) to LibreOffice. If None,
40 searches standard installation locations for each platform.
41 timeout: Maximum seconds for each LibreOffice process, including
42 the version check. Defaults to 120. Use None to disable.
44 Raises:
45 FileNotFoundError: If LibreOffice executable cannot be found.
46 ValueError: If timeout is invalid or the version cannot be parsed.
47 RuntimeError: If LibreOffice is too old, fails to start, or times out.
48 """
49 if timeout is not None and (not isfinite(timeout) or timeout <= 0):
50 raise ValueError("timeout must be a positive finite number or None.")
51 self.timeout = timeout
52 self.executable_path = self._resolve_executable_path(executable_path)
54 self._verify_version()
56 def _resolve_executable_path(self, executable_path: str | Path | None) -> Path:
57 """Resolve the LibreOffice executable path."""
58 if executable_path is None:
59 found_executable = self._find_executable()
60 if found_executable is None:
61 raise FileNotFoundError("Can't find LibreOffice executable.")
62 return found_executable
64 executable = os.fspath(executable_path)
65 expanded = os.path.expanduser(executable)
66 candidate = Path(expanded)
67 looks_like_path = (
68 isinstance(executable_path, Path)
69 or candidate.is_absolute()
70 or os.sep in expanded
71 or (os.altsep is not None and os.altsep in expanded)
72 )
73 if looks_like_path:
74 if candidate.is_file():
75 return candidate.absolute()
76 raise FileNotFoundError(
77 f"LibreOffice executable not found at: {candidate}."
78 )
80 resolved_executable = shutil.which(executable)
81 if resolved_executable is None:
82 raise FileNotFoundError(f"Can't find LibreOffice executable: {executable}.")
83 return Path(resolved_executable).absolute()
85 def _find_executable(self) -> Path | None:
86 """Find LibreOffice executable in default locations."""
87 system = platform.system()
88 # Windows needs the console launcher to capture output and wait for exit.
89 names: tuple[str, ...] = ("soffice", "libreoffice")
90 if system == "Windows":
91 names = ("soffice.com", *names)
92 for name in names:
93 resolved = shutil.which(name)
94 if resolved is not None:
95 return Path(resolved).absolute()
97 if system not in DEFAULT_PATHS:
98 raise RuntimeError(f"Unsupported operating system: {system}.")
100 for path in DEFAULT_PATHS[system]:
101 candidate = Path(path)
102 if candidate.is_file():
103 return candidate
104 return None
106 def _run_command(
107 self, cmd: list[str], action: str
108 ) -> subprocess.CompletedProcess[str]:
109 """Run LibreOffice with bounded execution and useful diagnostics."""
110 try:
111 return subprocess.run(
112 cmd,
113 capture_output=True,
114 text=True,
115 errors="replace",
116 check=True,
117 timeout=self.timeout,
118 )
119 except subprocess.TimeoutExpired as e:
120 raise RuntimeError(
121 f"LibreOffice {action} timed out after {self.timeout} seconds."
122 ) from e
123 except subprocess.CalledProcessError as e:
124 raise RuntimeError(
125 f"LibreOffice {action} failed (exit code {e.returncode}):\n"
126 f"Command output: {e.stdout}\n"
127 f"Error output: {e.stderr}"
128 ) from e
129 except OSError as e:
130 raise RuntimeError(
131 f"Failed to run LibreOffice at {self.executable_path}: {e}"
132 ) from e
134 def _verify_version(self) -> None:
135 """Verify LibreOffice version meets minimum requirement."""
136 result = self._run_command(
137 [str(self.executable_path), "--version"], "version check"
138 )
139 version_str = result.stdout.strip()
140 match = re.search(r"LibreOffice\s+(\d+\.\d+(?:\.\d+)*)", version_str)
141 if not match:
142 raise ValueError(f"Can't parse LibreOffice version from: {version_str}.")
144 # LibreOffice uses numeric versions, including calendar versions (24+).
145 # Compare numerically without requiring the optional packaging library.
146 current_version = tuple(int(part) for part in match.group(1).split("."))
147 min_version = tuple(int(part) for part in MIN_VERSION.split("."))
148 if current_version < min_version:
149 raise RuntimeError(
150 f"LibreOffice version {match.group(1)} is below minimum required "
151 f"version {MIN_VERSION}."
152 )
154 def convert(
155 self,
156 input_files: str | Path | Sequence[str | Path],
157 output_dir: str | Path,
158 format: str = "pdf",
159 overwrite: bool = False,
160 ) -> Path | Sequence[Path]:
161 """Convert RTF file(s) to specified format using LibreOffice.
163 Performs the actual conversion of RTF files to the target format using
164 LibreOffice in headless mode. Supports single file or batch conversion.
166 Args:
167 input_files: Path to input RTF file or list of paths. Can be string
168 or Path object. For batch conversion, provide a list/tuple.
169 output_dir: Directory where converted files will be saved. Created
170 if it doesn't exist. Can be string or Path object.
171 format: Target format for conversion. Supported formats:
173 - `'pdf'`: Portable Document Format (default)
174 - `'docx'`: Microsoft Word (Office Open XML)
175 - `'doc'`: Microsoft Word 97-2003
176 - `'html'`: HTML Document
177 - `'odt'`: OpenDocument Text
178 - `'txt'`: Plain Text
180 Also accepts LibreOffice's `extension:filter[:options]` syntax,
181 for example `'pdf:writer_pdf_Export'` or
182 `'txt:Text (encoded):UTF8'`. Filter names and options are passed
183 through unchanged; the extension determines the output filename.
184 overwrite: If `True`, overwrites existing files in output directory.
185 If `False`, raises error if output file already exists. Existing
186 output is preserved if LibreOffice fails to convert the input.
188 Returns:
189 Path | Sequence[Path]: For single file input, returns Path to the
190 converted file. For multiple files, returns list of Paths.
192 Raises:
193 FileNotFoundError: If an input file is missing or is not a file.
194 FileExistsError: If output file exists and overwrite=False.
195 ValueError: If format does not start with a valid file extension.
196 RuntimeError: If LibreOffice conversion fails or times out.
198 Note:
199 Each file is converted with a temporary, isolated LibreOffice user
200 profile, independent of an open desktop session or other conversions.
201 Personal LibreOffice settings and extensions are not used. Batch
202 inputs are processed sequentially, with a new process for each file.
204 Examples:
205 Single file conversion:
206 ```python
207 converter = LibreOfficeConverter()
208 pdf_path = converter.convert(
209 "report.rtf",
210 output_dir="pdfs/",
211 format="pdf"
212 )
213 print(f"Created: {pdf_path}")
214 ```
216 Batch conversion with overwrite:
217 ```python
218 rtf_files = ["report1.rtf", "report2.rtf", "report3.rtf"]
219 pdf_paths = converter.convert(
220 input_files=rtf_files,
221 output_dir="output/pdfs/",
222 format="pdf",
223 overwrite=True
224 )
225 for path in pdf_paths:
226 print(f"Converted: {path}")
227 ```
228 """
229 extension = format.split(":", 1)[0]
230 if not re.fullmatch(r"[A-Za-z0-9]+", extension):
231 raise ValueError(
232 "format must be an extension or extension:filter[:options], "
233 f"got {format!r}."
234 )
235 output_dir = Path(output_dir).expanduser()
236 output_dir.mkdir(parents=True, exist_ok=True)
238 # Handle single input file
239 if isinstance(input_files, (str, Path)):
240 input_path = Path(input_files).expanduser()
241 if not input_path.is_file():
242 raise FileNotFoundError(f"Input file not found: {input_path}.")
243 return self._convert_single_file(input_path, output_dir, format, overwrite)
245 # Handle multiple input files
246 input_paths = [Path(f).expanduser() for f in input_files]
247 for path in input_paths:
248 if not path.is_file():
249 raise FileNotFoundError(f"Input file not found: {path}.")
251 return [
252 self._convert_single_file(input_path, output_dir, format, overwrite)
253 for input_path in input_paths
254 ]
256 def _convert_single_file(
257 self, input_file: Path, output_dir: Path, format: str, overwrite: bool
258 ) -> Path:
259 """Convert a single file using LibreOffice."""
260 extension = format.split(":", 1)[0]
261 output_file = output_dir / f"{input_file.stem}.{extension}"
263 if output_file.exists() and not overwrite:
264 raise FileExistsError(
265 f"Output file already exists: {output_file}. "
266 "Use overwrite=True to force."
267 )
269 # Stage output so an old file cannot be mistaken for a successful export.
270 # Keep it on the destination filesystem for the final file replacement.
271 with tempfile.TemporaryDirectory(prefix=".rtflite-", dir=output_dir) as tmpdir:
272 work_dir = Path(tmpdir).resolve()
273 converted_dir = work_dir / "output"
274 converted_dir.mkdir()
275 cmd = [
276 str(self.executable_path),
277 f"-env:UserInstallation={(work_dir / 'profile').as_uri()}",
278 "--headless",
279 "--nologo",
280 "--norestore",
281 "--convert-to",
282 format,
283 "--outdir",
284 str(converted_dir),
285 str(input_file.absolute()),
286 ]
287 result = self._run_command(cmd, "conversion")
288 converted_file = converted_dir / output_file.name
289 if not converted_file.is_file():
290 raise RuntimeError(
291 f"Conversion failed: Output file not created.\n"
292 f"Command output: {result.stdout}\n"
293 f"Error output: {result.stderr}"
294 )
296 # Include companion files/directories created by formats such as HTML.
297 generated_paths = list(converted_dir.iterdir())
298 if not overwrite:
299 for path in generated_paths:
300 destination = output_dir / path.name
301 if destination.exists():
302 raise FileExistsError(
303 f"Output file already exists: {destination}. "
304 "Use overwrite=True to force."
305 )
306 for path in generated_paths:
307 if path == converted_file:
308 continue
309 destination = output_dir / path.name
310 if path.is_dir():
311 shutil.copytree(path, destination, dirs_exist_ok=overwrite)
312 else:
313 path.replace(destination)
314 converted_file.replace(output_file)
316 return output_file