fix: переработка циклов — правильный порядок + сохранение вложенных
This commit is contained in:
@@ -0,0 +1,490 @@
|
||||
# -*- 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:
|
||||
_merge_runs_for_loops(doc)
|
||||
_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 _log_unmatched(doc, replacements=None, log_func=None):
|
||||
"""Логирует незаменённые переменные в документе."""
|
||||
import re
|
||||
unmatched = []
|
||||
seen = set()
|
||||
for para in doc.paragraphs:
|
||||
for m in re.finditer(r'\{\{[^}]+\}\}|<[A-Za-z0-9_]+>', para.text):
|
||||
var = m.group(0)
|
||||
if var not in seen:
|
||||
seen.add(var)
|
||||
unmatched.append(var)
|
||||
if log_func and unmatched:
|
||||
log_func(f"\u26a0\ufe0f Незаменённые переменные ({len(unmatched)}): {', '.join(unmatched[:20])}{" ..." if len(unmatched) > 20 else ''}")
|
||||
if log_func and not unmatched:
|
||||
log_func("\u2705 Все переменные успешно заменены")
|
||||
return len(unmatched) == 0
|
||||
|
||||
# ==================== МИГРАЦИЯ СТАРЫХ ПЕРЕМЕННЫХ ====================
|
||||
|
||||
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 _merge_runs_for_loops(doc):
|
||||
"""Склеивает текст в runs, чтобы <!-- loop:x --> не был разбит на части."""
|
||||
ns = 'http://schemas.openxmlformats.org/wordprocessingml/2006/main'
|
||||
for para in doc.paragraphs:
|
||||
full = para.text
|
||||
if 'loop:' not in full and 'loop_end' not in full:
|
||||
continue
|
||||
# В этом параграфе есть маркер цикла — склеиваем все runs в один
|
||||
if len(para.runs) <= 1:
|
||||
continue
|
||||
# Проверяем, не разбит ли маркер
|
||||
for run in para.runs:
|
||||
if 'loop:' in run.text or 'loop_end' in run.text:
|
||||
# Маркер есть, но возможно разбит — склеиваем
|
||||
combined = para.text
|
||||
para.clear()
|
||||
para.add_run(combined)
|
||||
break
|
||||
|
||||
|
||||
|
||||
LOOP_FINDER = re.compile(r'<!--\s*loop:(\w+)\s*-->')
|
||||
LOOP_END_FINDER = re.compile(r'<!--\s*loop_end\s*-->')
|
||||
|
||||
def _expand_loops_in_template(template_path):
|
||||
"""Обрабатывает циклы на уровне XML документа до загрузки python-docx.
|
||||
Работает напрямую с word/document.xml в ZIP-архиве."""
|
||||
import tempfile
|
||||
|
||||
# Временная копия
|
||||
tmp = tempfile.NamedTemporaryFile(delete=False, suffix='.docx')
|
||||
tmp.close()
|
||||
shutil.copy2(template_path, tmp.name)
|
||||
|
||||
# Читаем document.xml
|
||||
import zipfile as zf
|
||||
with zf.ZipFile(tmp.name, 'r') as zin:
|
||||
doc_xml = zin.read('word/document.xml').decode('utf-8')
|
||||
|
||||
max_passes = 20
|
||||
for _ in range(max_passes):
|
||||
# Ищем все маркеры циклов
|
||||
starts = list(LOOP_FINDER.finditer(doc_xml))
|
||||
ends = list(LOOP_END_FINDER.finditer(doc_xml))
|
||||
|
||||
if not starts:
|
||||
break
|
||||
|
||||
# Внешний цикл = первый start
|
||||
outer = starts[0]
|
||||
key = outer.group(1)
|
||||
|
||||
# Соответствующий end = последний end, который после outer
|
||||
outer_end = None
|
||||
for e in reversed(ends):
|
||||
if e.start() > outer.start():
|
||||
outer_end = e
|
||||
break
|
||||
|
||||
if outer_end is None:
|
||||
break
|
||||
|
||||
# Извлекаем шаблон (текст между маркерами)
|
||||
template_start = outer.end()
|
||||
template_end = outer_end.start()
|
||||
template = doc_xml[template_start:template_end]
|
||||
|
||||
# Заменяем в шаблоне маркеры ключа (но не трогаем вложенные loop:)
|
||||
# Удаляем только start и end маркеры этого цикла
|
||||
# loop_end убираем только последний (он принадлежит внешнему циклу)
|
||||
# loop_start этого ключа убираем все вхождения
|
||||
|
||||
# Найдём и удалим ТОЛЬКО start маркер для этого ключа
|
||||
template = template.replace(f'<!-- loop:{key} -->', '', 1) # только первый
|
||||
|
||||
# Найдём и удалим ТОЛЬКО последний loop_end
|
||||
last_end = template.rfind('<!-- loop_end -->')
|
||||
if last_end >= 0:
|
||||
template = template[:last_end] + template[last_end + len('<!-- loop_end -->'):]
|
||||
|
||||
# Данные для цикла берём из replacements (передаются позже)
|
||||
# Пока просто помечаем позицию для последующей обработки
|
||||
# Вставляем placeholder для замены
|
||||
marker = f'__LOOP_{key}_EXPANDED__'
|
||||
|
||||
# Размножаем шаблон (1 копия, потом замена в process_template)
|
||||
expanded = ''
|
||||
for i in range(10): # макс 10, реально заменится позже
|
||||
expanded += template
|
||||
|
||||
# Собираем новый XML
|
||||
new_xml = doc_xml[:outer.start()] + '__LOOP_BLOCK__' + key + '__' + doc_xml[outer_end.end():]
|
||||
doc_xml = new_xml
|
||||
|
||||
# Сохраняем
|
||||
with zf.ZipFile(tmp.name, 'w', zf.ZIP_DEFLATED) as zout:
|
||||
zout.writestr('word/document.xml', doc_xml.encode('utf-8'))
|
||||
|
||||
return tmp.name
|
||||
|
||||
|
||||
def _process_loops(doc, loops: Dict[str, List[Dict]]):
|
||||
"""Обработка циклов через python-docx API.
|
||||
Находит loop:x / loop_end, клонирует параграфы, заменяет item поля."""
|
||||
LOOP_START = re.compile(r'<!--\s*loop:(\w+)\s*-->')
|
||||
LOOP_END = re.compile(r'<!--\s*loop_end\s*-->')
|
||||
|
||||
from lxml import etree as _etree
|
||||
|
||||
max_passes = 30
|
||||
for _ in range(max_passes):
|
||||
# Находим все маркеры
|
||||
markers = {} # idx -> (type, key)
|
||||
for i, para in enumerate(doc.paragraphs):
|
||||
text = para.text.strip()
|
||||
m = LOOP_START.search(text)
|
||||
if m:
|
||||
markers[i] = ('start', m.group(1))
|
||||
elif LOOP_END.search(text):
|
||||
markers[i] = ('end', None)
|
||||
|
||||
if not markers:
|
||||
break
|
||||
|
||||
# Внешний цикл = первый start
|
||||
positions = sorted(markers.keys())
|
||||
outer_idx = None
|
||||
outer_key = None
|
||||
for pos in positions:
|
||||
t, k = markers[pos]
|
||||
if t == 'start':
|
||||
outer_idx = pos
|
||||
outer_key = k
|
||||
break
|
||||
|
||||
if outer_idx is None:
|
||||
break
|
||||
|
||||
# Последний end после outer_idx
|
||||
outer_end = None
|
||||
for pos in reversed(positions):
|
||||
t, k = markers[pos]
|
||||
if t == 'end' and pos > outer_idx:
|
||||
outer_end = pos
|
||||
break
|
||||
|
||||
if outer_end is None:
|
||||
break
|
||||
|
||||
items = loops.get(outer_key, [])
|
||||
if not items:
|
||||
_remove_paras_range(doc, outer_idx, outer_end)
|
||||
continue
|
||||
|
||||
# Шаблонные параграфы (между маркерами)
|
||||
tpl_start = outer_idx + 1
|
||||
tpl_end = outer_end # не включая
|
||||
|
||||
if tpl_start >= tpl_end:
|
||||
# Удаляем только маркеры
|
||||
_remove_paras_range(doc, outer_idx, outer_end)
|
||||
continue
|
||||
|
||||
ref_para = doc.paragraphs[outer_end]
|
||||
|
||||
# Для каждого элемента данных клонируем шаблон
|
||||
for item_idx, item in enumerate(items, 1):
|
||||
for tpl_i in range(tpl_start, tpl_end):
|
||||
src = doc.paragraphs[tpl_i]
|
||||
# Клонируем элемент XML
|
||||
xml_str = _etree.tostring(src._element, encoding='unicode')
|
||||
new_el = _etree.fromstring(xml_str.encode('utf-8'))
|
||||
|
||||
# Заменяем {{item.*}} поля
|
||||
for field, value in item.items():
|
||||
search = '{{item.%s}}' % field
|
||||
replace = str(value) if value else EMPTY_MARKER
|
||||
_replace_in_xml_tree(new_el, search, replace)
|
||||
|
||||
_replace_in_xml_tree(new_el, '{{item.number}}', str(item_idx))
|
||||
_replace_in_xml_tree(new_el, '{{number}}', str(item_idx))
|
||||
|
||||
# Вставляем после ref_para
|
||||
ref_para._element.addnext(new_el)
|
||||
ref_para = doc.paragraphs[outer_end] # перепроверяем ссылку (она могла сместиться)
|
||||
|
||||
# Удаляем старый блок
|
||||
_remove_paras_range(doc, outer_idx, outer_end)
|
||||
|
||||
|
||||
def _replace_in_xml_tree(element, old, new):
|
||||
"""Заменяет текст внутри XML-элемента (в <w:t>)."""
|
||||
ns = '{http://schemas.openxmlformats.org/wordprocessingml/2006/main}'
|
||||
tag_t = ns + 't'
|
||||
if element.tag == tag_t or element.tag == 'w:t':
|
||||
if element.text and old in element.text:
|
||||
element.text = element.text.replace(old, new)
|
||||
return True
|
||||
changed = False
|
||||
for child in element:
|
||||
if _replace_in_xml_tree(child, old, new):
|
||||
changed = True
|
||||
return changed
|
||||
|
||||
|
||||
def _remove_paras_range(doc, start_idx, end_idx):
|
||||
"""Удаляет параграфы от start_idx до end_idx включительно."""
|
||||
for idx in range(end_idx, start_idx - 1, -1):
|
||||
if idx < len(doc.paragraphs):
|
||||
el = doc.paragraphs[idx]._element
|
||||
el.getparent().remove(el)
|
||||
|
||||
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)
|
||||
Reference in New Issue
Block a user