/
AxeOn
/
ProjectAi
Обзор
Документация
Войти
/
AxeOn
/
ProjectAi
Код
Запросы
0
Задачи
Вики
Пакеты
0
Релизы
0
CI/CD
Аналитика
Безопасность
master
main.py
775 строк
31 KB
Дмитрий
refactor: split deadlines_module into deadlines_ui and deadlines_service, completing SRP separation
25 апр 2026, 18:16
25 апр 2026, 18:16
ff36620
Код
Авторство
О чём код?
# main.py """ Офис-Планировщик v5.0 Главное окно приложения УЛУЧШЕНИЯ v5.0: - Исправлен thread safety в БД - Убран SQL из точки входа - Добавлено логирование - ConfigManager для настроек - Обработка ошибок - Защита от двойного запуска ЗАПУСК: python main.py ТРЕБОВАНИЯ: - Python 3.8+ - ttkbootstrap - pandas - requests - (опционально) speech_recognition для голосового ввода """ from dashboard_module import DashboardModule from deadlines_ui import DeadlinesControlModule from quality_module import QualityModule import tkinter as tk import ttkbootstrap as ttk from ttkbootstrap.constants import * from tkinter import messagebox import os import json import logging import logging.handlers import sys import shutil from datetime import datetime from pathlib import Path import atexit import database # === ЗАЩИТА ОТ ДВОЙНОГО ЗАПУСКА === LOCK_FILE = Path("app.lock") def check_single_instance(): """Проверка что программа не запущена""" if LOCK_FILE.exists(): try: # Проверяем что процесс действительно существует with open(LOCK_FILE, 'r') as f: pid = int(f.read().strip()) # Проверяем существование процесса (только для Windows) import psutil if psutil.pid_exists(pid): print("\n" + "="*60) print("⚠️ ВНИМАНИЕ: Программа уже запущена!") print("="*60) print(f"\nПроцесс ID: {pid}") print("\nЗакройте уже запущенную копию программы") print("или завершите процесс принудительно.\n") print("Если программа не запущена, удалите файл:") print(f" {LOCK_FILE.absolute()}\n") input("Нажмите Enter для выхода...") sys.exit(1) else: # Процесс не существует - удаляем старый lock LOCK_FILE.unlink() except (ValueError, FileNotFoundError, ImportError): # Если не можем проверить - удаляем lock LOCK_FILE.unlink() # Создаём lock-файл с PID текущего процесса with open(LOCK_FILE, 'w') as f: f.write(str(os.getpid())) # Регистрируем удаление при выходе atexit.register(cleanup_lock) def cleanup_lock(): """Удаление lock-файла при завершении""" try: if LOCK_FILE.exists(): LOCK_FILE.unlink() except Exception: pass # Проверяем одиночность запуска ПЕРЕД инициализацией try: check_single_instance() except Exception as e: print(f"\n⚠️ Ошибка при проверке запуска: {e}\n") # Продолжаем работу даже если проверка не сработала # Настройка логирования ПЕРЕД импортом модулей _log_formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s') _file_handler = logging.handlers.RotatingFileHandler( 'app.log', maxBytes=10 * 1024 * 1024, backupCount=5, encoding='utf-8' ) _file_handler.setLevel(logging.INFO) _file_handler.setFormatter(_log_formatter) _console_handler = logging.StreamHandler(sys.stdout) _console_handler.setLevel(logging.WARNING) _console_handler.setFormatter(_log_formatter) logging.basicConfig(level=logging.INFO, handlers=[_file_handler, _console_handler]) logger = logging.getLogger(__name__) # Импорт модулей try: from database import DBManager from telegram_service import TelegramService from ui import TaskView from controller import TaskController from vacation_module import VacationApp from employee_module import EmployeeApp from load_ui import LoadApp from calendar_module import CalendarApp from salary_module import SalaryApp from mail_module import MailApp except ImportError as e: logger.critical(f"Failed to import modules: {e}") print(f"ОШИБКА: Не удалось загрузить модули приложения: {e}") print("Убедитесь, что все файлы находятся в одной папке.") sys.exit(1) # Ключи для идентификации вкладок TAB_IDS = { "dashboard": "Дашборд", "tasks": "Задачи", "vacations": "Отпуска", "employees": "Сотрудники", "load": "Загрузка", "calendar": "Календарь", "salary": "Статистика", "deadlines": "Контроль сроков", "quality": "Контроль качества", "mail": "Почта", } # Порядок по умолчанию DEFAULT_TAB_CONFIG = [ {"id": "dashboard", "name": "📊 Дашборд"}, {"id": "tasks", "name": "📝 Текущие задачи"}, {"id": "deadlines", "name": "⏰ Контроль сроков"}, {"id": "vacations", "name": "🏖️ График отпусков"}, {"id": "employees", "name": "👥 Список сотрудников"}, {"id": "load", "name": "📈 Текущие загрузки"}, {"id": "calendar", "name": "📅 Календарь"}, {"id": "salary", "name": "💰 Статистика"}, {"id": "quality", "name": "🔍 Контроль качества"}, {"id": "mail", "name": "✉ Почта"}, ] def load_token_from_env(): """ Читает токен бота из файла .env Формат файла .env: BOT_TOKEN=your_bot_token_here """ try: env_file = Path(".env") if env_file.exists(): with open(env_file, "r", encoding="utf-8") as f: for line in f: line = line.strip() if line.startswith("BOT_TOKEN"): parts = line.split("=", 1) if len(parts) == 2: token = parts[1].strip() logger.info("Bot token loaded from .env") return token except Exception as e: logger.error(f"Error reading .env: {e}") logger.info("No bot token found in .env") return None class TabSettingsDialog(tk.Toplevel): """Окно для настройки порядка и имен вкладок""" def __init__(self, parent, current_config, on_save_callback): super().__init__(parent) self.title("Настройка вкладок") self.geometry("450x550") self.current_config = current_config self.on_save = on_save_callback self.resizable(False, False) # Центрирование окна self.transient(parent) self.grab_set() self._build_ui() self.refresh_list() def _build_ui(self): """Построение UI""" # Заголовок header = ttk.Frame(self, padding=10) header.pack(fill="x", side="top") ttk.Label( header, text="Настройка порядка и названий вкладок", font=("Segoe UI", 11, "bold") ).pack() # Панель инструментов tb = ttk.Frame(self, padding=5) tb.pack(fill="x", side="top") ttk.Button( tb, text="⬆ Вверх", command=self.move_up, bootstyle="secondary", width=10 ).pack(side="left", padx=2) ttk.Button( tb, text="⬇ Вниз", command=self.move_down, bootstyle="secondary", width=10 ).pack(side="left", padx=2) ttk.Button( tb, text="✎ Переименовать", command=self.rename, bootstyle="info", width=15 ).pack(side="left", padx=10) # Список list_frame = ttk.Frame(self) list_frame.pack(fill="both", expand=True, padx=10, pady=5) self.tree = ttk.Treeview( list_frame, columns=("name",), show="headings", selectmode="browse", bootstyle="primary" ) self.tree.heading("name", text="Название вкладки") self.tree.column("name", width=400) scrollbar = ttk.Scrollbar(list_frame, orient="vertical", command=self.tree.yview) self.tree.configure(yscrollcommand=scrollbar.set) self.tree.pack(side="left", fill="both", expand=True) scrollbar.pack(side="right", fill="y") # Подсказка hint = ttk.Label( self, text="💡 Выберите вкладку и используйте кнопки для изменения", foreground="gray" ) hint.pack(pady=5) # Кнопки сохранения bbox = ttk.Frame(self, padding=10) bbox.pack(fill="x", side="bottom") ttk.Button( bbox, text="Сброс по умолчанию", command=self.reset_defaults, bootstyle="danger-outline", width=20 ).pack(side="left") ttk.Button( bbox, text="Отмена", command=self.destroy, bootstyle="secondary", width=12 ).pack(side="right", padx=5) ttk.Button( bbox, text="Сохранить", command=self.save, bootstyle="success", width=12 ).pack(side="right") def refresh_list(self): """Обновить список вкладок""" self.tree.delete(*self.tree.get_children()) for i, item in enumerate(self.current_config, 1): self.tree.insert("", "end", values=(f"{i}. {item['name']}",), tags=(item['id'],)) def get_selected_index(self): """Получить индекс выбранной вкладки""" sel = self.tree.selection() if not sel: return None return self.tree.index(sel[0]) def move_up(self): """Переместить вкладку вверх""" idx = self.get_selected_index() if idx is not None and idx > 0: self.current_config[idx], self.current_config[idx-1] = \ self.current_config[idx-1], self.current_config[idx] self.refresh_list() child_id = self.tree.get_children()[idx-1] self.tree.selection_set(child_id) self.tree.see(child_id) def move_down(self): """Переместить вкладку вниз""" idx = self.get_selected_index() if idx is not None and idx < len(self.current_config) - 1: self.current_config[idx], self.current_config[idx+1] = \ self.current_config[idx+1], self.current_config[idx] self.refresh_list() child_id = self.tree.get_children()[idx+1] self.tree.selection_set(child_id) self.tree.see(child_id) def rename(self): """Переименовать вкладку""" from tkinter import simpledialog idx = self.get_selected_index() if idx is None: messagebox.showwarning("Внимание", "Выберите вкладку для переименования") return item = self.current_config[idx] new_name = simpledialog.askstring( "Переименование", f"Новое имя для '{item['name']}':", parent=self, initialvalue=item['name'] ) if new_name and new_name.strip(): item['name'] = new_name.strip() self.refresh_list() child_id = self.tree.get_children()[idx] self.tree.selection_set(child_id) def reset_defaults(self): """Сброс к настройкам по умолчанию""" if messagebox.askyesno( "Сброс", "Вернуть стандартный порядок и названия вкладок?", parent=self ): self.current_config = [x.copy() for x in DEFAULT_TAB_CONFIG] self.refresh_list() def save(self): """Сохранить изменения""" self.on_save(self.current_config) self.destroy() class MainApp(ttk.Window): """Главное окно приложения""" def __init__(self): super().__init__(themename="superhero") self.title("Офис-Планировщик v5.0") self.geometry("1400x900") logger.info("=" * 60) logger.info("Application starting: Офис-Планировщик v5.0") logger.info("=" * 60) # Определяем базовую директорию base_dir = os.path.dirname(os.path.abspath(__file__)) logger.info(f"Base directory: {base_dir}") logger.info(f"Working directory: {os.getcwd()}") # Инициализация БД db_path = os.path.join(base_dir, "planner_v4.db") logger.info(f"Database path: {db_path}") try: self.db = DBManager(db_path) logger.info("Database initialized successfully") except Exception as e: logger.critical(f"Failed to initialize database: {e}", exc_info=True) messagebox.showerror( "Ошибка БД", f"Не удалось подключиться к базе данных:\n{e}\n\nПриложение будет закрыто." ) self.destroy() return # Авто-бэкап при запуске (через 2 сек, чтобы не блокировать UI) self.after(2000, self._auto_backup) # Инициализация Telegram сервиса bot_token = load_token_from_env() self.tg_service = TelegramService(self.db, bot_token) if not bot_token: logger.warning("Telegram bot token not configured") # Хранилище фреймов модулей self.frames = {} # Основной Notebook self.notebook = ttk.Notebook(self) self.notebook.pack(fill="both", expand=True) # Статусная строка self._status_bar = ttk.Frame(self, bootstyle="secondary") self._status_bar.pack(fill=X, side=BOTTOM) self._status_lbl_date = ttk.Label(self._status_bar, text="", font=("Arial", 8), bootstyle="inverse-secondary") self._status_lbl_date.pack(side=LEFT, padx=10) ttk.Separator(self._status_bar, orient=VERTICAL).pack(side=LEFT, fill=Y, pady=2) self._status_lbl_cases = ttk.Label(self._status_bar, text="", font=("Arial", 8), bootstyle="inverse-secondary") self._status_lbl_cases.pack(side=LEFT, padx=10) ttk.Separator(self._status_bar, orient=VERTICAL).pack(side=LEFT, fill=Y, pady=2) self._status_lbl_time = ttk.Label(self._status_bar, text="", font=("Arial", 8), bootstyle="inverse-secondary") self._status_lbl_time.pack(side=RIGHT, padx=10) # Инициализация модулей self._init_modules() # Загрузка конфигурации и построение вкладок self.tab_config = self._load_tab_config() self.rebuild_tabs() # Контекстное меню self.notebook.bind("<Button-3>", self._show_context_menu) self._create_context_menu() # Глобальный поиск по Ctrl+F self.bind_all("<Control-f>", self._open_global_search) self.bind_all("<Control-F>", self._open_global_search) # Запуск обновления статусной строки self._update_status_bar() # Обработчик закрытия self.protocol("WM_DELETE_WINDOW", self._on_closing) logger.info("Application initialized successfully") def _init_modules(self): """Инициализация всех модулей приложения""" logger.info("Initializing application modules") try: # 0. Дашборд - ПЕРВЫМ! f_dashboard = ttk.Frame(self.notebook) DashboardModule(f_dashboard, self.db) self.frames["dashboard"] = f_dashboard logger.info("Dashboard module initialized") # 1. Задачи f_tasks = ttk.Frame(self.notebook) class TabbedTaskView(TaskView): def __init__(self, master, callbacks): super().__init__(f_tasks, callbacks) self.task_controller = TaskController( f_tasks, self.db, TabbedTaskView, self.tg_service ) self.frames["tasks"] = f_tasks logger.info("Tasks module initialized") # 1a. Контроль сроков f_deadlines = ttk.Frame(self.notebook) DeadlinesControlModule(f_deadlines, self.db) self.frames["deadlines"] = f_deadlines logger.info("Deadlines control module initialized") # 2. Отпуска f_vac = ttk.Frame(self.notebook) VacationApp(f_vac, self.db) self.frames["vacations"] = f_vac logger.info("Vacations module initialized") # 3. Сотрудники f_emp = ttk.Frame(self.notebook) EmployeeApp(f_emp, self.db) self.frames["employees"] = f_emp logger.info("Employees module initialized") # 4. Загрузка f_load = ttk.Frame(self.notebook) LoadApp(f_load, self.db, self.tg_service) self.frames["load"] = f_load logger.info("Load module initialized") # 5. Календарь f_cal = ttk.Frame(self.notebook) CalendarApp(f_cal, self.db) self.frames["calendar"] = f_cal logger.info("Calendar module initialized") # 6. Статистика f_sal = ttk.Frame(self.notebook) SalaryApp(f_sal, self.db) self.frames["salary"] = f_sal logger.info("Salary module initialized") # 7. Контроль качества f_quality = ttk.Frame(self.notebook) QualityModule(f_quality, self.db) self.frames["quality"] = f_quality logger.info("Quality control module initialized") # 8. Почта (IMAP) f_mail = ttk.Frame(self.notebook) MailApp(f_mail, self.db) self.frames["mail"] = f_mail logger.info("Mail module initialized") logger.info("All modules initialized successfully") except Exception as e: logger.error(f"Error initializing modules: {e}", exc_info=True) messagebox.showerror("Ошибка", f"Не удалось инициализировать модули:\n{e}") def _load_tab_config(self): """Загрузка конфигурации вкладок из БД""" raw_json = self.db.get_setting("tabs_config") if raw_json: try: config = json.loads(raw_json) # Проверка целостности: добавляем новые модули existing_ids = {item['id'] for item in config} for def_item in DEFAULT_TAB_CONFIG: if def_item['id'] not in existing_ids: config.append(def_item.copy()) logger.info(f"Added missing tab: {def_item['name']}") # Миграция: исправляем имена вкладок (включая опечатки и добавление эмодзи) _name_migrations = { "dashboard": "📊 Дашборд", "tasks": "📝 Текущие задачи", "deadlines": "⏰ Контроль сроков", "vacations": "🏖️ График отпусков", "employees": "👥 Список сотрудников", "load": "📈 Текущие загрузки", "calendar": "📅 Календарь", "salary": "💰 Статистика", "quality": "🔍 Контроль качества", } _migrated = False for item in config: tab_id = item.get('id') if tab_id in _name_migrations: correct = _name_migrations[tab_id] # Всегда приводим к эталонному имени if item['name'] != correct: logger.info(f"Migrate tab name: '{item['name']}' → '{correct}'") item['name'] = correct _migrated = True if _migrated: self.db.set_setting("tabs_config", json.dumps(config, ensure_ascii=False)) logger.info("Tab configuration loaded from database") return config except json.JSONDecodeError as e: logger.error(f"Invalid tabs_config JSON: {e}") logger.info("Using default tab configuration") return [x.copy() for x in DEFAULT_TAB_CONFIG] def _save_tab_config(self, new_config): """Сохранение конфигурации вкладок в БД""" self.tab_config = new_config self.db.set_setting("tabs_config", json.dumps(new_config, ensure_ascii=False)) self.rebuild_tabs() logger.info("Tab configuration saved") def rebuild_tabs(self): """Перестройка вкладок согласно конфигурации""" # Удаляем все текущие вкладки for tab in self.notebook.tabs(): self.notebook.forget(tab) # Добавляем согласно конфигу for item in self.tab_config: tab_id = item['id'] tab_name = item['name'] if tab_id in self.frames: self.notebook.add(self.frames[tab_id], text=tab_name) logger.debug(f"Tabs rebuilt: {[item['name'] for item in self.tab_config]}") def _create_context_menu(self): """Создание контекстного меню""" self.ctx_menu = tk.Menu(self, tearoff=0) self.ctx_menu.add_command( label="⚙ Настроить вкладки", command=self._open_tab_settings ) self.ctx_menu.add_separator() self.ctx_menu.add_command( label="💾 Резервная копия", command=self._manual_backup ) def _show_context_menu(self, event): """Показать контекстное меню""" self.ctx_menu.post(event.x_root, event.y_root) def _open_tab_settings(self): """Открыть окно настроек вкладок""" TabSettingsDialog( self, [x.copy() for x in self.tab_config], self._save_tab_config ) # ── Резервное копирование ───────────────────────────────────────────────── def _auto_backup(self): """Авто-бэкап БД при запуске, хранит 7 последних копий""" try: backup_dir = Path(self.db.db_path).parent / "backups" backup_dir.mkdir(exist_ok=True) ts = datetime.now().strftime("%Y-%m-%d_%H-%M") dst = backup_dir / f"planner_v4_{ts}.db" shutil.copy2(self.db.db_path, dst) # Оставляем только 7 последних backups = sorted(backup_dir.glob("planner_v4_*.db")) for old in backups[:-7]: old.unlink() logger.info(f"Auto-backup created: {dst}") except Exception as e: logger.error(f"Auto-backup failed: {e}") def _manual_backup(self): """Ручной бэкап с уведомлением""" try: self._auto_backup() messagebox.showinfo("Резервная копия", "Резервная копия успешно создана в папке backups/") except Exception as e: messagebox.showerror("Ошибка", f"Не удалось создать резервную копию:\n{e}") # ── Статусная строка ────────────────────────────────────────────────────── def _update_status_bar(self): """Обновляет статусную строку каждые 30 секунд.""" try: now = datetime.now() self._status_lbl_time.config( text=f"🕐 {now.strftime('%d.%m.%Y %H:%M:%S')}") self._status_lbl_date.config( text=f"📅 Сегодня: {now.strftime('%d.%m.%Y')}") # Дела из deadline_cases (последний снимок) try: cases = self.db.get_deadline_cases() threshold = int(self.db.get_setting('deadline_threshold', '3') or 3) over = sum(1 for c in cases if (c.deadline or 0) >= threshold) self._status_lbl_cases.config( text=f"⏰ Дел со сроком ≥{threshold}: {over} из {len(cases)}") except Exception: pass except Exception: pass self.after(30000, self._update_status_bar) # ── Глобальный поиск ────────────────────────────────────────────────────── def _open_global_search(self, event=None): """Ctrl+F — поиск дела по номеру в deadline_cases.""" win = tk.Toplevel(self) win.title("🔍 Поиск дела") win.geometry("640x420") win.grab_set() frm = ttk.Frame(win, padding=10) frm.pack(fill=X) ttk.Label(frm, text="Номер дела:").pack(side=LEFT) search_var = tk.StringVar() entry = ttk.Entry(frm, textvariable=search_var, width=30) entry.pack(side=LEFT, padx=6) cols = ('Номер дела', 'Дата загрузки', 'Срок', 'Эксперт (расчёт)', 'Комментарий') tree = ttk.Treeview(win, columns=cols, show='headings', height=14) for col in cols: tree.heading(col, text=col) tree.column('Номер дела', width=120, anchor='center') tree.column('Дата загрузки', width=100, anchor='center') tree.column('Срок', width=55, anchor='center') tree.column('Эксперт (расчёт)', width=190, anchor='w') tree.column('Комментарий', width=150, anchor='w') # Цветовые теги tree.tag_configure('dl_critical', background='#F2DCDC') tree.tag_configure('dl_high', background='#F8F0D4') vscroll = ttk.Scrollbar(win, orient='vertical', command=tree.yview) tree.configure(yscrollcommand=vscroll.set) tree.pack(side=LEFT, fill=BOTH, expand=True, padx=(10, 0), pady=6) vscroll.pack(side=RIGHT, fill=Y, padx=(0, 10), pady=6) status_lbl = ttk.Label(win, text="Введите номер дела и нажмите Enter", font=("Arial", 8), bootstyle="secondary") status_lbl.pack(pady=(0, 4)) def do_search(*_): for row in tree.get_children(): tree.delete(row) q = search_var.get().strip() if not q: return try: with self.db.pool.get_connection() as conn: cursor = conn.cursor() cursor.execute(""" SELECT case_number, load_date, deadline, expert_calculation, comment FROM deadline_cases WHERE case_number LIKE ? ORDER BY load_date DESC LIMIT 200 """, (f'%{q}%',)) rows = cursor.fetchall() for r in rows: d = r['deadline'] or 0 tag = 'dl_critical' if d >= 7 else ('dl_high' if d >= 5 else '') tree.insert('', 'end', tags=(tag,) if tag else (), values=(r['case_number'], r['load_date'], d, r['expert_calculation'] or '', r['comment'] or '')) status_lbl.config(text=f"Найдено: {len(rows)} записей") except Exception as e: status_lbl.config(text=f"Ошибка: {e}") entry.bind('<Return>', do_search) ttk.Button(frm, text="🔍 Найти", command=do_search, bootstyle="primary").pack(side=LEFT) ttk.Button(frm, text="✖", command=win.destroy, bootstyle="secondary", width=3).pack(side=LEFT, padx=4) entry.focus_set() def _on_closing(self): """Обработчик закрытия приложения""" logger.info("Application closing") # Закрываем Telegram сервис if hasattr(self, 'tg_service'): self.tg_service.close() # Закрываем БД if hasattr(self, 'db'): self.db.close() logger.info("Application closed gracefully") self.destroy() def main(): """Точка входа в приложение""" try: logger.info("Starting main application") app = MainApp() app.mainloop() except Exception as e: logger.critical(f"Application crashed: {e}", exc_info=True) messagebox.showerror( "Критическая ошибка", f"Приложение завершилось с ошибкой:\n{e}\n\nПодробности в app.log" ) if __name__ == "__main__": main()