Spaces:
Running
Running
import os | |
import logging | |
from pathlib import Path | |
import traceback | |
from modules.data_management.data_manager import DataManager | |
from modules.data_management.session_manager import SessionManager | |
from modules.data_management.index_manager import IndexManager | |
logger = logging.getLogger("jira_assistant_interface") | |
class LocalDataHelper: | |
""" | |
Клас для роботи з локальними CSV-файлами, сесіями та попереднім переглядом даних. | |
""" | |
def __init__(self, app, current_data_dir="current_data"): | |
self.app = app | |
self.current_data_dir = Path(current_data_dir) | |
self.current_data_dir.mkdir(exist_ok=True, parents=True) | |
# Ініціалізація менеджерів | |
self.session_manager = SessionManager() | |
self.data_manager = DataManager(current_data_dir, self.session_manager) | |
self.index_manager = IndexManager() | |
# Словник сесій для користувачів | |
self.user_sessions = {} | |
def get_or_create_session(self, user_id=None): | |
if not user_id: | |
import uuid | |
user_id = str(uuid.uuid4()) | |
if user_id in self.user_sessions: | |
return self.user_sessions[user_id] | |
session_id = self.session_manager.create_session(user_id) | |
self.user_sessions[user_id] = session_id | |
logger.info(f"Створено нову сесію {session_id} для користувача {user_id}") | |
return session_id | |
def list_local_files(self): | |
try: | |
files_info = self.data_manager.get_local_files() | |
if not files_info: | |
return [], "<p>Не знайдено файлів CSV у директорії current_data.</p>" | |
# Використовуємо реальні дані, якщо доступні, інакше fallback до preview | |
files_list = [ | |
f"{info['name']} ({info['size_kb']} KB, рядків: {info.get('rows_count', info.get('rows_preview', 'N/A'))}, колонок: {info.get('columns_count', info.get('columns_preview', 'N/A'))})" | |
for info in files_info | |
] | |
# Формуємо HTML | |
html_output = "<h3>Доступні файли в директорії current_data:</h3>" | |
html_output += "<table style='width:100%; border-collapse: collapse;'>" | |
html_output += "<tr style='background-color: #f2f2f2;'><th>Файл</th><th>Розмір</th><th>Змінено</th><th>Рядки</th><th>Колонки</th></tr>" | |
for info in files_info: | |
html_output += "<tr style='border-bottom: 1px solid #ddd;'>" | |
html_output += f"<td>{info['name']}</td>" | |
html_output += f"<td>{info['size_kb']} KB</td>" | |
html_output += f"<td>{info['modified']}</td>" | |
html_output += f"<td>{info.get('rows_count', info.get('rows_preview', 'N/A'))}</td>" | |
html_output += f"<td>{info.get('columns_count', info.get('columns_preview', 'N/A'))}</td>" | |
html_output += "</tr>" | |
html_output += "</table>" | |
# Приховане поле з шляхами | |
html_output += "<div id='file_paths' style='display:none;'>" | |
for info in files_info: | |
html_output += f"<div data-name='{info['name']}'>{info['path']}</div>" | |
html_output += "</div>" | |
return files_list, html_output | |
except Exception as e: | |
logger.error(f"Помилка при отриманні списку локальних файлів: {e}") | |
return [], f"<p>Помилка при отриманні списку файлів: {str(e)}</p>" | |
def get_file_preview(self, selected_file): | |
try: | |
if not selected_file: | |
return "<p>Виберіть файл для перегляду</p>" | |
local_files_info = self.data_manager.get_local_files() | |
local_files_dict = {info['name']: info['path'] for info in local_files_info} | |
file_name = selected_file.split(" (")[0].strip() if " (" in selected_file else selected_file.strip() | |
if file_name not in local_files_dict: | |
return f"<p>Файл {file_name} не знайдено</p>" | |
file_path = local_files_dict[file_name] | |
preview_info = self.data_manager.get_file_preview(file_path, max_rows=5) | |
if "error" in preview_info: | |
return f"<p style='color:red;'>Помилка при читанні файлу: {preview_info['error']}</p>" | |
# Формуємо HTML | |
html_output = f"<h3>Попередній перегляд файлу: {file_name}</h3>" | |
html_output += f"<p>Загальна кількість рядків: {preview_info['total_rows']}</p>" | |
html_output += f"<p>Кількість колонок: {preview_info['columns_count']}</p>" | |
html_output += "<table style='width:100%; border-collapse: collapse; font-size: 14px;'>" | |
# Заголовки | |
html_output += "<tr style='background-color: #4472C4; color: white;'>" | |
for col in preview_info['columns']: | |
html_output += f"<th style='padding: 8px; text-align: left;'>{col}</th>" | |
html_output += "</tr>" | |
# Дані | |
for i, row in enumerate(preview_info['preview_rows']): | |
row_style = "background-color: #E9EDF5;" if i % 2 == 0 else "" | |
html_output += f"<tr style='{row_style}'>" | |
for col in preview_info['columns']: | |
value = row.get(col, "") | |
if isinstance(value, str) and len(value) > 100: | |
value = value[:100] + "..." | |
html_output += f"<td style='padding: 8px; border-bottom: 1px solid #ddd;'>{value}</td>" | |
html_output += "</tr>" | |
html_output += "</table>" | |
return html_output | |
except Exception as e: | |
logger.error(f"Помилка при отриманні попереднього перегляду файлу: {e}") | |
return f"<p style='color:red;'>Помилка при перегляді файлу: {str(e)}</p>" | |
def initialize_data(self, selected_files, uploaded_file=None, user_id=None): | |
try: | |
session_id = self.get_or_create_session(user_id) | |
self.app.current_session_id = session_id | |
local_files_info = self.data_manager.get_local_files() | |
local_files_dict = {info['name']: info['path'] for info in local_files_info} | |
selected_paths = [] | |
for selected in selected_files: | |
file_name = selected.split(" (")[0].strip() if " (" in selected else selected.strip() | |
if file_name in local_files_dict: | |
selected_paths.append(local_files_dict[file_name]) | |
uploaded_file_path = None | |
if uploaded_file: | |
if hasattr(uploaded_file, 'name'): | |
uploaded_file_path = uploaded_file.name | |
else: | |
uploaded_file_path = uploaded_file | |
if not selected_paths and not uploaded_file_path: | |
return "<p style='color:red;'>Помилка: не вибрано жодного файлу для обробки</p>", None | |
success, result_info = self.data_manager.initialize_session_data( | |
session_id, | |
selected_paths, | |
uploaded_file_path | |
) | |
if not success: | |
error_msg = result_info.get("error", "Невідома помилка") | |
return f"<p style='color:red;'>Помилка при ініціалізації даних: {error_msg}</p>", None | |
merged_df = result_info.get("merged_df") | |
if merged_df is not None: | |
self.app.current_data = merged_df | |
self.app.last_loaded_csv = result_info.get("merged_file") | |
indices_dir = self.session_manager.get_session_indices_dir(session_id) | |
if indices_dir: | |
abs_indices_path = os.path.abspath(indices_dir) | |
self.app.indices_path = abs_indices_path | |
logger.info(f"Встановлено шлях до директорії для індексів в app: {abs_indices_path}") | |
# Спроба зберегти шлях глобально | |
try: | |
import builtins | |
if hasattr(builtins, 'app'): | |
builtins.app.indices_path = self.app.indices_path | |
logger.info("Збережено шлях до директорії індексів у глобальному об'єкті app") | |
if hasattr(builtins, 'index_manager'): | |
builtins.index_manager.last_indices_path = self.app.indices_path | |
logger.info("Збережено шлях до директорії індексів у глобальному об'єкті index_manager") | |
except Exception as e: | |
logger.warning(f"Не вдалося зберегти шлях глобально: {e}") | |
status_html = "<h3 style='color:green;'>Дані успішно ініціалізовано</h3>" | |
status_html += f"<p>Об'єднано {result_info.get('source_files_count', 0)} файлів</p>" | |
status_html += f"<p>Загальна кількість рядків: {result_info.get('rows_count', 0)}</p>" | |
status_html += f"<p>Кількість колонок: {result_info.get('columns_count', 0)}</p>" | |
files_info = { | |
"session_id": session_id, | |
"merged_file": result_info.get("merged_file"), | |
"rows_count": result_info.get("rows_count", 0), | |
"columns_count": result_info.get("columns_count", 0), | |
"source_files_count": result_info.get("source_files_count", 0), | |
"indices_dir": indices_dir if indices_dir else None | |
} | |
return status_html, files_info | |
except Exception as e: | |
logger.error(f"Помилка при ініціалізації даних: {e}") | |
error_details = traceback.format_exc() | |
logger.error(error_details) | |
return f"<p style='color:red;'>Помилка при ініціалізації даних: {str(e)}</p>", None | |