DocUA's picture
Єдиний коміт - очищення історії
4ad5efa
"""
Модуль утиліт для роботи з FAISS векторними індексами.
Цей файл повинен бути розміщений у modules/ai_analysis/faiss_utils.py
"""
import logging
import os
from pathlib import Path
import json
import hashlib
from datetime import datetime
import shutil
import tempfile
import sys
logger = logging.getLogger(__name__)
# Перевірка наявності змінних середовища Hugging Face
IS_HUGGINGFACE = os.environ.get("SPACE_ID") is not None
if IS_HUGGINGFACE:
logger.info("Виявлено середовище Hugging Face Spaces")
try:
import faiss
import numpy as np
from llama_index.vector_stores.faiss import FaissVectorStore
from llama_index.core import load_index_from_storage
from llama_index.core import StorageContext
FAISS_AVAILABLE = True
logger.info("FAISS успішно імпортовано")
except ImportError as e:
logger.warning(f"FAISS або llama-index-vector-stores-faiss не встановлено: {e}. Використання FAISS буде вимкнено.")
FAISS_AVAILABLE = False
def check_faiss_available():
"""
Перевірка доступності FAISS.
Returns:
bool: True, якщо FAISS доступний, False інакше
"""
return FAISS_AVAILABLE
def generate_file_hash(file_path):
"""
Генерує хеш для файлу на основі його вмісту.
Args:
file_path (str): Шлях до файлу
Returns:
str: Хеш файлу або None у випадку помилки
"""
try:
if not os.path.exists(file_path):
logger.error(f"Файл не знайдено: {file_path}")
return None
# Отримуємо базову інформацію про файл для додавання в хеш
file_stat = os.stat(file_path)
file_size = file_stat.st_size
file_mtime = file_stat.st_mtime
# Створюємо хеш на основі вмісту файлу
sha256 = hashlib.sha256()
# Додаємо базову інформацію про файл
sha256.update(f"{file_size}_{file_mtime}".encode())
# Додаємо вміст файлу
with open(file_path, "rb") as f:
for byte_block in iter(lambda: f.read(4096), b""):
sha256.update(byte_block)
return sha256.hexdigest()
except Exception as e:
logger.error(f"Помилка при генерації хешу файлу: {e}")
return None
def save_indices_metadata(directory_path, metadata):
"""
Зберігає метадані індексів у JSON файл.
Args:
directory_path (str): Шлях до директорії з індексами
metadata (dict): Метадані для збереження
Returns:
bool: True, якщо збереження успішне, False інакше
"""
try:
# Перевіряємо наявність директорії
if not os.path.exists(directory_path):
logger.warning(f"Директорія {directory_path} не існує. Створюємо...")
os.makedirs(directory_path, exist_ok=True)
metadata_path = Path(directory_path) / "metadata.json"
# Додаємо додаткову інформацію про оточення
metadata["environment"] = {
"is_huggingface": IS_HUGGINGFACE,
"python_version": sys.version,
"platform": sys.platform
}
# Додаємо логування для діагностики
logger.info(f"Збереження метаданих у {metadata_path}")
logger.info(f"Розмір метаданих: {len(str(metadata))} символів")
with open(metadata_path, "w", encoding="utf-8") as f:
json.dump(metadata, f, ensure_ascii=False, indent=2)
# Перевіряємо, що файл було створено
if os.path.exists(metadata_path):
logger.info(f"Метадані успішно збережено у {metadata_path}")
return True
else:
logger.error(f"Файл {metadata_path} не було створено")
return False
except Exception as e:
logger.error(f"Помилка при збереженні метаданих: {e}")
return False
def load_indices_metadata(directory_path):
"""
Завантажує метадані індексів з JSON файлу.
Args:
directory_path (str): Шлях до директорії з індексами
Returns:
dict: Метадані або пустий словник у випадку помилки
"""
try:
metadata_path = Path(directory_path) / "metadata.json"
if not metadata_path.exists():
logger.warning(f"Файл метаданих не знайдено: {metadata_path}")
return {}
with open(metadata_path, "r", encoding="utf-8") as f:
metadata = json.load(f)
logger.info(f"Метадані успішно завантажено з {metadata_path}")
return metadata
except Exception as e:
logger.error(f"Помилка при завантаженні метаданих: {e}")
return {}
def find_latest_indices(base_dir="temp/indices"):
"""
Знаходить найновіші збережені індекси.
Args:
base_dir (str): Базова директорія з індексами
Returns:
tuple: (bool, str) - (наявність індексів, шлях до найновіших індексів)
"""
try:
# Перевіряємо наявність базової директорії
indices_dir = Path(base_dir)
if not indices_dir.exists():
logger.info(f"Директорія {base_dir} не існує")
return False, None
if not os.path.isdir(indices_dir):
logger.warning(f"{base_dir} існує, але не є директорією")
return False, None
if not any(indices_dir.iterdir()):
logger.info(f"Директорія {base_dir} порожня")
return False, None
# Отримання списку піддиректорій з індексами
try:
subdirs = [d for d in indices_dir.iterdir() if d.is_dir()]
except Exception as iter_err:
logger.error(f"Помилка при перегляді директорії {base_dir}: {iter_err}")
return False, None
if not subdirs:
logger.info("Індекси не знайдено")
return False, None
# Знаходимо найновішу директорію
try:
latest_dir = max(subdirs, key=lambda x: x.stat().st_mtime)
except Exception as sort_err:
logger.error(f"Помилка при сортуванні директорій: {sort_err}")
return False, None
logger.info(f"Знайдено індекси у директорії {latest_dir}")
return True, str(latest_dir)
except Exception as e:
logger.error(f"Помилка при пошуку індексів: {e}")
return False, None
def find_indices_by_hash(csv_hash, base_dir="temp/indices"):
"""
Знаходить індекси, що відповідають вказаному хешу CSV файлу.
Args:
csv_hash (str): Хеш CSV файлу
base_dir (str): Базова директорія з індексами
Returns:
tuple: (bool, str) - (наявність індексів, шлях до відповідних індексів)
"""
try:
if not csv_hash:
logger.warning("Не вказано хеш CSV файлу")
return False, None
# Перевіряємо наявність базової директорії
indices_dir = Path(base_dir)
if not indices_dir.exists():
logger.info(f"Директорія {base_dir} не існує")
return False, None
if not any(indices_dir.iterdir()):
logger.info(f"Директорія {base_dir} порожня")
return False, None
# Отримання списку піддиректорій з індексами
subdirs = [d for d in indices_dir.iterdir() if d.is_dir()]
if not subdirs:
logger.info("Індекси не знайдено")
return False, None
# Перевіряємо кожну директорію на відповідність хешу
for directory in subdirs:
metadata_path = directory / "metadata.json"
if metadata_path.exists():
try:
with open(metadata_path, "r", encoding="utf-8") as f:
metadata = json.load(f)
if "csv_hash" in metadata and metadata["csv_hash"] == csv_hash:
# Додатково перевіряємо наявність файлів індексів
if (directory / "docstore.json").exists():
logger.info(f"Знайдено індекси для CSV з хешем {csv_hash} у {directory}")
return True, str(directory)
else:
logger.warning(f"Знайдено метадані для CSV з хешем {csv_hash}, але файли індексів відсутні у {directory}")
except Exception as md_err:
logger.warning(f"Помилка при читанні метаданих {metadata_path}: {md_err}")
# Якщо відповідних індексів не знайдено, повертаємо найновіші
logger.info(f"Не знайдено індексів для CSV з хешем {csv_hash}, спроба знайти найновіші")
return find_latest_indices(base_dir)
except Exception as e:
logger.error(f"Помилка при пошуку індексів за хешем: {e}")
return False, None
def create_indices_directory(csv_hash=None, base_dir="temp/indices"):
"""
Створює директорію для зберігання індексів з часовою міткою.
Args:
csv_hash (str, optional): Хеш CSV файлу для метаданих
base_dir (str): Базова директорія для індексів
Returns:
str: Шлях до створеної директорії
"""
try:
# Створення базової директорії, якщо вона не існує
indices_dir = Path(base_dir)
# Очищаємо старі індекси перед створенням нових, якщо ми на Hugging Face
if IS_HUGGINGFACE and indices_dir.exists():
logger.info("Очищення старих індексів перед створенням нових на Hugging Face")
try:
# Видаляємо тільки старі директорії, якщо їх більше 1
subdirs = [d for d in indices_dir.iterdir() if d.is_dir()]
if len(subdirs) > 1:
# Сортуємо за часом модифікації (від найстаріших до найновіших)
sorted_dirs = sorted(subdirs, key=lambda x: x.stat().st_mtime)
# Залишаємо тільки найновішу директорію
for directory in sorted_dirs[:-1]:
try:
shutil.rmtree(directory)
logger.info(f"Видалено стару директорію: {directory}")
except Exception as del_err:
logger.warning(f"Не вдалося видалити директорію {directory}: {del_err}")
except Exception as clean_err:
logger.warning(f"Помилка при очищенні старих індексів: {clean_err}")
# Створюємо базову директорію
indices_dir.mkdir(exist_ok=True, parents=True)
# Створення унікальної директорії з часовою міткою
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
index_dir = indices_dir / timestamp
# Спроба створити директорію
try:
index_dir.mkdir(exist_ok=True)
except Exception as mkdir_err:
logger.error(f"Не вдалося створити директорію {index_dir}: {mkdir_err}")
# Створюємо тимчасову директорію як запасний варіант
try:
temp_dir = tempfile.mkdtemp(prefix="faiss_indices_")
logger.info(f"Створено тимчасову директорію: {temp_dir}")
return temp_dir
except Exception as temp_err:
logger.error(f"Не вдалося створити тимчасову директорію: {temp_err}")
return str(indices_dir / "fallback")
# Зберігаємо базові метадані
metadata = {
"created_at": timestamp,
"timestamp": datetime.now().timestamp(),
"csv_hash": csv_hash
}
save_indices_metadata(str(index_dir), metadata)
logger.info(f"Створено директорію для індексів: {index_dir}")
return str(index_dir)
except Exception as e:
logger.error(f"Помилка при створенні директорії індексів: {e}")
# Створюємо тимчасову директорію як запасний варіант
try:
temp_dir = tempfile.mkdtemp(prefix="faiss_indices_")
logger.info(f"Створено тимчасову директорію для індексів: {temp_dir}")
return temp_dir
except Exception:
# Якщо і це не вдалося, використовуємо директорію temp
logger.error("Не вдалося створити навіть тимчасову директорію, використовуємо базову temp")
os.makedirs("temp", exist_ok=True)
return "temp"
def cleanup_old_indices(max_indices=3, base_dir="temp/indices"):
"""
Видаляє старі індекси, залишаючи тільки вказану кількість найновіших.
Args:
max_indices (int): Максимальна кількість індексів для зберігання
base_dir (str): Базова директорія з індексами
Returns:
int: Кількість видалених директорій
"""
try:
# На Hugging Face Space обмежуємо максимальну кількість індексів до 1
if IS_HUGGINGFACE:
max_indices = 1
logger.info("На Hugging Face Space обмежуємо кількість індексів до 1")
indices_dir = Path(base_dir)
if not indices_dir.exists():
logger.warning(f"Директорія {base_dir} не існує")
return 0
# Отримання списку піддиректорій з індексами
try:
subdirs = [d for d in indices_dir.iterdir() if d.is_dir()]
except Exception as iter_err:
logger.error(f"Помилка при скануванні директорії {base_dir}: {iter_err}")
return 0
if len(subdirs) <= max_indices:
logger.info(f"Кількість індексів ({len(subdirs)}) не перевищує ліміт ({max_indices})")
return 0
# Сортуємо директорії за часом модифікації (від найновіших до найстаріших)
try:
sorted_dirs = sorted(subdirs, key=lambda x: x.stat().st_mtime, reverse=True)
except Exception as sort_err:
logger.error(f"Помилка при сортуванні директорій: {sort_err}")
return 0
# Залишаємо тільки max_indices найновіших директорій
dirs_to_delete = sorted_dirs[max_indices:]
# Видаляємо старі директорії
deleted_count = 0
for directory in dirs_to_delete:
try:
shutil.rmtree(directory)
deleted_count += 1
logger.info(f"Видалено стару директорію індексів: {directory}")
except Exception as del_err:
logger.warning(f"Не вдалося видалити директорію {directory}: {del_err}")
return deleted_count
except Exception as e:
logger.error(f"Помилка при очищенні старих індексів: {e}")
return 0