markov/fckbot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Update 86 files

Browse Source 
- /rag/commands/expert.py
- /core/services/__init__.py
- /core/services/embedding_service.py
- /core/services/file_service.py
- /core/services/giga_client.py
- /core/services/kb_service.py
- /core/services/postgres_service.py
- /core/services/qdrant_service.py
- /core/services/reranker_service.py
- /core/functions/__init__.py
- /core/functions/check_consistency.py
- /core/functions/check_spelling.py
- /core/functions/critique_answer.py
- /core/functions/expand_query.py
- /core/functions/extract_metrics.py
- /core/functions/file_processor.py
- /core/functions/generate_document.py
- /core/functions/intent_classify.py
- /core/functions/rerank_context.py
- /core/functions/summarize_document.py
- /core/utils/__init__.py
- /core/utils/arg_parser.py
- /core/utils/config_loader.py
- /core/utils/layout_converter.py
- /core/utils/logger.py
- /core/utils/text_utils.py
- /core/utils/web_utils.py
- /bots/commands/__init__.py
- /bots/commands/base.py
- /bots/commands/create.py
- /core/commands/global_remove.py
- /core/commands/help.py
- /core/commands/info.py
- /core/commands/kb.py
- /core/commands/learn.py
- /core/commands/other.py
- /core/commands/registry.py
- /core/commands/stats.py
- /core/commands/template.py
- /core/handlers/__init__.py
- /core/handlers/message_handler.py
- /core/workers/__init__.py
- /core/workers/indexing_worker.py
- /core/xmpp/__init__.py
- /core/xmpp/client.py
- /bots/commands/global_remove.py
- /bots/commands/help.py
- /bots/commands/info.py
- /bots/commands/kb.py
- /bots/commands/registry.py
- /bots/commands/other.py
- /bots/commands/template.py
- /bots/commands/learn.py
- /bots/commands/stats.py
- /bots/handlers/message_handler.py
- /bots/handlers/__init__.py
- /bots/workers/__init__.py
- /bots/workers/indexing_worker.py
- /bots/xmpp/__init__.py
- /bots/xmpp/client.py
- /rag/services/qdrant_service.py
- /rag/services/giga_client.py
- /rag/services/embedding_service.py
- /rag/services/reranker_service.py
- /rag/services/__init__.py
- /rag/services/file_service.py
- /rag/services/postgres_service.py
- /rag/services/kb_service.py
- /rag/functions/check_spelling.py
- /rag/functions/__init__.py
- /rag/functions/extract_metrics.py
- /rag/functions/generate_document.py
- /rag/functions/expand_query.py
- /rag/functions/critique_answer.py
- /rag/functions/file_processor.py
- /rag/functions/check_consistency.py
- /rag/functions/summarize_document.py
- /rag/functions/rerank_context.py
- /rag/functions/intent_classify.py
- /rag/utils/__init__.py
- /rag/utils/layout_converter.py
- /rag/utils/text_utils.py
- /rag/utils/arg_parser.py
- /rag/utils/config_loader.py
- /rag/utils/logger.py
- /rag/utils/web_utils.py
This commit is contained in:
Markov Andrey
2026-06-30 10:33:28 +00:00
parent ea0d42de9c
commit 63900feece
45 changed files with 0 additions and 265 deletions
Download Patch File Download Diff File Expand all files Collapse all files

658
rag/services/file_service.py Normal file
View File

@@ -0,0 +1,658 @@
# -*- coding: utf-8 -*-
"""
Сервис обработки файлов: извлечение текста из различных форматов,
хирургическая замена в DOCX, создание DOCX из текста,
транскрипция аудио через SaluteSpeech.
Все методы синхронные (используются в asyncio.to_thread).
"""
import asyncio
import csv
import json
import logging
import os
import re
import shutil
import subprocess
import tempfile
import zipfile
from datetime import datetime
from typing import List, Tuple, Optional, Dict, Any
import requests
from docx import Document as DocxReader
from docx import Document as DocxWriter
from docx.shared import Pt
# Опциональные библиотеки
try:
from pypdf import PdfReader
HAS_PDF = True
except ImportError:
HAS_PDF = False
try:
import openpyxl
HAS_EXCEL = True
except ImportError:
HAS_EXCEL = False
try:
import pytesseract
from PIL import Image
HAS_OCR = True
except ImportError:
HAS_OCR = False
try:
import py7zr
HAS_7Z = True
except ImportError:
HAS_7Z = False
try:
from pptx import Presentation
HAS_PPTX = True
except ImportError:
HAS_PPTX = False
# Для чанкинга используем tiktoken, но здесь оставим только извлечение текста,
# а чанкинг вынесен в core/utils/text_utils.py
from core.utils.config_loader import BotConfig
logger = logging.getLogger(__name__)
class FileService:
def __init__(self, config: BotConfig):
self.config = config
self.temp_dir = config.temp_dir
self.temp_dir.mkdir(parents=True, exist_ok=True)
def cleanup_old_files(self, days: int = 3):
"""Удаляет старые файлы из temp_dir."""
now = datetime.now().timestamp()
for item in self.temp_dir.iterdir():
if item.is_file() and (now - item.stat().st_mtime) > days * 86400:
try:
item.unlink()
logger.info(f"Удалён старый файл: {item.name}")
except Exception as e:
logger.error(f"Ошибка удаления {item.name}: {e}")
# ---------- Извлечение текста из разных форматов ----------
def extract_text_from_docx(self, file_path: str) -> str:
"""
Извлекает текст из DOCX, включая колонтитулы, таблицы и сноски.
Возвращает текст, разделённый переносами строк.
"""
try:
doc = DocxReader(file_path)
full_text = []
# Вспомогательная функция для извлечения текста из XML-элемента
def get_xml_text(element):
text_nodes = element.xpath('.//*[local-name()="t"]')
return "".join([node.text for node in text_nodes if node.text])
# Колонтитулы
for section in doc.sections:
for header in [section.header, section.first_page_header, section.even_page_header]:
if header:
h_text = get_xml_text(header._element)
if h_text.strip():
full_text.append(f"[КОЛОНТИТУЛ ВЕРХНИЙ]: {h_text}")
for footer in [section.footer, section.first_page_footer, section.even_page_footer]:
if footer:
f_text = get_xml_text(footer._element)
if f_text.strip():
full_text.append(f"[КОЛОНТИТУЛ НИЖНИЙ]: {f_text}")
# Тело документа
for block in doc.element.body:
if block.tag.endswith('p'):
text = get_xml_text(block)
if text.strip():
full_text.append(text)
elif block.tag.endswith('tbl'):
from docx.table import Table
t = Table(block, doc)
full_text.append("\n[ТАБЛИЦА]")
for row in t.rows:
row_data = [get_xml_text(cell._element).strip().replace('\n', ' ') for cell in row.cells]
# Удаляем дублирование, если ячейки пустые
clean_row = []
for item in row_data:
if not clean_row or item != clean_row[-1]:
clean_row.append(item)
full_text.append(" | ".join(clean_row))
full_text.append("[КОНЕЦ ТАБЛИЦЫ]\n")
# Сноски
try:
for footnote in doc.footnotes:
fn_text = get_xml_text(footnote._element)
if fn_text.strip():
full_text.append(f"[СНОСКА]: {fn_text}")
except:
pass
return "\n".join(full_text)
except Exception as e:
logger.error(f"Ошибка чтения DOCX {file_path}: {e}")
return ""
def extract_text_from_pdf(self, file_path: str) -> str:
"""Извлекает текст из PDF через pypdf."""
if not HAS_PDF:
return "[Ошибка: библиотека pypdf не установлена]"
try:
reader = PdfReader(file_path)
pages = []
for i, page in enumerate(reader.pages):
text = page.extract_text()
if text.strip():
pages.append(f"--- Страница {i+1} ---\n{text}")
return "\n".join(pages)
except Exception as e:
logger.error(f"Ошибка чтения PDF {file_path}: {e}")
return ""
def extract_text_from_xlsx(self, file_path: str) -> str:
"""Извлекает текст из Excel (XLSX) через openpyxl."""
if not HAS_EXCEL:
return "[Ошибка: библиотека openpyxl не установлена]"
try:
wb = openpyxl.load_workbook(file_path, data_only=True)
output = []
for sheet in wb.sheetnames:
ws = wb[sheet]
output.append(f"\n--- ЛИСТ: {sheet} ---")
for row in ws.iter_rows(values_only=True):
if any(c is not None for c in row):
output.append(" | ".join(str(c) if c is not None else "" for c in row))
return "\n".join(output)
except Exception as e:
logger.error(f"Ошибка чтения XLSX {file_path}: {e}")
return ""
def extract_text_from_csv(self, file_path: str) -> str:
"""Извлекает текст из CSV."""
try:
with open(file_path, 'r', encoding='utf-8', errors='ignore') as f:
reader = csv.reader(f)
return "\n".join("\t".join(row) for row in reader)
except Exception as e:
logger.error(f"Ошибка чтения CSV {file_path}: {e}")
return ""
def extract_text_from_txt(self, file_path: str) -> str:
"""Извлекает текст из TXT."""
try:
with open(file_path, 'r', encoding='utf-8', errors='ignore') as f:
return f.read()
except Exception as e:
logger.error(f"Ошибка чтения TXT {file_path}: {e}")
return ""
def extract_text_from_json(self, file_path: str) -> str:
"""Извлекает текст из JSON (дамп в виде строки)."""
try:
with open(file_path, 'r', encoding='utf-8') as f:
data = json.load(f)
return json.dumps(data, ensure_ascii=False, indent=2)
except Exception as e:
logger.error(f"Ошибка чтения JSON {file_path}: {e}")
return ""
def extract_text_from_image(self, file_path: str) -> str:
"""Извлекает текст из изображения через OCR (Tesseract)."""
if not HAS_OCR:
return ""
try:
img = Image.open(file_path)
text = pytesseract.image_to_string(img, lang='rus+eng')
return f"--- ТЕКСТ С ИЗОБРАЖЕНИЯ ---\n{text}" if text.strip() else ""
except Exception as e:
logger.debug(f"OCR не удался: {e}")
return ""
def extract_text_from_pptx(self, file_path: str) -> str:
"""
Извлекает текст из презентации .pptx:
- текст всех слайдов (из текстовых полей, таблиц)
- заметки к слайдам
Возвращает строку с разделением по слайдам.
"""
if not HAS_PPTX:
return "[Ошибка: библиотека python-pptx не установлена]"
try:
prs = Presentation(file_path)
full_text = []
for slide_num, slide in enumerate(prs.slides, 1):
slide_content = []
# Текст из фигур (надписи, автофигуры)
for shape in slide.shapes:
if hasattr(shape, "text") and shape.text:
slide_content.append(shape.text.strip())
# Таблицы
if shape.has_table:
for row in shape.table.rows:
row_text = [cell.text.strip() for cell in row.cells]
slide_content.append(" | ".join(row_text))
# Заметки к слайду
if slide.has_notes_slide and slide.notes_slide.notes_text_frame:
notes = slide.notes_slide.notes_text_frame.text.strip()
if notes:
slide_content.append(f"[ЗАМЕТКИ]: {notes}")
if slide_content:
full_text.append(f"--- Слайд {slide_num} ---\n" + "\n".join(slide_content))
return "\n\n".join(full_text)
except Exception as e:
logger.error(f"Ошибка чтения PPTX {file_path}: {e}")
return ""
async def transcribe_audio(self, file_path: str) -> str:
"""
Асинхронно транскрибирует аудиофайл через SaluteSpeech API.
Все блокирующие вызовы (subprocess.run, requests.post) обёрнуты в asyncio.to_thread.
Требует наличия ffmpeg и переменной окружения SALUTE_SPEECH_AUTH.
"""
import uuid
# 1. Получаем токен асинхронно
token = await self._get_salute_token()
if not token:
return "[Ошибка авторизации в SaluteSpeech]"
# 2. Проверяем наличие ffmpeg
ffmpeg_bin = shutil.which('ffmpeg') or '/usr/local/bin/ffmpeg'
curl_bin = shutil.which('curl') or '/usr/local/bin/curl'
if not shutil.which('ffmpeg') and not os.path.exists(ffmpeg_bin):
return "[Ошибка: на сервере не установлен ffmpeg]"
pcm_output = file_path + ".raw"
try:
# ---- ОБЁРТКА: конвертация ffmpeg (subprocess.run) ----
def _run_ffmpeg():
cmd_ffmpeg = [
ffmpeg_bin, '-y', '-i', file_path,
'-ac', '1', '-ar', '16000', '-f', 's16le', '-filter:a', 'volume=2.0', pcm_output
]
return subprocess.run(cmd_ffmpeg, capture_output=True)
result = await asyncio.to_thread(_run_ffmpeg)
if result.returncode != 0:
err_msg = result.stderr.decode()
logger.error(f"FFmpeg ошибка: {err_msg}")
return f"[Ошибка конвертации аудио: {err_msg[:100]}]"
if not os.path.exists(pcm_output) or os.path.getsize(pcm_output) == 0:
return "[Ошибка: пустой аудиофайл после конвертации]"
# 3. Отправка запроса к SaluteSpeech через curl (subprocess.run)
url = "https://smartspeech.sber.ru/rest/v1/speech:recognize"
def _run_curl():
curl_cmd = [
curl_bin, '-s', '-k', '--http1.1', '-X', 'POST', url,
'-H', f'Authorization: Bearer {token}',
'-H', 'Content-Type: audio/x-pcm;bit=16;rate=16000',
'-H', f'RqUID: {str(uuid.uuid4())}',
'--data-binary', f'@{pcm_output}'
]
return subprocess.run(curl_cmd, capture_output=True, text=True)
result_curl = await asyncio.to_thread(_run_curl)
# Удаляем временный PCM-файл
if os.path.exists(pcm_output):
os.remove(pcm_output)
if result_curl.returncode != 0:
return f"[Ошибка сетевого соединения (curl): {result_curl.returncode}]"
# 4. Парсинг ответа (синхронная операция – быстрая, можно оставить без обёртки)
try:
response_data = json.loads(result_curl.stdout)
if 'result' in response_data:
if response_data['result']:
text = " ".join(response_data['result'])
return f"--- РАСПОЗНАННЫЙ ГОЛОС ---\n{text}"
else:
return "😶 Тишина на записи."
error_msg = response_data.get('message', 'Неизвестная ошибка API')
return f"--- ОШИБКА ГОЛОСА ---\n[Голос не распознан: {error_msg}]"
except Exception as e:
logger.error(f"Ошибка парсинга JSON: {result_curl.stdout}")
return f"[Ошибка парсинга ответа: {e}]"
except Exception as e:
logger.error(f"Ошибка в transcribe_audio: {e}")
return f"[Ошибка распознавания: {e}]"
async def _get_salute_token(self) -> Optional[str]:
"""
Асинхронно получает токен доступа к SaluteSpeech по Basic авторизации.
Блокирующий вызов requests.post обёрнут в asyncio.to_thread.
"""
import uuid
salute_auth = self.config.salute_speech_auth
if not salute_auth:
logger.error("Переменная окружения SALUTE_SPEECH_AUTH не задана")
return None
url = "https://ngw.devices.sberbank.ru.:9443/api/v2/oauth"
headers = {
'Content-Type': 'application/x-www-form-urlencoded',
'RqUID': str(uuid.uuid4()),
'Authorization': f'Basic {salute_auth}'
}
payload = {'scope': 'SALUTE_SPEECH_PERS'}
try:
# ---- ОБЁРТКА: синхронный requests.post выполняется в отдельном потоке ----
def _sync_request():
return requests.post(url, headers=headers, data=payload, verify=False)
response = await asyncio.to_thread(_sync_request)
token = response.json().get('access_token')
logger.info("SaluteSpeech token получен")
return token
except Exception as e:
logger.error(f"Ошибка получения токена SaluteSpeech: {e}")
return None
def process_archive(self, file_path: str) -> Tuple[List[Tuple[str, str]], int]:
"""
Распаковывает ZIP/7z и рекурсивно обрабатывает каждый файл.
Возвращает список (имя_файла, извлечённый_текст) и общее количество обработанных файлов.
"""
results = []
total_count = 0
extract_path = self.temp_dir / f"ext_{int(datetime.now().timestamp()*1000000)}"
extract_path.mkdir(parents=True, exist_ok=True)
max_total_bytes = self.config.max_file_size_mb * 2 * 1024 * 1024 # для архивов лимит удвоен
MAX_FILES = self.config.max_archive_files
total_files = 0
total_size = 0
try:
# ---------- ZIP ----------
if file_path.endswith('.zip'):
with zipfile.ZipFile(file_path, 'r') as z:
# 1. Предварительный подсчёт суммарного несжатого размера всех файлов в архиве
total_uncompressed = 0
for info in z.infolist():
if not info.is_dir():
total_uncompressed += info.file_size
if total_uncompressed > max_total_bytes:
logger.warning(
f"Архив {file_path} превышает лимит {max_total_bytes // (1024*1024)} МБ "
f"(несжатый размер {total_uncompressed // (1024*1024)} МБ)"
)
return [], 0
# 2. Распаковка и обработка
for info in z.infolist():
if info.is_dir():
continue
total_files += 1
if total_files > MAX_FILES:
logger.warning(f"Архив {file_path} содержит более {MAX_FILES} файлов, обработка прервана")
return [], 0
# Имя файла в правильной кодировке
try:
fname = info.filename.encode('cp437').decode('utf-8')
except:
try:
fname = info.filename.encode('cp437').decode('cp866')
except:
fname = info.filename
target = extract_path / os.path.basename(fname)
with open(target, 'wb') as f:
f.write(z.read(info))
text, count = self.process_any_file(str(target))
if text:
size = len(text.encode('utf-8'))
if total_size + size > max_total_bytes:
logger.warning(f"Архив {file_path} превышает лимит {self.config.max_file_size_mb*2} МБ")
return [], 0
results.append((fname, text))
total_count += count
total_size += size
# ---------- 7Z ----------
elif file_path.endswith('.7z') and HAS_7Z:
with py7zr.SevenZipFile(file_path, 'r') as z:
# Для 7z нет простого способа получить суммарный размер до распаковки,
# поэтому распаковываем с контролем в процессе
z.extractall(extract_path)
for root, dirs, files in os.walk(extract_path):
for f in files:
total_files += 1
if total_files > MAX_FILES:
logger.warning(f"Архив {file_path} содержит более {MAX_FILES} файлов, обработка прервана")
return [], 0
fpath = os.path.join(root, f)
text, count = self.process_any_file(fpath)
if text:
size = len(text.encode('utf-8'))
if total_size + size > max_total_bytes:
logger.warning(f"Архив {file_path} превышает лимит {self.config.max_file_size_mb*2} МБ")
return [], 0
results.append((f, text))
total_count += count
total_size += size
else:
# Неподдерживаемый формат архива (но по логике сюда не должны попасть)
logger.warning(f"Неподдерживаемый тип архива: {file_path}")
return [], 0
except Exception as e:
logger.error(f"Ошибка распаковки архива {file_path}: {e}")
return [], 0
finally:
shutil.rmtree(extract_path, ignore_errors=True)
return results, total_count
def process_any_file(self, file_path: str) -> Tuple[str, int]:
"""
Главный метод: определяет тип файла по расширению и вызывает соответствующий
метод извлечения текста. Возвращает (текст, количество_файлов_внутри).
Для архивов возвращает список файлов и общее количество.
"""
ext = os.path.splitext(file_path)[1].lower()
text = ""
count = 0
try:
size = os.path.getsize(file_path)
max_size = self.config.max_file_size_mb * 1024 * 1024
if size > max_size:
logger.warning(f"Файл {file_path} превышает лимит {self.config.max_file_size_mb} МБ")
return f"[Ошибка: файл превышает максимальный размер {self.config.max_file_size_mb} МБ]", 0
if ext == '.docx':
text = self.extract_text_from_docx(file_path)
count = 1 if text else 0
elif ext == '.pdf':
text = self.extract_text_from_pdf(file_path)
count = 1 if text else 0
elif ext == '.xlsx':
text = self.extract_text_from_xlsx(file_path)
count = 1 if text else 0
elif ext == '.csv':
text = self.extract_text_from_csv(file_path)
count = 1 if text else 0
elif ext == '.txt':
text = self.extract_text_from_txt(file_path)
count = 1 if text else 0
elif ext == '.json':
text = self.extract_text_from_json(file_path)
count = 1 if text else 0
elif ext in ('.jpg', '.jpeg', '.png', '.bmp', '.tiff'):
text = self.extract_text_from_image(file_path)
count = 1 if text else 0
elif ext in ('.ogg', '.wav', '.mp3', '.amr', '.m4a'):
# Асинхронный метод transcribe_audio запускаем синхронно через get_event_loop
# (так как process_any_file выполняется в отдельном потоке)
try:
loop = asyncio.get_running_loop()
except RuntimeError:
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
text = loop.run_until_complete(self.transcribe_audio(file_path))
count = 1 if text else 0
elif ext in ('.zip', '.7z'):
return self.process_archive(file_path)
elif ext == '.pptx':
text = self.extract_text_from_pptx(file_path)
count = 1 if text else 0
else:
logger.warning(f"Неподдерживаемый тип файла: {ext}")
return "", 0
except Exception as e:
logger.error(f"Ошибка обработки файла {file_path}: {e}")
return "", 0
return text, count
def surgical_replace(self, file_path: str, replacements: Dict[str, str]) -> Optional[str]:
"""
Умная замена слов в DOCX с учётом грамматики (падеж, число, род).
Сохраняет форматирование. При ошибке выполняет простую замену.
"""
import re
from datetime import datetime
from docx import Document
# ---- Попытка умной замены через mawo-pymorphy3 ----
try:
from mawo_pymorphy3 import create_analyzer
morph = create_analyzer()
logger.info("Морфологический анализатор загружен")
except Exception as e:
logger.warning(f"Не удалось загрузить mawo-pymorphy3: {e}. Выполняю простую замену.")
# Простая замена
try:
doc = Document(file_path)
new_path = self.temp_dir / f"surgical_{int(datetime.now().timestamp())}.docx"
for paragraph in doc.paragraphs:
for run in paragraph.runs:
for old, new in replacements.items():
if old in run.text:
run.text = run.text.replace(old, new)
doc.save(str(new_path))
logger.info(f"Простая замена выполнена: {new_path}")
return str(new_path)
except Exception as e2:
logger.error(f"Ошибка простой замены: {e2}")
return None
# ---- Умная замена ----
try:
doc = Document(file_path)
logger.info(f"Документ загружен, параграфов: {len(doc.paragraphs)}")
# Кэш для разобранных слов
parse_cache = {}
# Предвычисление форм старых слов
old_forms = {}
for old_word, new_word in replacements.items():
old_low = old_word.lower()
parsed = morph.parse(old_low)[0]
forms = {parsed.normal_form}
cases = ['nomn', 'gent', 'datv', 'accs', 'ablt', 'loct']
numbers = ['sing', 'plur']
for number in numbers:
for case in cases:
inflected = parsed.inflect({case, number})
if inflected:
forms.add(inflected.word)
old_forms[old_low] = (new_word, forms)
def get_new_form(new_word, old_form_text):
if old_form_text in parse_cache:
parsed_old = parse_cache[old_form_text]
else:
parsed_old = morph.parse(old_form_text)[0]
parse_cache[old_form_text] = parsed_old
tags = set()
if parsed_old.tag.case:
tags.add(parsed_old.tag.case)
if parsed_old.tag.number:
tags.add(parsed_old.tag.number)
if parsed_old.tag.gender:
tags.add(parsed_old.tag.gender)
parsed_new = morph.parse(new_word)[0]
inflected = parsed_new.inflect(tags)
return inflected.word if inflected else new_word
total_runs = 0
modified_runs = 0
for paragraph in doc.paragraphs:
for run in paragraph.runs:
total_runs += 1
original = run.text
if not original:
continue
tokens = re.split(r'(\W+)', original)
modified = False
for i, token in enumerate(tokens):
if token.isalpha():
token_low = token.lower()
for old_low, (new_word, forms) in old_forms.items():
if token_low in forms:
new_form = get_new_form(new_word, token)
tokens[i] = new_form
modified = True
break
if modified:
run.text = ''.join(tokens)
modified_runs += 1
logger.info(f"Умная замена: обработано {total_runs} runs, изменено {modified_runs}")
new_path = self.temp_dir / f"surgical_{int(datetime.now().timestamp())}.docx"
doc.save(str(new_path))
logger.info(f"Умная замена выполнена: {new_path}")
return str(new_path)
except Exception as e:
logger.error(f"Ошибка при умной замене: {e}. Выполняю простую замену.")
# Простая замена (fallback)
try:
doc = Document(file_path)
new_path = self.temp_dir / f"surgical_{int(datetime.now().timestamp())}.docx"
for paragraph in doc.paragraphs:
for run in paragraph.runs:
for old, new in replacements.items():
if old in run.text:
run.text = run.text.replace(old, new)
doc.save(str(new_path))
logger.info(f"Простая замена выполнена: {new_path}")
return str(new_path)
except Exception as e2:
logger.error(f"Ошибка простой замены: {e2}")
return None
def create_docx(self, content: str, title: str, user_name: str, mode: str = "CREATE") -> Optional[str]:
"""
Создаёт новый DOCX из текста и сохраняет во временную папку.
Возвращает путь к созданному файлу или None.
"""
try:
fname = f"output_{user_name}_{int(datetime.now().timestamp())}.docx"
target = self.temp_dir / fname
doc = DocxWriter()
doc.add_heading(title, 0)
for paragraph in content.split('\n'):
if paragraph.strip():
p = doc.add_paragraph(paragraph.strip())
p.runs[0].font.size = Pt(11)
doc.save(str(target))
return str(target)
except Exception as e:
logger.error(f"Ошибка создания DOCX: {e}")
return None