Update 2 files

- /rag/services/file_service.py
- /rag/services/embedding_service.py
This commit is contained in:
Markov Andrey
2026-06-30 14:45:46 +00:00
parent ab8cac768b
commit c00af40977
2 changed files with 114 additions and 117 deletions

View File

@@ -3,7 +3,7 @@
Сервис обработки файлов: извлечение текста из различных форматов,
хирургическая замена в DOCX, создание DOCX из текста,
транскрипция аудио через SaluteSpeech.
Все методы синхронные (используются в asyncio.to_thread).
Все методы, которые могут быть асинхронными, теперь async.
"""
import asyncio
@@ -24,7 +24,7 @@ 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
@@ -56,8 +56,6 @@ try:
except ImportError:
HAS_PPTX = False
# Для чанкинга используем tiktoken, но здесь оставим только извлечение текста,
# а чанкинг вынесен в core/utils/text_utils.py
from core.utils.config_loader import BotConfig
logger = logging.getLogger(__name__)
@@ -70,7 +68,7 @@ class FileService:
self.temp_dir.mkdir(parents=True, exist_ok=True)
def cleanup_old_files(self, days: int = 3):
"""Удаляет старые файлы из temp_dir."""
"""Удаляет старые файлы из 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:
@@ -80,22 +78,19 @@ class FileService:
except Exception as e:
logger.error(f"Ошибка удаления {item.name}: {e}")
# ---------- Извлечение текста из разных форматов ----------
# ---------- Синхронные методы извлечения (вызываются через to_thread) ----------
def extract_text_from_docx(self, file_path: str) -> str:
"""
Извлекает текст из DOCX, включая колонтитулы, таблицы и сноски.
Возвращает текст, разделённый переносами строк.
"""
"""Извлекает текст из 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:
@@ -108,7 +103,6 @@ class FileService:
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)
@@ -120,7 +114,6 @@ class FileService:
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]:
@@ -128,7 +121,6 @@ class FileService:
full_text.append(" | ".join(clean_row))
full_text.append("[КОНЕЦ ТАБЛИЦЫ]\n")
# Сноски
try:
for footnote in doc.footnotes:
fn_text = get_xml_text(footnote._element)
@@ -143,7 +135,6 @@ class FileService:
return ""
def extract_text_from_pdf(self, file_path: str) -> str:
"""Извлекает текст из PDF через pypdf."""
if not HAS_PDF:
return "[Ошибка: библиотека pypdf не установлена]"
try:
@@ -159,7 +150,6 @@ class FileService:
return ""
def extract_text_from_xlsx(self, file_path: str) -> str:
"""Извлекает текст из Excel (XLSX) через openpyxl."""
if not HAS_EXCEL:
return "[Ошибка: библиотека openpyxl не установлена]"
try:
@@ -177,7 +167,6 @@ class FileService:
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)
@@ -187,7 +176,6 @@ class FileService:
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()
@@ -196,7 +184,6 @@ class FileService:
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)
@@ -206,7 +193,6 @@ class FileService:
return ""
def extract_text_from_image(self, file_path: str) -> str:
"""Извлекает текст из изображения через OCR (Tesseract)."""
if not HAS_OCR:
return ""
try:
@@ -218,12 +204,6 @@ class FileService:
return ""
def extract_text_from_pptx(self, file_path: str) -> str:
"""
Извлекает текст из презентации .pptx:
- текст всех слайдов (из текстовых полей, таблиц)
- заметки к слайдам
Возвращает строку с разделением по слайдам.
"""
if not HAS_PPTX:
return "[Ошибка: библиотека python-pptx не установлена]"
try:
@@ -231,16 +211,13 @@ class FileService:
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:
@@ -252,48 +229,47 @@ class FileService:
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) ----
# Конвертация ffmpeg
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)
# Отправка запроса к 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,
@@ -303,17 +279,15 @@ class FileService:
'--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:
@@ -327,16 +301,12 @@ class FileService:
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:
@@ -350,7 +320,6 @@ class FileService:
}
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)
@@ -365,22 +334,21 @@ class FileService:
"""
Распаковывает 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_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():
@@ -392,7 +360,6 @@ class FileService:
)
return [], 0
# 2. Распаковка и обработка
for info in z.infolist():
if info.is_dir():
continue
@@ -401,7 +368,6 @@ class FileService:
logger.warning(f"Архив {file_path} содержит более {MAX_FILES} файлов, обработка прервана")
return [], 0
# Имя файла в правильной кодировке
try:
fname = info.filename.encode('cp437').decode('utf-8')
except:
@@ -414,7 +380,12 @@ class FileService:
with open(target, 'wb') as f:
f.write(z.read(info))
text, count = self.process_any_file(str(target))
# Для обработки каждого файла внутри архива используем синхронный метод,
# но мы не можем вызвать асинхронный 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:
@@ -424,11 +395,8 @@ class FileService:
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:
@@ -437,7 +405,7 @@ class FileService:
logger.warning(f"Архив {file_path} содержит более {MAX_FILES} файлов, обработка прервана")
return [], 0
fpath = os.path.join(root, f)
text, count = self.process_any_file(fpath)
text, count = self._process_any_file_sync(fpath)
if text:
size = len(text.encode('utf-8'))
if total_size + size > max_total_bytes:
@@ -447,8 +415,6 @@ class FileService:
total_count += count
total_size += size
else:
# Неподдерживаемый формат архива (но по логике сюда не должны попасть)
logger.warning(f"Неподдерживаемый тип архива: {file_path}")
return [], 0
except Exception as e:
@@ -459,11 +425,11 @@ class FileService:
return results, total_count
def process_any_file(self, file_path: str) -> Tuple[str, int]:
def _process_any_file_sync(self, file_path: str) -> Tuple[str, int]:
"""
Главный метод: определяет тип файла по расширению и вызывает соответствующий
метод извлечения текста. Возвращает (текст, количествоайлов_внутри).
Для архивов возвращает список файлов и общее количество.
Синхронная версия для использования внутри архива.
Определяет тип файла по расширению и вызывает соответствующий синхронный метод.
Не обрабатывает аудио (транскрибация) и изображения (OCR) внутри архива.
"""
ext = os.path.splitext(file_path)[1].lower()
text = ""
@@ -495,21 +461,72 @@ class FileService:
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
# ============================================================
# ОСНОВНОЙ МЕТОД теперь асинхронный!
# ============================================================
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
@@ -518,23 +535,20 @@ class FileService:
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"
@@ -550,15 +564,10 @@ class FileService:
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()
@@ -572,7 +581,7 @@ class FileService:
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]
@@ -589,7 +598,7 @@ class FileService:
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:
@@ -612,16 +621,15 @@ class FileService:
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"
@@ -638,10 +646,6 @@ class FileService:
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