Igor20264
1281 lines
40 KiB
Python
1281 lines
40 KiB
Python
#!/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())
|