/
kuzminaa
/
PDM
Обзор
Документация
Войти
/
kuzminaa
/
PDM
Код
Запросы
0
Задачи
Вики
Пакеты
0
Релизы
0
CI/CD
Аналитика
Безопасность
master
src/append_document.py
349 строк
17 KB
kuzminaa
refactor: migrate modules to src/, add logging system, add assembly workshop
29 июл 2026, 23:52
29 июл 2026, 23:52
a5323aa
Код
Авторство
О чём код?
# -*- coding: utf-8 -*- """ Модуль: append_document.py Назначение: Класс-контроллер для модального окна регистрации документов ЕСКД. Управляет логикой AppLayout и динамической доступностью полей дат. """ import os import logging import tkinter as tk from tkinter import ttk, filedialog, messagebox from . import config from . import db_manager as db from .layout import AppLayout, DateMaskEntry from .preview_engine import PDFPreviewEngine logger = logging.getLogger(__name__) class AppendDocumentWindow: """Контроллер модального окна регистрации, управляющий логикой AppLayout.""" def __init__(self, parent_menu, config_module, on_success_callback=None): self.parent = parent_menu self.on_success_callback = on_success_callback self.config = config_module # Создаем модальное окно верхнего уровня self.root = tk.Toplevel(parent_menu.root) self.root.transient(parent_menu.root) self.root.grab_set() # Переменные движка превью PyMuPDF self.pdf_doc = None self.current_page = 0 self.src_file_path = "" self.last_base_date = "" # Список для динамических строк согласующих лиц (Лист согласования) self.agreement_rows = [] # Инициализация и отрисовка интерфейса из layout.py self.ui = AppLayout(self.root, self) # Настройка и запуск бизнес-хуков self._init_logic_bindings() self._preload_cached_roles() self._update_all_date_states() # Первичная блокировка дат при старте self.preview_engine = PDFPreviewEngine(self) def _init_logic_bindings(self): """Связывание событийно-ориентированной логики и автозаполнения полей.""" # Автоматический расчет имени файла СЭД по мере ввода Обозначения ЕСКД self.ui.ent_des.bind("<KeyRelease>", self._update_calculated_paths) # Привязка умной синхронизации дат к полю "Разработал" dev_entry = self.ui.fixed_roles["dev"]["ent_d"] dev_entry.bind("<KeyRelease>", self._on_dev_date_changed, add="+") # Привязка динамической блокировки полей дат к фиксированным ролям for role_key, info in self.ui.fixed_roles.items(): cb = info["cb"] # Отслеживаем выбор мышкой и ручной ввод текста в комбобокс cb.bind("<<ComboboxSelected>>", lambda e, r=role_key: self._toggle_date_entry_state(r)) cb.bind("<KeyRelease>", lambda e, r=role_key: self._toggle_date_entry_state(r)) def _toggle_date_entry_state(self, role_key): """Переключает доступность поля даты в зависимости от выбранного сотрудника.""" info = self.ui.fixed_roles[role_key] fio = info["cb"].get().strip() ent_d = info["ent_d"] if fio in ("", "---"): ent_d.delete(0, tk.END) ent_d.config(state="disabled") else: # Возвращаем режим ввода, если поле было заблокировано if str(ent_d.cget("state")) == "disabled": ent_d.config(state="normal") # Подставляем базовую дату для ускорения потокового ввода, если поле пустое if not ent_d.get() and self.last_base_date: ent_d.insert(0, self.last_base_date) def _update_all_date_states(self): """Служебный метод для первоначальной настройки состояний полей дат.""" for role_key in self.ui.fixed_roles: self._toggle_date_entry_state(role_key) def _update_calculated_paths(self, event=None): """Интерактивный расчет имени файла и пути хранения на лету.""" designation = self.ui.ent_des.get().strip() if not designation: self._set_readonly_entry(self.ui.ent_fname, "") self._set_readonly_entry(self.ui.ent_fpath, "") return filename, filepath = db.generate_paths(designation, "dummy.pdf") self._set_readonly_entry(self.ui.ent_fname, filename) self._set_readonly_entry(self.ui.ent_fpath, filepath) def _set_readonly_entry(self, entry_widget, text): """Вспомогательный метод безопасного обновления защищенных полей.""" entry_widget.config(state="normal") entry_widget.delete(0, tk.END) entry_widget.insert(0, text) entry_widget.config(state="readonly") def _on_dev_date_changed(self, event): """Умная каскадная синхронизация пустых дат утверждения (только для активных полей).""" current_dev_date = self.ui.fixed_roles["dev"]["ent_d"].get() for role_key in ["chk", "ctl", "app"]: ent_widget = self.ui.fixed_roles[role_key]["ent_d"] # Синхронизируем дату только если поле активно (выбран сотрудник) if str(ent_widget.cget("state")) == "normal": val = ent_widget.get() if val == "" or val == self.last_base_date: ent_widget.delete(0, tk.END) ent_widget.insert(0, current_dev_date) self.last_base_date = current_dev_date def _preload_cached_roles(self): """Заполнение выпадающих списков комбобоксов из ОЗУ-кэша config.py.""" role_mapping = { "dev": "dev", "chk": "chk", "ctl": "ctl", "app": "app" } for ui_key, config_key in role_mapping.items(): cb = self.ui.fixed_roles[ui_key]["cb"] cb['values'] = self.config.get_role_staff(config_key) cb.set("---") def browse_pdf(self): """Действие при нажатии кнопки 'Выбрать PDF'.""" init_dir = self.config.get_last_doc_directory() path = filedialog.askopenfilename( title="Выберите документ ЕСКД в формате PDF", initialdir=init_dir, filetypes=[("Электронные документы (*.pdf, *.PDF)", "*.pdf *.PDF")] ) if not path: return # Передаем файл в движок превью if self.preview_engine.load_pdf(path): self.src_file_path = path self.config.save_last_doc_directory(os.path.dirname(path)) # Выводим статус и MD5 на UI self.ui.lbl_status.config(text=f"Выбран: {os.path.basename(path)}", foreground="green") self.ui.lbl_md5.config(text=f"MD5: {self.preview_engine.current_md5}", foreground="gray") # Запускаем рендеринг страницы self.preview_engine.render_current_page(self.ui.lbl_preview, self.ui.lbl_page_counter) # Автозаполнение наименования if not self.ui.ent_title.get().strip(): base_name, _ = os.path.splitext(os.path.basename(path)) self.ui.ent_title.insert(0, base_name) logger.info(f"PDF загружен: {os.path.basename(path)} (MD5: {self.preview_engine.current_md5})") else: logger.error(f"Не удалось загрузить PDF: {path}") messagebox.showerror("Ошибка", "Не удалось загрузить PDF-файл!", parent=self.root) def prev_page(self): """Вызывается из layout.py кнопкой '◀ Назад'.""" if self.preview_engine.prev_page(): self.preview_engine.render_current_page(self.ui.lbl_preview, self.ui.lbl_page_counter) def next_page(self): """Вызывается из layout.py кнопкой 'Вперед ▶'.""" if self.preview_engine.next_page(): self.preview_engine.render_current_page(self.ui.lbl_preview, self.ui.lbl_page_counter) def add_agreement_row(self): """Динамическое добавление строки согласующего лица с блокировкой даты.""" row_frame = ttk.Frame(self.ui.rows_container) row_frame.pack(fill="x", pady=2, anchor="w") cb_agr = ttk.Combobox(row_frame, width=25, values=self.config.get_role_staff("agr")) cb_agr.pack(side="left", padx=5) cb_agr.set("---") ent_d = DateMaskEntry(row_frame, width=15) ent_d.pack(side="left", padx=5) # Кнопка удаления строки btn_del = ttk.Button(row_frame, text="❌", width=3, command=lambda: self._delete_agreement_row(row_frame, (cb_agr, ent_d))) btn_del.pack(side="left", padx=5) # По умолчанию комбобокс равен "---", поэтому блокируем поле даты при создании ent_d.config(state="disabled") # Привязываем замыкание для отслеживания состояния конкретной строки def toggle_dyn_date(event=None): fio = cb_agr.get().strip() if fio in ("", "---"): ent_d.delete(0, tk.END) ent_d.config(state="disabled") else: if str(ent_d.cget("state")) == "disabled": ent_d.config(state="normal") dev_date = self.ui.fixed_roles["dev"]["ent_d"].get() if dev_date and not ent_d.get(): ent_d.insert(0, dev_date) cb_agr.bind("<<ComboboxSelected>>", toggle_dyn_date) cb_agr.bind("<KeyRelease>", toggle_dyn_date) self.agreement_rows.append((cb_agr, ent_d, row_frame)) def _delete_agreement_row(self, frame, controls): """Удаление строки из листа согласования.""" for item in self.agreement_rows: if item[:2] == controls: self.agreement_rows.remove(item) break frame.destroy() def save(self): """Валидация бизнес-логики и фиксация данных в БД по правилам СЭД.""" designation = self.ui.ent_des.get().strip() title = self.ui.ent_title.get().strip() ver_str = self.ui.ent_ver.get().strip() chg_str = self.ui.ent_chg.get().strip() # 1. Проверка заполнения обязательных реквизитов изделия if not self.src_file_path: messagebox.showwarning("Внимание", "Не выбран исходный PDF-файл документа!", parent=self.root) return if not designation: messagebox.showwarning("Внимание", "Поле 'Обозначение' является обязательным!", parent=self.root) return if not title: messagebox.showwarning("Внимание", "Поле 'Наименование' является обязательным!", parent=self.root) return # 2. Валидация форматов дат (только для АКТИВНЫХ полей) for role_key, info in self.ui.fixed_roles.items(): ent_d = info["ent_d"] if str(ent_d.cget("state")) == "normal": dt = ent_d.get().strip() if not dt: messagebox.showerror("Ошибка заполнения", f"Выбран сотрудник для роли '{info['lbl']}', но не указана дата подписи!", parent=self.root) return if not db.validate_date(dt): messagebox.showerror("Ошибка формата дат", f"Неверный формат даты у роли '{info['lbl']}'! Шаблон: ГГГГ-ММ-ДД", parent=self.root) return # Валидация дат динамических листов согласования agreements_payload = [] for cb_agr, ent_d, _ in self.agreement_rows: if str(ent_d.cget("state")) == "normal": fio = cb_agr.get().strip() dt = ent_d.get().strip() if not dt: messagebox.showerror("Ошибка заполнения", f"Для согласующего '{fio}' не указана дата подписи!", parent=self.root) return if not db.validate_date(dt): messagebox.showerror("Ошибка формата дат", f"Неверный формат даты у согласующего {fio}!", parent=self.root) return agreements_payload.append({"fio": fio, "date": dt}) # 3. Контроль дубликатов по правилам СУБД (v2) checksum = db.calculate_md5(self.src_file_path) try: version_int = int(ver_str) if ver_str.isdigit() else 1 change_int = int(chg_str) if chg_str.isdigit() else 0 except ValueError: version_int, change_int = 1, 0 dup_error = db.check_duplicates_v2(designation, change_int, checksum) if dup_error: messagebox.showerror("Контроль дубликатов", dup_error, parent=self.root) return # 4. Формирование путей назначения внутри хранилища filename, filepath = db.generate_paths(designation, self.src_file_path) main_data = self._build_main_data(designation, title, filename, filepath, checksum, version_int, change_int) # 5. Вызов транзакции записи в SQLite try: db.insert_document(main_data, agreements_payload) # Добавление новых уникальных сотрудников в кэш ОЗУ на лету for role_key in self.ui.fixed_roles: fio_val = self.ui.fixed_roles[role_key]["cb"].get().strip() if fio_val and fio_val != "---": self.config.add_fio_to_cache(role_key, fio_val) for item in agreements_payload: self.config.add_fio_to_cache("agr", item["fio"]) logger.info(f"Документ '{designation}' сохранен в архив") messagebox.showinfo("Успех", f"Документ '{designation}' успешно сохранен в архив.", parent=self.root) if self.on_success_callback: self.on_success_callback() self.root.destroy() except Exception as e: logger.error(f"Критический сбой при транзакции записи: {e}") messagebox.showerror("Ошибка СУБД", f"Критический сбой при транзакции записи:\n{str(e)}", parent=self.root) def _get_role_fio(self, role_key): """Извлекает ФИО сотрудника из роли, если она активна.""" info = self.ui.fixed_roles[role_key] if str(info["ent_d"].cget("state")) == "normal": return info["cb"].get().strip() return "" def _get_role_date(self, role_key): """Извлекает дату из роли, если она активна.""" info = self.ui.fixed_roles[role_key] if str(info["ent_d"].cget("state")) == "normal": return info["ent_d"].get().strip() return "" def _build_main_data(self, designation, title, filename, filepath, checksum, version, change): """Формирует словарь метаданных для записи в БД.""" return { "filename": filename, "filepath": filepath, "checksum": checksum, "source_path": self.src_file_path, "designation": designation, "title": title, "version": version, "last_change": change, "dev_fio": self._get_role_fio("dev"), "dev_date": self._get_role_date("dev"), "chk_fio": self._get_role_fio("chk"), "chk_date": self._get_role_date("chk"), "ctl_fio": self._get_role_fio("ctl"), "ctl_date": self._get_role_date("ctl"), "app_fio": self._get_role_fio("app"), "app_date": self._get_role_date("app"), }