Files
md_to_gost/md2gost/pipeline.py
T
Igor20264 818a044aa1
Python application / build (push) Has been cancelled
update 0.4.4
- update документация
- промт для ии полу конфигурируемый
2026-09-06 11:04:01 +03:00

332 lines
12 KiB
Python

"""Shared Markdown → DOCX pipeline for CLI and GUI."""
from __future__ import annotations
import logging
import os
import platform
import subprocess
import traceback
from dataclasses import dataclass
from datetime import datetime
from getpass import getuser
from typing import Callable
from docx import Document
from . import package_dir
from .checker import check_markdown, format_report
from .converter import Converter
from .profiles import (
DEFAULT_HEADING_NUMBERING,
DEFAULT_TABLE_CONTINUATION,
DEFAULT_LISTING_CONTINUATION,
DEFAULT_TOC_MODE,
)
LogFn = Callable[[str], None]
@dataclass
class ConvertRequest:
filename: str = ""
output: str | None = None
template: str | None = None
doc_type: str = "practice"
heading_numbering: str = DEFAULT_HEADING_NUMBERING
toc_mode: str = DEFAULT_TOC_MODE
table_continuation: str = DEFAULT_TABLE_CONTINUATION
listing_continuation: str = DEFAULT_LISTING_CONTINUATION
emdash_to_hyphen: bool = False
hr_pagebreak: bool = False
title: str | None = None
assignment: str | None = None
check: bool = False
check_only: bool = False
strict: bool = False
syntax_highlighting: bool = False
plantuml_jar: str | None = None
kroki_url: str | None = None
diagram_fallback: str = "remote"
diagram_format: str = "png"
diagram_scale: float = 2.0
schemes_path: str | None = None
styles_path: str | None = None
debug: bool = False
open_when_done: bool = False
check_pages: bool = False
table_repeat_header: bool = False
@dataclass
class ConvertResult:
ok: bool
exit_code: int = 0
output_path: str | None = None
check_report: str = ""
message: str = ""
def default_output_path(filename: str) -> str:
base = os.path.basename(filename)
if base.lower().endswith(".md"):
base = base[:-3]
return os.path.join(os.path.dirname(os.path.abspath(filename)), base + ".docx")
def timestamped_output_path(path: str) -> str:
"""report.docx → report_2026-09-05-10-05.docx (local time; seconds if taken)."""
directory, name = os.path.split(os.path.abspath(path))
stem, ext = os.path.splitext(name)
if ext.lower() != ".docx":
ext = ".docx"
stamp = datetime.now().strftime("%Y-%m-%d-%H-%M")
candidate = os.path.join(directory, f"{stem}_{stamp}{ext}")
if not os.path.exists(candidate):
return candidate
stamp = datetime.now().strftime("%Y-%m-%d-%H-%M-%S")
candidate = os.path.join(directory, f"{stem}_{stamp}{ext}")
n = 2
while os.path.exists(candidate):
candidate = os.path.join(directory, f"{stem}_{stamp}_{n}{ext}")
n += 1
return candidate
def default_template_path() -> str:
return os.path.join(package_dir(), "Template.docx")
def open_document(path: str) -> None:
system = platform.system()
if system == "Darwin":
subprocess.call(("open", path))
elif system == "Windows":
os.startfile(path) # type: ignore[attr-defined]
else:
subprocess.call(("xdg-open", path))
def _fix_front_matter_after_compose(document, *, had_title: bool, had_assignment: bool) -> None:
"""Clear PAGE on title/assignment sections; keep continuous page numbers."""
from .page_geometry import clear_section_footer, ensure_continuous_page_numbers
ensure_continuous_page_numbers(document)
n_front = int(bool(had_title)) + int(bool(had_assignment))
for i, section in enumerate(document.sections):
if i < n_front:
clear_section_footer(section)
else:
break
ensure_continuous_page_numbers(document)
def convert(req: ConvertRequest, log: LogFn | None = None) -> ConvertResult:
"""Run check and/or conversion. Does not re-raise; errors go into ConvertResult."""
emit: LogFn = log if log is not None else print
def fail(code: int, message: str, check_report: str = "") -> ConvertResult:
emit(message)
return ConvertResult(False, code, check_report=check_report, message=message)
filename = (req.filename or "").strip()
# Standalone page-fill check on an existing DOCX
if req.check_pages and filename.lower().endswith(".docx") and not filename.lower().endswith(".md"):
from .page_fill_check import check_docx_page_fill, format_page_fill_report
issues, status = check_docx_page_fill(filename)
report = format_page_fill_report(issues, status)
emit(report)
return ConvertResult(True, 0, check_report=report, message=report)
if not filename:
return fail(2, "Укажите исходный markdown-файл")
if not filename.lower().endswith(".md"):
return fail(1, "Исходный файл должен быть в формате .md")
if not os.path.isfile(filename):
return fail(2, f"Файл не найден: {filename}")
output = (req.output or "").strip() or None
if output and not output.lower().endswith(".docx"):
return fail(1, "Выходной файл должен быть в формате .docx")
for label, path in (("титул", req.title), ("задание", req.assignment), ("шаблон", req.template)):
if path and not os.path.isfile(path):
return fail(2, f"Файл ({label}) не найден: {path}")
if req.syntax_highlighting:
os.environ["SYNTAX_HIGHLIGHTING"] = "1"
else:
os.environ.pop("SYNTAX_HIGHLIGHTING", None)
from .diagram_renderer import configure_diagrams
md_dir = os.path.dirname(os.path.abspath(filename)) or "."
configure_diagrams(
plantuml_jar=req.plantuml_jar or None,
kroki_url=req.kroki_url or None,
fallback=req.diagram_fallback,
schemes_path=req.schemes_path or None,
md_dir=md_dir,
diagram_format=req.diagram_format if req.diagram_format in ("png", "svg") else "png",
diagram_scale=req.diagram_scale,
)
os.environ["WORKING_DIR"] = md_dir
with open(filename, encoding="utf-8") as f:
md_text = f.read()
check_report = ""
if req.check or req.check_only:
issues = check_markdown(
md_text,
req.doc_type,
table_continuation=req.table_continuation,
listing_continuation=req.listing_continuation,
)
check_report = format_report(issues)
emit(check_report)
errors = [i for i in issues if i.severity == "error"]
if req.strict and errors:
return fail(1, "Проверка ТЗ: есть ошибки (--strict)", check_report)
if req.check_only:
return ConvertResult(
True, 0, check_report=check_report,
message=check_report,
)
if not output:
output = default_output_path(filename)
template = (req.template or "").strip() or default_template_path()
handler = _CallbackLogHandler(emit)
handler.setFormatter(logging.Formatter("%(levelname)s: %(message)s"))
root_log = logging.getLogger("md2gost")
root_log.addHandler(handler)
attached_diag = logging.getLogger("md2gost.diagram_renderer")
prev_diag_level = attached_diag.level
attached_diag.setLevel(logging.INFO)
attached_diag.addHandler(handler)
try:
converter = Converter(
filename, output, template, req.debug,
doc_type=req.doc_type,
heading_numbering=req.heading_numbering,
emdash_to_hyphen=req.emdash_to_hyphen,
toc_mode=req.toc_mode,
table_continuation=req.table_continuation,
listing_continuation=req.listing_continuation,
hr_pagebreak=req.hr_pagebreak,
styles_path=req.styles_path or None,
)
converter.convert()
document = converter.document
if req.title or req.assignment:
try:
from docxcompose.composer import Composer
except ImportError:
return fail(3, "Для титула/задания нужен пакет docxcompose", check_report)
from .styles import apply_style_config
style_cfg = converter.style_config
shell = Document(template)
apply_style_config(shell, style_cfg)
body = shell.element.body
for child in list(body):
if not child.tag.endswith("}sectPr"):
body.remove(child)
composer = Composer(shell)
if req.title:
composer.append(Document(req.title))
shell.add_page_break()
if req.assignment:
composer.append(Document(req.assignment))
shell.add_page_break()
composer.append(document)
document = composer.doc
apply_style_config(document, style_cfg)
_fix_front_matter_after_compose(
document,
had_title=bool(req.title),
had_assignment=bool(req.assignment),
)
document.core_properties.author = getuser()
document.core_properties.comments = "Создано при помощи md2gost (ТЗ МИРЭА)"
try:
document.save(output)
except PermissionError:
alt = timestamped_output_path(output)
emit(f"Не удалось записать {output} (файл занят). Сохраняю как {alt}")
document.save(alt)
output = alt
except Exception as exc:
emit(traceback.format_exc())
return ConvertResult(
False, 1, check_report=check_report,
message=f"Ошибка конвертации: {exc}",
)
finally:
root_log.removeHandler(handler)
attached_diag.removeHandler(handler)
attached_diag.setLevel(prev_diag_level)
abs_out = os.path.abspath(output)
emit(f"Generated document: {abs_out}")
if req.table_continuation == "word" or req.listing_continuation == "word":
from .word_fix import fix_continuations
fix = fix_continuations(
abs_out,
tables=(req.table_continuation == "word"),
listings=(req.listing_continuation == "word"),
repeat_header=bool(req.table_repeat_header),
)
emit(fix.message)
if fix.details:
for line in fix.details:
emit(line)
if not fix.ok:
check_report = (
(check_report + "\n" + fix.message).strip() if check_report else fix.message
)
if req.check_pages:
from .page_fill_check import check_docx_page_fill, format_page_fill_report
issues, status = check_docx_page_fill(abs_out)
page_report = format_page_fill_report(issues, status)
emit(page_report)
check_report = (check_report + "\n" + page_report).strip() if check_report else page_report
if req.debug or req.open_when_done:
try:
open_document(abs_out)
except Exception as exc:
emit(f"Не удалось открыть файл: {exc}")
return ConvertResult(
True, 0, output_path=abs_out, check_report=check_report,
message=f"Generated document: {abs_out}",
)
class _CallbackLogHandler(logging.Handler):
def __init__(self, emit: LogFn):
super().__init__()
self._emit = emit
def emit(self, record: logging.LogRecord) -> None:
try:
self._emit(self.format(record))
except Exception:
pass
def should_launch_gui(filename: str | None, gui_flag: bool) -> bool:
"""GUI if --gui, or if no input file was given (interactive default)."""
return bool(gui_flag or not filename)