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 [], "

Не знайдено файлів CSV у директорії current_data.

" # Використовуємо реальні дані, якщо доступні, інакше 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 = "

Доступні файли в директорії current_data:

" html_output += "" html_output += "" for info in files_info: html_output += "" html_output += f"" html_output += f"" html_output += f"" html_output += f"" html_output += f"" html_output += "" html_output += "
ФайлРозмірЗміненоРядкиКолонки
{info['name']}{info['size_kb']} KB{info['modified']}{info.get('rows_count', info.get('rows_preview', 'N/A'))}{info.get('columns_count', info.get('columns_preview', 'N/A'))}
" # Приховане поле з шляхами html_output += "" return files_list, html_output except Exception as e: logger.error(f"Помилка при отриманні списку локальних файлів: {e}") return [], f"

Помилка при отриманні списку файлів: {str(e)}

" def get_file_preview(self, selected_file): try: if not selected_file: return "

Виберіть файл для перегляду

" 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"

Файл {file_name} не знайдено

" 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"

Помилка при читанні файлу: {preview_info['error']}

" # Формуємо HTML html_output = f"

Попередній перегляд файлу: {file_name}

" html_output += f"

Загальна кількість рядків: {preview_info['total_rows']}

" html_output += f"

Кількість колонок: {preview_info['columns_count']}

" html_output += "" # Заголовки html_output += "" for col in preview_info['columns']: html_output += f"" html_output += "" # Дані for i, row in enumerate(preview_info['preview_rows']): row_style = "background-color: #E9EDF5;" if i % 2 == 0 else "" html_output += f"" for col in preview_info['columns']: value = row.get(col, "") if isinstance(value, str) and len(value) > 100: value = value[:100] + "..." html_output += f"" html_output += "" html_output += "
{col}
{value}
" return html_output except Exception as e: logger.error(f"Помилка при отриманні попереднього перегляду файлу: {e}") return f"

Помилка при перегляді файлу: {str(e)}

" 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 "

Помилка: не вибрано жодного файлу для обробки

", 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"

Помилка при ініціалізації даних: {error_msg}

", 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 = "

Дані успішно ініціалізовано

" status_html += f"

Об'єднано {result_info.get('source_files_count', 0)} файлів

" status_html += f"

Загальна кількість рядків: {result_info.get('rows_count', 0)}

" status_html += f"

Кількість колонок: {result_info.get('columns_count', 0)}

" 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"

Помилка при ініціалізації даних: {str(e)}

", None