v0.6.0 BigUpdate
Example document / build (push) Has been cancelled
Python application / build (push) Has been cancelled

This commit is contained in:
Igor20264
2026-10-20 11:38:06 +03:00
parent daa84f8bce
commit 8223e0ac14
77 changed files with 12158 additions and 117 deletions
+21 -6
View File
@@ -454,7 +454,7 @@ _TABLE_BLOCK_RE = re.compile(
r"(?:^[ \t]*\|.+\|[ \t]*\n?)*",
re.M,
)
_MERGE_ONLY_RE = re.compile(r"^\s*(\^{1,2}|>{1,2})\s*$")
_MERGE_ONLY_RE = re.compile(r"^\s*(\^{1,2}|>{1,2}|<{1,2})\s*$")
def _split_table_row(line: str) -> list[str]:
@@ -467,7 +467,11 @@ def _split_table_row(line: str) -> list[str]:
def check_table_merge(text: str) -> list[Issue]:
"""Validate ^ / > merge markers in pipe tables."""
"""Validate ^ / > / < merge markers in pipe tables.
`<` continues the cell on the left, same as `>`, and may sit in the header
so a title can span the row. `^` and `>` stay forbidden in the header.
"""
issues: list[Issue] = []
for tm in _TABLE_BLOCK_RE.finditer(text):
block = tm.group(0)
@@ -481,8 +485,18 @@ def check_table_merge(text: str) -> list[Issue]:
n_cols = len(header)
header_line = _line_of(text, block_start)
for cell in header:
if _MERGE_ONLY_RE.match(cell):
for c_i, cell in enumerate(header):
m = _MERGE_ONLY_RE.match(cell)
if not m:
continue
if m.group(1)[:1] in (">", "<") and c_i == 0:
marker = m.group(1)[:1]
issues.append(Issue(
"table.merge_col", "error",
f"Маркер «{marker}» нельзя ставить в первом столбце (нет ячейки слева)",
header_line,
))
elif not m.group(1).startswith("<"):
issues.append(Issue(
"table.merge_header", "error",
"Маркер склеивания ^/> нельзя использовать в первой (заголовочной) строке таблицы",
@@ -497,10 +511,11 @@ def check_table_merge(text: str) -> list[Issue]:
m = _MERGE_ONLY_RE.match(cell)
if not m:
continue
if m.group(1).startswith(">") and c_i == 0:
if m.group(1)[:1] in (">", "<") and c_i == 0:
marker = m.group(1)[:1]
issues.append(Issue(
"table.merge_col", "error",
"Маркер «>» нельзя ставить в первом столбце (нет ячейки слева)",
f"Маркер «{marker}» нельзя ставить в первом столбце (нет ячейки слева)",
line_no,
))
+961
View File
@@ -0,0 +1,961 @@
"""DFD в нотации Гейна–Сарсона (лист Ramus).
Оградка ```dfd-gs (алиасы ```gane-sarson, ```uml-dfd-gs, ```uml-gane-sarson).
Локальный Pillow/SVG, без Graphviz. Йордон остаётся в ```dfd.
Раскладка: сущности слева, процессы по центру, хранилища справа.
Стрелки между соседними колонками идут по зазору. Сквозной поток через
промежуточную колонку может пересечь блок — обхода препятствий нет.
"""
from __future__ import annotations
import logging
import math
import xml.sax.saxutils
from dataclasses import dataclass, field
from io import BytesIO
from pathlib import Path
from PIL import Image, ImageDraw
from .idef0 import _load_font, _text_wh, _wrap
_log = logging.getLogger(__name__)
CACHE_KEY_PREFIX = "dfd-gs-v1\n"
_META = {
"title": "title",
"node": "node",
"number": "number",
"author": "author",
"project": "project",
"date": "date",
"rev": "rev",
"revision": "rev",
"status": "status",
"used_at": "used_at",
"used at": "used_at",
"used-at": "used_at",
"form": "form",
}
_META_ALIASES = sorted(_META, key=len, reverse=True)
_KINDS = ("entity", "process", "store")
_KIND_W = {"entity": 200.0, "process": 248.0, "store": 256.0}
_KIND_MIN_H = {"entity": 78.0, "process": 96.0, "store": 70.0}
class DfdGsParseError(ValueError):
"""Неверный DSL DFD Гейна–Сарсона."""
@dataclass
class GsNode:
kind: str
id: str
label: str
badge: str = ""
@dataclass
class GsFlow:
src: str
dst: str
label: str = ""
@dataclass
class GsDiagram:
title: str = ""
node: str = ""
number: str = ""
author: str = ""
project: str = ""
date: str = ""
rev: str = ""
status: str = ""
used_at: str = ""
form: str = "kit"
nodes: list[GsNode] = field(default_factory=list)
flows: list[GsFlow] = field(default_factory=list)
@dataclass
class _Placed:
node: GsNode
col: int
x: float
y: float
w: float
h: float
@property
def r(self) -> float:
return self.x + self.w
@property
def b(self) -> float:
return self.y + self.h
@property
def cx(self) -> float:
return self.x + self.w / 2
@property
def cy(self) -> float:
return self.y + self.h / 2
@dataclass
class _Route:
flow: GsFlow
pts: list[tuple[float, float]]
@dataclass
class _Layout:
width: int
height: int
frame: tuple[float, float, float, float] | None
placed: dict[str, _Placed]
routes: list[_Route]
def _is_proc_num(token: str) -> bool:
if not token or token[0] not in "0123456789":
return False
parts = token.split(".")
return all(part.isdigit() and part != "" for part in parts)
def _is_store_code(token: str) -> bool:
if len(token) < 2 or token[0] not in "Dd":
return False
return token[1:].isdigit()
def _meta_of(line: str) -> tuple[str, str] | None:
low = line.lower()
for alias in _META_ALIASES:
if low.startswith(alias) and (len(line) == len(alias) or line[len(alias)].isspace()):
return _META[alias], line[len(alias):].strip()
return None
def _assign_badges(nodes: list[GsNode]) -> None:
processes = [n for n in nodes if n.kind == "process"]
for index, node in enumerate(processes, 1):
node.badge = node.id if _is_proc_num(node.id) else str(index)
stores = [n for n in nodes if n.kind == "store"]
used = {n.id.upper() for n in stores if _is_store_code(n.id)}
seq = 1
for node in stores:
if _is_store_code(node.id):
node.badge = node.id.upper()
continue
while f"D{seq}" in used:
seq += 1
node.badge = f"D{seq}"
used.add(node.badge)
seq += 1
entities = [n for n in nodes if n.kind == "entity"]
for index, node in enumerate(entities, 1):
node.badge = str(index)
def parse_dfd_gs(source: str) -> GsDiagram:
"""Разобрать DSL. Пустой текст и неизвестные узлы — DfdGsParseError."""
text = (source or "").lstrip("\ufeff")
if not text.strip():
raise DfdGsParseError("пустой DFD")
diagram = GsDiagram()
seen: dict[str, int] = {}
for lineno, raw in enumerate(text.splitlines(), 1):
line = raw.strip()
if not line or line.startswith("#") or line.startswith("//"):
continue
kind = next((k for k in _KINDS if line.lower().startswith(k) and (
len(line) == len(k) or line[len(k)].isspace()
)), "")
if kind:
rest = line[len(kind):].strip()
parts = rest.split(None, 1)
if len(parts) < 2 or not parts[1].strip():
raise DfdGsParseError(f"строка {lineno}: у {kind} нет подписи")
node_id, label = parts[0], parts[1].strip()
if node_id in seen:
raise DfdGsParseError(
f"строка {lineno}: повтор идентификатора «{node_id}»"
)
seen[node_id] = lineno
diagram.nodes.append(GsNode(kind, node_id, label))
continue
if "-->" in line:
left, _, right = line.partition("-->")
src = left.strip()
tail = right.strip()
if not src or not tail:
raise DfdGsParseError(f"строка {lineno}: неполный поток")
dst, _, label = tail.partition(" ")
label = label.strip()
if label.startswith(":"):
label = label[1:].strip()
if not dst:
raise DfdGsParseError(f"строка {lineno}: неполный поток")
diagram.flows.append(GsFlow(src, dst, label))
continue
meta = _meta_of(line)
if meta:
key, value = meta
if key == "form":
form = value.lower()
if form not in {"kit", "plain"}:
raise DfdGsParseError(
f"строка {lineno}: form «{value}» — ожидается kit или plain"
)
diagram.form = form
else:
setattr(diagram, key, value)
continue
raise DfdGsParseError(f"строка {lineno}: неизвестная запись «{line}»")
if not diagram.nodes:
raise DfdGsParseError("DFD без узлов")
for flow in diagram.flows:
if flow.src == flow.dst:
raise DfdGsParseError(f"поток из «{flow.src}» в себя")
for end in (flow.src, flow.dst):
if end not in seen:
raise DfdGsParseError(f"неизвестный узел «{end}»")
_assign_badges(diagram.nodes)
return diagram
# --- scene -----------------------------------------------------------------
def _svg_anchor(anchor: str) -> tuple[str, str, str]:
a = (anchor or "lt").lower()
horiz = a[0] if a else "l"
vert = a[1] if len(a) > 1 else "t"
ta = {"l": "start", "m": "middle", "r": "end"}.get(horiz, "start")
dy = {"t": "0.9em", "m": "0.35em", "b": "0"}.get(vert, "0.9em")
return ta, "0", dy
class _Scene:
def __init__(self, width: int, height: int):
self.width = width
self.height = height
self.ops: list[tuple] = []
def line(self, x1, y1, x2, y2, width: float = 1.4) -> None:
self.ops.append(("line", float(x1), float(y1), float(x2), float(y2), float(width)))
def polyline(self, pts: list[tuple[float, float]], width: float = 1.4) -> None:
if len(pts) < 2:
return
self.ops.append(("polyline", [(float(x), float(y)) for x, y in pts], float(width)))
def rect(
self,
x, y, w, h,
width: float = 1.4,
fill: str | None = None,
outline: str | None = "#000000",
) -> None:
self.ops.append(
("rect", float(x), float(y), float(w), float(h), float(width), fill, outline)
)
def roundrect(
self, x, y, w, h, radius: float,
width: float = 1.4,
fill: str | None = "#ffffff",
outline: str | None = "#000000",
) -> None:
self.ops.append((
"roundrect", float(x), float(y), float(w), float(h), float(radius),
float(width), fill, outline,
))
def polygon(self, pts: list[tuple[float, float]], fill: str = "#000000") -> None:
if len(pts) < 3:
return
self.ops.append(("polygon", [(float(x), float(y)) for x, y in pts], fill))
def text(
self, x, y, text: str, size: float, *,
bold: bool = False, anchor: str = "lt", fill: str = "#000000",
) -> None:
if not text:
return
self.ops.append(("text", float(x), float(y), text, float(size), bold, anchor, fill))
def text_rot(
self, x, y, text: str, size: float, *,
bold: bool = False, angle: float = 90,
) -> None:
if not text:
return
self.ops.append(("text_rot", float(x), float(y), text, float(size), bold, float(angle)))
def to_png(self) -> bytes:
img = Image.new("RGB", (self.width, self.height), "#ffffff")
draw = ImageDraw.Draw(img)
for op in self.ops:
kind = op[0]
if kind == "line":
_, x1, y1, x2, y2, w = op
draw.line([(x1, y1), (x2, y2)], fill="#000000", width=max(1, int(round(w))))
elif kind == "polyline":
_, pts, w = op
draw.line(pts, fill="#000000", width=max(1, int(round(w))))
elif kind == "rect":
_, x, y, w, h, lw, fill, outline = op
stroke = max(1, int(round(lw))) if outline else 0
draw.rectangle([x, y, x + w, y + h], fill=fill, outline=outline, width=stroke)
elif kind == "roundrect":
_, x, y, w, h, radius, lw, fill, outline = op
stroke = max(1, int(round(lw))) if outline else 0
rad = max(0, min(radius, w / 2 - 1, h / 2 - 1))
draw.rounded_rectangle(
[x, y, x + w, y + h], radius=rad, fill=fill, outline=outline, width=stroke,
)
elif kind == "polygon":
_, pts, fill = op
draw.polygon(pts, fill=fill)
elif kind == "text":
_, x, y, text, size, bold, anchor, fill = op
font = _load_font(int(round(size)), bold)
draw.text((x, y), text, font=font, fill=fill, anchor=anchor)
elif kind == "text_rot":
_, x, y, text, size, bold, angle = op
_paste_rotated(img, text, x, y, int(round(size)), bold, angle)
buf = BytesIO()
img.save(buf, format="PNG", optimize=True)
return buf.getvalue()
def to_svg(self) -> str:
esc = xml.sax.saxutils.escape
parts = [
f'<svg xmlns="http://www.w3.org/2000/svg" width="{self.width}" height="{self.height}" '
f'viewBox="0 0 {self.width} {self.height}">',
'<rect width="100%" height="100%" fill="#fff"/>',
]
for op in self.ops:
kind = op[0]
if kind == "line":
_, x1, y1, x2, y2, w = op
parts.append(
f'<line x1="{x1:.1f}" y1="{y1:.1f}" x2="{x2:.1f}" y2="{y2:.1f}" '
f'stroke="#000" stroke-width="{w:.2f}"/>'
)
elif kind == "polyline":
_, pts, w = op
d = " ".join(f"{x:.1f},{y:.1f}" for x, y in pts)
parts.append(
f'<polyline points="{d}" fill="none" stroke="#000" '
f'stroke-width="{w:.2f}" stroke-linejoin="miter"/>'
)
elif kind == "rect":
_, x, y, w, h, lw, fill, outline = op
parts.append(
f'<rect x="{x:.1f}" y="{y:.1f}" width="{w:.1f}" height="{h:.1f}" '
f'fill="{fill or "none"}" stroke="{outline or "none"}" '
f'stroke-width="{0 if outline is None else lw:.2f}"/>'
)
elif kind == "roundrect":
_, x, y, w, h, radius, lw, fill, outline = op
parts.append(
f'<rect x="{x:.1f}" y="{y:.1f}" width="{w:.1f}" height="{h:.1f}" '
f'rx="{radius:.1f}" ry="{radius:.1f}" fill="{fill or "none"}" '
f'stroke="{outline or "none"}" '
f'stroke-width="{0 if outline is None else lw:.2f}"/>'
)
elif kind == "polygon":
_, pts, fill = op
d = " ".join(f"{x:.1f},{y:.1f}" for x, y in pts)
parts.append(f'<polygon points="{d}" fill="{fill}"/>')
elif kind == "text":
_, x, y, text, size, bold, anchor, fill = op
ta, dx, dy = _svg_anchor(anchor)
weight = "700" if bold else "400"
parts.append(
f'<text x="{x:.1f}" y="{y:.1f}" dx="{dx}" dy="{dy}" text-anchor="{ta}" '
f'font-family="Arial, DejaVu Sans, sans-serif" font-size="{size:.1f}" '
f'font-weight="{weight}" fill="{fill}">{esc(text)}</text>'
)
elif kind == "text_rot":
_, x, y, text, size, bold, angle = op
weight = "700" if bold else "400"
parts.append(
f'<text x="{x:.1f}" y="{y:.1f}" text-anchor="middle" '
f'dominant-baseline="middle" '
f'transform="rotate({-angle:.0f} {x:.1f} {y:.1f})" '
f'font-family="Arial, DejaVu Sans, sans-serif" font-size="{size:.1f}" '
f'font-weight="{weight}" fill="#000">{esc(text)}</text>'
)
parts.append("</svg>")
return "\n".join(parts) + "\n"
def _paste_rotated(
base: Image.Image, text: str, cx: float, cy: float,
size: int, bold: bool, angle: float,
) -> None:
font = _load_font(size, bold)
left, top, right, bottom = font.getbbox(text)
pad = 2
patch = Image.new("RGBA", (right - left + pad * 2, bottom - top + pad * 2), (0, 0, 0, 0))
ImageDraw.Draw(patch).text((pad - left, pad - top), text, font=font, fill=(0, 0, 0, 255))
rotated = patch.rotate(angle, expand=True, resample=Image.Resampling.BICUBIC)
base.paste(
rotated,
(int(round(cx - rotated.width / 2)), int(round(cy - rotated.height / 2))),
rotated,
)
# --- layout ----------------------------------------------------------------
def _node_size(node: GsNode, scale: float) -> tuple[float, float]:
s = scale
width = _KIND_W[node.kind] * s
font = _load_font(int(round(15 * s)), False)
if node.kind == "store":
inner = width - 44 * s
else:
inner = width - 28 * s
lines = _wrap(node.label, font, max(40, int(inner)))
line_h = 18 * s
text_h = max(1, len(lines)) * line_h
extra = 30 * s if node.kind == "process" else 22 * s
height = max(_KIND_MIN_H[node.kind] * s, text_h + extra)
return width, height
def _simplify(pts: list[tuple[float, float]]) -> list[tuple[float, float]]:
out: list[tuple[float, float]] = []
for point in pts:
if out and abs(point[0] - out[-1][0]) < 0.4 and abs(point[1] - out[-1][1]) < 0.4:
continue
if len(out) >= 2:
a, b = out[-2], out[-1]
horiz = abs(a[1] - b[1]) < 0.4 and abs(b[1] - point[1]) < 0.4
vert = abs(a[0] - b[0]) < 0.4 and abs(b[0] - point[0]) < 0.4
if horiz or vert:
out[-1] = point
continue
out.append(point)
return out
def _slot_xs(left: float, right: float, n: int) -> list[float]:
if n <= 0:
return []
if right < left:
left, right = right, left
span = right - left
margin = min(6.0, span / 4) if span else 0.0
a, b = left + margin, right - margin
if b <= a or n == 1:
return [(left + right) / 2] * n
return [a + (b - a) * (k + 1) / (n + 1) for k in range(n)]
def _arrange(diagram: GsDiagram, scale: float) -> _Layout:
s = max(float(scale), 1.0)
groups: list[list[GsNode]] = []
for kind in _KINDS:
chunk = [n for n in diagram.nodes if n.kind == kind]
if chunk:
groups.append(chunk)
sizes = {n.id: _node_size(n, s) for n in diagram.nodes}
# Vertical gap fits a one-line label between stacked boxes.
gap_y = 50 * s
# Horizontal gap fits a wrapped flow label without covering the boxes.
gap_x = 132 * s
col_w = [max(sizes[n.id][0] for n in group) for group in groups]
col_h = []
for group in groups:
heights = [sizes[n.id][1] for n in group]
col_h.append(sum(heights) + gap_y * (len(group) - 1))
col_of = {n.id: i for i, group in enumerate(groups) for n in group}
index_in_col = {
node.id: (col_i, pos)
for col_i, group in enumerate(groups)
for pos, node in enumerate(group)
}
def _neighbors(flow: GsFlow) -> bool:
a, b = index_in_col[flow.src], index_in_col[flow.dst]
return a[0] == b[0] and abs(a[1] - b[1]) == 1
last = len(groups) - 1
needs_channel = any(
col_of[flow.src] == col_of[flow.dst] == last and not _neighbors(flow)
for flow in diagram.flows
)
extra_r = 56 * s if needs_channel else 0.0
content_w = sum(col_w) + gap_x * (len(groups) - 1) + extra_r
content_h = max(col_h) if col_h else 0.0
inner_pad = 24 * s
outer = 8 * s
kit = diagram.form != "plain"
header_h = 76 * s if kit else 0.0
footer_h = 56 * s if kit else 0.0
need_w = content_w + 2 * inner_pad
need_h = content_h + 2 * inner_pad
# Stamp needs a readable width; height follows the columns, not a tall empty sheet.
width = max((760 if kit else 480) * s, need_w + 2 * outer)
height = need_h + header_h + footer_h + 2 * outer
width_i, height_i = int(round(width)), int(round(height))
frame = (outer, outer, width - 2 * outer, height - 2 * outer)
if kit:
work_x, work_y = frame[0], frame[1] + header_h
work_w, work_h = frame[2], frame[3] - header_h - footer_h
else:
work_x, work_y, work_w, work_h = frame
block_w = content_w
ox = work_x + (work_w - block_w) / 2
oy = work_y + max(inner_pad, (work_h - content_h) / 2)
placed: dict[str, _Placed] = {}
col_box: list[tuple[float, float]] = []
x = ox
for index, group in enumerate(groups):
col_box.append((x, col_w[index]))
y = oy + (content_h - col_h[index]) / 2
for node in group:
nh = sizes[node.id][1]
placed[node.id] = _Placed(node, index, x, y, col_w[index], nh)
y += nh + gap_y
x += col_w[index] + gap_x
gaps = [
(col_box[i][0] + col_box[i][1], col_box[i + 1][0])
for i in range(len(col_box) - 1)
]
# Side each flow touches. Ports on one side are spread together, so a
# pair of opposite arrows does not share the same point.
# Neighbours stacked in one column are linked through the gap between them.
touch: list[tuple[str, str]] = []
side_hits: dict[tuple[str, str], list[tuple[int, str]]] = {}
neighbor_flows: list[int] = []
for i, flow in enumerate(diagram.flows):
if _neighbors(flow):
touch.append(("v", "v"))
neighbor_flows.append(i)
continue
sc, dc = col_of[flow.src], col_of[flow.dst]
if sc < dc:
sides = ("right", "left")
elif sc > dc:
sides = ("left", "right")
else:
sides = ("right", "right")
touch.append(sides)
side_hits.setdefault((flow.src, sides[0]), []).append((i, "src"))
side_hits.setdefault((flow.dst, sides[1]), []).append((i, "dst"))
neighbor_dx: dict[int, float] = {}
neighbor_groups: dict[tuple[int, int, int], list[int]] = {}
for i in neighbor_flows:
flow = diagram.flows[i]
col_i, pos_a = index_in_col[flow.src]
_col_b, pos_b = index_in_col[flow.dst]
key = (col_i, min(pos_a, pos_b), max(pos_a, pos_b))
neighbor_groups.setdefault(key, []).append(i)
for idxs in neighbor_groups.values():
count = len(idxs)
for k, fi in enumerate(idxs):
neighbor_dx[fi] = (k - (count - 1) / 2) * 18 * s
port_y: dict[tuple[int, str], float] = {}
for (nid, _side), hits in side_hits.items():
box = placed[nid]
ordered = sorted(
hits,
key=lambda hit: (
placed[
diagram.flows[hit[0]].dst
if hit[1] == "src"
else diagram.flows[hit[0]].src
].cy,
hit[0],
),
)
n = len(ordered)
for k, (fi, role) in enumerate(ordered):
y = box.cy if n == 1 else box.y + box.h * (k + 1) / (n + 1)
port_y[(fi, role)] = y
thresh = 10 * s
elbow_gap: dict[tuple, list[int]] = {}
for i, flow in enumerate(diagram.flows):
if i in neighbor_dx:
continue
sc, dc = col_of[flow.src], col_of[flow.dst]
y1, y2 = port_y[(i, "src")], port_y[(i, "dst")]
if sc != dc and abs(y1 - y2) <= thresh:
continue
if sc == dc:
elbow_gap.setdefault(("same", sc), []).append(i)
else:
gap_i = (dc - 1) if sc < dc else (sc - 1)
elbow_gap.setdefault(("gap", gap_i), []).append(i)
gutter_x: dict[int, float] = {}
for key, idxs in elbow_gap.items():
ordered = sorted(idxs, key=lambda fi: (port_y[(fi, "src")], fi))
kind, index = key
if kind == "gap":
left, right = gaps[index]
else:
left = col_box[index][0] + col_box[index][1] + 8 * s
right = left + (40 * s if index == last else gap_x - 16 * s)
for fi, gx in zip(ordered, _slot_xs(left, right, len(ordered))):
gutter_x[fi] = gx
routes: list[_Route] = []
for i, flow in enumerate(diagram.flows):
src, dst = placed[flow.src], placed[flow.dst]
if i in neighbor_dx:
x = src.cx + neighbor_dx[i]
if src.y < dst.y:
pts = [(x, src.b), (x, dst.y)]
else:
pts = [(x, src.y), (x, dst.b)]
routes.append(_Route(flow, pts))
continue
src_side, dst_side = touch[i]
x1 = src.r if src_side == "right" else src.x
x2 = dst.r if dst_side == "right" else dst.x
y1, y2 = port_y[(i, "src")], port_y[(i, "dst")]
if src.col != dst.col and abs(y1 - y2) <= thresh:
pts = [(x1, y1), (x2, y2)]
else:
gx = gutter_x[i]
pts = _simplify([(x1, y1), (gx, y1), (gx, y2), (x2, y2)])
routes.append(_Route(flow, pts))
return _Layout(
width_i,
height_i,
frame if kit else None,
placed,
routes,
)
# --- draw ------------------------------------------------------------------
def _draw_centered(
scene: _Scene, text: str, x: float, y: float, w: float, h: float,
size: float, max_width: float,
) -> None:
font = _load_font(int(round(size)), False)
lines = _wrap(text, font, max(20, int(max_width)))
line_h = size * 1.2
block = line_h * len(lines)
top = y + (h - block) / 2
for i, line in enumerate(lines):
scene.text(x + w / 2, top + i * line_h + line_h / 2, line, size, anchor="mm")
def _draw_node(scene: _Scene, box: _Placed, scale: float, lw: float) -> None:
s = scale
node = box.node
if node.kind == "entity":
sh = 5 * s
scene.rect(box.x + sh, box.y + sh, box.w, box.h, width=0, fill="#c4c4c4", outline=None)
scene.rect(box.x, box.y, box.w, box.h, width=lw, fill="#ffffff")
scene.text(box.x + 8 * s, box.y + 6 * s, node.badge, 13 * s, bold=True, anchor="lt")
_draw_centered(
scene, node.label, box.x, box.y + 16 * s, box.w, box.h - 22 * s,
15 * s, box.w - 28 * s,
)
elif node.kind == "process":
radius = min(18 * s, box.h * 0.2, box.w * 0.12)
scene.roundrect(box.x, box.y, box.w, box.h, radius, width=lw, fill="#ffffff")
_draw_centered(
scene, node.label, box.x, box.y + 4 * s, box.w, box.h - 24 * s,
15 * s, box.w - 28 * s,
)
scene.text(box.r - 8 * s, box.b - 6 * s, node.badge, 14 * s, bold=True, anchor="rb")
else:
scene.rect(box.x, box.y, box.w, box.h, width=lw, fill="#ffffff")
strip = 28 * s
scene.line(box.x + strip, box.y, box.x + strip, box.b, lw)
scene.text_rot(box.x + strip / 2, box.cy, node.badge, 13 * s, bold=True, angle=90)
_draw_centered(
scene, node.label, box.x + strip, box.y, box.w - strip, box.h,
15 * s, box.w - strip - 16 * s,
)
def _draw_arrow(scene: _Scene, pts: list[tuple[float, float]], lw: float, scale: float) -> None:
if len(pts) < 2:
return
p1, p2 = pts[-2], pts[-1]
dx, dy = p2[0] - p1[0], p2[1] - p1[1]
length = math.hypot(dx, dy) or 1.0
size = min(12 * scale, length * 0.55)
ux, uy = dx / length, dy / length
base = (p2[0] - ux * size, p2[1] - uy * size)
shaft = list(pts[:-1]) + [base]
scene.polyline(shaft, lw)
wing = size * 0.42
px, py = -uy * wing, ux * wing
scene.polygon([p2, (base[0] + px, base[1] + py), (base[0] - px, base[1] - py)])
def _point_along(pts: list[tuple[float, float]], frac: float) -> tuple[float, float, float, float]:
segs = []
total = 0.0
for a, b in zip(pts, pts[1:]):
length = math.hypot(b[0] - a[0], b[1] - a[1])
segs.append((a, b, length))
total += length
if total <= 0:
return pts[0][0], pts[0][1], 0.0, -1.0
target = total * min(1.0, max(0.0, frac))
acc = 0.0
a, b = segs[-1][0], segs[-1][1]
for a, b, length in segs:
if acc + length >= target or length == 0:
t = 0 if length == 0 else (target - acc) / length
x = a[0] + (b[0] - a[0]) * t
y = a[1] + (b[1] - a[1]) * t
dx, dy = b[0] - a[0], b[1] - a[1]
norm = math.hypot(dx, dy) or 1.0
px, py = -dy / norm, dx / norm
if py > 0 or (abs(py) < 1e-6 and px < 0):
px, py = -px, -py
return x, y, px, py
acc += length
return a[0], a[1], 0.0, -1.0
def _label_overlap(
cx: float, cy: float, bw: float, bh: float,
boxes: list[_Placed],
busy: list[tuple[float, float, float, float]],
pad: float,
) -> float:
ax, ay = cx - bw / 2 - pad, cy - bh / 2 - pad
ar, ab = ax + bw + 2 * pad, ay + bh + 2 * pad
total = 0.0
for box in boxes:
ix = max(0.0, min(ar, box.r) - max(ax, box.x))
iy = max(0.0, min(ab, box.b) - max(ay, box.y))
total += ix * iy * 4
for x, y, w, h in busy:
ix = max(0.0, min(ar, x + w) - max(ax, x))
iy = max(0.0, min(ab, y + h) - max(ay, y))
total += ix * iy
return total
def _clamp_label(
cx: float, cy: float, bw: float, bh: float,
canvas_w: float, canvas_h: float, pad: float,
) -> tuple[float, float]:
half_w, half_h = bw / 2 + pad, bh / 2 + pad
cx = min(max(cx, half_w), max(half_w, canvas_w - half_w))
cy = min(max(cy, half_h), max(half_h, canvas_h - half_h))
return cx, cy
def _draw_label(
scene: _Scene,
text: str,
pts: list[tuple[float, float]],
scale: float,
boxes: list[_Placed],
canvas_w: float,
canvas_h: float,
busy: list[tuple[float, float, float, float]],
) -> None:
label = (text or "").strip()
if not label or len(pts) < 2:
return
s = scale
size = 13 * s
font = _load_font(int(round(size)), False)
# Narrow wrap: the label stays in the column gap instead of covering a box.
lines = _wrap(label, font, max(72, int(108 * s)))
line_h = size * 1.2
block_w = max(_text_wh(font, line)[0] for line in lines)
block_h = line_h * len(lines)
pad = 3 * s
best: tuple[float, float, float] | None = None
for frac in (0.5, 0.34, 0.66, 0.2, 0.8):
x, y, px, py = _point_along(pts, frac)
horizontal = abs(py) >= abs(px)
shift = (5 * s + block_h / 2) if horizontal else (8 * s)
for sign in (1.0, -1.0):
cx, cy = _clamp_label(
x + px * shift * sign, y + py * shift * sign,
block_w, block_h, canvas_w, canvas_h, pad,
)
overlap = _label_overlap(cx, cy, block_w, block_h, boxes, busy, 1.0)
rank = overlap + (0 if frac == 0.5 and sign > 0 else 0.01)
if best is None or rank < best[0]:
best = (rank, cx, cy)
if overlap <= 0 and frac == 0.5 and sign > 0:
best = (0.0, cx, cy)
break
if best is not None and best[0] <= 0:
break
if best is None:
return
cx, cy = best[1], best[2]
scene.rect(
cx - block_w / 2 - pad, cy - block_h / 2 - pad,
block_w + 2 * pad, block_h + 2 * pad,
width=0, fill="#ffffff", outline=None,
)
top = cy - block_h / 2
busy.append((cx - block_w / 2 - pad, top - pad, block_w + 2 * pad, block_h + 2 * pad))
for i, line in enumerate(lines):
scene.text(cx, top + i * line_h + line_h / 2, line, size, anchor="mm")
def _cell_text(
scene: _Scene, x: float, y: float, w: float, h: float,
caption: str, value: str, scale: float,
) -> None:
s = scale
scene.text(x + 6 * s, y + 4 * s, caption, 9 * s, anchor="lt", fill="#444444")
if not value:
return
font = _load_font(int(round(12 * s)), False)
lines = _wrap(value, font, max(20, int(w - 12 * s)))[:2]
line_h = 14 * s
for i, line in enumerate(lines):
scene.text(x + w / 2, y + 26 * s + i * line_h, line, 12 * s, anchor="mt")
def _draw_frame(scene: _Scene, diagram: GsDiagram, frame: tuple[float, float, float, float], scale: float) -> None:
s = scale
x, y, w, h = frame
lw = max(1.2, 1.3 * s)
scene.rect(x, y, w, h, width=lw, fill=None)
header_h = 76 * s
footer_h = 56 * s
scene.line(x, y + header_h, x + w, y + header_h, lw)
scene.line(x, y + h - footer_h, x + w, y + h - footer_h, lw)
fracs = (0.14, 0.18, 0.14, 0.08, 0.28, 0.18)
cells = (
("USED AT", diagram.used_at),
("AUTHOR", diagram.author),
("DATE", diagram.date),
("REV", diagram.rev),
("PROJECT", diagram.project),
("STATUS", diagram.status),
)
cursor = x
for (caption, value), frac in zip(cells, fracs):
cw = w * frac
scene.line(cursor, y, cursor, y + header_h, lw)
_cell_text(scene, cursor, y, cw, header_h, caption, value, s)
cursor += cw
num_w = 130 * s
scene.line(x + w - num_w, y + h - footer_h, x + w - num_w, y + h, lw)
scene.text(x + 8 * s, y + h - footer_h + 4 * s, "NODE / TITLE", 9 * s, anchor="lt", fill="#444444")
title = " ".join(part for part in (diagram.node, diagram.title) if part)
if title:
font = _load_font(int(round(13 * s)), True)
lines = _wrap(title, font, max(40, int(w - num_w - 16 * s)))[:2]
for i, line in enumerate(lines):
scene.text(x + 8 * s, y + h - footer_h + 24 * s + i * 16 * s, line, 13 * s, bold=True, anchor="lt")
scene.text(
x + w - num_w / 2, y + h - footer_h + 8 * s, "NUMBER", 9 * s, anchor="mt", fill="#444444",
)
if diagram.number:
scene.text(x + w - num_w / 2, y + h - footer_h / 2 + 6 * s, diagram.number, 14 * s, bold=True, anchor="mm")
def render_dfd_gs(
diagram: GsDiagram,
*,
scale: float = 2.0,
want_svg: bool = True,
) -> tuple[bytes, str | None]:
"""Вернуть (png_bytes, svg или None)."""
s = max(float(scale), 1.0)
layout = _arrange(diagram, s)
scene = _Scene(layout.width, layout.height)
lw = max(1.2, 1.35 * s)
if layout.frame is not None:
_draw_frame(scene, diagram, layout.frame, s)
for box in layout.placed.values():
_draw_node(scene, box, s, lw)
for route in layout.routes:
_draw_arrow(scene, route.pts, lw, s)
busy: list[tuple[float, float, float, float]] = []
for route in layout.routes:
_draw_label(
scene, route.flow.label, route.pts, s,
list(layout.placed.values()), layout.width, layout.height, busy,
)
svg = scene.to_svg() if want_svg else None
return scene.to_png(), svg
def write_dfd_gs_outputs(
source: str,
out_png: Path,
out_svg: Path | None,
*,
scale: float = 2.0,
) -> bool:
"""Разобрать и нарисовать. DfdGsParseError при ошибке DSL."""
diagram = parse_dfd_gs(source)
png, svg = render_dfd_gs(diagram, scale=scale, want_svg=out_svg is not None)
out_png.parent.mkdir(parents=True, exist_ok=True)
out_png.write_bytes(png)
if out_svg is not None and svg is not None:
out_svg.write_text(svg, encoding="utf-8")
_log.info("DFD Гейна–Сарсона → %s", out_png.name)
return True
SAMPLE = """\
title Зафиксировать решение приёмки
node A44
number 1
author Тарасов И.А.
project ИС ArtFreelance
date 05.10.2026
rev 01
status WORKING
used_at учебный проект
entity Artist Художник
process 1 Принять файлы результата
store D2 Файлы работ
Artist --> 1 файлы работы
1 --> D2 превью и хеш
"""
+21 -3
View File
@@ -37,12 +37,15 @@ DIAGRAM_LANGS = frozenset(
"mmd",
"idef0",
"dfd",
"dfd-gs",
"gane-sarson",
}
)
# Native Pillow renderers — never jar / Kroki.
LOCAL_ONLY_TYPES = frozenset({"idef0", "dfd"})
# Handled in this module, never as a Kroki diagram type.
# idef0 and dfd-gs are Pillow; dfd draws itself (Graphviz, Kroki only inside dfd.py).
LOCAL_ONLY_TYPES = frozenset({"idef0", "dfd", "dfd-gs"})
# Local hi-res path uses a scale-specific cache key.
SCALED_LOCAL_TYPES = frozenset({"mermaid", "idef0", "dfd"})
SCALED_LOCAL_TYPES = frozenset({"mermaid", "idef0", "dfd", "dfd-gs"})
FallbackMode = Literal["local", "remote", "off"]
DiagramFormat = Literal["png", "svg"]
@@ -267,6 +270,7 @@ def diagram_engine_status(explicit_jar: str | None = None) -> str:
except Exception as exc:
lines = lines + f"\nMermaid: статус недоступен ({exc})"
lines = lines + "\nIDEF0: локальный рендер (Pillow), оградка ```idef0"
lines = lines + "\nDFD Гейна–Сарсона: локальный рендер (Pillow), оградка ```dfd-gs"
try:
from .dfd import dfd_engine_status
@@ -618,6 +622,10 @@ def render_diagram(
from .dfd import CACHE_KEY_PREFIX as DFD_CACHE_PREFIX
cache_prefix = DFD_CACHE_PREFIX
elif kroki_type == "dfd-gs":
from .dfd_gs import CACHE_KEY_PREFIX as DFD_GS_CACHE_PREFIX
cache_prefix = DFD_GS_CACHE_PREFIX
# Local hi-res uses a scale-specific cache key; Kroki uses unscaled key.
local_prefix = cache_prefix
@@ -722,6 +730,16 @@ def render_diagram(
raise RuntimeError(str(exc)) from exc
raise RuntimeError("Не удалось отрендерить DFD-диаграмму")
if kroki_type == "dfd-gs":
from .dfd_gs import DfdGsParseError, write_dfd_gs_outputs
try:
if write_dfd_gs_outputs(prepared, out_png, svg_target, scale=scale):
return _result(pscale=scale if scale > 1 else 1.0)
except DfdGsParseError as exc:
raise RuntimeError(str(exc)) from exc
raise RuntimeError("Не удалось отрендерить DFD Гейна–Сарсона")
raise RuntimeError(f"Неизвестный локальный тип диаграммы: {kroki_type}")
# 1b) Mermaid local: system browser → Playwright Chromium → QuickJS
+4
View File
@@ -35,6 +35,10 @@ NATIVE_DIAGRAM_LANGS = {
"data-flow-diagram": "dfd",
"yourdon": "dfd",
"uml-yourdon": "dfd",
"dfd-gs": "dfd-gs",
"uml-dfd-gs": "dfd-gs",
"gane-sarson": "dfd-gs",
"uml-gane-sarson": "dfd-gs",
}
+1 -1
View File
@@ -52,7 +52,7 @@
"docs": "LAYOUT_TOP_DOWN() / LAYOUT_LEFT_RIGHT()\nMotivation_Stakeholder/Driver/Assessment/Goal/Outcome/Principle/Requirement/Constraint/Meaning/Value\nStrategy_Resource/Capability/ValueStream/CourseOfAction\nBusiness_Actor/Role/Collaboration/Interface/Process/Function/Interaction/Event/Service/Object/Contract/Representation/Product\nApplication_Component/Collaboration/Interface/Function/Interaction/Process/Event/Service/DataObject\nTechnology_Node/Device/SystemSoftware/Collaboration/Interface/Path/CommunicationNetwork/Function/Process/Interaction/Event/Service/Artifact\nPhysical_Equipment/Facility/DistributionNetwork/Material\nImplementation_WorkPackage/Deliverable/Event/Plateau/Gap\nGrouping / Other_Grouping / Group / Other_Location / Junction_And / Junction_Or\nRel_Composition/Aggregation/Assignment/Realization/Serving/Access(_r|_w|_rw)/Influence/Triggering/Flow/Specialization/Association(_dir)\nRel_Type_Up/Down/Left/Right(from, to, \"label?\")\nLay_U/D/L/R — скрытое выравнивание",
"ai-prompt": "Рисуй ArchiMate 3.2 (Open Group C226). В блоке ```uml-archimate (или ```archimate) только макросы Archimate-PlantUML. Без @startuml и без !include — схема добавит сама.\n\nПравила: один рисунок = один viewpoint. Поведение по слоям (Business_Process ≠ Application_Process ≠ Technology_Process). Цепочка: Active —Assignment→ Process/Function —Realization→ Service —Serving→ потребитель; Access к пассивному объекту. Serving от поставщика к потребителю; Realization от конкретного к абстрактному; Composition/Aggregation целое→часть. Не Association «на всё». Не UML/C4/BPMN. Alias латиница (не end/start/group). 8–18 элементов. Сначала элементы, потом Rel_*. Подписи по-русски.\n\nМакросы: Motivation_*, Strategy_*, Business_*, Application_*, Technology_*, Physical_*, Implementation_*, Grouping, Junction_And/Or. Связи: Rel_Composition/Aggregation/Assignment/Realization/Serving/Access_r|w|rw/Influence/Triggering/Flow/Specialization/Association + _Up/_Down/_Left/_Right. Макет: LAYOUT_TOP_DOWN() или LAYOUT_LEFT_RIGHT(). Полный гайд: docs/archimate.md.",
"includes": [],
"prefix": "@startuml\n!include <archimate/Archimate>\n!theme archimate-standard from <archimate/themes>\n",
"prefix": "@startuml\n!include <archimate/Archimate>\n",
"postfix": "\n@enduml"
}
}
+1 -1
View File
@@ -45,4 +45,4 @@ class Caption(BlockElement):
def parse(cls, source: Source) -> Match[str] | None:
m = source.match
source.consume()
return m
return m
+3 -2
View File
@@ -5,11 +5,12 @@ from marko.ext.gfm.elements import TableCell as GfmTableCell
MERGE_V_MARKERS = {"^", "^^"}
MERGE_H_MARKERS = {">", ">>"}
# `<` is the same colspan-continue as `>`, and is allowed in the header row.
MERGE_H_MARKERS = {">", ">>", "<", "<<"}
class TableCell(GfmTableCell):
"""GFM table cell with rowspan/colspan continue markers (^ / >)."""
"""GFM table cell with rowspan/colspan continue markers (^ / > / <)."""
def __init__(self, text: str, position: int | None = None) -> None:
stripped = text.strip()
+79 -16
View File
@@ -189,6 +189,65 @@ def _tip(widget, text: str) -> None:
_Tooltip(widget, text)
def import_output_path(docx_path: str) -> str:
"""Markdown path written by DOCX → MD for this document."""
return os.path.splitext(os.path.abspath(docx_path))[0] + ".md"
def _same_path(left: str, right: str) -> bool:
return os.path.normcase(os.path.abspath(left)) == os.path.normcase(os.path.abspath(right))
def output_tracks_source(source: str, current: str) -> bool:
"""True when the output field is still the automatic path for this source.
Markdown sources track ``name.docx``. DOCX sources (import) track ``name.md``.
``default_output_path`` always appends ``.docx``, so it must not be used to
decide whether an import path is still automatic.
"""
if not source:
return False
current = (current or "").strip()
if not current:
return True
if source.lower().endswith(".docx"):
auto = import_output_path(source)
else:
auto = default_output_path(source)
return _same_path(current, auto)
def next_markdown_output(source_md: str, current: str, auto: bool) -> tuple[str, bool]:
"""DOCX output after the user picks a markdown source.
A non-``.docx`` path is leftover from DOCX → MD and is not a chosen
conversion target, even if the auto-output flag was cleared.
"""
current = (current or "").strip()
if auto or not current.lower().endswith(".docx"):
return default_output_path(source_md), True
return current, auto
def next_import_output(docx_path: str, current: str, auto: bool) -> tuple[str, bool]:
"""Markdown output after the user picks a DOCX to import.
A ``.docx`` path is the previous MD → DOCX target and must not be reused.
"""
current = (current or "").strip()
if auto or not current.lower().endswith(".md"):
return import_output_path(docx_path), True
return current, auto
def coerce_docx_output(source_md: str, current: str) -> str:
"""Output path for an MD → DOCX run. Non-docx paths fall back to the default."""
current = (current or "").strip()
if current.lower().endswith(".docx"):
return current
return default_output_path(source_md) if source_md else current
class Md2GostApp:
def __init__(self, initial: ConvertRequest | None = None):
self.req = initial or ConvertRequest()
@@ -1693,18 +1752,21 @@ class Md2GostApp:
def _on_output_edited(self, *_args) -> None:
if not self._last_md:
return
current = self.output_var.get().strip()
auto = default_output_path(self._last_md)
self._auto_output = (not current) or os.path.normcase(os.path.abspath(current)) == os.path.normcase(auto)
self._auto_output = output_tracks_source(self._last_md, self.output_var.get())
def _apply_output(self, path: str, auto: bool) -> None:
if path != self.output_var.get().strip():
self.output_var.set(path)
self._auto_output = auto
def _set_markdown(self, path: str, from_user_output: bool = False) -> None:
path = os.path.abspath(path)
self._last_md = path
self.zone_label.configure(text=os.path.basename(path))
self.zone_path.configure(text=path)
if not from_user_output and self._auto_output:
self.output_var.set(default_output_path(path))
self._auto_output = True
if not from_user_output:
output, auto = next_markdown_output(path, self.output_var.get(), self._auto_output)
self._apply_output(output, auto)
self._append_log(f"Файл: {path}")
def _set_docx(self, path: str) -> None:
@@ -1712,10 +1774,8 @@ class Md2GostApp:
self._last_md = path
self.zone_label.configure(text=os.path.basename(path))
self.zone_path.configure(text=path)
if self._auto_output:
stem = os.path.splitext(path)[0]
self.output_var.set(stem + ".md")
self._auto_output = True
output, auto = next_import_output(path, self.output_var.get(), self._auto_output)
self._apply_output(output, auto)
self._append_log(f"DOCX для импорта: {path}")
def _import_docx(self) -> None:
@@ -1847,10 +1907,10 @@ class Md2GostApp:
return
if is_docx:
target = (req.output or "").strip()
if target.lower().endswith(".docx"):
target = os.path.splitext(req.filename)[0] + ".md"
if not target.lower().endswith(".md"):
target = import_output_path(req.filename)
req.output = target
self.output_var.set(target)
self._apply_output(target, True)
if target and os.path.isfile(target):
if not messagebox.askyesno(
"md2gost", f"Файл уже есть:\n{target}\n\nПерезаписать?"
@@ -1862,9 +1922,12 @@ class Md2GostApp:
self.output_var.set(target)
self._append_log("Импорт DOCX → Markdown…")
elif not req.check_only:
target = (req.output or "").strip() or (
default_output_path(req.filename) if req.filename else ""
)
raw = (req.output or "").strip()
target = coerce_docx_output(req.filename, raw)
if raw and not raw.lower().endswith(".docx"):
self._append_log(f"После импорта выход был не .docx, для сборки беру {target}")
req.output = target
self._apply_output(target, True)
if target and os.path.isfile(target):
if not messagebox.askyesno(
"md2gost", f"Файл уже есть:\n{target}\n\nПерезаписать?"
+19 -4
View File
@@ -87,8 +87,9 @@ CLI (тот же движок)
|---|---|
| 1 | 2 |
Ссылка: @Таблица:tbl1
Склеивание: ^ — ячейка сверху (rowspan), > — ячейка слева (colspan).
Не ставить ^/> в шапке; > — не в первом столбце.
Склеивание: ^ — ячейка сверху (rowspan), > или < — ячейка слева (colspan).
< можно в шапке, чтобы заголовок занял всю строку. ^ и > в шапке нельзя;
< и > — не в первом столбце.
Листинг
%code1 Название
@@ -140,9 +141,21 @@ CLI (тот же движок)
System --> Client статус
```
DFD Гейна–Сарсона (локальный Pillow, без Graphviz):
%dfd_gs Приёмка
```dfd-gs
title Зафиксировать решение приёмки
node A44
entity Artist Художник
process 1 Принять файлы результата
store D2 Файлы работ
Artist --> 1 файлы работы
1 --> D2 превью и хеш
```
+listing — ещё и листинг с исходником.
+landscape — отдельная альбомная страница (A4, повёрт на 90°) вокруг рисунка/таблицы, затем снова книжная.
Языки: uml, plantuml, mermaid / mmd, idef0, dfd / data-flow-diagram, или uml-<id> / <id> из файла схем (c4, usecase, archimate, …).
Языки: uml, plantuml, mermaid / mmd, idef0, dfd / data-flow-diagram, dfd-gs / gane-sarson, или uml-<id> / <id> из файла схем (c4, usecase, archimate, …).
Формат в Word: PNG по умолчанию (PlantUML, локальный Mermaid, IDEF0 и DFD ~2× для чёткости); --diagram-format svg — вектор + PNG-запасной (Word 2016+).
Подробнее — Справка → Схемы и раздел ниже в инструкции.
@@ -205,8 +218,10 @@ SCHEMES_HELP = """СХЕМЫ ДИАГРАММ (PlantUML)
```uml-archimate — ArchiMate 3.2 (макросы Business_*, Rel_Serving, …)
```mermaid / ```mmd — Mermaid локально (браузер / Chromium / QuickJS) или через Kroki
```idef0 — IDEF0 (FIPS 183): боксы, стрелки ICOM, рамка NIST; локальный Pillow
```dfd / ```data-flow-diagram — DFD (pbauermeister/dfd): process/entity/store, --> ;
```dfd / ```data-flow-diagram — DFD Йордона (pbauermeister/dfd): process/entity/store, --> ;
Graphviz (вшитый / «Скачать Graphviz» / PATH) или Kroki
```dfd-gs / ```gane-sarson — DFD Гейна–Сарсона: те же узлы, скруглённый процесс,
хранилище с полосой D1, штамп Ramus; локальный Pillow, без Graphviz
Встроенные пресеты PlantUML: c4, c4context, c4component, usecase.
Пример
+48 -15
View File
@@ -22,6 +22,11 @@ class OfficeInstall:
python_path: str | None = None
# First successful ``auto`` resolution is reused for the process lifetime so a
# second convert does not re-probe Word and fall through to LibreOffice.
_cached_auto_resolve: tuple[str, OfficeInstall] | None = None
def _program_dirs(kind: str) -> list[Path]:
roots: list[Path] = []
for env in ("ProgramFiles", "ProgramFiles(x86)"):
@@ -133,41 +138,69 @@ def detect_office_install(engine: str) -> OfficeInstall | None:
return None
def _detect_word() -> OfficeInstall | None:
def _word_com_available() -> bool:
"""True when pywin32 is importable and Word.Application is registered."""
if sys.platform != "win32":
return None
return False
try:
import win32com.client # type: ignore
import win32com.client # type: ignore # noqa: F401
except ImportError:
return None
word = None
return False
try:
word = win32com.client.DispatchEx("Word.Application")
return OfficeInstall(engine="word", label="Microsoft Word")
except Exception:
import winreg
with winreg.OpenKey(winreg.HKEY_CLASSES_ROOT, r"Word.Application"):
pass
return True
except OSError:
return False
def _detect_word() -> OfficeInstall | None:
if not _word_com_available():
return None
finally:
if word is not None:
try:
word.Quit()
except Exception:
pass
return OfficeInstall(engine="word", label="Microsoft Word")
def _install_still_available(install: OfficeInstall) -> bool:
if install.engine == "word":
return _word_com_available()
if install.soffice_path and install.python_path:
return Path(install.soffice_path).is_file() and Path(install.python_path).is_file()
return False
def clear_office_engine_cache() -> None:
"""Drop cached ``auto`` resolution (tests / long-running hosts)."""
global _cached_auto_resolve
_cached_auto_resolve = None
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.
``auto`` tries Word, then LibreOffice, then OpenOffice. The first successful
``auto`` pick is cached for subsequent calls in the same process.
"""
global _cached_auto_resolve
if mode not in OFFICE_ENGINE_MODES:
mode = DEFAULT_OFFICE_ENGINE
if mode == "auto":
if _cached_auto_resolve is not None:
cached_engine, cached_install = _cached_auto_resolve
if _install_still_available(cached_install):
return _cached_auto_resolve
_cached_auto_resolve = None
for candidate in ("word", "libre", "openoffice"):
install = detect_office_install(candidate)
if install is not None:
_cached_auto_resolve = (candidate, install)
return candidate, install
_cached_auto_resolve = None
return None, None
install = detect_office_install(mode)
+10 -4
View File
@@ -3,7 +3,7 @@
from __future__ import annotations
from ..word_fix import FixResult, fix_continuations
from .detect import DEFAULT_OFFICE_ENGINE, resolve_office_engine
from .detect import DEFAULT_OFFICE_ENGINE, OfficeInstall, resolve_office_engine
from .uno_fix import fix_continuations_uno
@@ -11,6 +11,7 @@ def fix_continuations_with_engine(
path: str,
*,
engine: str = DEFAULT_OFFICE_ENGINE,
install: OfficeInstall | None = None,
tables: bool = True,
listings: bool = True,
repeat_header: bool = False,
@@ -18,10 +19,15 @@ def fix_continuations_with_engine(
"""
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.
``engine`` is ``auto|word|libre|openoffice``. Pass a concrete ``engine`` plus
``install`` from ``resolve_office_engine`` to skip a second resolve. On
missing backend returns ``ok=False`` with a reason; conversion should
continue.
"""
resolved, install = resolve_office_engine(engine)
if install is not None and engine in ("word", "libre", "openoffice"):
resolved = engine
else:
resolved, install = resolve_office_engine(engine)
if resolved is None or install is None:
if engine == "auto":
return FixResult(
+10 -7
View File
@@ -34,16 +34,19 @@ def _open_in_word(path: str) -> bool:
import win32com.client # type: ignore
except ImportError:
return False
word = None
from ..word_fix import word_com_apartment
try:
word = win32com.client.DispatchEx("Word.Application")
word.Visible = True
word.Documents.Open(path)
return True
with word_com_apartment():
word = win32com.client.DispatchEx("Word.Application")
try:
word.Visible = True
word.Documents.Open(path)
return True
finally:
word = None
except Exception:
return False
finally:
pass # keep Word open for the user
def _open_in_soffice(soffice_path: str, path: str) -> bool:
+10 -3
View File
@@ -37,13 +37,20 @@ def _office_program_dir(install: OfficeInstall) -> Path | 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()
# System Python's PYTHONHOME/PYTHONPATH confuse LO's embedded interpreter
# ("Could not find platform independent libraries <prefix>").
for key in (
"PYTHONHOME",
"PYTHONPATH",
"PYTHONUSERBASE",
"VIRTUAL_ENV",
"CONDA_PREFIX",
):
env.pop(key, None)
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:
+125
View File
@@ -39,6 +39,28 @@ DOCX_SAVE_FILTER_SPECS = {
}
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
@@ -162,6 +184,109 @@ def find_page_break_row(page_numbers):
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)
+85 -13
View File
@@ -11,7 +11,8 @@ import sys
import time
from uno_helpers import (
MAX_PASSES,
PASS_SAFETY_CAP,
continuation_pass_limit,
CaptionInfo,
as_unicode,
caption_style_name,
@@ -22,6 +23,8 @@ from uno_helpers import (
BREAK_PAGE_BEFORE,
CONTROL_PARAGRAPH_BREAK,
find_page_break_row,
locate_break_row,
resume_page_before_cut,
is_caption_orphaned,
is_orphan_header_break,
parse_caption_text,
@@ -489,6 +492,20 @@ def _caption_for_split(doc, table):
)
def _count_split_targets(doc, tables, listings):
"""Captioned tables and listings this pass is allowed to split."""
found = 0
for table in _body_tables(doc):
caption = _caption_before_table(doc, table)
if caption is None:
continue
if caption.kind == "table" and tables:
found += 1
elif caption.kind == "listing" and listings:
found += 1
return found
def _caption_page_before_table(doc, table):
try:
anchor = _uno_call(table, "Anchor")
@@ -506,6 +523,26 @@ def _caption_page_before_table(doc, table):
return None
def _page_at_row(doc, table, row_1based):
"""Page number of one row, or None. Rows are 1-based, UNO cells are 0-based."""
row = int(row_1based) - 1
if row < 0:
return None
try:
ncols = int(table.getColumns().getCount())
except Exception:
ncols = 1
for col in range(max(1, ncols)):
try:
cell = table.getCellByPosition(col, row)
page = _page_at_range(doc, _uno_call(cell, "Start"))
if page is not None:
return int(page)
except Exception:
continue
return None
def _row_page_numbers(doc, table):
pages = []
try:
@@ -1049,10 +1086,16 @@ def _copy_header_to_continuation(first_table, cont_table):
pass
def _fix_pass(uno, ctx, doc, tables, listings, repeat_header, details):
def _fix_pass(uno, ctx, doc, tables, listings, repeat_header, details, start_page=1):
"""Split the first crossing table. Returns (made, cut_page)."""
made = 0
cut_page = None
try:
start_page = int(start_page)
except (TypeError, ValueError):
start_page = 1
body_tables = _body_tables(doc)
for table in reversed(body_tables):
for table in body_tables:
caption = _caption_for_split(doc, table)
if caption is None:
continue
@@ -1061,9 +1104,18 @@ def _fix_pass(uno, ctx, doc, tables, listings, repeat_header, details):
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:
row_count = int(table.getRows().getCount())
except Exception:
continue
if start_page > 1 and row_count > 0:
last_page = _page_at_row(doc, table, row_count)
if last_page is not None and last_page < start_page:
continue
first_page = _page_at_row(doc, table, 1) if row_count else None
if caption.kind == "table" and first_page is not None and not caption.is_continuation:
if is_caption_orphaned(_caption_page_before_table(doc, table), first_page):
try:
if _page_break_before_table_caption(doc, table):
details.append(
@@ -1082,10 +1134,14 @@ def _fix_pass(uno, ctx, doc, tables, listings, repeat_header, details):
details.append("skip: orphan-caption page break: {0}".format(exc))
continue
break_at = find_page_break_row(pages)
break_at, cut_page = locate_break_row(
row_count,
lambda row, current=table: _page_at_row(doc, current, row),
start_page,
)
if break_at is None:
continue
if break_at > int(table.getRows().getCount()):
if break_at > row_count:
continue
had_header = _table_has_header_row(table)
@@ -1169,9 +1225,9 @@ def _fix_pass(uno, ctx, doc, tables, listings, repeat_header, details):
details.append(
"split: {0} {1} @row {2}".format(caption.kind, caption.number, break_at)
)
return made
return made, cut_page
return made
return made, cut_page
def run_worker(port, path, engine, tables, listings, repeat_header):
@@ -1187,14 +1243,30 @@ def run_worker(port, path, engine, tables, listings, repeat_header):
doc, uno, ctx, tables, listings, details
)
_repair_split_table_borders(doc, tables, listings, details)
for _ in range(MAX_PASSES):
targets = _count_split_targets(doc, tables, listings)
passes_done = 0
start_page = 1
while passes_done < continuation_pass_limit(targets, splits):
if passes_done >= PASS_SAFETY_CAP:
details.append("skip: pass safety cap")
break
_ensure_pagination(uno, ctx, doc)
made = _fix_pass(
uno, ctx, doc, tables, listings, repeat_header, details
made, cut_page = _fix_pass(
uno, ctx, doc, tables, listings, repeat_header, details, start_page
)
passes_done += 1
if made == 0:
if start_page > 1:
start_page = 1
details.append("recheck: from page 1")
continue
break
splits += made
if cut_page:
start_page = resume_page_before_cut(cut_page)
details.append(
"resume: page {0} after cut at {1}".format(start_page, cut_page)
)
doc = _reload_document(uno, ctx, doc, path, engine)
_ensure_pagination(uno, ctx, doc)
repaired += _repair_missing_continuations(
+11 -1
View File
@@ -95,10 +95,20 @@ def check_docx_page_fill(path: str) -> tuple[list[PageFillIssue], str]:
"(pip install pywin32) и Microsoft Word."
)
from .word_fix import word_com_apartment
try:
with word_com_apartment():
return _check_docx_page_fill_in_word(path, win32com.client)
except Exception as exc:
return [], f"Проверка вёрстки пропущена: не удалось открыть Word ({exc})."
def _check_docx_page_fill_in_word(path: str, win32com_client):
word = None
doc = None
try:
word = win32com.client.DispatchEx("Word.Application")
word = win32com_client.DispatchEx("Word.Application")
word.Visible = False
word.DisplayAlerts = 0
abs_path = str(path)
+2 -1
View File
@@ -436,7 +436,8 @@ def convert(req: ConvertRequest, log: LogFn | None = None) -> ConvertResult:
split_engine, split_install = resolve_office_engine(req.office_engine)
fix = fix_continuations_with_engine(
abs_out,
engine=req.office_engine,
engine=split_engine or req.office_engine,
install=split_install,
tables=(req.table_continuation == "word"),
listings=(req.listing_continuation == "word"),
repeat_header=bool(req.table_repeat_header),
+10 -3
View File
@@ -142,9 +142,13 @@ class Image(Renderable, RequiresNumbering):
need = Length(int(height) + int(caption_reserve))
remaining = layout_state.remaining_page_height
if remaining < need:
if self.landscape:
# Already on a fresh landscape section — shrink instead of soft page-break
# (soft break + section break = empty landscape page before the figure).
at_top_of_landscape = (
self.landscape and layout_state.current_page_height == 0
)
if at_top_of_landscape:
# Too tall for a whole landscape page. Shrink here: a soft break
# on this fresh section used to stack with the section break and
# leave an empty page before the figure.
avail = Length(max(0, int(remaining) - int(caption_reserve)))
if avail > 0 and self._image.height > avail:
ratio = int(avail) / int(self._image.height)
@@ -152,6 +156,9 @@ class Image(Renderable, RequiresNumbering):
self._image.width = Length(int(self._image.width * ratio))
height = self._image.height
else:
# Does not fit the leftover of the current page (including a
# second landscape figure after the first). Keep the fitted size
# and let Word move it; do not squash it into the gap.
height = Length(int(height) + int(remaining))
yield (rendered_image := RenderedInfo(self._docx_paragraph, Length(height)))
+52 -9
View File
@@ -52,32 +52,75 @@ class List(Renderable):
self._paragraphs.append(paragraph)
return paragraph
@staticmethod
def _is_hyphen_run(run) -> bool:
"""True for a run that carries only w:noBreakHyphen (text is empty)."""
if run.text:
return False
return bool(run._element.xpath("./w:noBreakHyphen"))
@classmethod
def _logical_text(cls, runs) -> str:
parts = []
for run in runs:
if cls._is_hyphen_run(run):
parts.append("-")
else:
parts.append(run.text or "")
return "".join(parts)
def finalize_punctuation(self) -> None:
"""Apply TZ list punctuation: bullet → lowercase + ';' (last '.'); numbered → Capital + '.'."""
"""Apply TZ list punctuation: bullet → lowercase + ';' (last '.'); numbered → Capital + '.'.
Case and the final mark are written into existing runs. Runs are not merged:
merging drops w:noBreakHyphen and inline code formatting, so «REST-интерфейс»
became «RESTинтерфейс».
"""
for i, paragraph in enumerate(self._paragraphs):
runs = paragraph._docx_paragraph.runs
if len(runs) < 2:
continue
# Text after marker run
text_runs = runs[1:]
full = "".join(r.text or "" for r in text_runs).strip()
logical = self._logical_text(text_runs)
full = logical.strip()
if not full:
continue
is_last = i == len(self._paragraphs) - 1
if self._ordered:
fixed = full[0].upper() + full[1:] if full else full
fixed = full[0].upper() + full[1:]
if not fixed.endswith("."):
fixed = fixed.rstrip(";,. ") + "."
else:
fixed = full[0].lower() + full[1:] if full else full
fixed = full[0].lower() + full[1:]
if is_last:
fixed = fixed.rstrip(";,. ") + "."
else:
fixed = fixed.rstrip(";,. ") + ";"
# Put all text into first content run, clear others
text_runs[0].text = fixed
for r in text_runs[1:]:
r.text = ""
if logical == fixed:
continue
self._apply_list_punctuation(text_runs, full, fixed)
@classmethod
def _apply_list_punctuation(cls, text_runs, full: str, fixed: str) -> None:
if full[0] != fixed[0]:
for run in text_runs:
if cls._is_hyphen_run(run) or not run.text:
continue
run.text = fixed[0] + run.text[1:]
break
desired_end = fixed[-1]
last = None
for run in text_runs:
if cls._is_hyphen_run(run) or not run.text:
continue
last = run
if last is None:
return
trimmed = last.text.rstrip(";,. ")
if trimmed.endswith(desired_end):
last.text = trimmed
else:
last.text = trimmed + desired_end
def render(self, previous_rendered: RenderedInfo, layout_state: LayoutState) -> Generator[
RenderedInfo | Renderable, None, None]:
+34
View File
@@ -56,6 +56,10 @@ class Renderer:
self._after_toc = False
self._body_section_started = False
self._landscape_depth = 0
# Exit from landscape is delayed so the next +landscape table/figure
# stays in the same section. Closing and reopening here is two
# next-page breaks with nothing between them: a blank portrait page.
self._portrait_return_pending = False
max_height, max_width = content_size(landscape=False)
self._layout_tracker = LayoutTracker(max_height, max_width)
@@ -87,6 +91,11 @@ class Renderer:
apply_centered_page_footer(new_section)
def _enter_landscape(self) -> None:
if self._portrait_return_pending:
# Previous block already left us in the landscape section.
self._portrait_return_pending = False
self._landscape_depth += 1
return
if self._landscape_depth > 0:
self._landscape_depth += 1
return
@@ -109,12 +118,34 @@ class Renderer:
self._landscape_depth -= 1
if self._landscape_depth > 0:
return
self._portrait_return_pending = True
def _commit_portrait_return(self) -> None:
"""Close the landscape section once the next block is portrait content."""
if not self._portrait_return_pending:
return
self._portrait_return_pending = False
section = self._document.add_section(WD_SECTION.NEW_PAGE)
apply_section_geometry(section, landscape=False)
apply_centered_page_footer(section)
max_height, max_width = content_size(landscape=False)
self._layout_tracker.set_page_size(max_height, max_width)
@staticmethod
def _hosts_only_landscape_figures(renderable: Renderable) -> bool:
"""Paragraph whose only content is +landscape figures.
The paragraph itself is not marked landscape; the figure opens the
section. Returning to portrait before that figure inserts a blank page
between two photos in a row.
"""
images = getattr(renderable, "_images", None)
if not images or not all(getattr(img, "landscape", False) for img in images):
return False
paragraph = getattr(renderable, "_docx_paragraph", None)
text = (paragraph.text or "").strip() if paragraph is not None else ""
return text == ""
def process(self, renderables: list[Renderable]):
items = self._ensure_appendix_index(list(renderables))
for i in range(len(items)):
@@ -124,6 +155,7 @@ class Renderer:
self.render(items[i])
self._flush_to_new_screen()
self._commit_portrait_return()
# If document had no TOC, still add page numbers to the only section
if not self._body_section_started:
section = self._document.sections[0]
@@ -316,6 +348,8 @@ class Renderer:
if wants_landscape:
self._enter_landscape()
elif not self._hosts_only_landscape_figures(renderable):
self._commit_portrait_return()
try:
if not self._skip_numbering and isinstance(renderable, RequiresNumbering):
+157 -17
View File
@@ -9,6 +9,7 @@ from __future__ import annotations
import os
import re
import sys
from contextlib import contextmanager
from dataclasses import dataclass, field
# wdActiveEndPageNumber
@@ -22,7 +23,28 @@ _CAPTION_RE = re.compile(
)
_CONTINUATION_RE = re.compile(r"^Продолжение\s+(?:Таблицы|Листинга)\b", re.IGNORECASE)
MAX_PASSES = 5
# Stop if Word keeps reporting a page break after every single split.
PASS_SAFETY_CAP = 200
@contextmanager
def word_com_apartment():
"""Initialize COM on this thread and release that initialization afterwards.
Each GUI conversion runs on a new thread. Importing ``win32com`` initializes
COM only on the first thread that imports it, so the next conversion raises
``CO_E_NOTINITIALIZED`` (-2147221008) at ``DispatchEx``.
"""
import pythoncom # type: ignore
pythoncom.CoInitialize()
try:
yield
finally:
try:
pythoncom.CoUninitialize()
except Exception:
pass
@dataclass
@@ -65,6 +87,20 @@ def parse_caption_text(text: str) -> CaptionInfo | None:
)
def resume_page_before_cut(cut_page: int) -> int:
"""Next scan starts on the page before the cut. See uno_helpers."""
from .office.uno_helpers import resume_page_before_cut as impl
return impl(cut_page)
def locate_break_row(n_rows: int, page_at, min_page: int = 1):
"""Binary-search the next page jump. See uno_helpers.locate_break_row."""
from .office.uno_helpers import locate_break_row as impl
return impl(n_rows, page_at, min_page)
def find_page_break_row(page_numbers: list[int]) -> int | None:
"""
Return 1-based Word row index where a new page starts.
@@ -90,6 +126,16 @@ def is_orphan_header_break(break_at: int | None, *, had_header: bool) -> bool:
return bool(had_header and break_at == 2)
def continuation_pass_limit(object_count: int, splits_done: int = 0) -> int:
"""How many passes to run. One pass performs one split.
``10 + (tables + listings) + 1`` for each split already made.
"""
count = max(0, int(object_count))
extra = max(0, int(splits_done))
return 10 + count + extra
def is_caption_orphaned(caption_page: int | None, first_row_page: int | None) -> bool:
"""True if «Таблица N — …» sits on an earlier page than the table body.
@@ -142,27 +188,68 @@ def fix_continuations(
if not os.path.isfile(abs_path):
return FixResult(False, message=f"Файл не найден: {abs_path}")
try:
with word_com_apartment():
return _fix_continuations_in_word(
abs_path,
win32com.client,
tables=tables,
listings=listings,
repeat_header=repeat_header,
)
except Exception as exc:
return FixResult(
False,
message=f"Пост-разрыв таблиц не удался: {exc}",
)
def _fix_continuations_in_word(
abs_path: str,
win32com_client,
*,
tables: bool,
listings: bool,
repeat_header: bool,
) -> FixResult:
word = None
doc = None
splits = 0
details: list[str] = []
try:
word = win32com.client.DispatchEx("Word.Application")
word = win32com_client.DispatchEx("Word.Application")
word.Visible = False
word.DisplayAlerts = 0
doc = word.Documents.Open(abs_path, ReadOnly=False)
for _pass in range(MAX_PASSES):
targets = _count_split_targets(doc, tables=tables, listings=listings)
passes_done = 0
# After a cut, pages above it stay put. The next pass starts on the
# page before that cut; a final pass from page 1 checks nothing was missed.
start_page = 1
while passes_done < continuation_pass_limit(targets, splits):
if passes_done >= PASS_SAFETY_CAP:
details.append("skip: pass safety cap")
break
doc.Repaginate()
made = _fix_pass(
made, cut_page = _fix_pass(
doc,
tables=tables,
listings=listings,
repeat_header=repeat_header,
details=details,
start_page=start_page,
)
passes_done += 1
if made == 0:
if start_page > 1:
start_page = 1
details.append("recheck: from page 1")
continue
break
splits += made
if cut_page:
start_page = resume_page_before_cut(cut_page)
details.append(f"resume: page {start_page} after cut at {cut_page}")
skipped = sum(1 for d in details if d.startswith("skip:"))
_clear_section_page_restarts(doc)
@@ -530,11 +617,18 @@ def _fix_pass(
listings: bool,
repeat_header: bool,
details: list[str],
) -> int:
"""One pass over tables (bottom-up). Returns number of splits performed."""
start_page: int = 1,
) -> tuple[int, int | None]:
"""One pass in document order. Splits the first table that crosses a page.
Tables that end before ``start_page`` are skipped. The page jump inside a
table is binary-searched, so a long remainder is not read from row 1.
Returns ``(splits_made, cut_page)``.
"""
made = 0
cut_page: int | None = None
count = int(doc.Tables.Count)
for ti in range(count, 0, -1):
for ti in range(1, count + 1):
try:
table = doc.Tables(ti)
except Exception:
@@ -554,18 +648,28 @@ def _fix_pass(
if caption.kind == "listing" and not listings:
continue
row_count = int(table.Rows.Count)
# A one-cell listing can cross pages inside the cell; its row page is
# only the start, so it must not be skipped on that alone.
single_cell_listing = caption.kind == "listing" and row_count == 1
if start_page > 1 and row_count > 0 and not single_cell_listing:
last_page = _page_at_row(table, row_count)
if last_page is not None and last_page < start_page:
continue
# Legacy single-cell listings: expand to rows so Split works
if caption.kind == "listing" and int(table.Rows.Count) == 1:
if single_cell_listing:
para_pages = _paragraph_page_numbers_in_cell(table)
if find_page_break_row(para_pages) is not None:
if _expand_single_cell_listing_to_rows(table):
details.append(f"expand: listing {caption.number} → {table.Rows.Count} rows")
doc.Repaginate()
row_count = int(table.Rows.Count)
pages = _row_page_numbers(table)
first_page = _page_at_row(table, 1) if row_count else None
# Method: only the title («Таблица N») at page bottom is forbidden.
if caption.kind == "table" and pages:
if is_caption_orphaned(_caption_page_before_table(table), pages[0]):
if caption.kind == "table" and first_page is not None:
if is_caption_orphaned(_caption_page_before_table(table), first_page):
try:
if _page_break_before_table_caption(table):
details.append(
@@ -582,10 +686,14 @@ def _fix_pass(
details.append(f"skip: orphan-caption page break: {exc}")
continue
break_at = find_page_break_row(pages)
break_at, cut_page = locate_break_row(
row_count,
lambda row, current=table: _page_at_row(current, row),
start_page,
)
if break_at is None:
continue
if break_at > int(table.Rows.Count):
if break_at > row_count:
continue
# Vertical merges block Rows()/Table.Split — unmerge, split, then restore.
@@ -600,8 +708,12 @@ def _fix_pass(
)
doc.Repaginate()
table = doc.Tables(ti)
pages = _row_page_numbers(table)
break_at = find_page_break_row(pages)
row_count = int(table.Rows.Count)
break_at, cut_page = locate_break_row(
row_count,
lambda row, current=table: _page_at_row(current, row),
start_page,
)
if break_at is None:
details.append(
f"skip: {caption.kind} {caption.number} "
@@ -650,7 +762,7 @@ def _fix_pass(
except Exception as exc:
details.append(f"skip: after split cannot get continuation table[{ti}+1]: {exc}")
made += 1
continue
return made, cut_page
if vmerge_spans:
try:
@@ -698,8 +810,36 @@ def _fix_pass(
made += 1
details.append(f"split: {caption.kind} {caption.number} @row {break_at}")
return made, cut_page
return made, None
def _count_split_targets(doc, *, tables: bool, listings: bool) -> int:
"""Captioned tables and listings this pass is allowed to split."""
found = 0
try:
count = int(doc.Tables.Count)
except Exception:
return 0
for ti in range(1, count + 1):
try:
table = doc.Tables(ti)
except Exception:
continue
try:
if int(table.NestingLevel) > 1:
continue
except Exception:
pass
caption = _caption_before_table(table)
if caption is None:
continue
if caption.kind == "table" and tables:
found += 1
elif caption.kind == "listing" and listings:
found += 1
return found
return made
def _caption_before_table(table) -> CaptionInfo | None:
text = _paragraph_text_before_table(table)