430 lines
15 KiB
Python
430 lines
15 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""DokoGen — Генератор документов из шаблонов DOCX"""
|
||
|
||
import os
|
||
import re
|
||
import io
|
||
from datetime import datetime
|
||
from typing import Dict, List, Optional, Callable
|
||
|
||
try:
|
||
from docx import Document
|
||
from docx.oxml.ns import qn
|
||
from docx.oxml import OxmlElement
|
||
from lxml import etree
|
||
except ImportError:
|
||
Document = None
|
||
|
||
from .variables import OLD_VAR_MAP, OLD_VAR_PATTERN, VAR_PATTERN
|
||
|
||
# Стандартный шаблон для незаполненных переменных
|
||
EMPTY_MARKER = '—'
|
||
|
||
|
||
def process_template(
|
||
template_path: str,
|
||
replacements: Dict[str, str],
|
||
output_path: Optional[str] = None,
|
||
log_func: Optional[Callable] = None,
|
||
) -> str:
|
||
"""Обрабатывает шаблон DOCX: подстановка переменных, циклы, сохранение."""
|
||
if Document is None:
|
||
raise ImportError("python-docx не установлен. Установите: pip install python-docx")
|
||
|
||
if not os.path.exists(template_path):
|
||
raise FileNotFoundError(f"Шаблон не найден: {template_path}")
|
||
|
||
if output_path is None:
|
||
base, ext = os.path.splitext(template_path)
|
||
output_path = f"{base}_result{ext}"
|
||
|
||
doc = Document(template_path)
|
||
|
||
# 1. Миграция старых переменных <...> → {{...}}
|
||
_migrate_old_vars(doc)
|
||
|
||
# 2. Удаление proofErr
|
||
_strip_prooferr(doc)
|
||
|
||
# 3. Обработка циклов (loop/loop_end)
|
||
loops = replacements.pop('_loops', {})
|
||
if loops:
|
||
_process_loops(doc, loops)
|
||
|
||
# 4. Перезагрузка после циклов
|
||
buf = io.BytesIO()
|
||
doc.save(buf)
|
||
buf.seek(0)
|
||
doc = Document(buf)
|
||
|
||
# 5. Плоская замена переменных (3 прохода)
|
||
for _ in range(3):
|
||
_replace_in_paragraphs(doc.paragraphs, replacements)
|
||
_replace_in_tables(doc.tables, replacements)
|
||
_replace_in_headers_footers(doc, replacements)
|
||
|
||
# 6. Логирование незаменённых переменных
|
||
_log_unmatched(doc, replacements, log_func if log_func else lambda x: None)
|
||
|
||
# 7. Подсветка незаполненных
|
||
_highlight_unmatched(doc)
|
||
|
||
# 8. Сохранение
|
||
doc.save(output_path)
|
||
|
||
if log_func:
|
||
log_func(f"✅ {os.path.basename(output_path)}")
|
||
return output_path
|
||
|
||
|
||
# ==================== МИГРАЦИЯ СТАРЫХ ПЕРЕМЕННЫХ ====================
|
||
|
||
def _migrate_old_vars(doc):
|
||
"""Заменяет <OldVar> на {{new_var}} в документе."""
|
||
def _migrate_text(text):
|
||
return OLD_VAR_PATTERN.sub(lambda m: OLD_VAR_MAP.get(m.group(0), m.group(0)), text)
|
||
|
||
for para in doc.paragraphs:
|
||
new_text = _migrate_text(para.text)
|
||
if new_text != para.text:
|
||
if para.runs:
|
||
first = para.runs[0]
|
||
para.clear()
|
||
run = para.add_run(new_text)
|
||
_copy_run_style(first, run)
|
||
else:
|
||
para.clear()
|
||
para.add_run(new_text)
|
||
|
||
for table in doc.tables:
|
||
for row in table.rows:
|
||
for cell in row.cells:
|
||
for para in cell.paragraphs:
|
||
new_text = _migrate_text(para.text)
|
||
if new_text != para.text:
|
||
if para.runs:
|
||
first = para.runs[0]
|
||
para.clear()
|
||
run = para.add_run(new_text)
|
||
_copy_run_style(first, run)
|
||
else:
|
||
para.clear()
|
||
para.add_run(new_text)
|
||
|
||
for section in doc.sections:
|
||
for hf in [section.header, section.footer,
|
||
section.first_page_header, section.first_page_footer]:
|
||
if hf:
|
||
for para in hf.paragraphs:
|
||
new_text = _migrate_text(para.text)
|
||
if new_text != para.text:
|
||
if para.runs:
|
||
first = para.runs[0]
|
||
para.clear()
|
||
run = para.add_run(new_text)
|
||
_copy_run_style(first, run)
|
||
else:
|
||
para.clear()
|
||
para.add_run(new_text)
|
||
|
||
|
||
# ==================== ЗАМЕНА В ЭЛЕМЕНТАХ ====================
|
||
|
||
def _replace_in_paragraphs(paragraphs, replacements):
|
||
for para in paragraphs:
|
||
full = para.text
|
||
# Пропускаем строки циклов
|
||
if 'loop:' in full or 'loop_end' in full:
|
||
continue
|
||
matches = VAR_PATTERN.findall(full)
|
||
if not matches:
|
||
continue
|
||
new_text = full
|
||
for var in matches:
|
||
val = replacements.get(var)
|
||
if val is not None:
|
||
new_text = new_text.replace(var, str(val))
|
||
if new_text != full:
|
||
if para.runs:
|
||
first = para.runs[0]
|
||
para.clear()
|
||
run = para.add_run(new_text)
|
||
_copy_run_style(first, run)
|
||
else:
|
||
para.clear()
|
||
para.add_run(new_text)
|
||
|
||
|
||
def _replace_in_tables(tables, replacements):
|
||
for table in tables:
|
||
for row in table.rows:
|
||
for cell in row.cells:
|
||
for para in cell.paragraphs:
|
||
full = para.text
|
||
matches = VAR_PATTERN.findall(full)
|
||
if not matches:
|
||
continue
|
||
new_text = full
|
||
for var in matches:
|
||
val = replacements.get(var)
|
||
if val is not None:
|
||
new_text = new_text.replace(var, str(val))
|
||
if new_text != full:
|
||
if para.runs:
|
||
first = para.runs[0]
|
||
para.clear()
|
||
run = para.add_run(new_text)
|
||
_copy_run_style(first, run)
|
||
else:
|
||
para.clear()
|
||
para.add_run(new_text)
|
||
|
||
|
||
def _replace_in_headers_footers(doc, replacements):
|
||
for section in doc.sections:
|
||
for hf in [section.header, section.footer,
|
||
section.first_page_header, section.first_page_footer]:
|
||
if hf:
|
||
_replace_in_paragraphs(hf.paragraphs, replacements)
|
||
|
||
|
||
def _copy_run_style(source, target):
|
||
"""Копирует форматирование из одного run в другой."""
|
||
try:
|
||
target.bold = source.bold
|
||
target.italic = source.italic
|
||
target.underline = source.underline
|
||
if source.font:
|
||
target.font.size = source.font.size
|
||
target.font.name = source.font.name
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
def _strip_prooferr(doc):
|
||
ns = 'http://schemas.openxmlformats.org/wordprocessingml/2006/main'
|
||
for elem in doc.element.iter():
|
||
for child in list(elem):
|
||
if child.tag == f'{{{ns}}}proofErr':
|
||
elem.remove(child)
|
||
|
||
|
||
# ==================== ОБРАБОТКА ЦИКЛОВ ====================
|
||
|
||
LOOP_START = re.compile(r'(?:<!--|<!--)?\s*loop:(\w+)(?:\s*(?:-->|-->))?')
|
||
LOOP_END = re.compile(r'(?:<!--|<!--)?\s*loop_end(?:\s*(?:-->|-->))?')
|
||
|
||
|
||
def _process_loops(doc, loops: Dict[str, List[Dict]]):
|
||
body = doc.element.body
|
||
max_passes = 20
|
||
for _ in range(max_passes):
|
||
body_str = etree.tostring(body, encoding='unicode')
|
||
# Находим все ключи циклов
|
||
found = set(LOOP_START.findall(body_str))
|
||
if not found:
|
||
break
|
||
changed = False
|
||
for key in sorted(found):
|
||
items = loops.get(key, [])
|
||
if not items:
|
||
_remove_loop_block(body, key)
|
||
changed = True
|
||
else:
|
||
_expand_loop(body, key, items)
|
||
changed = True
|
||
body = doc.element.body
|
||
if not changed:
|
||
break
|
||
|
||
|
||
def _expand_loop(body, key: str, items: List[Dict]):
|
||
body_str = etree.tostring(body, encoding='unicode')
|
||
|
||
markers = []
|
||
for m in LOOP_START.finditer(body_str):
|
||
if m.group(1) == key:
|
||
markers.append((m.start(), True, m.end()))
|
||
for m in LOOP_END.finditer(body_str):
|
||
markers.append((m.start(), False, m.end()))
|
||
|
||
if len(markers) < 2:
|
||
return
|
||
markers.sort()
|
||
|
||
start_pos = end_pos = start_end = end_end = None
|
||
for pos, is_start, end in markers:
|
||
if is_start and start_pos is None:
|
||
start_pos = pos
|
||
start_end = end
|
||
elif not is_start and start_pos is not None:
|
||
end_pos = pos
|
||
end_end = end
|
||
break
|
||
if start_pos is None or end_pos is None:
|
||
return
|
||
|
||
# Определяем структурный элемент (параграф или строка таблицы)
|
||
s_start, s_end = _find_element_bounds(body_str, start_pos)
|
||
e_start, e_end = _find_element_bounds(body_str, end_pos)
|
||
|
||
if s_start is None or e_end is None:
|
||
return
|
||
|
||
# Извлекаем шаблон (содержимое между маркерами)
|
||
template_xml = body_str[start_end:end_pos]
|
||
# Убираем ТОЛЬКО маркеры текущего цикла, не трогая вложенные
|
||
template_xml = template_xml.replace('<!-- loop:%s -->' % key, '')
|
||
template_xml = template_xml.replace('<!-- loop:%s -->' % key, '')
|
||
# loop_end убираем только те, что без вложенности (последний)
|
||
template_xml = template_xml.replace('<!-- loop_end -->', '')
|
||
template_xml = template_xml.replace('<!-- loop_end -->', '')
|
||
|
||
# Размножаем
|
||
expanded = []
|
||
for idx, item in enumerate(items, 1):
|
||
clone = template_xml
|
||
for field, value in item.items():
|
||
clone = clone.replace(f'{{{{item.{field}}}}}', str(value) if value else EMPTY_MARKER)
|
||
clone = clone.replace(f'{{{{{field}}}}}', str(value) if value else EMPTY_MARKER)
|
||
clone = clone.replace('{{item.number}}', str(idx))
|
||
clone = clone.replace('{{number}}', str(idx))
|
||
clone = clone.replace('\u27f5BR\u27f5', '</w:t><w:br/><w:t xml:space="preserve">')
|
||
expanded.append(clone)
|
||
|
||
new_str = body_str[:s_start] + ''.join(expanded) + body_str[e_end:]
|
||
try:
|
||
new_body = etree.fromstring(new_str.encode('utf-8'))
|
||
parent = body.getparent()
|
||
if parent is not None:
|
||
parent.replace(body, new_body)
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
def _remove_loop_block(body, key: str):
|
||
body_str = etree.tostring(body, encoding='unicode')
|
||
start_pos = None
|
||
for m in LOOP_START.finditer(body_str):
|
||
if m.group(1) == key:
|
||
start_pos = m.start()
|
||
break
|
||
if start_pos is None:
|
||
return
|
||
s_start, s_end = _find_element_bounds(body_str, start_pos)
|
||
end_match = LOOP_END.search(body_str, start_pos)
|
||
if end_match:
|
||
_, e_end = _find_element_bounds(body_str, end_match.start())
|
||
new_str = body_str[:s_start] + body_str[e_end:]
|
||
else:
|
||
new_str = body_str[:s_start] + body_str[s_end:]
|
||
try:
|
||
new_body = etree.fromstring(new_str.encode('utf-8'))
|
||
parent = body.getparent()
|
||
if parent is not None:
|
||
parent.replace(body, new_body)
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
def _find_element_bounds(xml_str: str, pos: int):
|
||
"""Находит границы элемента (table row или paragraph), содержащего позицию."""
|
||
# Пробуем tr
|
||
tr_open = max(xml_str.rfind('<w:tr ', 0, pos), xml_str.rfind('<w:tr>', 0, pos))
|
||
if tr_open != -1:
|
||
tr_close = xml_str.find('</w:tr>', pos)
|
||
if tr_close != -1:
|
||
return tr_open, tr_close + len('</w:tr>')
|
||
# Пробуем p
|
||
p_open = max(xml_str.rfind('<w:p ', 0, pos), xml_str.rfind('<w:p>', 0, pos))
|
||
if p_open != -1:
|
||
p_close = xml_str.find('</w:p>', pos)
|
||
if p_close != -1:
|
||
return p_open, p_close + len('</w:p>')
|
||
return None, None
|
||
|
||
|
||
# ==================== ЛОГИРОВАНИЕ НЕЗАМЕНЁННЫХ ====================
|
||
|
||
UNMATCHED_PATTERN = re.compile(r'\{\{[^}]+\}\}|<[A-Za-z0-9_]+>')
|
||
|
||
|
||
def _log_unmatched(doc, replacements: Dict[str, str], log_func):
|
||
"""Логирует оставшиеся незаменённые переменные."""
|
||
unmatched = set()
|
||
|
||
def _scan_text(text):
|
||
for m in UNMATCHED_PATTERN.finditer(text):
|
||
var = m.group(0)
|
||
# Проверяем, не является ли это уже обработанной переменной
|
||
if var.startswith('{{'):
|
||
if var not in replacements:
|
||
unmatched.add(var)
|
||
elif var.startswith('<'):
|
||
if var not in replacements:
|
||
unmatched.add(var)
|
||
|
||
for para in doc.paragraphs:
|
||
_scan_text(para.text)
|
||
for table in doc.tables:
|
||
for row in table.rows:
|
||
for cell in row.cells:
|
||
for para in cell.paragraphs:
|
||
_scan_text(para.text)
|
||
for section in doc.sections:
|
||
for hf in [section.header, section.footer,
|
||
section.first_page_header, section.first_page_footer]:
|
||
if hf:
|
||
for para in hf.paragraphs:
|
||
_scan_text(para.text)
|
||
|
||
if unmatched:
|
||
log_func(f"⚠️ Незаменённые переменные ({len(unmatched)}): {', '.join(sorted(unmatched)[:20])}")
|
||
if len(unmatched) > 20:
|
||
log_func(f" ... и ещё {len(unmatched) - 20}")
|
||
else:
|
||
log_func("✅ Все переменные успешно заменены")
|
||
|
||
|
||
# ==================== ПОДСВЕТКА НЕЗАПОЛНЕННЫХ ====================
|
||
|
||
def _highlight_unmatched(doc):
|
||
"""Подсвечивает незаполненные переменные красным ({{...}} и старые <...>)."""
|
||
ns = 'http://schemas.openxmlformats.org/wordprocessingml/2006/main'
|
||
|
||
def _has_unmatched(text):
|
||
return bool(UNMATCHED_PATTERN.search(text))
|
||
|
||
def _highlight_run(run):
|
||
rpr = run._element.find(f'{{{ns}}}rPr')
|
||
if rpr is None:
|
||
rpr = OxmlElement('w:rPr')
|
||
run._element.insert(0, rpr)
|
||
color = OxmlElement('w:color')
|
||
color.set(f'{{{ns}}}val', 'FF0000')
|
||
rpr.append(color)
|
||
sz = OxmlElement('w:sz')
|
||
sz.set(f'{{{ns}}}val', '22')
|
||
rpr.append(sz)
|
||
|
||
for para in doc.paragraphs:
|
||
for run in para.runs:
|
||
if _has_unmatched(run.text):
|
||
_highlight_run(run)
|
||
|
||
for table in doc.tables:
|
||
for row in table.rows:
|
||
for cell in row.cells:
|
||
for para in cell.paragraphs:
|
||
for run in para.runs:
|
||
if _has_unmatched(run.text):
|
||
_highlight_run(run)
|
||
|
||
for section in doc.sections:
|
||
for hf in [section.header, section.footer]:
|
||
if hf:
|
||
for para in hf.paragraphs:
|
||
for run in para.runs:
|
||
if _has_unmatched(run.text):
|
||
_highlight_run(run)
|