From 1461a1b86415caf5ec7f03e34187fc0fa470e9da Mon Sep 17 00:00:00 2001 From: Markov Andrey Date: Tue, 30 Jun 2026 09:16:15 +0000 Subject: [PATCH] Add new file --- core/services/file_service.py | 658 ++++++++++++++++++++++++++++++++++ 1 file changed, 658 insertions(+) create mode 100644 core/services/file_service.py diff --git a/core/services/file_service.py b/core/services/file_service.py new file mode 100644 index 0000000..4a1489a --- /dev/null +++ b/core/services/file_service.py @@ -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 \ No newline at end of file