From daa84f8bce668adfaf192c929e002b76f0949371 Mon Sep 17 00:00:00 2001 From: Igor20264 Date: Tue, 22 Sep 2026 18:49:09 +0300 Subject: [PATCH] v0.5.6 --- build-exe-uvx.bat | 2 +- docs/cli-gui.md | 3 +- md2gost.fast.spec | 2 + md2gost.spec | 2 + md2gost/README.md | 11 +- md2gost/__main__.py | 22 +- md2gost/checker/__init__.py | 4 +- md2gost/gui.py | 123 +++- md2gost/office/__init__.py | 19 + md2gost/office/detect.py | 176 +++++ md2gost/office/dispatch.py | 59 ++ md2gost/office/open_doc.py | 70 ++ md2gost/office/uno_cli.py | 125 ++++ md2gost/office/uno_fix.py | 343 +++++++++ md2gost/office/uno_helpers.py | 209 ++++++ md2gost/office/uno_profile.py | 41 ++ md2gost/office/uno_worker.py | 1280 +++++++++++++++++++++++++++++++++ md2gost/pipeline.py | 21 +- md2gost/word_fix.py | 1 + pyproject.toml | 2 + tests/test_office.py | 269 +++++++ 21 files changed, 2761 insertions(+), 23 deletions(-) create mode 100644 md2gost/office/__init__.py create mode 100644 md2gost/office/detect.py create mode 100644 md2gost/office/dispatch.py create mode 100644 md2gost/office/open_doc.py create mode 100644 md2gost/office/uno_cli.py create mode 100644 md2gost/office/uno_fix.py create mode 100644 md2gost/office/uno_helpers.py create mode 100644 md2gost/office/uno_profile.py create mode 100644 md2gost/office/uno_worker.py create mode 100644 tests/test_office.py diff --git a/build-exe-uvx.bat b/build-exe-uvx.bat index 7cf8b45..c311c62 100644 --- a/build-exe-uvx.bat +++ b/build-exe-uvx.bat @@ -51,7 +51,7 @@ if exist "%USERPROFILE%\.cargo\bin\uvx.exe" set "UVX_BIN=%USERPROFILE%\.cargo\bi if defined UVX_BIN goto :got_uvx echo uvx не найден. Установите uv: https://docs.astral.sh/uv/getting-started/installation/ -echo powershell: irm https://astral.sh/uv/install.ps1 ^| iex +echo powershell -ExecutionPolicy ByPass -c "irm https://astral.sh/uv/install.ps1 | iex" goto :fail :got_uvx diff --git a/docs/cli-gui.md b/docs/cli-gui.md index 9fb69a3..164b4cb 100644 --- a/docs/cli-gui.md +++ b/docs/cli-gui.md @@ -34,6 +34,7 @@ python -m word2md report.docx -o report.md | `--check-pages` | Проверить вёрстку в Word | эвристика полупустых страниц (Word + pywin32) | | `--table-continuation` | Основные | `word` (по умолчанию), `off`, `legacy`, `caption`, `soft` | | `--listing-continuation` | Основные | то же для листингов | +| `--office-engine` | Основные → Движок пост-разрыва | `auto` (Word → LibreOffice → OpenOffice), `word`, `libre`, `openoffice` | | `--table-repeat-header` | Галочка | шапка на фрагментах после word-split | | `--plantuml-jar` / `--kroki-url` | Настройки → Диаграммы | локальный рендер UML / запасной Kroki | | `--diagram-fallback` | Диаграммы | `remote` / `local` / `off` (отключает только kroki.io) | @@ -49,7 +50,7 @@ python -m word2md report.docx -o report.md ## Продолжение таблиц и листингов -DOCX не умеет сам писать «Продолжение Таблицы N» только со 2-й страницы. Режим **`word`** (по умолчанию): после сохранения Word COM режет по реальной пагинации. Нужны Windows + Word + pywin32. Без Word: `--table-continuation off` (и то же для листингов). +DOCX не умеет сам писать «Продолжение Таблицы N» только со 2-й страницы. Режим **`word`** (по умолчанию): после сохранения режет по реальной пагинации выбранным движком (`--office-engine`): Microsoft Word COM (Windows + pywin32) или LibreOffice/OpenOffice Writer через UNO (`soffice` + `program/python`). По умолчанию `auto` пробует Word, затем LibreOffice, затем OpenOffice. Без офиса: `--table-continuation off` (и то же для листингов). ## Меню GUI diff --git a/md2gost.fast.spec b/md2gost.fast.spec index b9020f3..b5ce07f 100644 --- a/md2gost.fast.spec +++ b/md2gost.fast.spec @@ -30,6 +30,8 @@ datas = [ (str(ROOT / "md2gost" / "TitleTemplate.docx"), "md2gost"), (str(ROOT / "md2gost" / "mml2omml"), "md2gost"), (str(ROOT / "md2gost" / "diagrams"), "md2gost/diagrams"), + (str(ROOT / "md2gost" / "office" / "uno_worker.py"), "md2gost/office"), + (str(ROOT / "md2gost" / "office" / "uno_helpers.py"), "md2gost/office"), (str(ROOT / "prompts"), "prompts"), (str(ROOT / "docs"), "docs"), ] diff --git a/md2gost.spec b/md2gost.spec index 501d405..ddcff32 100644 --- a/md2gost.spec +++ b/md2gost.spec @@ -19,6 +19,8 @@ datas = [ (str(ROOT / "md2gost" / "TitleTemplate.docx"), "md2gost"), (str(ROOT / "md2gost" / "mml2omml"), "md2gost"), (str(ROOT / "md2gost" / "diagrams"), "md2gost/diagrams"), + (str(ROOT / "md2gost" / "office" / "uno_worker.py"), "md2gost/office"), + (str(ROOT / "md2gost" / "office" / "uno_helpers.py"), "md2gost/office"), (str(ROOT / "prompts"), "prompts"), (str(ROOT / "docs"), "docs"), ] diff --git a/md2gost/README.md b/md2gost/README.md index 2878e72..f8a858d 100644 --- a/md2gost/README.md +++ b/md2gost/README.md @@ -132,28 +132,33 @@ python -m md2gost report.md -o report.docx --toc manual Наша оценка высоты строк ≠ вёрстка Word → если резать таблицу в скрипте, получается mid-page «Продолжение…» (как было на 2.4). По умолчанию режим **`word`**: после save -Word COM режет по реальной пагинации. Без Word — укажите `off` или поставьте Word + pywin32. +живой движок вёрстки режет по реальной пагинации (`--office-engine`, по умолчанию `auto`: +Word COM → LibreOffice UNO → OpenOffice UNO). Без офиса — `off`. | Режим | Поведение | |--------|-----------| -| **`word`** (по умолчанию) | Как `off` при рендере; после save Word COM: `Split` + «Продолжение Таблицы N». Нужны Windows, Word, pywin32. Шапка на продолжении не повторяется (вкл: `--table-repeat-header`). Таблицы с вертикальным merge (`^`): Word не умеет `Split` при vMerge — merge временно снимается, после разреза восстанавливается в обоих фрагментах | +| **`word`** (по умолчанию) | Как `off` при рендере; после save: `Split` + «Продолжение Таблицы N» через Word COM или LibreOffice/OpenOffice UNO (`--office-engine`). Шапка на продолжении не повторяется (вкл: `--table-repeat-header`). В Word при vMerge (`^`): merge временно снимается, после разреза восстанавливается в обоих фрагментах | | **`off`** / **`soft`** | Одна таблица Word; перенос строк делает Word. Без автоподписи. Первая строка — повторяющаяся шапка (`tblHeader`). «Продолжение…» — вручную в markdown, если нужно | | **`legacy`** | Режем по нашей оценке высоты + «Продолжение…» с `page_break_before` (могут быть дыры) | | **`caption`** | Режем по оценке + явный PageBreak + «Продолжение…» (то же ограничение точности) | ```bash python -m md2gost report.md -o report.docx --table-continuation word +python -m md2gost report.md -o report.docx --table-continuation word --office-engine libre +python -m md2gost report.md -o report.docx --table-continuation word --office-engine openoffice python -m md2gost report.md -o report.docx --table-continuation off python -m md2gost report.md -o report.docx --table-continuation caption ``` +OpenOffice/LibreOffice UNO используют постоянный профиль `%LOCALAPPDATA%\md2gost\office-profiles\` (без мастера регистрации на каждый запуск). Если OpenOffice всё же просит регистрацию один раз — завершите её; дальше профиль сохранится. + ### Продолжение листинга (`--listing-continuation`) Те же режимы, что у таблиц. По умолчанию **`word`**. | Режим | Поведение | |--------|-----------| -| **`word`** (по умолчанию) | После save Word COM + «Продолжение Листинга N» (Windows + Word + pywin32) | +| **`word`** (по умолчанию) | После save + «Продолжение Листинга N» (Word COM или LibreOffice/OpenOffice UNO) | | **`off`** / **`soft`** | Один блок кода; пагинацию делает Word. «Продолжение…» — вручную в markdown, если нужно | | **`legacy`** | Режем по оценке высоты + «Продолжение Листинга N» с `page_break_before` | | **`caption`** | Режем по оценке + явный PageBreak + «Продолжение Листинга N» | diff --git a/md2gost/__main__.py b/md2gost/__main__.py index 9a41977..0559f6e 100644 --- a/md2gost/__main__.py +++ b/md2gost/__main__.py @@ -4,6 +4,7 @@ import sys from .doc_metadata import AUTHOR_SOURCES, DEFAULT_DOC_COMMENTS from .pipeline import ConvertRequest, convert, should_launch_gui +from .office import DEFAULT_OFFICE_ENGINE, OFFICE_ENGINE_MODES from .profiles import ( DEFAULT_HEADING_NUMBERING, DEFAULT_TABLE_CONTINUATION, @@ -69,8 +70,8 @@ def build_parser() -> ArgumentParser: default=DEFAULT_TABLE_CONTINUATION, help=( "Таблицы длиннее страницы: " - "word — после сохранения Word COM режет по реальной пагинации + «Продолжение…» " - "(по умолчанию; нужны Windows, Word, pywin32); " + "word — после сохранения режет по реальной пагинации + «Продолжение…» " + "(по умолчанию; движок: --office-engine auto|word|libre|openoffice); " "off/soft — одна таблица, пагинация Word, без авто«Продолжение»; " "legacy/caption — режем по оценке высоты и вставляем «Продолжение…» " "(оценка ≠ Word, возможны артефакты). " @@ -83,8 +84,8 @@ def build_parser() -> ArgumentParser: default=DEFAULT_LISTING_CONTINUATION, help=( "Листинги длиннее страницы: " - "word — после сохранения Word COM + «Продолжение Листинга…» " - "(по умолчанию; нужны Windows, Word, pywin32); " + "word — после сохранения + «Продолжение Листинга…» " + "(по умолчанию; движок: --office-engine auto|word|libre|openoffice); " "off/soft — один блок, пагинация Word, без авто«Продолжение»; " "legacy/caption — режем по оценке высоты и вставляем «Продолжение Листинга…» " "(оценка ≠ Word, возможны артефакты). " @@ -100,6 +101,18 @@ def build_parser() -> ArgumentParser: "после разрыва. По умолчанию выкл." ), ) + parser.add_argument( + "--office-engine", + choices=OFFICE_ENGINE_MODES, + default=DEFAULT_OFFICE_ENGINE, + help=( + "Движок пост-разрыва таблиц/листингов при --table-continuation word: " + "auto — Word, иначе LibreOffice, иначе OpenOffice; " + "word — только Microsoft Word COM (Windows + pywin32); " + "libre / openoffice — UNO Writer (soffice + program/python). " + f"По умолчанию: {DEFAULT_OFFICE_ENGINE}." + ), + ) parser.add_argument( "--emdash-to-hyphen", action=BooleanOptionalAction, @@ -280,6 +293,7 @@ def request_from_args(args) -> ConvertRequest: table_continuation=args.table_continuation, listing_continuation=args.listing_continuation, table_repeat_header=bool(getattr(args, "table_repeat_header", False)), + office_engine=getattr(args, "office_engine", DEFAULT_OFFICE_ENGINE), emdash_to_hyphen=args.emdash_to_hyphen, hr_pagebreak=args.hr_pagebreak, title=args.title, diff --git a/md2gost/checker/__init__.py b/md2gost/checker/__init__.py index 8580433..2164e4b 100644 --- a/md2gost/checker/__init__.py +++ b/md2gost/checker/__init__.py @@ -423,8 +423,8 @@ def check_continuation_hints( issues.append(Issue( "table.continuation", "warning", "Word сам не вставит «Продолжение Таблицы N» при переносе. " - "Варианты: --table-continuation word (точный разрыв через Word COM) " - "или caption (оценка высоты ≠ вёрстка Word, возможны артефакты).", + "Варианты: --table-continuation word (Word COM или LibreOffice/OpenOffice UNO; " + "см. --office-engine) или caption (оценка высоты ≠ вёрстка, возможны артефакты).", )) # Long fenced code blocks (likely listings) long_listing = False diff --git a/md2gost/gui.py b/md2gost/gui.py index 4617fd7..09ec7b3 100644 --- a/md2gost/gui.py +++ b/md2gost/gui.py @@ -29,6 +29,7 @@ from .help_content import ( load_docs_catalog, load_prompt_catalog, ) +from .office import DEFAULT_OFFICE_ENGINE, OFFICE_ENGINE_MODES from .pipeline import ConvertRequest, convert, default_output_path, timestamped_output_path from .profiles import ( DEFAULT_HEADING_NUMBERING, @@ -64,6 +65,14 @@ TABLE_LABELS = { "legacy": "Резать + «Продолжение»", "caption": "Резать + разрыв страницы + «Продолжение»", } +OFFICE_ENGINE_LABELS = { + "auto": "Авто: Word → LibreOffice", + "word": "Только Microsoft Word", + "libre": "LibreOffice Writer (UNO)", +} +GUI_OFFICE_ENGINE_MODES = tuple( + key for key in OFFICE_ENGINE_MODES if key != "openoffice" +) FALLBACK_LABELS = { "remote": "Авто: локально, иначе интернет (kroki.io)", "local": "Только этот компьютер (без kroki.io)", @@ -245,6 +254,7 @@ class Md2GostApp: self.toc_var = tk.StringVar() self.table_var = tk.StringVar() self.listing_var = tk.StringVar() + self.office_engine_var = tk.StringVar() self.emdash_var = tk.BooleanVar(value=False) self.hr_pagebreak_var = tk.BooleanVar(value=False) self.auto_title_var = tk.BooleanVar(value=False) @@ -1067,10 +1077,18 @@ class Md2GostApp: tip="off — Word сам переносит; word — Split + «Продолжение» после сборки (нужен Word)") self._combo(main, 4, "Продолжение листингов", self.listing_var, _combo_values(TABLE_LABELS, LISTING_CONTINUATION_MODES), tip="То же для листингов; word требует Windows + Word + pywin32") + self._combo( + main, + 5, + "Движок пост-разрыва", + self.office_engine_var, + _combo_values(OFFICE_ENGINE_LABELS, GUI_OFFICE_ENGINE_MODES), + tip="При режиме word: Word COM или LibreOffice UNO (soffice)", + ) - ttk.Label(main, text="Смещение страниц").grid(row=5, column=0, sticky="w", padx=(0, 8), pady=3) + ttk.Label(main, text="Смещение страниц").grid(row=6, column=0, sticky="w", padx=(0, 8), pady=3) page_row = ttk.Frame(main) - page_row.grid(row=5, column=1, sticky="w", pady=3) + page_row.grid(row=6, column=1, sticky="w", pady=3) page_entry = ttk.Entry(page_row, textvariable=self.page_start_var, width=8) page_entry.pack(side=tk.LEFT) _tip( @@ -1085,7 +1103,7 @@ class Md2GostApp: ).pack(side=tk.LEFT) flags = ttk.Frame(main) - flags.grid(row=6, column=0, columnspan=2, sticky="w", pady=(8, 0)) + flags.grid(row=7, column=0, columnspan=2, sticky="w", pady=(8, 0)) checks = [ ("Заменять «—» на «-»", self.emdash_var, "Типографское тире → дефис (методичка обычно требует «—»)"), ("--- → разрыв страницы", self.hr_pagebreak_var, "Строка --- станет page break в Word"), @@ -1139,17 +1157,41 @@ class Md2GostApp: log_frame = ttk.LabelFrame(outer, text="Журнал", padding=4) log_frame.pack(fill=tk.BOTH, expand=True, pady=(0, 0)) + log_toolbar = ttk.Frame(log_frame) + log_toolbar.pack(fill=tk.X, pady=(0, 4)) + ttk.Button( + log_toolbar, text="Копировать выделение", command=self._log_copy_selection + ).pack(side=tk.LEFT) + ttk.Button( + log_toolbar, text="Копировать весь журнал", command=self._log_copy_all + ).pack(side=tk.LEFT, padx=(6, 0)) + log_body = ttk.Frame(log_frame) + log_body.pack(fill=tk.BOTH, expand=True) self.log = tk.Text( - log_frame, height=7, wrap=tk.WORD, relief=tk.FLAT, + log_body, height=7, wrap=tk.WORD, relief=tk.FLAT, bg="#fffdf8", fg=INK, padx=8, pady=6, font=("Consolas", 9) if os.name == "nt" else ("monospace", 9), + exportselection=True, undo=False, ) - scroll = ttk.Scrollbar(log_frame, command=self.log.yview) + scroll = ttk.Scrollbar(log_body, command=self.log.yview) self.log.configure(yscrollcommand=scroll.set) self.log.pack(side=tk.LEFT, fill=tk.BOTH, expand=True) scroll.pack(side=tk.RIGHT, fill=tk.Y) - self.log.insert("1.0", "Готово к работе. Перетащите .md (сборка DOCX) или .docx (импорт в Markdown).\n") - self.log.configure(state=tk.DISABLED) + self.log.insert( + "1.0", + "Готово к работе. Перетащите .md (сборка DOCX) или .docx (импорт в Markdown).\n", + ) + self._log_menu = tk.Menu(self.root, tearoff=0) + self._log_menu.add_command( + label="Копировать", command=self._log_copy_selection + ) + self._log_menu.add_command( + label="Копировать весь журнал", command=self._log_copy_all + ) + self.log.bind("", self._log_context_menu) + self.log.bind("", self._log_context_menu) + self.log.bind("", self._log_keypress) + self.log.bind("<>", self._log_copy_event) def _build_schemes_tab(self, parent) -> None: ttk.Label( @@ -1697,6 +1739,10 @@ class Md2GostApp: self.toc_var.set(_display(TOC_LABELS, req.toc_mode or DEFAULT_TOC_MODE)) self.table_var.set(_display(TABLE_LABELS, req.table_continuation or DEFAULT_TABLE_CONTINUATION)) self.listing_var.set(_display(TABLE_LABELS, req.listing_continuation or DEFAULT_LISTING_CONTINUATION)) + office_engine = req.office_engine or DEFAULT_OFFICE_ENGINE + if office_engine not in GUI_OFFICE_ENGINE_MODES: + office_engine = DEFAULT_OFFICE_ENGINE + self.office_engine_var.set(_display(OFFICE_ENGINE_LABELS, office_engine)) self.fallback_var.set(_display(FALLBACK_LABELS, req.diagram_fallback or "remote")) self.format_var.set(_display(FORMAT_LABELS, req.diagram_format or "png")) scale = getattr(req, "diagram_scale", 2.0) or 2.0 @@ -1737,6 +1783,11 @@ class Md2GostApp: toc_mode=_combo_key(self.toc_var.get(), TOC_MODES, DEFAULT_TOC_MODE), table_continuation=_combo_key(self.table_var.get(), TABLE_CONTINUATION_MODES, DEFAULT_TABLE_CONTINUATION), listing_continuation=_combo_key(self.listing_var.get(), LISTING_CONTINUATION_MODES, DEFAULT_LISTING_CONTINUATION), + office_engine=_combo_key( + self.office_engine_var.get(), + GUI_OFFICE_ENGINE_MODES, + DEFAULT_OFFICE_ENGINE, + ), emdash_to_hyphen=self.emdash_var.get(), hr_pagebreak=self.hr_pagebreak_var.get(), auto_title=self.auto_title_var.get(), @@ -1860,17 +1911,71 @@ class Md2GostApp: self._append_log(result.message or "Ошибка") messagebox.showerror("md2gost", result.message or "Ошибка конвертации") + def _log_to_clipboard(self, text: str) -> None: + self.root.clipboard_clear() + self.root.clipboard_append(text) + self.root.update_idletasks() + + def _log_copy_selection(self) -> None: + try: + if not self.log.tag_ranges(tk.SEL): + return + text = self.log.get(tk.SEL_FIRST, tk.SEL_LAST) + except tk.TclError: + return + if text: + self._log_to_clipboard(text) + + def _log_copy_all(self) -> None: + text = self.log.get("1.0", tk.END).strip() + if text: + self._log_to_clipboard(text) + + def _log_context_menu(self, event) -> str: + try: + self._log_menu.tk_popup(event.x_root, event.y_root) + finally: + self._log_menu.grab_release() + return "break" + + def _log_copy_event(self, _event=None) -> str | None: + self._log_copy_selection() + return "break" + + def _log_keypress(self, event) -> str | None: + """Read-only log: allow selection/copy, block edits.""" + if event.state & 0x4: + key = event.keysym.lower() + if key == "c": + self._log_copy_selection() + return "break" + if key == "a": + self.log.tag_add(tk.SEL, "1.0", tk.END) + self.log.mark_set(tk.INSERT, tk.END) + self.log.see(tk.INSERT) + return "break" + if key == "insert": + self._log_copy_selection() + return "break" + if event.keysym in ( + "Left", "Right", "Up", "Down", "Home", "End", + "Prior", "Next", "Shift_L", "Shift_R", "Control_L", "Control_R", + "Tab", "ISO_Left_Tab", + ): + return None + if not event.char: + return None + return "break" + def _append_log(self, text: str) -> None: if not text: return if not hasattr(self, "log"): return - self.log.configure(state=tk.NORMAL) if not text.endswith("\n"): text += "\n" self.log.insert(tk.END, text) self.log.see(tk.END) - self.log.configure(state=tk.DISABLED) def run(self) -> None: self.root.mainloop() diff --git a/md2gost/office/__init__.py b/md2gost/office/__init__.py new file mode 100644 index 0000000..84337d6 --- /dev/null +++ b/md2gost/office/__init__.py @@ -0,0 +1,19 @@ +"""Office-suite backends for post-save table/listing continuation splits.""" + +from .detect import ( + DEFAULT_OFFICE_ENGINE, + OFFICE_ENGINE_MODES, + OfficeInstall, + detect_office_install, + resolve_office_engine, +) +from .dispatch import fix_continuations_with_engine + +__all__ = [ + "DEFAULT_OFFICE_ENGINE", + "OFFICE_ENGINE_MODES", + "OfficeInstall", + "detect_office_install", + "resolve_office_engine", + "fix_continuations_with_engine", +] diff --git a/md2gost/office/detect.py b/md2gost/office/detect.py new file mode 100644 index 0000000..034a341 --- /dev/null +++ b/md2gost/office/detect.py @@ -0,0 +1,176 @@ +"""Detect installed Word / LibreOffice / OpenOffice for post-save table splits.""" + +from __future__ import annotations + +import os +import shutil +import sys +from dataclasses import dataclass +from pathlib import Path + +from .uno_cli import soffice_cli_path + +OFFICE_ENGINE_MODES = ("auto", "word", "libre", "openoffice") +DEFAULT_OFFICE_ENGINE = "auto" + + +@dataclass(frozen=True) +class OfficeInstall: + engine: str # "word" | "libre" | "openoffice" + label: str + soffice_path: str | None = None + python_path: str | None = None + + +def _program_dirs(kind: str) -> list[Path]: + roots: list[Path] = [] + for env in ("ProgramFiles", "ProgramFiles(x86)"): + base = os.environ.get(env) + if base: + roots.append(Path(base)) + if sys.platform != "win32": + roots.extend( + [ + Path("/usr/lib/libreoffice"), + Path("/usr/lib64/libreoffice"), + Path("/opt/libreoffice"), + Path("/opt/openoffice4"), + Path("/Applications/LibreOffice.app/Contents"), + ] + ) + candidates: list[Path] = [] + if kind == "libre": + for root in roots: + candidates.extend( + [ + root / "LibreOffice" / "program", + root / "libreoffice" / "program", + root / "program", + root / "MacOS", + ] + ) + else: + for root in roots: + candidates.extend( + [ + root / "OpenOffice 4" / "program", + root / "openoffice4" / "program", + ] + ) + return candidates + + +def _soffice_names() -> tuple[str, ...]: + if sys.platform == "win32": + # soffice.com = CLI/headless; soffice.exe opens the Start Center GUI. + return ("soffice.com", "soffice.exe") + return ("soffice", "soffice.bin") + + +def _python_names() -> tuple[str, ...]: + if sys.platform == "win32": + return ("python.exe",) + return ("python", "python3") + + +def _find_soffice_install(kind: str) -> OfficeInstall | None: + label = "LibreOffice" if kind == "libre" else "OpenOffice" + seen: set[str] = set() + + for directory in _program_dirs(kind): + for soffice_name in _soffice_names(): + soffice = directory / soffice_name + key = str(soffice.resolve()) if soffice.exists() else "" + if not key or key in seen: + continue + seen.add(key) + python_path = None + for py_name in _python_names(): + candidate = directory / py_name + if candidate.is_file(): + python_path = str(candidate) + break + if python_path is None: + continue + return OfficeInstall( + engine=kind, + label=label, + soffice_path=soffice_cli_path(str(soffice)), + python_path=python_path, + ) + + which = shutil.which("soffice") or shutil.which("soffice.bin") + if which: + program_dir = Path(which).resolve().parent + python_path = None + for py_name in _python_names(): + candidate = program_dir / py_name + if candidate.is_file(): + python_path = str(candidate) + break + if python_path: + lower = str(program_dir).lower() + detected = "openoffice" if "openoffice" in lower else "libre" + if kind != detected: + return None + return OfficeInstall( + engine=kind, + label=label, + soffice_path=soffice_cli_path(str(Path(which).resolve())), + python_path=python_path, + ) + return None + + +def detect_office_install(engine: str) -> OfficeInstall | None: + """Return install info for a concrete engine id.""" + if engine == "word": + return _detect_word() + if engine == "libre": + return _find_soffice_install("libre") + if engine == "openoffice": + return _find_soffice_install("openoffice") + return None + + +def _detect_word() -> OfficeInstall | None: + if sys.platform != "win32": + return None + try: + import win32com.client # type: ignore + except ImportError: + return None + word = None + try: + word = win32com.client.DispatchEx("Word.Application") + return OfficeInstall(engine="word", label="Microsoft Word") + except Exception: + return None + finally: + if word is not None: + try: + word.Quit() + except Exception: + pass + + +def resolve_office_engine(mode: str) -> tuple[str | None, OfficeInstall | None]: + """ + Resolve ``mode`` to a concrete engine id and optional install record. + + ``auto`` tries Word, then LibreOffice, then OpenOffice. + """ + if mode not in OFFICE_ENGINE_MODES: + mode = DEFAULT_OFFICE_ENGINE + + if mode == "auto": + for candidate in ("word", "libre", "openoffice"): + install = detect_office_install(candidate) + if install is not None: + return candidate, install + return None, None + + install = detect_office_install(mode) + if install is None: + return None, None + return mode, install diff --git a/md2gost/office/dispatch.py b/md2gost/office/dispatch.py new file mode 100644 index 0000000..78a58d5 --- /dev/null +++ b/md2gost/office/dispatch.py @@ -0,0 +1,59 @@ +"""Route table/listing continuation post-processing to Word or UNO backends.""" + +from __future__ import annotations + +from ..word_fix import FixResult, fix_continuations +from .detect import DEFAULT_OFFICE_ENGINE, resolve_office_engine +from .uno_fix import fix_continuations_uno + + +def fix_continuations_with_engine( + path: str, + *, + engine: str = DEFAULT_OFFICE_ENGINE, + tables: bool = True, + listings: bool = True, + repeat_header: bool = False, +) -> FixResult: + """ + Split cross-page tables/listings using the selected office engine. + + ``engine`` is ``auto|word|libre|openoffice``. On missing backend returns + ``ok=False`` with a reason; conversion should continue. + """ + resolved, install = resolve_office_engine(engine) + if resolved is None or install is None: + if engine == "auto": + return FixResult( + False, + message=( + "Пост-разрыв таблиц пропущен: не найден Word, LibreOffice " + "или OpenOffice." + ), + ) + labels = { + "word": "Microsoft Word (Windows + pywin32)", + "libre": "LibreOffice (soffice + program/python)", + "openoffice": "OpenOffice (soffice + program/python)", + } + return FixResult( + False, + message=f"Пост-разрыв таблиц пропущен: нужен {labels.get(engine, engine)}.", + ) + + if resolved == "word": + return fix_continuations( + path, + tables=tables, + listings=listings, + repeat_header=repeat_header, + ) + + assert install.soffice_path and install.python_path + return fix_continuations_uno( + path, + install=install, + tables=tables, + listings=listings, + repeat_header=repeat_header, + ) diff --git a/md2gost/office/open_doc.py b/md2gost/office/open_doc.py new file mode 100644 index 0000000..a1abf33 --- /dev/null +++ b/md2gost/office/open_doc.py @@ -0,0 +1,70 @@ +"""Open a finished DOCX with the same office engine used for post-split.""" + +from __future__ import annotations + +import os +import subprocess +import sys +from pathlib import Path + +from .detect import OfficeInstall, resolve_office_engine +from .uno_cli import soffice_cli_path + + +def open_document_with_engine( + path: str, + *, + engine: str, + install: OfficeInstall | None = None, +) -> None: + """Open *path* in Word / LibreOffice / OpenOffice instead of the default app.""" + abs_path = os.path.abspath(path) + resolved, detected = resolve_office_engine(engine) if install is None else (engine, install) + if resolved == "word" and sys.platform == "win32": + if _open_in_word(abs_path): + return + if resolved in ("libre", "openoffice") and detected and detected.soffice_path: + if _open_in_soffice(detected.soffice_path, abs_path): + return + _open_with_os_default(abs_path) + + +def _open_in_word(path: str) -> bool: + try: + import win32com.client # type: ignore + except ImportError: + return False + word = None + try: + word = win32com.client.DispatchEx("Word.Application") + word.Visible = True + word.Documents.Open(path) + return True + except Exception: + return False + finally: + pass # keep Word open for the user + + +def _open_in_soffice(soffice_path: str, path: str) -> bool: + try: + cli = soffice_cli_path(soffice_path) + subprocess.Popen( + [cli, os.path.abspath(path)], + cwd=str(Path(cli).parent), + ) + return True + except Exception: + return False + + +def _open_with_os_default(path: str) -> None: + import platform + + system = platform.system() + if system == "Darwin": + subprocess.call(("open", path)) + elif system == "Windows": + os.startfile(path) # type: ignore[attr-defined] + else: + subprocess.call(("xdg-open", path)) diff --git a/md2gost/office/uno_cli.py b/md2gost/office/uno_cli.py new file mode 100644 index 0000000..fa7f6bf --- /dev/null +++ b/md2gost/office/uno_cli.py @@ -0,0 +1,125 @@ +"""Headless soffice CLI helpers (convert/import) for UNO post-process.""" + +from __future__ import annotations + +import os +import subprocess +import sys +from pathlib import Path + +from .uno_helpers import file_url_from_path +from .uno_profile import ensure_profile_seeded, profile_dir + + +def soffice_cli_path(soffice_path: str) -> str: + """On Windows prefer ``soffice.com`` — ``soffice.exe`` opens the Start Center GUI.""" + path = Path(soffice_path) + if sys.platform == "win32": + com = path.with_name("soffice.com") + if com.is_file(): + return str(com) + return str(path) + + +def _base_args(soffice_path: str, profile_root: Path) -> list[str]: + ensure_profile_seeded(profile_root) + return [ + soffice_cli_path(soffice_path), + f"-env:UserInstallation={file_url_from_path(str(profile_root))}", + "--headless", + "--nologo", + "--norestore", + "--nodefault", + "--nofirststartwizard", + "--nolockcheck", + ] + + +def convert_file( + soffice_path: str, + *, + engine: str, + src: str, + out_dir: str, + convert_to: str, + timeout_s: float = 300.0, +) -> None: + """Run ``soffice --convert-to …`` and wait until it finishes.""" + profile_root = profile_dir(engine) + cmd = [ + *_base_args(soffice_path, profile_root), + "--convert-to", + convert_to, + "--outdir", + os.path.abspath(out_dir), + os.path.abspath(src), + ] + completed = subprocess.run( + cmd, + capture_output=True, + text=True, + timeout=timeout_s, + cwd=str(Path(soffice_path).parent), + creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0) if sys.platform == "win32" else 0, + ) + if completed.returncode != 0: + tail = (completed.stderr or completed.stdout or "").strip() + if len(tail) > 500: + tail = tail[-500:] + raise RuntimeError( + f"convert-to {convert_to} failed (code {completed.returncode})" + + (f": {tail}" if tail else "") + ) + stem = Path(src).stem + expected = Path(out_dir) / f"{stem}.docx" + if convert_to.startswith("docx") and not expected.is_file(): + found = sorted(Path(out_dir).glob("*.docx")) + if len(found) == 1: + return + raise RuntimeError( + f"convert-to {convert_to}: output missing ({expected.name})" + + (f", got {[p.name for p in found]}" if found else "") + ) + + +def import_docx_to_odt(soffice_path: str, *, engine: str, docx_path: str, out_dir: str) -> Path: + """OpenOffice imports DOCX more reliably via native ODT round-trip.""" + for spec in ("odt", 'odt:"writer8"'): + try: + convert_file( + soffice_path, + engine=engine, + src=docx_path, + out_dir=out_dir, + convert_to=spec, + ) + odt = Path(out_dir) / f"{Path(docx_path).stem}.odt" + if odt.is_file(): + return odt + except RuntimeError: + continue + raise RuntimeError(f"OpenOffice: не удалось импортировать DOCX → ODT: {docx_path}") + + +def export_odt_to_docx(soffice_path: str, *, engine: str, odt_path: str, out_dir: str) -> Path: + # OpenOffice on Windows treats spaces in ``docx:MS Word 2007 XML`` as extra + # file arguments (Word / 2007 / XML). Use plain ``docx`` there. + if engine == "openoffice": + specs = ("docx",) + else: + specs = ("docx", "MS Word 2007 XML") + for spec in specs: + try: + convert_file( + soffice_path, + engine=engine, + src=odt_path, + out_dir=out_dir, + convert_to=spec, + ) + docx = Path(out_dir) / f"{Path(odt_path).stem}.docx" + if docx.is_file(): + return docx + except RuntimeError: + continue + raise RuntimeError(f"OpenOffice: не удалось экспортировать ODT → DOCX: {odt_path}") diff --git a/md2gost/office/uno_fix.py b/md2gost/office/uno_fix.py new file mode 100644 index 0000000..7fae509 --- /dev/null +++ b/md2gost/office/uno_fix.py @@ -0,0 +1,343 @@ +"""Host-side launcher for LibreOffice/OpenOffice UNO table-split worker.""" + +from __future__ import annotations + +import json +import os +import shutil +import socket +import subprocess +import sys +import tempfile +import time +from pathlib import Path + +from ..word_fix import FixResult +from .detect import OfficeInstall +from .uno_cli import export_odt_to_docx, soffice_cli_path +from .uno_helpers import file_url_from_path +from .uno_profile import ensure_profile_seeded, profile_dir + +_STARTUP_TIMEOUT_S = 90.0 +_WORKER_TIMEOUT_S = 600.0 +_WORKER_FILES = ("uno_worker.py", "uno_helpers.py") + + +def _office_program_dir(install: OfficeInstall) -> Path | None: + """Directory with ``uno.py`` / ``pyuno`` (LibreOffice/OpenOffice ``program``).""" + for candidate in (install.python_path, install.soffice_path): + if not candidate: + continue + program = Path(candidate).resolve().parent + if (program / "uno.py").is_file() or (program / "pyuno.pyd").is_file(): + return program + return None + + +def _uno_worker_env(worker_dir: Path, install: OfficeInstall) -> tuple[dict[str, str], Path]: + """Build env + cwd so bundled ``program/python.exe`` can ``import uno``.""" + env = os.environ.copy() + program_dir = _office_program_dir(install) + path_parts = [str(worker_dir)] + if program_dir is not None: + path_parts.append(str(program_dir)) + existing = env.get("PYTHONPATH", "") + if existing: + path_parts.append(existing) + env["PYTHONPATH"] = os.pathsep.join(path_parts) + env.setdefault("PYTHONIOENCODING", "utf-8") + if install.soffice_path: + env["MD2GOST_SOFFICE"] = soffice_cli_path(install.soffice_path) + cwd = program_dir if program_dir is not None else worker_dir + return env, cwd + + +def _office_dir() -> Path: + return Path(__file__).resolve().parent + + +def _bundled_office_dir() -> Path | None: + if not getattr(sys, "frozen", False): + return None + meipass = getattr(sys, "_MEIPASS", None) + if not meipass: + return None + bundled = Path(meipass) / "md2gost" / "office" + if (bundled / "uno_worker.py").is_file(): + return bundled + return None + + +def _prepare_worker_dir() -> Path: + """Return a directory with uno_worker.py (+ helpers) on disk for LO python.""" + bundled = _bundled_office_dir() + if bundled is not None: + target = Path(tempfile.mkdtemp(prefix="md2gost-uno-worker-")) + for name in _WORKER_FILES: + src = bundled / name + if not src.is_file(): + raise FileNotFoundError(f"UNO worker bundle incomplete: missing {src}") + shutil.copy2(src, target / name) + return target + + dev_dir = _office_dir() + worker = dev_dir / "uno_worker.py" + if not worker.is_file(): + raise FileNotFoundError(f"UNO worker не найден: {worker}") + return dev_dir + + +def _free_port() -> int: + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: + sock.bind(("127.0.0.1", 0)) + return int(sock.getsockname()[1]) + + +def _wait_for_port(port: int, _proc: subprocess.Popen[str], timeout_s: float) -> bool: + """Wait until UNO port accepts connections. + + On Windows ``soffice.exe`` is a launcher that exits after spawning + ``soffice.bin``; do not treat wrapper exit as failure. + """ + deadline = time.monotonic() + timeout_s + while time.monotonic() < deadline: + try: + with socket.create_connection(("127.0.0.1", port), timeout=0.25): + return True + except OSError: + time.sleep(0.15) + return False + + +def _parse_worker_output(stdout: str) -> FixResult | None: + for line in reversed(stdout.splitlines()): + line = line.strip() + if not line.startswith("{"): + continue + try: + payload = json.loads(line) + except json.JSONDecodeError: + continue + return FixResult( + ok=bool(payload.get("ok")), + splits=int(payload.get("splits", 0)), + skipped=int(payload.get("skipped", 0)), + message=str(payload.get("message", "")), + details=list(payload.get("details", [])), + odt_temp=payload.get("odt_temp") or None, + ) + return None + + +def _export_odt_temp_to_docx( + install: OfficeInstall, odt_temp: str, docx_path: str +) -> None: + """Convert worker-side ODT (ASCII temp) back to target DOCX on the host (Py3).""" + time.sleep(0.5) + bridge_dir = Path(tempfile.mkdtemp(prefix="md2gost-oo-export-")) + local_odt = bridge_dir / "export.odt" + try: + shutil.copy2(odt_temp, local_odt) + engines = [install] + if install.engine == "openoffice": + from .detect import detect_office_install + + libre = detect_office_install("libre") + if libre is not None and libre.soffice_path: + engines.append(libre) + last_exc: Exception | None = None + out_docx: Path | None = None + for eng in engines: + try: + out_docx = export_odt_to_docx( + eng.soffice_path or "", + engine=eng.engine, + odt_path=str(local_odt), + out_dir=str(bridge_dir), + ) + break + except Exception as exc: + last_exc = exc + if out_docx is None: + raise RuntimeError(str(last_exc or "ODT export failed")) + shutil.copy2(out_docx, docx_path) + finally: + shutil.rmtree(bridge_dir, ignore_errors=True) + try: + os.remove(odt_temp) + except OSError: + pass + + +def _soffice_server_cmd(soffice_path: str, profile_root: Path, port: int) -> list[str]: + ensure_profile_seeded(profile_root) + return [ + soffice_cli_path(soffice_path), + f"-env:UserInstallation={file_url_from_path(str(profile_root))}", + "-headless", + "-invisible", + "-nologo", + "-norestore", + "-nodefault", + "-nofirststartwizard", + "-nolockcheck", + f"-accept=socket,host=127.0.0.1,port={port};urp;", + ] + + +def _run_uno_worker( + *, + install: OfficeInstall, + worker_dir: Path, + worker_path: str, + tables: bool, + listings: bool, + repeat_header: bool, +) -> FixResult: + soffice = install.soffice_path + python_exe = install.python_path + if not soffice or not python_exe: + return FixResult( + False, + message=f"Пост-разрыв ({install.label}): не задан путь к soffice/python.", + ) + + port = _free_port() + user_profile = profile_dir(install.engine) + proc: subprocess.Popen[str] | None = None + try: + proc = subprocess.Popen( + _soffice_server_cmd(soffice, user_profile, port), + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0) if sys.platform == "win32" else 0, + ) + if not _wait_for_port(port, proc, _STARTUP_TIMEOUT_S): + hint = "" + if install.engine == "openoffice": + hint = ( + " Закройте все окна OpenOffice (в т.ч. «Центр запуска») " + "и повторите." + ) + return FixResult( + False, + message=( + f"Пост-разрыв ({install.label}): не удалось запустить soffice " + f"на порту {port}.{hint}" + ), + ) + + env, worker_cwd = _uno_worker_env(worker_dir, install) + + worker_cmd = [ + python_exe, + str(worker_dir / "uno_worker.py"), + "--port", + str(port), + "--path", + worker_path, + "--engine", + install.engine, + "--tables", + "1" if tables else "0", + "--listings", + "1" if listings else "0", + "--repeat-header", + "1" if repeat_header else "0", + ] + completed = subprocess.run( + worker_cmd, + capture_output=True, + timeout=_WORKER_TIMEOUT_S, + env=env, + cwd=str(worker_cwd), + creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0) if sys.platform == "win32" else 0, + ) + stdout = (completed.stdout or b"").decode("utf-8", errors="replace") + stderr = (completed.stderr or b"").decode("utf-8", errors="replace") + parsed = _parse_worker_output(stdout) + if parsed is not None: + if parsed.ok and parsed.odt_temp: + try: + _export_odt_temp_to_docx(install, parsed.odt_temp, worker_path) + parsed.details.append("bridge: ODT→DOCX (host)") + parsed.odt_temp = None + except Exception as exc: + if parsed.splits > 0: + msg = ( + f"Пост-разрыв ({install.label}): разрезов {parsed.splits}, " + f"но экспорт DOCX не удался: {exc}" + ) + else: + msg = ( + f"Пост-разрыв ({install.label}): " + f"экспорт DOCX не удался: {exc}" + ) + return FixResult( + False, + splits=parsed.splits, + skipped=parsed.skipped, + message=msg, + details=parsed.details, + ) + return parsed + + tail = (stderr or stdout or "").strip() + if len(tail) > 400: + tail = tail[-400:] + return FixResult( + False, + message=( + f"Пост-разрыв ({install.label}) не удался " + f"(код {completed.returncode})." + + (f" {tail}" if tail else "") + ), + ) + except subprocess.TimeoutExpired: + return FixResult( + False, + message=f"Пост-разрыв ({install.label}): превышено время ожидания UNO worker.", + ) + finally: + if proc is not None: + try: + proc.terminate() + proc.wait(timeout=5) + except Exception: + try: + proc.kill() + except Exception: + pass + + +def fix_continuations_uno( + path: str, + *, + install: OfficeInstall, + tables: bool = True, + listings: bool = True, + repeat_header: bool = False, +) -> FixResult: + """Run UNO worker via LibreOffice/OpenOffice bundled Python.""" + abs_path = os.path.abspath(path) + if not os.path.isfile(abs_path): + return FixResult(False, message=f"Файл не найден: {abs_path}") + + worker_dir: Path | None = None + try: + worker_dir = _prepare_worker_dir() + except FileNotFoundError as exc: + return FixResult(False, message=str(exc)) + + try: + return _run_uno_worker( + install=install, + worker_dir=worker_dir, + worker_path=abs_path, + tables=tables, + listings=listings, + repeat_header=repeat_header, + ) + finally: + if worker_dir is not None and _bundled_office_dir() is not None: + shutil.rmtree(worker_dir, ignore_errors=True) diff --git a/md2gost/office/uno_helpers.py b/md2gost/office/uno_helpers.py new file mode 100644 index 0000000..bb95312 --- /dev/null +++ b/md2gost/office/uno_helpers.py @@ -0,0 +1,209 @@ +# -*- coding: utf-8 -*- +"""Pure table-split helpers for UNO worker (Py2.7 + Py3, standalone, no pywin32).""" + +from __future__ import print_function, unicode_literals + +import os +import re +import sys + +try: + from urllib.parse import quote +except ImportError: + from urllib import quote as _urllib_quote + + def quote(value, safe="/:@"): + if isinstance(value, bytes): + value = value.decode("utf-8") + return _urllib_quote(value.encode("utf-8"), safe) + + +DOCX_FILTER_CANDIDATES = { + "libre": ("Office Open XML Text", "MS Word 2007 XML", "Word 2007 XML"), + "openoffice": ( + "MS Word 2007 XML", + "Office Open XML Text", + "Word 2007 XML", + ), +} + +DOCX_SAVE_FILTER_SPECS = { + "libre": ( + "Office Open XML Text", + "MS Word 2007 XML", + ), + "openoffice": ( + "MS Word 2007 XML", + "Office Open XML Text", + ), +} + +MAX_PASSES = 20 + +# UNO enum numeric values (``from com.sun.star.*`` is unavailable in some LO/OO Pythons). +BREAK_NONE = 0 +BREAK_PAGE_BEFORE = 3 +CONTROL_PARAGRAPH_BREAK = 0 + +_UNO_MISSING = object() + +try: + _TEXT_TYPE = unicode +except NameError: + _TEXT_TYPE = str + + +def as_unicode(value): + """Coerce value to a Unicode string (safe on Python 2.7 + 3).""" + if value is None: + return _TEXT_TYPE() + if isinstance(value, _TEXT_TYPE): + return value + if isinstance(value, bytes): + for encoding in ("utf-8", sys.getfilesystemencoding(), "mbcs"): + if not encoding: + continue + try: + return value.decode(encoding) + except (UnicodeDecodeError, LookupError): + continue + return value.decode("utf-8", "replace") + try: + return _TEXT_TYPE(value) + except Exception: + return _TEXT_TYPE(repr(value)) + + +def uno_error_text(exc): + """Readable UNO/Python exception text (Py2 often has empty str(exc)).""" + for attr in ("Message", "Context", "Name"): + try: + val = getattr(exc, attr, None) + if val: + text = as_unicode(val) + if text: + return text + except Exception: + continue + text = as_unicode(exc) + if text: + return text + return repr(exc) + + +def uno_call(obj, base, default=_UNO_MISSING): + """Invoke UNO getter ``getBase()`` or read attribute ``Base`` (OO Py2 vs LO Py3).""" + getter = "get" + base + for access in (getter, base): + try: + member = getattr(obj, access) + return member() if callable(member) else member + except Exception: + continue + if default is not _UNO_MISSING: + return default + raise AttributeError(base) + +_RE_FLAGS = re.IGNORECASE +if sys.version_info[0] < 3: + _RE_FLAGS |= re.UNICODE + +_CAPTION_RE = re.compile( + r"^(?:Продолжение\s+)?" + r"(?PТаблица|Таблицы|Листинг|Листинга)\s+" + r"(?P[\d.]+)", + _RE_FLAGS, +) +_CONTINUATION_RE = re.compile( + r"^Продолжение\s+(?:Таблицы|Листинга)\b", _RE_FLAGS +) + + +class CaptionInfo(object): + __slots__ = ("kind", "number", "is_continuation") + + def __init__(self, kind, number, is_continuation=False): + self.kind = kind + self.number = number + self.is_continuation = is_continuation + + +def parse_caption_text(text): + raw = (text or "").replace("\r", "").replace("\x07", "").strip() + if not raw: + return None + first = raw.split("\n", 1)[0].strip() + m = _CAPTION_RE.match(first) + if not m: + return None + kind_raw = m.group("kind").lower() + if kind_raw.startswith("табл"): + kind = "table" + elif kind_raw.startswith("лист"): + kind = "listing" + else: + return None + return CaptionInfo( + kind=kind, + number=m.group("number"), + is_continuation=bool(_CONTINUATION_RE.match(first)), + ) + + +def find_page_break_row(page_numbers): + if len(page_numbers) < 2: + return None + for i in range(1, len(page_numbers)): + if page_numbers[i] > page_numbers[i - 1]: + row_1based = i + 1 + if row_1based <= 1: + return None + return row_1based + return None + + +def is_orphan_header_break(break_at, had_header): + return bool(had_header and break_at == 2) + + +def is_caption_orphaned(caption_page, first_row_page): + if caption_page is None or first_row_page is None: + return False + return caption_page < first_row_page + + +def continuation_label(kind, number): + if kind == "listing": + return "Продолжение Листинга {0}".format(number) + return "Продолжение Таблицы {0}".format(number) + + +def caption_style_name(kind): + return "Caption Listing" if kind == "listing" else "Название таблицы" + + +def file_url_from_path(path): + abs_path = os.path.abspath(path) + posix = abs_path.replace("\\", "/") + if len(posix) >= 2 and posix[1] == ":": + return "file:///" + quote(posix, safe="/:@") + return "file://" + quote(posix, safe="/:@") + + +def path_to_file_url(uno, path): + abs_path = os.path.abspath(path) + try: + url = uno.systemPathToFileUrl(abs_path) + if url: + return url + except Exception: + pass + return file_url_from_path(abs_path) + + +def docx_filter_candidates(engine): + return DOCX_FILTER_CANDIDATES.get(engine, DOCX_FILTER_CANDIDATES["libre"]) + + +def docx_save_filter_specs(engine): + return DOCX_SAVE_FILTER_SPECS.get(engine, DOCX_SAVE_FILTER_SPECS["libre"]) diff --git a/md2gost/office/uno_profile.py b/md2gost/office/uno_profile.py new file mode 100644 index 0000000..4399893 --- /dev/null +++ b/md2gost/office/uno_profile.py @@ -0,0 +1,41 @@ +"""Persistent LibreOffice/OpenOffice user profiles for headless UNO runs.""" + +from __future__ import annotations + +import os +import tempfile +from pathlib import Path + +# Marks first-start / registration wizard as done (OpenOffice + LibreOffice). +_REGISTRY_SEED = """ + + + + true + + + true + + + +""" + + +def profile_dir(engine: str) -> Path: + """Writable profile reused across conversions (avoids OO registration every run).""" + root = os.environ.get("LOCALAPPDATA") or os.environ.get("APPDATA") or tempfile.gettempdir() + path = Path(root) / "md2gost" / "office-profiles" / engine + path.mkdir(parents=True, exist_ok=True) + return path + + +def ensure_profile_seeded(profile_root: Path) -> None: + """Pre-seed registry so OpenOffice does not show the registration wizard.""" + marker = profile_root / ".md2gost-seeded" + user_dir = profile_root / "user" + user_dir.mkdir(parents=True, exist_ok=True) + xcu = user_dir / "registrymodifications.xcu" + if not xcu.is_file(): + xcu.write_text(_REGISTRY_SEED, encoding="utf-8") + if not marker.is_file(): + marker.write_text("1", encoding="utf-8") diff --git a/md2gost/office/uno_worker.py b/md2gost/office/uno_worker.py new file mode 100644 index 0000000..148552d --- /dev/null +++ b/md2gost/office/uno_worker.py @@ -0,0 +1,1280 @@ +#!/usr/bin/env python +# -*- coding: utf-8 -*- +"""UNO post-process worker - run with LibreOffice/OpenOffice program/python.exe.""" + +from __future__ import print_function, unicode_literals + +import argparse +import json +import os +import sys +import time + +from uno_helpers import ( + MAX_PASSES, + CaptionInfo, + as_unicode, + caption_style_name, + continuation_label, + docx_filter_candidates, + docx_save_filter_specs, + BREAK_NONE, + BREAK_PAGE_BEFORE, + CONTROL_PARAGRAPH_BREAK, + find_page_break_row, + is_caption_orphaned, + is_orphan_header_break, + parse_caption_text, + path_to_file_url, + uno_call as _uno_call, + uno_error_text, +) + + +def _document_text(doc): + return _uno_call(doc, "Text", default=None) + + +def _document_controller(doc): + return _uno_call(doc, "CurrentController", default=None) + + +def _text_tables(doc): + return _uno_call(doc, "TextTables") + + +def _format_exc(exc): + name = type(exc).__name__ + try: + if getattr(exc, "args", None): + text = as_unicode(exc.args[0]) + else: + text = as_unicode(exc) + except Exception: + text = as_unicode(repr(exc)) + if text: + return "{0}: {1}".format(name, text) + return name + + +def _fs_path(value): + """Normalize argv path to unicode on Python 2.7 (Cyrillic paths on Windows).""" + if sys.version_info[0] >= 3: + return value + try: + text_type = unicode + except NameError: + return value + if isinstance(value, text_type): + return value + for encoding in (sys.getfilesystemencoding(), "utf-8", "mbcs"): + if not encoding: + continue + try: + return value.decode(encoding) + except (UnicodeDecodeError, LookupError): + continue + return value.decode("utf-8", "replace") + + +def _emit_result(result): + if sys.version_info[0] < 3: + result = _unicode_result(result) + payload = json.dumps(result, ensure_ascii=False) + if sys.version_info[0] < 3: + try: + if isinstance(payload, unicode): + payload = payload.encode("utf-8") + except NameError: + pass + if sys.version_info[0] >= 3: + sys.stdout.write(payload + "\n") + else: + sys.stdout.write(payload) + if not payload.endswith(b"\n"): + sys.stdout.write(b"\n") + sys.stdout.flush() + + +def _unicode_result(value): + if isinstance(value, dict): + return dict((as_unicode(k), _unicode_result(v)) for k, v in value.items()) + if isinstance(value, list): + return [_unicode_result(item) for item in value] + if isinstance(value, tuple): + return tuple(_unicode_result(item) for item in value) + if isinstance(value, (int, float, bool)) or value is None: + return value + return as_unicode(value) + + +def _connect(port): + import uno + + local_ctx = uno.getComponentContext() + resolver = local_ctx.ServiceManager.createInstanceWithContext( + "com.sun.star.bridge.UnoUrlResolver", local_ctx + ) + last_exc = None + for _ in range(int(45 / 0.2)): + try: + ctx = resolver.resolve( + "uno:socket,host=127.0.0.1,port={0};urp;StarOffice.ComponentContext".format( + port + ) + ) + return uno, ctx + except Exception as exc: + last_exc = exc + time.sleep(0.2) + raise RuntimeError( + "cannot connect to soffice on port {0}: {1}".format(port, last_exc) + ) + + +def _property(name, value): + from com.sun.star.beans import PropertyValue + + return PropertyValue(name, 0, value, 0) + + +def _is_writer_document(doc): + return _document_text(doc) is not None + + +def _load_document(uno, ctx, path, engine): + smgr = ctx.ServiceManager + desktop = smgr.createInstanceWithContext("com.sun.star.frame.Desktop", ctx) + file_url = path_to_file_url(uno, path) + ext = os.path.splitext(path)[1].lower() + if ext == ".docx": + filter_names = docx_filter_candidates(engine) + (None,) + else: + filter_names = (None,) + + errors = [] + for filter_name in filter_names: + props = [ + _property("Hidden", True), + _property("ReadOnly", False), + _property("MacroExecutionMode", 4), + ] + if filter_name: + props.append(_property("FilterName", filter_name)) + label = filter_name or "auto" + try: + doc = desktop.loadComponentFromURL(file_url, "_blank", 0, tuple(props)) + except Exception as exc: + errors.append("{0}: {1}".format(label, exc)) + continue + if doc is None: + errors.append("{0}: loadComponentFromURL returned None".format(label)) + continue + if not _is_writer_document(doc): + errors.append("{0}: not a Writer document".format(label)) + try: + doc.close(True) + except Exception: + pass + continue + return doc + + detail = "; ".join(errors) if errors else "unknown error" + raise RuntimeError("cannot open document: {0} ({1})".format(path, detail)) + + +def _reload_document(uno, ctx, doc, path, engine): + """Save and reopen so LO refreshes row page numbers after SplitTable.""" + _save_document(uno, doc, path, engine) + try: + doc.close(True) + except Exception: + pass + return _load_document(uno, ctx, path, engine) + + +def _save_odt_temp(uno, doc): + """Save to an ASCII-only temp ODT; host converts to DOCX (Py2 + Cyrillic paths).""" + import tempfile + + fd, odt_path = tempfile.mkstemp(prefix="md2gost-", suffix=".odt") + os.close(fd) + odt_url = path_to_file_url(uno, odt_path) + doc.storeToURL(odt_url, (_property("Overwrite", True),)) + if not os.path.isfile(odt_path): + raise RuntimeError("temp ODT not created: {0}".format(odt_path)) + return odt_path + + +def _save_document(uno, doc, path, engine): + """Save document. Returns temp ODT path when host must export DOCX.""" + ext = os.path.splitext(path)[1].lower() + if ext != ".docx": + doc.store() + return None + if engine == "openoffice": + return _save_odt_temp(uno, doc) + + file_url = path_to_file_url(uno, path) + errors = [] + for filter_name in docx_save_filter_specs(engine) + (None,): + props = [_property("Overwrite", True)] + if filter_name: + props.append(_property("FilterName", filter_name)) + try: + doc.storeToURL(file_url, tuple(props)) + return None + except Exception as exc: + errors.append( + "{0}: {1}".format(filter_name or "auto", uno_error_text(exc)) + ) + try: + doc.store() + return None + except Exception as exc: + errors.append("store: {0}".format(uno_error_text(exc))) + detail = "; ".join(errors) if errors else "unknown error" + raise RuntimeError("cannot save document: {0} ({1})".format(path, detail)) + + +def _dispatch(uno, ctx, doc, command, args=()): + smgr = ctx.ServiceManager + dispatcher = smgr.createInstanceWithContext( + "com.sun.star.frame.DispatchHelper", ctx + ) + controller = _document_controller(doc) + if controller is None: + raise RuntimeError("no document controller") + frame = _uno_call(controller, "Frame") + dispatcher.executeDispatch(frame, command, "", 0, args) + + +def _repaginate(uno, ctx, doc): + for command in (".uno:UpdateAll", ".uno:Repaginate"): + try: + _dispatch(uno, ctx, doc, command) + return + except Exception: + continue + try: + controller = _document_controller(doc) + if controller is not None: + view = _uno_call(controller, "View", default=None) + if view is not None: + view.Invalidate() + except Exception: + pass + + +def _page_at_range(doc, text_range): + try: + controller = _document_controller(doc) + if controller is None: + return None + view_cursor = _uno_call(controller, "ViewCursor") + view_cursor.gotoRange(text_range, False) + page = _view_cursor_page(view_cursor) + if page is None: + return None + return int(page) + except Exception: + return None + + +def _view_cursor_page(view_cursor): + try: + import uno + + from com.sun.star.text import XPageCursor + + page_cursor = uno.queryInterface(XPageCursor, view_cursor) + if page_cursor is not None: + return int(page_cursor.getPage()) + except Exception: + pass + try: + return int(_uno_call(view_cursor, "Page")) + except Exception: + return None + + +def _ensure_pagination(uno, ctx, doc): + for command in (".uno:UpdateAll", ".uno:Repaginate", ".uno:ZoomPageWidth"): + try: + _dispatch(uno, ctx, doc, command) + except Exception: + continue + try: + _uno_call(doc, "PageCount") + except Exception: + pass + time.sleep(0.3) + + +def _diagnose_pagination(doc, details): + try: + collection = _text_tables(doc) + count = int(_uno_call(collection, "Count")) + details.append("diag: tables={0}".format(count)) + for idx in range(min(count, 4)): + table = collection.getByIndex(idx) + caption = _caption_before_table(doc, table) + pages = _row_page_numbers(doc, table) + prev = _paragraph_text_before_table(doc, table) + if caption: + label = "{0} {1}".format(caption.kind, caption.number) + else: + preview = prev[:48] + ("..." if len(prev) > 48 else "") + label = "no-caption prev={0!r}".format(preview) + if pages: + page_info = "{0}-{1} (n={2})".format( + min(pages), max(pages), len(pages) + ) + else: + page_info = "none" + details.append( + "diag: table[{0}] {1} pages={2}".format(idx, label, page_info) + ) + except Exception as exc: + details.append("diag: {0}".format(uno_error_text(exc))) + + +def _supports_service(obj, name): + try: + return bool(obj.supportsService(name)) + except Exception: + return False + + +def _element_text(elem): + try: + return (_uno_call(elem, "String", default="") or "").replace("\r", "").strip() + except Exception: + return "" + + +def _same_text_table(a, b): + try: + if a == b: + return True + except Exception: + pass + try: + return _uno_call(a, "Anchor") == _uno_call(b, "Anchor") + except Exception: + return False + + +def _is_target_table(elem, table): + try: + if elem == table: + return True + except Exception: + pass + return _same_text_table(elem, table) + + +def _body_tables(doc): + """All body tables in document order (LO ``getByIndex`` can alias split fragments).""" + text = _document_text(doc) + if text is None: + return [] + try: + enum = text.createEnumeration() + except Exception: + return [] + tables = [] + while enum.hasMoreElements(): + elem = enum.nextElement() + if _supports_service(elem, "com.sun.star.text.TextTable"): + tables.append(elem) + return tables + + +def _next_table_in_body(doc, table): + """Return the next body table after *table* (LO ``getByIndex(i+1)`` can alias).""" + text = _document_text(doc) + if text is None: + return None + try: + enum = text.createEnumeration() + except Exception: + return None + + seen = False + while enum.hasMoreElements(): + elem = enum.nextElement() + if not _supports_service(elem, "com.sun.star.text.TextTable"): + continue + if not seen: + if _is_target_table(elem, table): + seen = True + continue + return elem + return None + + +def _paragraphs_before_table(doc, table, max_scan=6): + """Walk body content in order; LO/OO table anchors are not navigable.""" + text = _document_text(doc) + if text is None: + return [] + try: + enum = text.createEnumeration() + except Exception: + return [] + recent = [] + while enum.hasMoreElements(): + elem = enum.nextElement() + if _supports_service(elem, "com.sun.star.text.TextTable"): + if _is_target_table(elem, table): + return recent[-max_scan:] + recent = [] + continue + if _supports_service(elem, "com.sun.star.text.Paragraph"): + para_text = _element_text(elem) + if para_text: + recent.append(para_text) + return [] + + +def _paragraph_text_before_table(doc, table): + paras = _paragraphs_before_table(doc, table, max_scan=1) + return paras[-1] if paras else "" + + +def _prev_body_element_before_table(doc, table): + """Body element immediately before *table* (paragraph or table).""" + text = _document_text(doc) + if text is None: + return None + try: + enum = text.createEnumeration() + except Exception: + return None + prev_elem = None + while enum.hasMoreElements(): + elem = enum.nextElement() + if _supports_service(elem, "com.sun.star.text.TextTable") and _is_target_table( + elem, table + ): + return prev_elem + prev_elem = elem + return None + + +def _caption_before_table(doc, table): + for para_text in reversed(_paragraphs_before_table(doc, table, max_scan=6)): + caption = parse_caption_text(para_text) + if caption is not None: + return caption + return None + + +def _caption_for_split(doc, table): + """Caption for split decisions, including orphan continuation fragments.""" + caption = _caption_before_table(doc, table) + if caption is not None: + return caption + prev_table = _first_table_fragment_before(doc, table) + if prev_table is None: + return None + upstream = _caption_before_table(doc, prev_table) + if upstream is None: + return None + return CaptionInfo( + kind=upstream.kind, + number=upstream.number, + is_continuation=True, + ) + + +def _caption_page_before_table(doc, table): + try: + anchor = _uno_call(table, "Anchor") + text = _document_text(doc) + if text is None: + return None + cursor = text.createTextCursorByRange(anchor) + if not cursor.goLeft(1, False): + return None + cursor.gotoRange(anchor, False) + cursor.goLeft(1, False) + cursor.gotoStartOfParagraph(False) + return _page_at_range(doc, cursor) + except Exception: + return None + + +def _row_page_numbers(doc, table): + pages = [] + try: + nrows = int(table.getRows().getCount()) + except Exception: + return pages + for row in range(nrows): + page = None + for col in range(max(1, int(table.getColumns().getCount()))): + try: + cell = table.getCellByPosition(col, row) + page = _page_at_range(doc, _uno_call(cell, "Start")) + if page is not None: + break + except Exception: + continue + if page is None: + page = pages[-1] if pages else 1 + pages.append(page) + return pages + + +def _table_has_header_row(table): + try: + if bool(table.getPropertyValue("RepeatHeadline")): + return True + except Exception: + pass + try: + row = table.getRows().getByIndex(0) + return bool(row.getPropertyValue("IsRepeatHeading")) + except Exception: + return False + + +def _page_break_before_table_caption(doc, table): + try: + anchor = _uno_call(table, "Anchor") + text = _document_text(doc) + if text is None: + return False + cursor = text.createTextCursorByRange(anchor) + if not cursor.goLeft(1, False): + return False + cursor.gotoRange(anchor, False) + cursor.goLeft(1, False) + cursor.gotoStartOfParagraph(False) + info = parse_caption_text((_uno_call(cursor, "String", default="") or "").strip()) + if info is None or info.is_continuation: + return False + cursor.setPropertyValue("BreakType", BREAK_PAGE_BEFORE) + cursor.setPropertyValue("ParaKeepTogether", True) + return True + except Exception: + return False + + +def _cursor_after_element(text, elem, cursor): + """Move *cursor* to the end of a body-text element (paragraph or table).""" + if _supports_service(elem, "com.sun.star.text.TextTable"): + try: + nrows = int(elem.getRows().getCount()) + ncols = int(elem.getColumns().getCount()) + if nrows > 0 and ncols > 0: + cell = elem.getCellByPosition(ncols - 1, nrows - 1) + cursor.gotoRange(_uno_call(cell, "End"), False) + return True + except Exception: + pass + try: + if _supports_service(elem, "com.sun.star.text.TextTable"): + cursor.gotoRange(_uno_call(elem, "Anchor"), False) + else: + cursor.gotoRange(elem, False) + cursor.collapseToEnd() + return True + except Exception: + pass + try: + cursor.gotoRange(_uno_call(elem, "Anchor"), False) + cursor.gotoEndOfParagraph(False) + return True + except Exception: + return False + + +def _apply_caption_style(cursor, caption): + style_name = caption_style_name(caption.kind) + for name in (style_name, "Caption Table", "Caption"): + try: + cursor.setPropertyValue("ParaStyleName", name) + break + except Exception: + continue + try: + cursor.setPropertyValue("BreakType", BREAK_NONE) + cursor.setPropertyValue("ParaKeepTogether", True) + except Exception: + pass + + +def _first_table_fragment_before(doc, table): + """Return the first table fragment before *table* (skip empty paragraphs).""" + text = _document_text(doc) + if text is None: + return None + try: + enum = text.createEnumeration() + except Exception: + return None + + seen = [] + while enum.hasMoreElements(): + elem = enum.nextElement() + if _supports_service(elem, "com.sun.star.text.TextTable") and _is_target_table( + elem, table + ): + for prev in reversed(seen): + if _supports_service(prev, "com.sun.star.text.TextTable"): + return prev + if _supports_service(prev, "com.sun.star.text.Paragraph"): + para_text = _element_text(prev) + if not para_text: + continue + info = parse_caption_text(para_text) + if info is not None and not info.is_continuation: + return None + return None + seen.append(elem) + return None + + +def _write_continuation_paragraph(text, paragraph, label, caption): + """Fill or replace a body paragraph with a continuation caption.""" + start = _uno_call(paragraph, "Start") + end = _uno_call(paragraph, "End") + cursor = text.createTextCursorByRange(start) + cursor.gotoRange(end, True) + cursor.setString(label) + _apply_caption_style(cursor, caption) + + +def _insert_paragraph_before_table(uno, ctx, doc, table, label, caption): + """Insert a caption paragraph before *table* via dispatch (anchor range is often invalid).""" + try: + cell = table.getCellByPosition(0, 0) + start = _uno_call(cell, "Start") + except Exception as exc: + raise RuntimeError("cannot reach continuation table cell: {0}".format(exc)) + + controller = _document_controller(doc) + if controller is None: + raise RuntimeError("no document controller") + view_cursor = _uno_call(controller, "ViewCursor") + view_cursor.gotoRange(start, False) + _dispatch(uno, ctx, doc, ".uno:InsertPara", ()) + try: + view_cursor.goLeft(1, False) + except Exception: + pass + view_cursor.setString(label) + _apply_caption_style(view_cursor, caption) + + +def _insert_continuation_at(doc, prev_elem, cont_table, label, caption, uno=None, ctx=None): + text = _document_text(doc) + if text is None: + raise RuntimeError("no document text") + if prev_elem is None: + raise RuntimeError("continuation table at document start") + + if _supports_service(prev_elem, "com.sun.star.text.Paragraph"): + prev_text = _element_text(prev_elem) + if prev_text.strip(): + info = parse_caption_text(prev_text) + if info is not None and info.is_continuation: + return + raise RuntimeError("non-empty paragraph before continuation table") + _write_continuation_paragraph(text, prev_elem, label, caption) + return + if _supports_service(prev_elem, "com.sun.star.text.TextTable"): + if uno is None or ctx is None: + raise RuntimeError("dispatch insert requires UNO context") + _insert_paragraph_before_table(uno, ctx, doc, cont_table, label, caption) + return + raise RuntimeError("unsupported block before continuation table") + + +def _insert_continuation_after_fragment(doc, first_table, caption, uno=None, ctx=None): + """Insert «Продолжение…» before the table fragment that follows *first_table*.""" + label = continuation_label(caption.kind, caption.number) + text = _document_text(doc) + if text is None: + raise RuntimeError("no document text") + try: + enum = text.createEnumeration() + except Exception as exc: + raise RuntimeError("cannot enumerate body text: {0}".format(exc)) + + prev_elem = None + seen_first = False + while enum.hasMoreElements(): + elem = enum.nextElement() + if _supports_service(elem, "com.sun.star.text.TextTable"): + if seen_first: + return _insert_continuation_at( + doc, prev_elem, elem, label, caption, uno, ctx + ) + if _is_target_table(elem, first_table): + seen_first = True + prev_elem = elem + continue + prev_elem = elem + raise RuntimeError("continuation fragment not found after split") + + +def _insert_continuation_before(doc, table, caption, uno=None, ctx=None): + """Insert «Продолжение…» before an already-split continuation fragment.""" + label = continuation_label(caption.kind, caption.number) + text = _document_text(doc) + if text is None: + raise RuntimeError("no document text") + try: + enum = text.createEnumeration() + except Exception as exc: + raise RuntimeError("cannot enumerate body text: {0}".format(exc)) + + prev_elem = None + match = None + while enum.hasMoreElements(): + elem = enum.nextElement() + if _supports_service(elem, "com.sun.star.text.TextTable"): + if elem == table: + match = (prev_elem, elem) + elif match is None and _same_text_table(elem, table): + match = (prev_elem, elem) + prev_elem = elem + continue + prev_elem = elem + if match is None: + raise RuntimeError("continuation table not found in body text") + prev_elem, cont_table = match + if not _needs_continuation_caption(doc, cont_table): + return + return _insert_continuation_at( + doc, prev_elem, cont_table, label, caption, uno, ctx + ) + + +def _needs_continuation_caption(doc, cont_table): + prev_elem = _prev_body_element_before_table(doc, cont_table) + if prev_elem is None: + return False + if _supports_service(prev_elem, "com.sun.star.text.TextTable"): + return True + if _supports_service(prev_elem, "com.sun.star.text.Paragraph"): + prev_text = _element_text(prev_elem) + if not prev_text.strip(): + return True + info = parse_caption_text(prev_text) + return info is None or not info.is_continuation + return True + + +def _has_continuation_before_table(doc, cont_table): + return not _needs_continuation_caption(doc, cont_table) + + +def _repair_split_table_borders(doc, tables, listings, details): + """Clear bottom borders on first fragments of already-split tables.""" + if not tables: + return 0 + text = _document_text(doc) + if text is None: + return 0 + try: + enum = text.createEnumeration() + except Exception: + return 0 + + fixed = 0 + last_table = None + while enum.hasMoreElements(): + elem = enum.nextElement() + if _supports_service(elem, "com.sun.star.text.TextTable"): + if last_table is not None and _has_continuation_before_table(doc, elem): + caption = _caption_for_split(doc, elem) or _caption_before_table( + doc, last_table + ) + if caption is not None and caption.kind == "table": + try: + _clear_table_bottom_border(last_table) + fixed += 1 + details.append( + "repair: clear bottom border table {0}".format( + caption.number + ) + ) + except Exception as exc: + details.append( + "skip: repair bottom border table {0}: {1}".format( + caption.number, exc + ) + ) + last_table = elem + continue + + if _supports_service(elem, "com.sun.star.text.Paragraph"): + para_text = _element_text(elem) + if not para_text: + continue + info = parse_caption_text(para_text) + if info is not None and not info.is_continuation: + last_table = None + return fixed + + +def _repair_missing_continuations(doc, uno, ctx, tables, listings, details): + """Insert captions between already-split table pairs (re-run after failed insert).""" + text = _document_text(doc) + if text is None: + return 0 + try: + enum = text.createEnumeration() + except Exception: + return 0 + + repaired = 0 + last_table = None + while enum.hasMoreElements(): + elem = enum.nextElement() + if _supports_service(elem, "com.sun.star.text.TextTable"): + if last_table is not None and _needs_continuation_caption(doc, elem): + caption = _caption_for_split(doc, elem) + if caption is not None: + if caption.kind == "table" and not tables: + caption = None + elif caption.kind == "listing" and not listings: + caption = None + if caption is not None: + try: + _insert_continuation_before(doc, elem, caption, uno, ctx) + repaired += 1 + details.append( + "repair: {0} {1} continuation caption".format( + caption.kind, caption.number + ) + ) + except Exception as exc: + details.append( + "skip: repair continuation {0} {1}: {2}".format( + caption.kind, caption.number, exc + ) + ) + last_table = elem + continue + + if _supports_service(elem, "com.sun.star.text.Paragraph"): + para_text = _element_text(elem) + if not para_text: + continue + info = parse_caption_text(para_text) + if info is not None and not info.is_continuation: + last_table = None + return repaired + + +def _refresh_table_layout(doc, table): + """Touch each row so LO recalculates page numbers after SplitTable.""" + controller = _document_controller(doc) + if controller is None: + return + try: + view_cursor = _uno_call(controller, "ViewCursor") + except Exception: + return + try: + nrows = int(table.getRows().getCount()) + ncols = max(1, int(table.getColumns().getCount())) + except Exception: + return + for row in range(nrows): + try: + cell = table.getCellByPosition(0, row) + view_cursor.gotoRange(_uno_call(cell, "Start"), False) + _view_cursor_page(view_cursor) + except Exception: + continue + + +def _split_table_at_row(uno, ctx, doc, table, break_at_1based): + from com.sun.star.beans import PropertyValue + + # Match Word ``SplitBeforeRow``: split before 1-based *break_at_1based*. + row_0based = break_at_1based - 1 + if row_0based < 1: + return False + try: + cell = table.getCellByPosition(0, row_0based) + except Exception: + return False + controller = _document_controller(doc) + if controller is None: + return False + view_cursor = _uno_call(controller, "ViewCursor") + view_cursor.gotoRange(_uno_call(cell, "Start"), False) + arg = PropertyValue("Type", 0, 3, 0) + _dispatch(uno, ctx, doc, ".uno:SplitTable", (arg,)) + return True + + +def _border_line_style_none(): + try: + from com.sun.star.table import BorderLineStyle + + return BorderLineStyle.NONE + except Exception: + return 32767 + + +def _none_border_line(): + """BorderLine2 with LineStyle NONE (matches Word wdLineStyleNone for DOCX export).""" + try: + from com.sun.star.table import BorderLine2 + + border = BorderLine2() + except Exception: + import uno + + border = uno.createUnoStruct("com.sun.star.table.BorderLine2") + border.LineStyle = _border_line_style_none() + border.LineWidth = 0 + border.OuterLineWidth = 0 + border.InnerLineWidth = 0 + border.LineDistance = 0 + return border + + +def _clear_border_line(border): + try: + border.LineStyle = _border_line_style_none() + except Exception: + pass + border.LineWidth = 0 + border.OuterLineWidth = 0 + border.InnerLineWidth = 0 + border.LineDistance = 0 + return border + + +def _clear_table_bottom_border(table): + """Remove the visible bottom edge of a table fragment (LO keeps it on last-row cells).""" + touched = False + + try: + nrows = int(table.getRows().getCount()) + ncols = int(table.getColumns().getCount()) + except Exception: + nrows = 0 + ncols = 0 + + if nrows > 0 and ncols > 0: + last_row = nrows - 1 + try: + row = table.getRows().getByIndex(last_row) + row.setPropertyValue("BottomBorder", _none_border_line()) + touched = True + except Exception: + pass + for col in range(ncols): + try: + cell = table.getCellByPosition(col, last_row) + cell.setPropertyValue("BottomBorder", _none_border_line()) + touched = True + except Exception: + continue + + for prop in ("BottomBorder",): + try: + border = table.getPropertyValue(prop) + table.setPropertyValue(prop, _clear_border_line(border)) + touched = True + except Exception: + pass + + try: + table_border = table.getPropertyValue("TableBorder") + table_border.BottomLine = _clear_border_line(table_border.BottomLine) + table.setPropertyValue("TableBorder", table_border) + touched = True + except Exception: + pass + + if not touched: + raise RuntimeError("no bottom border property accessible") + + +def _apply_listing_box_borders(table): + for edge in ("TopBorder", "BottomBorder", "LeftBorder", "RightBorder"): + try: + border = table.getPropertyValue(edge) + border.LineWidth = 1 + table.setPropertyValue(edge, border) + except Exception: + pass + for edge in ("HBorder", "VBorder"): + try: + border = table.getPropertyValue(edge) + border.LineWidth = 0 + table.setPropertyValue(edge, border) + except Exception: + pass + + +def _copy_header_to_continuation(first_table, cont_table): + try: + src_row = first_table.getRows().getByIndex(0) + rows = cont_table.getRows() + if int(rows.getCount()) < 1: + return + try: + if bool(rows.getByIndex(0).getPropertyValue("IsRepeatHeading")): + return + except Exception: + pass + rows.insertByIndex(0, 1) + new_row = rows.getByIndex(0) + ncols = int(cont_table.getColumns().getCount()) + for col in range(ncols): + try: + src_cell = src_row.getCellByPosition(col) + dst_cell = new_row.getCellByPosition(col) + dst_cell.setString((src_cell.getString() or "").strip()) + except Exception: + continue + try: + new_row.setPropertyValue("IsRepeatHeading", True) + except Exception: + pass + except Exception: + pass + + +def _fix_pass(uno, ctx, doc, tables, listings, repeat_header, details): + made = 0 + body_tables = _body_tables(doc) + for table in reversed(body_tables): + caption = _caption_for_split(doc, table) + if caption is None: + continue + if caption.kind == "table" and not tables: + continue + if caption.kind == "listing" and not listings: + continue + + pages = _row_page_numbers(doc, table) + if caption.kind == "table" and pages and not caption.is_continuation: + if is_caption_orphaned(_caption_page_before_table(doc, table), pages[0]): + try: + if _page_break_before_table_caption(doc, table): + details.append( + "orphan-caption: {0} {1} -> page break before caption".format( + caption.kind, caption.number + ) + ) + _repaginate(uno, ctx, doc) + else: + details.append( + "skip: orphan-caption {0} {1}: no caption para".format( + caption.kind, caption.number + ) + ) + except Exception as exc: + details.append("skip: orphan-caption page break: {0}".format(exc)) + continue + + break_at = find_page_break_row(pages) + if break_at is None: + continue + if break_at > int(table.getRows().getCount()): + continue + + had_header = _table_has_header_row(table) + if is_orphan_header_break(break_at, had_header): + try: + if _page_break_before_table_caption(doc, table): + details.append( + "orphan-header: {0} {1} -> page break before caption".format( + caption.kind, caption.number + ) + ) + _repaginate(uno, ctx, doc) + else: + details.append( + "skip: orphan-header {0} {1}: no caption para".format( + caption.kind, caption.number + ) + ) + except Exception as exc: + details.append("skip: orphan-header page break: {0}".format(exc)) + continue + + try: + if not _split_table_at_row(uno, ctx, doc, table, break_at): + details.append( + "skip: split {0} {1} row {2}: dispatch failed".format( + caption.kind, caption.number, break_at + ) + ) + continue + except Exception as exc: + details.append( + "skip: split {0} {1} row {2}: {3}".format( + caption.kind, caption.number, break_at, exc + ) + ) + continue + + first_table = table + cont_table = _next_table_in_body(doc, first_table) + if cont_table is None: + details.append( + "skip: after split {0} {1}: no continuation fragment".format( + caption.kind, caption.number + ) + ) + made += 1 + continue + + try: + _insert_continuation_after_fragment( + doc, first_table, caption, uno, ctx + ) + except Exception as exc: + details.append("skip: insert caption after split: {0}".format(exc)) + + if caption.kind == "table": + try: + _clear_table_bottom_border(first_table) + except Exception as exc: + details.append("skip: clear bottom border: {0}".format(exc)) + + if caption.kind == "listing": + try: + _apply_listing_box_borders(first_table) + _apply_listing_box_borders(cont_table) + except Exception as exc: + details.append("skip: listing borders: {0}".format(exc)) + + if repeat_header and had_header and caption.kind == "table": + try: + _copy_header_to_continuation(first_table, cont_table) + except Exception as exc: + details.append("skip: header copy: {0}".format(exc)) + + _refresh_table_layout(doc, first_table) + _refresh_table_layout(doc, cont_table) + _repaginate(uno, ctx, doc) + _ensure_pagination(uno, ctx, doc) + made += 1 + details.append( + "split: {0} {1} @row {2}".format(caption.kind, caption.number, break_at) + ) + return made + + return made + + +def run_worker(port, path, engine, tables, listings, repeat_header): + label = "LibreOffice" if engine == "libre" else "OpenOffice" + uno, ctx = _connect(port) + doc = _load_document(uno, ctx, path, engine) + splits = 0 + details = [] + odt_temp = None + try: + _ensure_pagination(uno, ctx, doc) + repaired = _repair_missing_continuations( + doc, uno, ctx, tables, listings, details + ) + _repair_split_table_borders(doc, tables, listings, details) + for _ in range(MAX_PASSES): + _ensure_pagination(uno, ctx, doc) + made = _fix_pass( + uno, ctx, doc, tables, listings, repeat_header, details + ) + if made == 0: + break + splits += made + doc = _reload_document(uno, ctx, doc, path, engine) + _ensure_pagination(uno, ctx, doc) + repaired += _repair_missing_continuations( + doc, uno, ctx, tables, listings, details + ) + if splits == 0 and repaired == 0: + _diagnose_pagination(doc, details) + elif repaired: + _ensure_pagination(uno, ctx, doc) + odt_temp = _save_document(uno, doc, path, engine) + finally: + try: + doc.close(True) + except Exception: + pass + try: + smgr = ctx.ServiceManager + desktop = smgr.createInstanceWithContext( + "com.sun.star.frame.Desktop", ctx + ) + desktop.terminate() + except Exception: + pass + + skipped = sum(1 for line in details if line.startswith("skip:")) + repaired = sum(1 for line in details if line.startswith("repair:")) + msg = "Пост-разрыв {0}: разрезов {1}".format(label, splits) + if repaired: + msg += ", подписей {0}".format(repaired) + if skipped: + msg += ", пропусков {0}".format(skipped) + msg += "." + result = { + "ok": True, + "splits": splits, + "skipped": skipped, + "message": msg, + "details": details, + } + if odt_temp: + result["odt_temp"] = odt_temp + return result + + +def main(): + parser = argparse.ArgumentParser(description="md2gost UNO table split worker") + parser.add_argument("--port", type=int, required=True) + parser.add_argument("--path", required=True) + parser.add_argument("--engine", choices=("libre", "openoffice"), default="libre") + parser.add_argument("--tables", choices=("0", "1"), default="1") + parser.add_argument("--listings", choices=("0", "1"), default="1") + parser.add_argument("--repeat-header", choices=("0", "1"), default="0") + args = parser.parse_args() + doc_path = _fs_path(args.path) + try: + result = run_worker( + args.port, + doc_path, + args.engine, + args.tables == "1", + args.listings == "1", + args.repeat_header == "1", + ) + _emit_result(result) + return 0 + except Exception as exc: + label = "LibreOffice" if args.engine == "libre" else "OpenOffice" + _emit_result( + { + "ok": False, + "splits": 0, + "skipped": 0, + "message": "Пост-разрыв ({0}) не удался: {1}".format( + label, _format_exc(exc) + ), + "details": [], + } + ) + return 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/md2gost/pipeline.py b/md2gost/pipeline.py index 07b2c24..ea39327 100644 --- a/md2gost/pipeline.py +++ b/md2gost/pipeline.py @@ -17,6 +17,7 @@ from docx import Document from . import package_dir from .checker import check_markdown, format_report from .converter import Converter +from .office import DEFAULT_OFFICE_ENGINE from .profiles import ( DEFAULT_HEADING_NUMBERING, DEFAULT_TABLE_CONTINUATION, @@ -60,6 +61,7 @@ class ConvertRequest: open_when_done: bool = False check_pages: bool = False table_repeat_header: bool = False + office_engine: str = DEFAULT_OFFICE_ENGINE # PAGE field start on body section; None = continuous from document start. page_number_start: int | None = None # DOCX File → Info; None = take from md2gost.user.json (see doc_metadata). @@ -425,12 +427,16 @@ def convert(req: ConvertRequest, log: LogFn | None = None) -> ConvertResult: abs_out = os.path.abspath(output) emit(f"Generated document: {abs_out}") + split_engine: str | None = None + split_install = None if req.table_continuation == "word" or req.listing_continuation == "word": - from .word_fix import fix_continuations + from .office import fix_continuations_with_engine, resolve_office_engine from .page_geometry import patch_docx_page_starts - fix = fix_continuations( + split_engine, split_install = resolve_office_engine(req.office_engine) + fix = fix_continuations_with_engine( abs_out, + engine=req.office_engine, tables=(req.table_continuation == "word"), listings=(req.listing_continuation == "word"), repeat_header=bool(req.table_repeat_header), @@ -463,7 +469,16 @@ def convert(req: ConvertRequest, log: LogFn | None = None) -> ConvertResult: if req.debug or req.open_when_done: try: - open_document(abs_out) + if split_engine in ("word", "libre", "openoffice"): + from .office.open_doc import open_document_with_engine + + open_document_with_engine( + abs_out, + engine=split_engine, + install=split_install, + ) + else: + open_document(abs_out) except Exception as exc: emit(f"Не удалось открыть файл: {exc}") diff --git a/md2gost/word_fix.py b/md2gost/word_fix.py index 5360861..2f0824d 100644 --- a/md2gost/word_fix.py +++ b/md2gost/word_fix.py @@ -39,6 +39,7 @@ class FixResult: skipped: int = 0 message: str = "" details: list[str] = field(default_factory=list) + odt_temp: str | None = None def parse_caption_text(text: str) -> CaptionInfo | None: diff --git a/pyproject.toml b/pyproject.toml index d8a5607..04680ca 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -52,6 +52,8 @@ packages = ["md2gost", "mdexport", "md2latex", "md2fodt", "word2md"] "md2gost/TitleTemplate.docx" = "md2gost/TitleTemplate.docx" "md2gost/mml2omml" = "md2gost/mml2omml" "md2gost/diagrams" = "md2gost/diagrams" +"md2gost/office/uno_worker.py" = "md2gost/office/uno_worker.py" +"md2gost/office/uno_helpers.py" = "md2gost/office/uno_helpers.py" "latex/mirea" = "latex/mirea" "prompts/generate-md.md" = "md2gost/prompts/generate-md.md" "prompts/generate-mirea-report.md" = "md2gost/prompts/generate-mirea-report.md" diff --git a/tests/test_office.py b/tests/test_office.py new file mode 100644 index 0000000..5ca8d9b --- /dev/null +++ b/tests/test_office.py @@ -0,0 +1,269 @@ +"""Tests for office-engine detection and dispatch.""" + +from __future__ import annotations + +import os +import sys +from pathlib import Path +from unittest.mock import patch + +import pytest + +from md2gost.office import ( + DEFAULT_OFFICE_ENGINE, + OFFICE_ENGINE_MODES, + OfficeInstall, + detect_office_install, + fix_continuations_with_engine, + resolve_office_engine, +) +from md2gost.office import uno_fix +from md2gost.office import uno_profile +from md2gost.office.uno_cli import soffice_cli_path +from md2gost.office.uno_helpers import ( + as_unicode, + docx_filter_candidates, + file_url_from_path, + uno_call, +) +from md2gost.__main__ import build_parser, request_from_args +from md2gost.word_fix import FixResult + + +def test_office_engine_modes(): + assert DEFAULT_OFFICE_ENGINE == "auto" + assert OFFICE_ENGINE_MODES == ("auto", "word", "libre", "openoffice") + + +def test_parser_office_engine_default(): + parser = build_parser() + args = parser.parse_args(["report.md"]) + req = request_from_args(args) + assert req.office_engine == "auto" + + +def test_parser_office_engine_libre(): + parser = build_parser() + args = parser.parse_args(["report.md", "--office-engine", "libre"]) + req = request_from_args(args) + assert req.office_engine == "libre" + + +@patch("md2gost.office.detect._detect_word") +@patch("md2gost.office.detect._find_soffice_install") +def test_resolve_auto_prefers_word(mock_lo, mock_word): + mock_lo.return_value = OfficeInstall( + engine="libre", + label="LibreOffice", + soffice_path=r"C:\LibreOffice\program\soffice.exe", + python_path=r"C:\LibreOffice\program\python.exe", + ) + mock_word.return_value = OfficeInstall(engine="word", label="Microsoft Word") + engine, install = resolve_office_engine("auto") + assert engine == "word" + assert install is not None + assert install.engine == "word" + + +@patch("md2gost.office.detect._detect_word") +@patch("md2gost.office.detect._find_soffice_install") +def test_resolve_auto_falls_back_to_libre(mock_lo, mock_word): + mock_word.return_value = None + mock_lo.return_value = OfficeInstall( + engine="libre", + label="LibreOffice", + soffice_path=r"C:\LibreOffice\program\soffice.exe", + python_path=r"C:\LibreOffice\program\python.exe", + ) + + def _lo(kind: str): + return mock_lo.return_value if kind == "libre" else None + + with patch("md2gost.office.detect._find_soffice_install", side_effect=_lo): + engine, install = resolve_office_engine("auto") + assert engine == "libre" + assert install is not None + assert install.engine == "libre" + + +@patch("md2gost.office.detect._detect_word") +@patch("md2gost.office.detect._find_soffice_install") +def test_resolve_auto_missing(mock_lo, mock_word): + mock_word.return_value = None + mock_lo.return_value = None + engine, install = resolve_office_engine("auto") + assert engine is None + assert install is None + + +def test_detect_libre_from_custom_path(tmp_path): + program = tmp_path / "program" + program.mkdir() + soffice = program / ("soffice.exe" if sys.platform == "win32" else "soffice") + python_bin = program / ("python.exe" if sys.platform == "win32" else "python") + soffice.write_text("", encoding="utf-8") + python_bin.write_text("", encoding="utf-8") + + with patch("md2gost.office.detect._program_dirs", return_value=[program]): + install = detect_office_install("libre") + assert install is not None + assert install.engine == "libre" + assert install.soffice_path == str(soffice) + assert install.python_path == str(python_bin) + + +@patch("md2gost.office.dispatch.fix_continuations") +@patch("md2gost.office.dispatch.resolve_office_engine") +def test_dispatch_word_backend(mock_resolve, mock_word_fix): + mock_resolve.return_value = ("word", OfficeInstall(engine="word", label="Microsoft Word")) + mock_word_fix.return_value = FixResult(True, splits=2, message="ok") + result = fix_continuations_with_engine("report.docx", engine="word") + assert result.ok is True + mock_word_fix.assert_called_once() + + +@patch("md2gost.office.dispatch.fix_continuations_uno") +@patch("md2gost.office.dispatch.resolve_office_engine") +def test_dispatch_libre_backend(mock_resolve, mock_uno): + install = OfficeInstall( + engine="libre", + label="LibreOffice", + soffice_path=r"C:\LibreOffice\program\soffice.exe", + python_path=r"C:\LibreOffice\program\python.exe", + ) + mock_resolve.return_value = ("libre", install) + mock_uno.return_value = FixResult(True, splits=1, message="uno ok") + result = fix_continuations_with_engine("report.docx", engine="libre") + assert result.ok is True + mock_uno.assert_called_once() + + +@patch("md2gost.office.dispatch.resolve_office_engine") +def test_dispatch_missing_engine(mock_resolve): + mock_resolve.return_value = (None, None) + result = fix_continuations_with_engine("report.docx", engine="libre") + assert result.ok is False + assert "пропущен" in result.message.lower() + + +def test_soffice_cli_path_prefers_com_on_windows(tmp_path): + program = tmp_path / "program" + program.mkdir() + exe = program / "soffice.exe" + com = program / "soffice.com" + exe.write_text("", encoding="utf-8") + com.write_text("", encoding="utf-8") + if sys.platform == "win32": + assert soffice_cli_path(str(exe)) == str(com) + else: + assert soffice_cli_path(str(exe)) == str(exe) + + +def test_file_url_from_path_cyrillic(): + url = file_url_from_path(r"G:\Sync_File\Практическая работа 2.test.docx") + assert url.startswith("file:///") + assert "%" in url or "Практическая" in url + + +def test_docx_filters_openoffice_first(): + assert docx_filter_candidates("openoffice")[0] == "MS Word 2007 XML" + + +def test_profile_seeded_once(tmp_path, monkeypatch): + monkeypatch.setenv("LOCALAPPDATA", str(tmp_path)) + root = uno_profile.profile_dir("openoffice") + uno_profile.ensure_profile_seeded(root) + xcu = root / "user" / "registrymodifications.xcu" + assert xcu.is_file() + assert "FirstStartWizardCompleted" in xcu.read_text(encoding="utf-8") + mtime = xcu.stat().st_mtime + uno_profile.ensure_profile_seeded(root) + assert xcu.stat().st_mtime == mtime + + +def test_as_unicode_utf8_bytes(): + raw = "Практическая".encode("utf-8") + text = as_unicode(raw) + assert "Практ" in text + + +def test_uno_call_method_or_attribute(): + class _DocMethod: + def getText(self): + return "method" + + class _DocAttr: + Text = "attr" + + assert uno_call(_DocMethod(), "Text") == "method" + assert uno_call(_DocAttr(), "Text") == "attr" + + +def test_uno_worker_env_includes_program_dir(): + install = OfficeInstall( + engine="openoffice", + label="OpenOffice", + soffice_path=r"C:\OpenOffice 4\program\soffice.exe", + python_path=r"C:\OpenOffice 4\program\python.exe", + ) + worker_dir = Path(r"C:\temp\worker") + with patch.object(uno_fix, "_office_program_dir", return_value=Path(r"C:\OpenOffice 4\program")): + env, cwd = uno_fix._uno_worker_env(worker_dir, install) + assert r"C:\OpenOffice 4\program" in env["PYTHONPATH"] + assert str(worker_dir) in env["PYTHONPATH"] + assert cwd == Path(r"C:\OpenOffice 4\program") + + +def test_prepare_worker_dir_dev(): + worker_dir = uno_fix._prepare_worker_dir() + assert (worker_dir / "uno_worker.py").is_file() + assert (worker_dir / "uno_helpers.py").is_file() + + +def test_prepare_worker_dir_frozen_extracts_to_temp(tmp_path, monkeypatch): + bundled = tmp_path / "md2gost" / "office" + bundled.mkdir(parents=True) + (bundled / "uno_worker.py").write_text("# worker\n", encoding="utf-8") + (bundled / "uno_helpers.py").write_text("# helpers\n", encoding="utf-8") + + monkeypatch.setattr(sys, "frozen", True, raising=False) + monkeypatch.setattr(sys, "_MEIPASS", str(tmp_path), raising=False) + + worker_dir = uno_fix._prepare_worker_dir() + assert worker_dir != bundled + assert (worker_dir / "uno_worker.py").read_text(encoding="utf-8") == "# worker\n" + assert (worker_dir / "uno_helpers.py").is_file() + + +@pytest.mark.skipif(os.environ.get("MD2GOST_TEST_LIBRE") != "1", reason="set MD2GOST_TEST_LIBRE=1") +def test_libre_uno_smoke(tmp_path): + install = detect_office_install("libre") + if install is None: + pytest.skip("LibreOffice not installed") + + from docx import Document + from docx.shared import Pt + + path = tmp_path / "tall_table.docx" + doc = Document() + doc.add_paragraph("Таблица 1 — Длинная") + table = doc.add_table(rows=1, cols=2) + table.rows[0].cells[0].text = "A" + table.rows[0].cells[1].text = "B" + for i in range(80): + row = table.add_row() + row.cells[0].text = f"row {i}" + row.cells[1].text = "x" * 20 + for cell in row.cells: + for para in cell.paragraphs: + para.paragraph_format.space_after = Pt(6) + doc.save(str(path)) + + result = fix_continuations_with_engine( + str(path), + engine="libre", + tables=True, + listings=False, + ) + assert result.ok, result.message + assert path.is_file()