Add new file
This commit is contained in:
Markov Andrey
658
core/services/file_service.py
Normal file
658
core/services/file_service.py
Normal 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
|
||||
Reference in New Issue
Block a user