jira-ai-assistant / modules /interface /local_data_helper.py
DocUA's picture
Єдиний коміт - очищення історії
4ad5efa
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