335 lines
8.8 KiB
Python
335 lines
8.8 KiB
Python
# -*- 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
|
|
# Stop if pagination never settles (each pass is one split, budget can grow).
|
|
PASS_SAFETY_CAP = 200
|
|
|
|
|
|
def continuation_pass_limit(object_count, splits_done=0):
|
|
"""10 + (tables + listings) + 1 per split already made.
|
|
|
|
One pass performs one split.
|
|
"""
|
|
try:
|
|
count = int(object_count)
|
|
except (TypeError, ValueError):
|
|
count = 0
|
|
try:
|
|
done = int(splits_done)
|
|
except (TypeError, ValueError):
|
|
done = 0
|
|
if count < 0:
|
|
count = 0
|
|
if done < 0:
|
|
done = 0
|
|
return 10 + count + done
|
|
|
|
# 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<kind>Таблица|Таблицы|Листинг|Листинга)\s+"
|
|
r"(?P<number>[\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 resume_page_before_cut(cut_page):
|
|
"""Page to scan from after a split: the page before the cut, not page 1.
|
|
|
|
Rows that move to the continuation start on ``cut_page``. Only that page
|
|
and the pages after it can shift, so the next pass starts at ``cut_page - 1``.
|
|
"""
|
|
try:
|
|
page = int(cut_page)
|
|
except (TypeError, ValueError):
|
|
return 1
|
|
if page <= 2:
|
|
return 1
|
|
return page - 1
|
|
|
|
|
|
def locate_break_row(n_rows, page_at, min_page=1):
|
|
"""Return ``(break_row_1based, cut_page)`` or ``(None, None)``.
|
|
|
|
``page_at(row_1based)`` returns a page number or None. Pages are treated as
|
|
non-decreasing. The search skips a table that ends before ``min_page`` and
|
|
binary-searches the first later jump, instead of reading every row from
|
|
the start. A failed probe falls back to a forward scan.
|
|
"""
|
|
try:
|
|
n_rows = int(n_rows)
|
|
except (TypeError, ValueError):
|
|
return None, None
|
|
try:
|
|
min_page = int(min_page)
|
|
except (TypeError, ValueError):
|
|
min_page = 1
|
|
if n_rows < 2:
|
|
return None, None
|
|
if min_page < 1:
|
|
min_page = 1
|
|
|
|
cache = {}
|
|
|
|
def probe(row):
|
|
if row in cache:
|
|
return cache[row]
|
|
try:
|
|
page = page_at(row)
|
|
except Exception:
|
|
page = None
|
|
if page is not None:
|
|
try:
|
|
page = int(page)
|
|
except (TypeError, ValueError):
|
|
page = None
|
|
cache[row] = page
|
|
return page
|
|
|
|
first = probe(1)
|
|
last = probe(n_rows)
|
|
if first is None or last is None:
|
|
return _linear_break_row(n_rows, probe, min_page)
|
|
if last < min_page or first == last:
|
|
return None, None
|
|
threshold = (first + 1) if first >= min_page else min_page
|
|
row = _first_row_at_or_after(n_rows, probe, threshold)
|
|
if row is None or row <= 1:
|
|
return _linear_break_row(n_rows, probe, min_page)
|
|
cut = probe(row)
|
|
prev = probe(row - 1)
|
|
if cut is None or prev is None or prev >= cut or cut < threshold:
|
|
return _linear_break_row(n_rows, probe, min_page)
|
|
return row, cut
|
|
|
|
|
|
def _first_row_at_or_after(n_rows, probe, threshold):
|
|
lo = 1
|
|
hi = n_rows
|
|
found = None
|
|
while lo <= hi:
|
|
mid = (lo + hi) // 2
|
|
page = probe(mid)
|
|
if page is None:
|
|
return None
|
|
if page >= threshold:
|
|
found = mid
|
|
hi = mid - 1
|
|
else:
|
|
lo = mid + 1
|
|
return found
|
|
|
|
|
|
def _linear_break_row(n_rows, probe, min_page):
|
|
prev = probe(1)
|
|
if prev is None:
|
|
return None, None
|
|
row = 2
|
|
while row <= n_rows:
|
|
page = probe(row)
|
|
if page is None:
|
|
page = prev
|
|
if page > prev and page >= min_page:
|
|
return row, page
|
|
prev = page
|
|
row += 1
|
|
return None, 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"])
|