670 lines
32 KiB
Python
670 lines
32 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
Сервис обработки файлов: извлечение текста из различных форматов,
|
||
хирургическая замена в DOCX, создание DOCX из текста,
|
||
транскрипция аудио через SaluteSpeech.
|
||
Все методы, которые могут быть асинхронными, теперь async.
|
||
Добавлено экранирование путей через shlex.quote.
|
||
"""
|
||
|
||
import asyncio
|
||
import csv
|
||
import json
|
||
import logging
|
||
import os
|
||
import re
|
||
import shutil
|
||
import shlex
|
||
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
|
||
|
||
# Опциональные библиотеки (синхронные, обёрнуты в asyncio.to_thread при необходимости)
|
||
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
|
||
|
||
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}")
|
||
|
||
# ---------- Синхронные методы извлечения (вызываются через to_thread) ----------
|
||
def extract_text_from_docx(self, file_path: str) -> str:
|
||
"""Извлекает текст из DOCX, включая колонтитулы, таблицы и сноски."""
|
||
try:
|
||
doc = DocxReader(file_path)
|
||
full_text = []
|
||
|
||
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:
|
||
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:
|
||
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:
|
||
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:
|
||
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:
|
||
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:
|
||
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:
|
||
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.
|
||
Пути экранируются через shlex.quote.
|
||
"""
|
||
import uuid
|
||
|
||
token = await self._get_salute_token()
|
||
if not token:
|
||
return "[Ошибка авторизации в SaluteSpeech]"
|
||
|
||
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]"
|
||
|
||
# Экранируем пути
|
||
safe_file_path = shlex.quote(file_path)
|
||
pcm_output = file_path + ".raw"
|
||
safe_pcm_output = shlex.quote(pcm_output)
|
||
|
||
try:
|
||
# Конвертация ffmpeg
|
||
def _run_ffmpeg():
|
||
cmd_ffmpeg = [
|
||
ffmpeg_bin, '-y', '-i', safe_file_path,
|
||
'-ac', '1', '-ar', '16000', '-f', 's16le', '-filter:a', 'volume=2.0', safe_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 "[Ошибка: пустой аудиофайл после конвертации]"
|
||
|
||
# Отправка запроса к SaluteSpeech через curl
|
||
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'@{safe_pcm_output}'
|
||
]
|
||
return subprocess.run(curl_cmd, capture_output=True, text=True)
|
||
|
||
result_curl = await asyncio.to_thread(_run_curl)
|
||
|
||
if os.path.exists(pcm_output):
|
||
os.remove(pcm_output)
|
||
|
||
if result_curl.returncode != 0:
|
||
return f"[Ошибка сетевого соединения (curl): {result_curl.returncode}]"
|
||
|
||
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:
|
||
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 и рекурсивно обрабатывает каждый файл.
|
||
Возвращает список (имя_файла, извлечённый_текст) и общее количество обработанных файлов.
|
||
Этот метод синхронный, вызывается через to_thread.
|
||
"""
|
||
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:
|
||
if file_path.endswith('.zip'):
|
||
with zipfile.ZipFile(file_path, 'r') as z:
|
||
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
|
||
|
||
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))
|
||
|
||
# Для обработки каждого файла внутри архива используем синхронный метод,
|
||
# но мы не можем вызвать асинхронный process_any_file здесь.
|
||
# Поэтому мы вызываем синхронные методы напрямую, а для аудио не поддерживаем.
|
||
# Вместо этого мы просто извлекаем текст синхронно.
|
||
# В будущем можно переписать process_archive асинхронно.
|
||
text, count = self._process_any_file_sync(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
|
||
|
||
elif file_path.endswith('.7z') and HAS_7Z:
|
||
with py7zr.SevenZipFile(file_path, 'r') as z:
|
||
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_sync(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:
|
||
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_sync(self, file_path: str) -> Tuple[str, int]:
|
||
"""
|
||
Синхронная версия для использования внутри архива.
|
||
Определяет тип файла по расширению и вызывает соответствующий синхронный метод.
|
||
Не обрабатывает аудио (транскрибация) и изображения (OCR) внутри архива.
|
||
"""
|
||
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 == '.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
|
||
|
||
# ============================================================
|
||
# ОСНОВНОЙ МЕТОД
|
||
# ============================================================
|
||
|
||
async def process_any_file(self, file_path: str) -> Tuple[str, int]:
|
||
"""
|
||
Главный асинхронный метод: определяет тип файла по расширению и вызывает соответствующий
|
||
метод извлечения текста. Для аудио использует transcribe_audio (асинхронно).
|
||
Для синхронных операций использует asyncio.to_thread.
|
||
Возвращает (текст, количество_файлов_внутри).
|
||
Для архивов возвращает список файлов и общее количество.
|
||
"""
|
||
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 = await asyncio.to_thread(self.extract_text_from_docx, file_path)
|
||
count = 1 if text else 0
|
||
elif ext == '.pdf':
|
||
text = await asyncio.to_thread(self.extract_text_from_pdf, file_path)
|
||
count = 1 if text else 0
|
||
elif ext == '.xlsx':
|
||
text = await asyncio.to_thread(self.extract_text_from_xlsx, file_path)
|
||
count = 1 if text else 0
|
||
elif ext == '.csv':
|
||
text = await asyncio.to_thread(self.extract_text_from_csv, file_path)
|
||
count = 1 if text else 0
|
||
elif ext == '.txt':
|
||
text = await asyncio.to_thread(self.extract_text_from_txt, file_path)
|
||
count = 1 if text else 0
|
||
elif ext == '.json':
|
||
text = await asyncio.to_thread(self.extract_text_from_json, file_path)
|
||
count = 1 if text else 0
|
||
elif ext in ('.jpg', '.jpeg', '.png', '.bmp', '.tiff'):
|
||
text = await asyncio.to_thread(self.extract_text_from_image, file_path)
|
||
count = 1 if text else 0
|
||
elif ext in ('.ogg', '.wav', '.mp3', '.amr', '.m4a'):
|
||
# Асинхронный метод транскрибации – используем await!
|
||
text = await self.transcribe_audio(file_path)
|
||
count = 1 if text else 0
|
||
elif ext in ('.zip', '.7z'):
|
||
# Архив обрабатывается синхронно в отдельном потоке
|
||
result = await asyncio.to_thread(self.process_archive, file_path)
|
||
return result # (список, количество)
|
||
elif ext == '.pptx':
|
||
text = await asyncio.to_thread(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]:
|
||
# (код без изменений, полная версия ранее)
|
||
# Я приведу его целиком для целостности.
|
||
import re
|
||
from datetime import datetime
|
||
from docx import Document
|
||
|
||
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}. Выполняю простую замену.")
|
||
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]:
|
||
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 |