/
AxeOn
/
ProjectAi
Обзор
Документация
Войти
/
AxeOn
/
ProjectAi
Код
Запросы
0
Задачи
Вики
Пакеты
0
Релизы
0
CI/CD
Аналитика
Безопасность
master
quality_module.py
1 252 строки
60 KB
Дмитрий
chore: base state before refactoring (office_planner_v5)
25 апр 2026, 15:01
25 апр 2026, 15:01
bc896b2
Код
Авторство
О чём код?
# quality_module.py """ Модуль «Контроль качества» Пользователь выбирает пути к файлам вручную через диалог; пути сохраняются в настройках БД и восстанавливаются при следующем запуске. Файлы: • Основной — 2026.xlsx (Раздел = ПСО Имущество) • ПСО ИФЛ — источник дат (col3=Номер полиса, col6=Дата закрытия) • ПСО ИЮЛ — источник дат (col0=Номер дела/полиса, col21=Дата выполнения) • РИ — источник дат (col0=Номер дела/полиса, col21=Дата выполнения) """ import os import re import logging import tkinter as tk from tkinter import messagebox, filedialog from datetime import datetime import ttkbootstrap as ttk from ttkbootstrap.constants import * import pandas as pd logger = logging.getLogger(__name__) # Кириллическое слово (допускается дефис для двойных фамилий/имён) _CYR_WORD_RE = re.compile(r'^[А-ЯЁа-яё]+(?:-[А-ЯЁа-яё]+)?$') def _extract_fio(s: str) -> str: """Вернуть только ФИО из строки — первые 2–3 кириллических слова. Если не удалось выделить минимум 2 кириллических слова подряд с начала, возвращает исходную строку без изменений. """ if not s: return s parts = s.strip().split() name_parts = [] for p in parts: if _CYR_WORD_RE.match(p): name_parts.append(p) if len(name_parts) == 3: break else: break return ' '.join(name_parts) if len(name_parts) >= 2 else s.strip() ISPRAVLENIE_VALUES = ['Нет', 'Ошибка категории №1', 'Ошибка категории №2'] # Ключи хранения путей в таблице settings _SETTING_KEYS = { 'main': 'quality_file_main', 'ifl': 'quality_file_pso_ifl', 'iul': 'quality_file_pso_iul', 'ri': 'quality_file_ri', } # Метки для UI _FILE_LABELS = { 'main': 'Основной файл (Контроль.xlsx)', 'ifl': 'ПСО ИФЛ', 'iul': 'ПСО ИЮЛ', 'ri': 'РИ', } class QualityModule(ttk.Frame): """Модуль контроля качества осмотров ПСО Имущество""" COLUMNS = [ ('nomer_polisa', 'Номер полиса', 130), ('data_zakr', 'Дата закрытия', 90), ('mes_zakr', 'Мес', 40), ('god_zakr', 'Год', 50), ('osmotrowshik', 'Осмотровщик', 240), ('region_osm', 'Регион', 120), ('narusheniya', 'Нарушения', 360), ('naim_podr', 'Наим. подр. НЭ', 140), ('opisanie', 'Описание причины', 200), ('ispravlenie', 'Исправление рассм.', 160), ('data_osm', 'Дата осмотра', 90), ('prosrochka', 'Просрочка', 60), ('povtor', 'Повтор', 70), ('kto_obnaruzhil', 'Кто обнаружил', 140), ] def __init__(self, master, db): super().__init__(master) self.db = db self._combo_widget = None # inline combobox (Исправление) self._combo_item = None self._entry_widget = None # inline entry (Дата осмотра) self._entry_item = None self._sort_reverse: dict = {} # Словарь путей: ключ → StringVar (отображается в UI и сохраняется в БД) self._path_vars: dict[str, tk.StringVar] = {} self.pack(fill=BOTH, expand=YES) self._setup_ui() self._load_saved_paths() self.refresh_data() # ══════════════════════════════════════════════════════ UI setup ═══════ def _setup_ui(self): # ── Верхняя панель: выбор файлов ────────────────────────────────── files_lf = ttk.Labelframe(self, text='📁 Файлы для загрузки', padding=(10, 6)) files_lf.pack(fill=X, padx=8, pady=(8, 4)) for row_idx, key in enumerate(('main', 'ifl', 'iul', 'ri')): var = tk.StringVar() self._path_vars[key] = var ttk.Label( files_lf, text=_FILE_LABELS[key] + ':', width=28, anchor=E ).grid(row=row_idx, column=0, sticky=E, padx=(0, 6), pady=2) ttk.Entry( files_lf, textvariable=var, state='readonly', width=60 ).grid(row=row_idx, column=1, sticky=EW, padx=(0, 4), pady=2) ttk.Button( files_lf, text='📁', width=4, bootstyle='secondary-outline', command=lambda k=key: self._browse_file(k) ).grid(row=row_idx, column=2, pady=2) files_lf.columnconfigure(1, weight=1) # Кнопки действий справа от блока файлов btn_frame = ttk.Frame(files_lf) btn_frame.grid(row=0, column=3, rowspan=4, sticky=NS, padx=(12, 0)) ttk.Button( btn_frame, text='🔄 Загрузить данные', bootstyle=SUCCESS, command=self._load_all_files, width=20 ).pack(pady=(0, 6)) ttk.Button( btn_frame, text='📊 Экспорт', bootstyle='info-outline', command=self._export, width=20 ).pack(pady=(0, 6)) ttk.Button( btn_frame, text='📥 Импорт', bootstyle='warning-outline', command=self._import_ispravlenie, width=20 ).pack(pady=(0, 6)) ttk.Button( btn_frame, text='🗑 Удалить строку', bootstyle='danger-outline', command=self._delete_selected_row, width=20 ).pack() # ── Панель фильтров + статистика ────────────────────────────────── flt = ttk.Frame(self, padding=(8, 4)) flt.pack(fill=X) ttk.Label(flt, text='Осмотровщик:').pack(side=LEFT, padx=(0, 2)) self._cmb_osm = ttk.Combobox(flt, width=22, state='readonly') self._cmb_osm.pack(side=LEFT, padx=(0, 8)) ttk.Label(flt, text='Наим. подр. НЭ:').pack(side=LEFT, padx=(0, 2)) self._cmb_naim = ttk.Combobox(flt, width=20, state='readonly') self._cmb_naim.pack(side=LEFT, padx=(0, 8)) ttk.Label(flt, text='Месяц:').pack(side=LEFT, padx=(0, 2)) self._cmb_mes = ttk.Combobox(flt, width=5, state='readonly') self._cmb_mes.pack(side=LEFT, padx=(0, 8)) ttk.Label(flt, text='Год:').pack(side=LEFT, padx=(0, 2)) self._cmb_god = ttk.Combobox(flt, width=7, state='readonly') self._cmb_god.pack(side=LEFT, padx=(0, 8)) ttk.Label(flt, text='Исправление:').pack(side=LEFT, padx=(0, 2)) self._cmb_isp = ttk.Combobox( flt, values=['Все'] + ISPRAVLENIE_VALUES, width=22, state='readonly' ) self._cmb_isp.set('Все') self._cmb_isp.pack(side=LEFT, padx=(0, 6)) ttk.Button(flt, text='🔍', bootstyle='secondary-outline', command=self._apply_filter, width=4).pack(side=LEFT) ttk.Button(flt, text='✖', bootstyle='secondary-outline', command=self._clear_filter, width=4).pack(side=LEFT, padx=(4, 0)) self._lbl_stats = ttk.Label(flt, text='', bootstyle=SECONDARY) self._lbl_stats.pack(side=RIGHT, padx=8) # ── Таблица ──────────────────────────────────────────────────────── tbl = ttk.Frame(self) tbl.pack(fill=BOTH, expand=YES, padx=8, pady=(0, 8)) col_ids = [c[0] for c in self.COLUMNS] self.tree = ttk.Treeview( tbl, columns=col_ids, show='headings', selectmode='browse', bootstyle=PRIMARY ) for cid, cname, cwidth in self.COLUMNS: self.tree.heading(cid, text=cname, command=lambda c=cid: self._sort(c)) anchor = CENTER if cid in ('mes_zakr', 'god_zakr', 'prosrochka', 'povtor') else W self.tree.column(cid, width=cwidth, anchor=anchor, stretch=False) vsb = ttk.Scrollbar(tbl, orient=VERTICAL, command=self.tree.yview) hsb = ttk.Scrollbar(tbl, orient=HORIZONTAL, command=self.tree.xview) self.tree.configure(yscrollcommand=vsb.set, xscrollcommand=hsb.set) self.tree.tag_configure('net', background='#C8E8C0', foreground='#1a3a1a') self.tree.tag_configure('no_date', background='#F8F0D4', foreground='#3a2a00') self.tree.tag_configure('error1', background='#9090B8', foreground='white') self.tree.tag_configure('error2', background='#B89090', foreground='white') self.tree.grid(row=0, column=0, sticky=NSEW) vsb.grid(row=0, column=1, sticky=NS) hsb.grid(row=1, column=0, sticky=EW) tbl.rowconfigure(0, weight=1) tbl.columnconfigure(0, weight=1) self.tree.bind('<Double-Button-1>', self._on_double_click) self.tree.bind('<Button-1>', self._on_single_click) self.tree.bind('<Control-c>', self._copy_polisa) self.tree.bind('<Control-C>', self._copy_polisa) # ══════════════════════════════════════════════════════ Paths ══════════ def _load_saved_paths(self): """Восстановить сохранённые пути из БД""" for key, setting_key in _SETTING_KEYS.items(): val = self.db.get_setting(setting_key, '') if val: self._path_vars[key].set(val) def _browse_file(self, key: str): """Открыть диалог выбора файла и сохранить путь""" current = self._path_vars[key].get() init_dir = os.path.dirname(current) if current else os.path.join( os.path.dirname(__file__), 'Excel', 'Контроль' ) path = filedialog.askopenfilename( title=f'Выбрать файл — {_FILE_LABELS[key]}', initialdir=init_dir if os.path.isdir(init_dir) else os.path.dirname(__file__), filetypes=[('Excel файлы', '*.xlsx *.xls'), ('Все файлы', '*.*')] ) if path: self._path_vars[key].set(path) self.db.set_setting(_SETTING_KEYS[key], path) # ══════════════════════════════════════════════════════ Data ═══════════ @staticmethod def _parse_qdate(s): """Разобрать дату в формате ДД.ММ.ГГГГ или ГГГГ-ММ-ДД, вернуть datetime или None.""" if not s: return None s = str(s).strip() for fmt in ('%d.%m.%Y', '%Y-%m-%d'): try: return datetime.strptime(s, fmt) except ValueError: continue return None @staticmethod def _fmt_date(s): """ГГГГ-ММ-ДД → ДД.ММ.ГГГГ для отображения в таблице.""" if not s: return '' d = QualityModule._parse_qdate(str(s)) return d.strftime('%d.%m.%Y') if d else str(s) @staticmethod def _tree_date_to_db(s): """ДД.ММ.ГГГГ (из ячейки таблицы) → ГГГГ-ММ-ДД для запросов к БД.""" if not s: return None d = QualityModule._parse_qdate(str(s)) return d.strftime('%Y-%m-%d') if d else None def _compute_derived(self, rows): """Вычислить Просрочку и Повтор для списка строк из БД. Возвращает (prosrochka_vals, povtor_vals) — списки той же длины, что rows. prosrochka_vals: int (дни) или None. povtor_vals: строка из ⚡ или ''. """ from collections import defaultdict # Просрочка: data_zakr − data_osm в календарных днях prosrochka_vals = [] for r in rows: d_zakr = self._parse_qdate(r.get('data_zakr')) d_osm = self._parse_qdate(r.get('data_osm')) prosrochka_vals.append((d_zakr - d_osm).days if d_zakr and d_osm else None) # Повтор: тот же эксперт + то же нарушение в окне [d−90, d] expert_nar_dates: dict = defaultdict(list) row_dates = [] for r in rows: d = self._parse_qdate(r.get('data_osm')) or self._parse_qdate(r.get('data_zakr')) row_dates.append(d) osm = (r.get('osmotrowshik') or '').strip() nar = (r.get('narusheniya') or '').strip() if osm and nar and d: expert_nar_dates[(osm, nar)].append(d) povtor_vals = [] for i, r in enumerate(rows): osm = (r.get('osmotrowshik') or '').strip() nar = (r.get('narusheniya') or '').strip() d = row_dates[i] key = (osm, nar) if not osm or not nar or not d or key not in expert_nar_dates: povtor_vals.append('') continue count = sum(1 for dd in expert_nar_dates[key] if 0 <= (d - dd).days <= 90) povtor_vals.append('⚡' * count if count >= 2 else '') return prosrochka_vals, povtor_vals def refresh_data(self, osmotrowshik=None, naim_podr=None, ispravlenie=None, mes=None, god=None): """Перезаполнить таблицу из БД""" self._hide_combo() for item in self.tree.get_children(): self.tree.delete(item) try: rows = self.db.get_quality_cases( osmotrowshik, naim_podr, ispravlenie, mes, god ) except Exception as e: logger.error(f'QualityModule.refresh_data: {e}', exc_info=True) messagebox.showerror('Ошибка', f'Ошибка загрузки данных:\n{e}') return prosrochka_vals, povtor_vals = self._compute_derived(rows) col_ids = [c[0] for c in self.COLUMNS] total = len(rows) with_date = 0 for i, r in enumerate(rows): values = [] for cid in col_ids: if cid == 'prosrochka': v = prosrochka_vals[i] values.append('' if v is None else v) elif cid == 'povtor': values.append(povtor_vals[i]) elif r.get(cid) is None: values.append('') elif cid in ('mes_zakr', 'god_zakr'): values.append(int(r[cid]) if r[cid] else '') elif cid in ('data_zakr', 'data_osm'): values.append(self._fmt_date(r.get(cid))) else: values.append(str(r[cid])) if r.get('data_zakr'): with_date += 1 isp = r.get('ispravlenie', '') if isp == 'Ошибка категории №1': tag = 'error1' elif isp == 'Ошибка категории №2': tag = 'error2' elif isp == 'Нет': tag = 'net' elif not r.get('data_zakr'): tag = 'no_date' else: tag = '' self.tree.insert('', END, values=values, tags=(tag,) if tag else ()) self._lbl_stats.config( text=(f'Всего: {total} | ' f'С датой закрытия: {with_date} | ' f'Без даты: {total - with_date}') ) self._refresh_filter_lists() def _refresh_filter_lists(self): try: osm_list = ['Все'] + self.db.get_quality_osmotrowshiki() naim_list = ['Все'] + self.db.get_quality_naim_podr() mes_list = ['Все'] + self.db.get_quality_months() god_list = ['Все'] + self.db.get_quality_years() except Exception: return cur_osm = self._cmb_osm.get() cur_naim = self._cmb_naim.get() cur_mes = self._cmb_mes.get() cur_god = self._cmb_god.get() self._cmb_osm['values'] = osm_list self._cmb_naim['values'] = naim_list self._cmb_mes['values'] = mes_list self._cmb_god['values'] = god_list if cur_osm not in osm_list: self._cmb_osm.set('Все') if cur_naim not in naim_list: self._cmb_naim.set('Все') if cur_mes not in mes_list: self._cmb_mes.set('Все') if cur_god not in god_list: self._cmb_god.set('Все') # ══════════════════════════════════════════════════════ Filters ════════ def _get_filter_params(self) -> dict: """Текущие значения фильтров как kwargs для get_quality_cases.""" def _val(v): return v if v and v != 'Все' else None return dict( osmotrowshik=_val(self._cmb_osm.get()), naim_podr=_val(self._cmb_naim.get()), mes=_val(self._cmb_mes.get()), god=_val(self._cmb_god.get()), ispravlenie=_val(self._cmb_isp.get()), ) def _apply_filter(self): self.refresh_data(**self._get_filter_params()) def _clear_filter(self): self._cmb_osm.set('Все') self._cmb_naim.set('Все') self._cmb_mes.set('Все') self._cmb_god.set('Все') self._cmb_isp.set('Все') self.refresh_data() # ══════════════════════════════════════════════ Inline Combobox ════════ def _on_single_click(self, event): """Скрывает открытые inline-виджеты при одиночном клике.""" self._hide_combo() self._hide_entry() def _on_double_click(self, event): region = self.tree.identify_region(event.x, event.y) if region != 'cell': return col = self.tree.identify_column(event.x) item = self.tree.identify_row(event.y) if not item: return col_ids = [c[0] for c in self.COLUMNS] col_idx = int(col.replace('#', '')) - 1 if col_idx < 0 or col_idx >= len(col_ids): return cid = col_ids[col_idx] if cid == 'ispravlenie': self._open_ispravlenie_combo(item, col) elif cid == 'data_osm': self._open_data_osm_entry(item, col) def _open_ispravlenie_combo(self, item, col): self._hide_combo() self._hide_entry() self._combo_item = item bbox = self.tree.bbox(item, col) if not bbox: return x, y, w, h = bbox _isp_opts = [''] + ISPRAVLENIE_VALUES current_val = self.tree.set(item, 'ispravlenie') self._combo_widget = ttk.Combobox( self.tree, values=_isp_opts, state='readonly', width=w // 8 ) self._combo_widget.set( current_val if current_val in _isp_opts else '' ) self._combo_widget.place(x=x, y=y, width=w, height=h) self._combo_widget.focus_set() self._combo_widget.bind('<<ComboboxSelected>>', self._on_combo_select) self._combo_widget.bind('<Escape>', lambda e: self._hide_combo()) self._combo_widget.bind('<FocusOut>', lambda e: self._hide_combo()) def _open_data_osm_entry(self, item, col): self._hide_combo() self._hide_entry() self._entry_item = item bbox = self.tree.bbox(item, col) if not bbox: return x, y, w, h = bbox current_val = self.tree.set(item, 'data_osm') self._entry_widget = ttk.Entry(self.tree, width=w // 8) self._entry_widget.insert(0, current_val) self._entry_widget.select_range(0, 'end') self._entry_widget.place(x=x, y=y, width=w, height=h) self._entry_widget.focus_set() self._entry_widget.bind('<Return>', self._on_entry_confirm) self._entry_widget.bind('<KP_Enter>', self._on_entry_confirm) self._entry_widget.bind('<Escape>', lambda e: self._hide_entry()) self._entry_widget.bind('<FocusOut>', lambda e: self._hide_entry()) def _on_combo_select(self, event): if not self._combo_widget or not self._combo_item: return new_val = self._combo_widget.get() item = self._combo_item self._hide_combo() nomer_polisa = self.tree.set(item, 'nomer_polisa') if not nomer_polisa: return data_osm = self._tree_date_to_db(self.tree.set(item, 'data_osm')) try: self.db.update_quality_ispravlenie(nomer_polisa, new_val, data_osm) self.tree.set(item, 'ispravlenie', new_val) if new_val == 'Ошибка категории №1': tag = 'error1' elif new_val == 'Ошибка категории №2': tag = 'error2' elif new_val == 'Нет': tag = 'net' elif not self.tree.set(item, 'data_zakr'): tag = 'no_date' else: tag = '' self.tree.item(item, tags=(tag,) if tag else ()) except Exception as e: logger.error(f'update_quality_ispravlenie: {e}', exc_info=True) messagebox.showerror('Ошибка', f'Не удалось сохранить:\n{e}') def _hide_combo(self, event=None): if self._combo_widget: self._combo_widget.destroy() self._combo_widget = None self._combo_item = None def _hide_entry(self, event=None): if self._entry_widget: self._entry_widget.destroy() self._entry_widget = None self._entry_item = None def _on_entry_confirm(self, event=None): if not self._entry_widget or not self._entry_item: return raw = self._entry_widget.get().strip() item = self._entry_item self._hide_entry() # Валидация и нормализация: принимаем ДД.ММ.ГГГГ или ГГГГ-ММ-ДД d = self._parse_qdate(raw) if raw else None if raw and not d: messagebox.showwarning( 'Неверная дата', f'Введите дату в формате ДД.ММ.ГГГГ\n(получено: «{raw}»)', parent=self ) return display_val = d.strftime('%d.%m.%Y') if d else '' db_val = d.strftime('%Y-%m-%d') if d else None nomer_polisa = self.tree.set(item, 'nomer_polisa') old_data_osm = self._tree_date_to_db(self.tree.set(item, 'data_osm')) try: self.db.update_quality_data_osm(nomer_polisa, old_data_osm, db_val) self.tree.set(item, 'data_osm', display_val) # Пересчёт просрочки: data_zakr − data_osm d_zakr = self._parse_qdate(self.tree.set(item, 'data_zakr')) prosrochka = (d_zakr - d).days if d_zakr and d else '' self.tree.set(item, 'prosrochka', prosrochka) except Exception as e: logger.error(f'update_quality_data_osm: {e}', exc_info=True) messagebox.showerror('Ошибка', f'Не удалось сохранить дату:\n{e}', parent=self) def _copy_polisa(self, event=None): """Копировать Номер полиса выбранной строки в буфер обмена (Ctrl+C).""" item = self.tree.focus() if not item: return val = self.tree.set(item, 'nomer_polisa') if val: self.clipboard_clear() self.clipboard_append(val) # ══════════════════════════════════════════════════════ Sort ══════════ def _sort(self, col_id): rev = self._sort_reverse.get(col_id, False) data = [(self.tree.set(it, col_id), it) for it in self.tree.get_children('')] data.sort(key=lambda x: (x[0] == '', x[0]), reverse=rev) for idx, (_, it) in enumerate(data): self.tree.move(it, '', idx) self._sort_reverse[col_id] = not rev # ══════════════════════════════════════════════════ Load Excel ══════════ def _load_all_files(self): """Загрузить данные из всех выбранных файлов и добавить только новые строки. Уникальность строки определяется составным ключом (Номер полиса, Дата осмотра). Существующие строки никогда не изменяются. ПСО ИФЛ/ИЮЛ/РИ используются только для дозаполнения новых строк. """ main_path = self._path_vars['main'].get().strip() if not main_path: messagebox.showwarning( 'Файл не выбран', 'Укажите основной файл (2026.xlsx) с помощью кнопки 📁.' ) return if not os.path.exists(main_path): messagebox.showerror( 'Файл не найден', f'Основной файл не найден:\n{main_path}' ) return try: rows, with_date, total, excluded_sa, skipped_no_date = self._parse_files( main_path = main_path, pso_ifl_path = self._path_vars['ifl'].get().strip() or None, pso_iul_path = self._path_vars['iul'].get().strip() or None, ri_path = self._path_vars['ri'].get().strip() or None, ) except Exception as e: logger.error(f'_load_all_files parse error: {e}', exc_info=True) messagebox.showerror('Ошибка чтения', f'Ошибка при чтении файлов:\n{e}') return if not rows: messagebox.showinfo( 'Нет данных', 'Не найдено строк с признаком ПСО Имущество в основном файле.' ) return # ── Определяем новые строки по составному ключу (полис + дата осмотра) ── existing_keys = { (r['nomer_polisa'], r.get('data_osm') or '') for r in self.db.get_quality_cases() if r.get('nomer_polisa') } logger.info(f'В базе найдено уникальных ключей: {len(existing_keys)}') new_rows = [r for r in rows if (r['nomer_polisa'], r.get('data_osm') or '') not in existing_keys] skipped_exist = len(rows) - len(new_rows) logger.info(f'Из файла: {len(rows)}, новых: {len(new_rows)}, уже в базе: {skipped_exist}') if not new_rows: messagebox.showinfo( 'Нет новых строк', f'Все {total} строк из файла уже присутствуют в базе данных.\n' f'Новых строк не обнаружено.' ) return # ── Сообщение пользователю ───────────────────────────────────────── msg_lines = [ f'Всего строк ПСО в файле: {total}', f' · Уже в базе (пропустить): {skipped_exist}', f' · Без даты закрытия (пропуск): {skipped_no_date}', f' · Новых для добавления: {len(new_rows)}', ] if excluded_sa: msg_lines.append(f' · Исключено «Сторонний акт»: {excluded_sa}') for src_key, src_label in [('ifl', 'ПСО ИФЛ'), ('iul', 'ПСО ИЮЛ'), ('ri', 'РИ')]: p = self._path_vars[src_key].get().strip() if not p: msg_lines.append(f'⚠️ {src_label}: файл не выбран — даты не дополнялись.') elif not os.path.exists(p): msg_lines.append(f'❌ {src_label}: файл НЕ НАЙДЕН (путь устарел?):\n {p}') msg_lines.append('\nДобавить новые строки в базу данных?') if not messagebox.askyesno('Загрузка данных', '\n'.join(msg_lines)): return try: inserted = self.db.upsert_quality_cases(new_rows) msg = f'Добавлено {inserted} новых записей.' if skipped_exist: msg += f'\nПропущено (уже в базе): {skipped_exist}.' if excluded_sa: msg += f'\nИсключено «Сторонний акт»: {excluded_sa}.' messagebox.showinfo('Готово', msg) self.refresh_data() except Exception as e: logger.error(f'_load_all_files upsert error: {e}', exc_info=True) messagebox.showerror('Ошибка сохранения', f'Ошибка при сохранении в БД:\n{e}') # ──────────────────────────────────────────────── parsing logic ───────── def _parse_files(self, main_path: str, pso_ifl_path: str | None, pso_iul_path: str | None, ri_path: str | None): """ Разбирает основной файл и три источника дат. Возвращает (rows, with_date, total, excluded_sa). """ import warnings warnings.filterwarnings('ignore') # ── Основной файл ───────────────────────────────────────────────── # Выбираем строку заголовка: берём ту из row0/row1, в которой # больше ключевых слов, характерных для заголовков таблицы. _raw = pd.read_excel(main_path, header=None, nrows=2) _HDR_KWS = ('номер', 'полис', 'заявк', 'раздел', 'риск', 'дата', 'осмотр', 'нарушен', 'регион') def _hdr_score(row): text = ' '.join(str(v).strip() for v in row if pd.notna(v)).lower() return sum(1 for kw in _HDR_KWS if kw in text) _s0 = _hdr_score(_raw.iloc[0]) _s1 = _hdr_score(_raw.iloc[1]) if len(_raw) > 1 else 0 _header_row = 1 if _s1 > _s0 else 0 logger.info(f'Основной файл: header row={_header_row} (score0={_s0}, score1={_s1})') df = pd.read_excel(main_path, header=_header_row) # Маппинг колонок по ИМЕНАМ (не по позициям) — независим от порядка в файле # nomer_polisa = «Номер полиса» (SYS-ID) — ключ поиска в ПСО ИФЛ/ИЮЛ/РИ # «Уч № заявки» — бухгалтерский номер заявки, не используется в поиске _MAIN_COL_MAP = { 'Номер полиса': 'nomer_polisa', 'Риск': 'risk', 'Раздел': 'razdel', 'Дата закрытия заявки': 'data_zakr', 'Осмотровщик': 'osmotrowshik', 'Регион осмотра': 'region_osm', 'Нарушения при осмотре': 'narusheniya', 'Наименование подр. НЭ': 'naim_podr', 'Описание причины изм. актa': 'opisanie', 'Дата осмотра': 'data_osm', 'Кто обнаружил': 'kto_obnaruzhil', } _missing = [k for k in _MAIN_COL_MAP if k not in df.columns] if _missing: raise ValueError(f'В основном файле не найдены колонки: {_missing}') df = df.rename(columns=_MAIN_COL_MAP) logger.info(f'Основной файл: {len(df)} строк, маппинг применён успешно') # Фильтр 1: только ПСО Имущество (по именованным колонкам) pso_mask = ( df['risk'].astype(str).str.contains('ПСО', case=False, na=False) | df['razdel'].astype(str).str.contains('ПСО', case=False, na=False) ) logger.info(f'После фильтра ПСО: {pso_mask.sum()} строк из {len(df)}') df = df[pso_mask].copy() # Фильтр 2: исключаем строки с «Сторонний акт» в Нарушениях before = len(df) mask_sa = df['narusheniya'].astype(str).str.contains( 'Сторонний акт', na=False, case=False ) df = df[~mask_sa].copy() excluded_sa = before - len(df) if excluded_sa: logger.info(f'Исключено строк «Сторонний акт»: {excluded_sa}') # ── Справочники из source-файлов ────────────────────────────────── date_lookup: dict = {} # key → дата выполнения услуги executor_lookup: dict = {} # key → Выполнил услугу (для Осмотровщик) naim_podr_lookup: dict = {} # key → Наименование подр. НЭ def _load_source(path: str, label: str): """ Загружает дату и исполнителя из файла-источника. Ищет join-ключ, дату и «Выполнил услугу» по НАЗВАНИЮ колонки, при неудаче — по стандартным позициям (col 0, col 18, col 21). Поддерживаемые имена ключа: 'Номер дела/полиса', 'Номер полиса' Поддерживаемые имена даты: 'Дата выполнения услуги', 'Дата закрытия заявки', 'Дата осмотра' Поддерживаемые имена исп.: 'Выполнил услугу' """ if not path or not os.path.exists(path): if path: logger.warning(f'Source file not found: {path}') return try: # Автоопределение строки заголовка (0 или 1) _r = pd.read_excel(path, header=None, nrows=2) _r0 = [str(v).strip() for v in _r.iloc[0] if pd.notna(v)] _hdr = 0 if any( kw in ' '.join(_r0).lower() for kw in ('номер', 'дата', 'подразд', 'статус', 'регион') ) else 1 src = pd.read_excel(path, header=_hdr) cols = list(src.columns) cols_lower = [str(c).lower() for c in cols] # Ищем ключевую колонку (по названию) key_col = None for kw in ('номер дела/полиса', 'номер полиса', 'номер'): for i, cl in enumerate(cols_lower): if kw in cl: key_col = i break if key_col is not None: break if key_col is None: key_col = 0 # fallback # Контентная проверка: убеждаемся, что в найденной колонке # действительно лежат ID-значения вида «Буквы+цифры» (SYS..., POL...). # Если нет — ищем другую колонку с таким форматом. _id_pat = re.compile(r'^[A-Za-zА-Яа-я]{2,}\d{4,}') _sample_key = src.iloc[:20, key_col].dropna().astype(str) if not _sample_key.str.match(_id_pat).any(): for _ci in range(len(cols)): if _ci == key_col: continue _s2 = src.iloc[:20, _ci].dropna().astype(str) if _s2.str.match(_id_pat).any(): logger.info( f'{label}: key_col исправлен по содержимому: ' f'{key_col}({cols[key_col]!r}) → {_ci}({cols[_ci]!r})' ) key_col = _ci break # Ищем колонку с датой date_col = None for kw in ('дата выполнения', 'дата закрытия', 'дата осмотра'): for i, cl in enumerate(cols_lower): if kw in cl: date_col = i break if date_col is not None: break if date_col is None: date_col = 21 # fallback # Ищем колонку «Выполнил услугу» exec_col = None for kw in ('выполнил услугу', 'выполнил'): for i, cl in enumerate(cols_lower): if kw in cl: exec_col = i break if exec_col is not None: break if exec_col is None: exec_col = 18 # fallback exec_col_name = cols[exec_col] if exec_col < len(cols) else '?' # Ищем колонку «Наименование подр. НЭ» / «Подр. НЭ» # Варианты названий: «Подр. НЭ», «Подразделение НЭ», # «Наим. подр. НЭ», «Наименование подразделения НЭ» naim_col = None for i, cl in enumerate(cols_lower): if ('подр' in cl or 'наим' in cl) and 'нэ' in cl: naim_col = i break # Фильтр по статусу: берём только строки «Выполнено» # (у одного полиса может быть несколько строк с разными статусами) status_col = None for _i, _cl in enumerate(cols_lower): if 'статус' in _cl: # ловим «Статус услуги», «Статус» и т.п. status_col = _i break _rows_before = len(src) if status_col is not None: # Используем contains('выполнен') вместо точного сравнения: # устойчиво к невидимым символам, разному регистру, «Выполнена» и пр. _status_vals = (src.iloc[:, status_col] .astype(str) .str.replace('\u00a0', ' ', regex=False) .str.strip() .str.lower()) _mask = _status_vals.str.contains('выполнен', na=False) src = src[_mask].copy() logger.info( f'{label}: фильтр «выполнен*» по «{cols[status_col]}» — ' f'оставлено {len(src)} из {_rows_before} строк' ) else: logger.warning( f'{label}: колонка «Статус» не найдена, ' f'фильтрация по статусу не применена' ) logger.info( f'{label}: key_col={key_col} ({cols[key_col]!r}), ' f'date_col={date_col} ({cols[date_col]!r}), ' f'exec_col={exec_col} ({exec_col_name!r})' ) loaded_dates = 0 loaded_exec = 0 for _, row in src.iterrows(): # Нормализуем ключ: убираем пробелы (включая неразрывные), # приводим к верхнему регистру — это устраняет выборочные # несовпадения из-за разного форматирования ячеек в Excel. key_raw = str(row.iloc[key_col]) key = key_raw.strip().replace('\u00a0', '').replace(' ', '').upper() if not key or key == 'NAN': continue # Дата val = row.iloc[date_col] if pd.notna(val) and key not in date_lookup: date_lookup[key] = val loaded_dates += 1 # Выполнил услугу → executor_lookup if key not in executor_lookup and exec_col < len(row): exec_val = row.iloc[exec_col] if pd.notna(exec_val): s_exec = str(exec_val).strip() if s_exec and s_exec.lower() != 'nan': executor_lookup[key] = s_exec loaded_exec += 1 # Наименование подр. НЭ → naim_podr_lookup if key not in naim_podr_lookup and naim_col is not None and naim_col < len(row): naim_val = row.iloc[naim_col] if pd.notna(naim_val): s_naim = str(naim_val).strip() if s_naim and s_naim.lower() != 'nan': naim_podr_lookup[key] = s_naim logger.info( f'{label}: добавлено дат={loaded_dates}, ' f'исполнителей={loaded_exec} в справочник' ) except Exception as exc: logger.warning(f'Cannot read {label} ({path}): {exc}') # Порядок сверки: ИФЛ → ИЮЛ → РИ (первый нашедший запись выигрывает) _load_source(pso_ifl_path, 'ПСО ИФЛ') _load_source(pso_iul_path, 'ПСО ИЮЛ') _load_source(ri_path, 'РИ') # ── Текущие значения 'Исправление' в БД (не затираем) ───────────── db_cache = {r['nomer_polisa']: r for r in self.db.get_quality_cases()} # ── Формируем строки ─────────────────────────────────────────────── rows = [] with_date = 0 def _s(v): if v is None or (isinstance(v, float) and pd.isna(v)): return '' s = str(v).replace('\u00a0', ' ').strip() return '' if s.lower() == 'nan' else s def _nk(s: str) -> str: """Нормализованный ключ для поиска в lookup-словарях.""" return s.strip().replace('\u00a0', '').replace(' ', '').upper() for _, r in df.iterrows(): polisa = _s(r.get('nomer_polisa')) # Номер полиса (SYS-ID) → в БД и поиск if not polisa: continue polisa_key = _nk(polisa) # нормализован → совпадает с ПСО ИФЛ/ИЮЛ/РИ # Дата осмотра из основного файла (нужна ниже как fallback) d_osm_raw = r.get('data_osm') data_osm_str = None if pd.notna(d_osm_raw): try: data_osm_str = pd.Timestamp(d_osm_raw).strftime('%Y-%m-%d') except Exception: data_osm_str = str(d_osm_raw) # Дата закрытия: приоритеты # 1. Колонка «Дата закрытия заявки» основного файла # 2. Справочник из источников (ПСО ИФЛ → ПСО ИЮЛ → РИ) # 3. «Дата осмотра» из основного файла (тот же день выполнения) d_zakr_raw = r.get('data_zakr') if pd.isna(d_zakr_raw) and polisa_key in date_lookup: d_zakr_raw = date_lookup[polisa_key] if pd.isna(d_zakr_raw) and pd.notna(d_osm_raw): d_zakr_raw = d_osm_raw # fallback: дата осмотра = дата выполнения data_zakr_str = None mes = None god = None if pd.notna(d_zakr_raw): try: ts = pd.Timestamp(d_zakr_raw) data_zakr_str = ts.strftime('%Y-%m-%d') mes = int(ts.month) god = int(ts.year) with_date += 1 except Exception: data_zakr_str = str(d_zakr_raw) # Осмотровщик: берём из основного файла, при пустом — из «Выполнил услугу» osmotrowshik_val = _s(r.get('osmotrowshik')) if not osmotrowshik_val and polisa_key in executor_lookup: osmotrowshik_val = executor_lookup[polisa_key] osmotrowshik_val = _extract_fio(osmotrowshik_val) # Наим. подр. НЭ: берём из основного файла, при пустом — из источников naim_podr_val = _s(r.get('naim_podr')) if not naim_podr_val and polisa_key in naim_podr_lookup: naim_podr_val = naim_podr_lookup[polisa_key] # Значение «Исправление» — берём из БД, если уже задано вручную ispravlenie = db_cache.get(polisa, {}).get('ispravlenie', 'Нет') rows.append({ 'nomer_polisa': polisa, 'risk': _s(r.get('risk')), 'razdel': _s(r.get('razdel')), 'data_zakr': data_zakr_str, 'mes_zakr': mes, 'god_zakr': god, 'osmotrowshik': osmotrowshik_val, 'region_osm': _s(r.get('region_osm')), 'narusheniya': _s(r.get('narusheniya')), 'naim_podr': naim_podr_val, 'opisanie': _s(r.get('opisanie')), 'ispravlenie': ispravlenie, 'data_osm': data_osm_str, 'kto_obnaruzhil': _s(r.get('kto_obnaruzhil')), }) # Убираем строки без даты закрытия — в БД не сохраняем rows_with_date = [r for r in rows if r.get('data_zakr')] skipped_no_date = len(rows) - len(rows_with_date) return rows_with_date, len(rows_with_date), len(rows), excluded_sa, skipped_no_date # ══════════════════════════════════════════════════════ Delete row ══════ def _delete_selected_row(self): """Удалить выбранную строку из БД и таблицы.""" item = self.tree.focus() if not item: messagebox.showwarning('Удаление', 'Выберите строку для удаления.', parent=self) return nomer_polisa = self.tree.set(item, 'nomer_polisa') if not nomer_polisa: return data_osm = self._tree_date_to_db(self.tree.set(item, 'data_osm')) if not messagebox.askyesno( 'Подтверждение удаления', f'Удалить запись:\n{nomer_polisa}?', parent=self ): return try: self.db.delete_quality_case(nomer_polisa, data_osm) except Exception as e: logger.error(f'delete_quality_case: {e}', exc_info=True) messagebox.showerror('Ошибка', f'Не удалось удалить запись:\n{e}', parent=self) return self.tree.delete(item) # ══════════════════════════════════════════════════════ Import ══════════ def _import_ispravlenie(self): """Импортировать значения 'Исправление рассм.' из экспортированного Excel-файла.""" path = filedialog.askopenfilename( title='Импорт Исправление рассм.', filetypes=[('Excel files', '*.xlsx')], parent=self, ) if not path: return try: import openpyxl wb = openpyxl.load_workbook(path, read_only=True, data_only=True) ws = wb.active updated = 0 skipped = 0 # Row 1 is header; group-header rows have no value in column B (Осмотровщик) for row in ws.iter_rows(min_row=2, values_only=True): # col index 0 = nomer_polisa, col index 1 = osmotrowshik (None in group headers) # col index 8 = ispravlenie nomer_polisa = row[0] if len(row) > 0 else None osmotrowshik = row[1] if len(row) > 1 else None ispravlenie = row[8] if len(row) > 8 else None # Skip group-header rows (no osmotrowshik) and rows without policy number if not nomer_polisa or not osmotrowshik: skipped += 1 continue nomer_polisa = str(nomer_polisa).strip() ispravlenie_val = str(ispravlenie).strip() if ispravlenie is not None else '' # Allow only known values or empty string allowed = [''] + ISPRAVLENIE_VALUES if ispravlenie_val not in allowed: ispravlenie_val = '' try: self.db.update_quality_ispravlenie(nomer_polisa, ispravlenie_val) updated += 1 except Exception as e: logger.error(f'import_ispravlenie update error for {nomer_polisa}: {e}') skipped += 1 wb.close() except Exception as e: logger.error(f'_import_ispravlenie: {e}', exc_info=True) messagebox.showerror('Ошибка импорта', f'Не удалось прочитать файл:\n{e}', parent=self) return messagebox.showinfo( 'Импорт завершён', f'Обновлено записей: {updated}\nПропущено строк: {skipped}', parent=self, ) self.refresh_data() # ══════════════════════════════════════════════════════ Export ══════════ def _export(self): try: rows = self.db.get_quality_cases(**self._get_filter_params()) if not rows: messagebox.showinfo('Экспорт', 'Нет данных для экспорта.') return path = filedialog.asksaveasfilename( title='Экспорт контроля качества', defaultextension='.xlsx', filetypes=[('Excel', '*.xlsx'), ('Все файлы', '*.*')] ) if not path: return import openpyxl from openpyxl.styles import PatternFill, Font, Alignment from openpyxl.utils import get_column_letter from openpyxl.worksheet.datavalidation import DataValidation from collections import defaultdict prosrochka_vals, povtor_vals = self._compute_derived(rows) # Группировка по Наим. подр. НЭ groups: dict = defaultdict(list) for i, r in enumerate(rows): dept = (r.get('naim_podr') or '').strip() or '(не указано)' groups[dept].append(i) EXPORT_COLS = [ ('nomer_polisa', 'Номер полиса', 18), ('osmotrowshik', 'Осмотровщик', 28), ('data_osm', 'Дата осмотра', 14), ('data_zakr', 'Дата закрытия', 14), ('narusheniya', 'Нарушения', 35), ('opisanie', 'Описание причины', 45), ('prosrochka', 'Просрочка', 11), ('povtor', 'Повтор', 9), ('ispravlenie', 'Исправление рассм.', 28), ] n_cols = len(EXPORT_COLS) ISP_COL = n_cols # номер столбца Исправление (1-based) # ── Стили ────────────────────────────────────────────────────────── hdr_fill = PatternFill('solid', fgColor='2F5496') # тёмно-синий dept_fill = PatternFill('solid', fgColor='D6DCE4') # светло-серый warn_fill = PatternFill('solid', fgColor='FFAAAA') # красноватый hdr_font = Font(bold=True, color='FFFFFF', size=10) dept_font = Font(bold=True, size=10) center = Alignment(horizontal='center', vertical='center', wrap_text=False) wrap_left = Alignment(horizontal='left', vertical='top', wrap_text=True) wb = openpyxl.Workbook() ws = wb.active ws.title = 'Контроль качества' # ── Шапка (строка 1) ─────────────────────────────────────────────── for col_idx, (_, cname, cw) in enumerate(EXPORT_COLS, 1): cell = ws.cell(row=1, column=col_idx, value=cname) cell.fill = hdr_fill cell.font = hdr_font cell.alignment = center ws.column_dimensions[get_column_letter(col_idx)].width = cw ws.row_dimensions[1].height = 20 # ── Data Validation: выпадающий список для «Исправление рассм.» ─── isp_formula = '"' + ','.join(ISPRAVLENIE_VALUES) + '"' dv = DataValidation( type='list', formula1=isp_formula, allow_blank=True, showDropDown=False, showErrorMessage=False, ) ws.add_data_validation(dv) current_row = 2 for dept in sorted(groups.keys()): indices = groups[dept] # Строка-заголовок группы (Наим. подр. НЭ) ws.merge_cells( start_row=current_row, start_column=1, end_row=current_row, end_column=n_cols ) cell = ws.cell(row=current_row, column=1, value=dept) cell.fill = dept_fill cell.font = dept_font cell.alignment = Alignment(horizontal='left', vertical='center', indent=1) ws.row_dimensions[current_row].height = 16 current_row += 1 for idx in indices: r = rows[idx] for col_idx, (cid, _, _) in enumerate(EXPORT_COLS, 1): if cid == 'prosrochka': v = prosrochka_vals[idx] cell = ws.cell(row=current_row, column=col_idx, value=v if v is not None else '') cell.alignment = center if isinstance(v, int) and v > 0: cell.fill = warn_fill elif cid == 'povtor': pv = povtor_vals[idx] pv_num = len(pv) if pv else '' cell = ws.cell(row=current_row, column=col_idx, value=pv_num) cell.alignment = center if pv_num: cell.fill = warn_fill elif cid == 'ispravlenie': cell = ws.cell(row=current_row, column=col_idx, value=r.get(cid) or '') cell.alignment = center dv.add(cell) elif cid in ('data_osm', 'data_zakr'): cell = ws.cell(row=current_row, column=col_idx, value=self._fmt_date(r.get(cid))) cell.alignment = center else: cell = ws.cell(row=current_row, column=col_idx, value=r.get(cid) or '') cell.alignment = wrap_left current_row += 1 ws.freeze_panes = 'A2' wb.save(path) messagebox.showinfo('Экспорт', f'Файл сохранён:\n{path}') except Exception as e: logger.error(f'_export: {e}', exc_info=True) messagebox.showerror('Ошибка', f'Ошибка экспорта:\n{e}')