"""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, get_profile, ) 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 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, ) 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_document_styles shell = Document(template) apply_document_styles(shell, get_profile(req.doc_type).style_preset) 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_document_styles(document, get_profile(req.doc_type).style_preset) _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)