# -*- coding: utf-8 -*- """DokoGen — Модуль импорта из Excel (опросный лист 152-ФЗ)""" import re import traceback from datetime import datetime, timedelta from typing import Callable, Optional try: import openpyxl except ImportError: openpyxl = None from .import_settings import load_settings, parse_cell_ref, cell_value def clean_value(val): """Очистка и нормализация значения ячейки Excel.""" if val is None: return '' if isinstance(val, (int, float)): # Попытка распознать дату (серийный номер Excel) if 10000 < val < 80000: serial = int(val) if serial > 60: serial -= 1 dt = datetime(1899, 12, 30) + timedelta(days=serial) return dt.strftime('%d.%m.%Y') if isinstance(val, float) and val == int(val): return str(int(val)) return str(val) text = str(val).strip() text = text.replace('\n', ' ').replace('\r', ' ') text = ' '.join(text.split()) # Приведение регистра: если текст содержит только заглавные буквы РУ # НО не трогаем аббревиатуры/номера (АРМ №1, ИС, ЕМСЭД, СКУД и т.п.) if len(text) > 4 and any('А' <= c <= 'Я' or c == 'Ё' for c in text): has_lower = any('а' <= c <= 'я' or c == 'ё' for c in text) has_digit_or_num = bool(re.search(r'[0-9№«»"\']', text)) if not has_lower and not has_digit_or_num: words = text.split() # если есть короткая аббревиатура целиком заглавными — не трогаем has_abbr = any(len(w) <= 4 and w.isupper() and any('А' <= c <= 'Я' for c in w) for w in words) if not has_abbr: if len(words) >= 3: text = text.title() else: text = text.capitalize() return text def import_152fz(filepath: str, log_fn: Optional[Callable] = None, settings_path: Optional[str] = None) -> dict: """Импорт данных из опросного листа Excel (152-ФЗ). settings_path — путь к JSON-файлу настроек импорта (что и откуда брать). Если None — файл import_settings.json рядом с программой; если его нет — используется встроенный словарь по умолчанию. """ if openpyxl is None: raise ImportError("openpyxl не установлен. Установите: pip install openpyxl") if log_fn is None: log_fn = lambda msg: None settings = load_settings(settings_path) log_fn(f"Загрузка опросного листа: {filepath}") data = { 'fullName': '', 'shortName': '', 'addressLegal': '', 'addressActual': '', 'email': '', 'phone': '', 'inn': '', 'ogrn': '', 'kpp': '', 'okved': '', 'chiefPosition': '', 'chiefFio': '', 'ispdnFio': '', 'ispdnPosition': '', 'paperDocuments': '', 'paperStorage': '', 'ispdnDepartment': '', 'ispdnEmail': '', 'ispdnPhone': '', 'informationSystems': [], 'commission': [], 'employeesAccess': [], } try: wb = openpyxl.load_workbook(filepath, data_only=True, read_only=False) except Exception as e: log_fn(f"❌ Ошибка открытия файла: {e}") return data try: _import_152_common(wb, data, settings, log_fn) _import_152_is_list(wb, data, settings, log_fn) _import_152_employees(wb, data, settings, log_fn) except Exception as e: log_fn(f"❌ Ошибка при импорте: {e}") log_fn(traceback.format_exc()) finally: wb.close() # Копирование полей ИСПДн в админа, если админ не заполнен for src, dst in [('ispdnFio', 'administratorFio'), ('ispdnFio', 'adminFio'), ('ispdnPosition', 'administratorPosition'), ('ispdnPosition', 'adminPosition'), ('ispdnPhone', 'phone'), ('ispdnEmail', 'email')]: if data.get(src) and not data.get(dst): data[dst] = data[src] _validate_company_data(data, log_fn) return data def _import_152_common(wb, data, settings, log_fn): """Импорт общих сведений по словарю настроек. Для каждого поля: сначала пробуем конкретную ячейку (cell), если она задана и заполнена; иначе ищем строку по ключевым словам (keywords) в колонке A листа «Общие сведения». """ sheet_name = None sheet_kw = settings.get('common_sheet_keyword', 'общие') for sn in wb.sheetnames: if sheet_kw in sn.lower(): sheet_name = sn break if not sheet_name: log_fn(f"⚠ Лист по ключу '{sheet_kw}' не найден") return ws = wb[sheet_name] value_col = int(settings.get('value_column', 1)) # Кэш значений: для каждого поля ищем по строке один раз row_cache = {} for row in ws.iter_rows(min_row=1, values_only=True): field_raw = clean_value(row[0]) if row[0] is not None else '' if not field_raw: continue row_cache[field_raw.lower()] = ( clean_value(row[value_col]) if len(row) > value_col else '' ) for rule in settings.get('fields', []): key = rule.get('var', '') if not key: continue name = rule.get('name', key) cell_ref = rule.get('cell', '') keywords = rule.get('keywords', '') # 1) Конкретная ячейка if cell_ref: sheet_part, coord = parse_cell_ref(cell_ref) target_ws = ws if sheet_part: for sn in wb.sheetnames: if sn.lower() == sheet_part.lower(): target_ws = wb[sn] break val = clean_value(cell_value(target_ws, coord)) if val: data[key] = val log_fn(f" {name} [ячейка {coord}]: {val}") continue # 2) Поиск по ключевым словам if keywords: try: rx = re.compile(keywords, re.IGNORECASE) except re.error: rx = None if rx: for field_lower, val in row_cache.items(): if rx.search(field_lower): if val: data[key] = val log_fn(f" {name} [по ключу]: {val}") break def _import_152_is_list(wb, data, settings, log_fn): """Импорт информационных систем из остальных листов.""" exclude_kws = settings.get('is_sheet_exclude_keywords', ['общие сведения', 'сотрудники']) is_sheets = [] for sn in wb.sheetnames: sn_lower = sn.lower() if any(kw in sn_lower for kw in exclude_kws): continue is_sheets.append(sn) if not is_sheets: log_fn("⚠ Листы ИС не найдены") return is_list = [] for sheet_name in is_sheets: ws = wb[sheet_name] rows = [] for row in ws.iter_rows(min_row=1, values_only=True): vals = [clean_value(c) for c in row] field = (vals[0] or '').strip() if vals else '' if not field: continue rows.append(vals) if len(rows) < 2: continue is_obj = { 'name': '', 'description': '', 'room': '', 'employee': '', 'position': '', 'personalDataList': [], 'pd_subjects_list': [], 'pd_actions': [], 'usersList': [], 'isInternet': False, 'personalDataCount': 'менее 100 000', 'defense_tools': [], 'processing_mode': '', } current_section = 'header' pd_items_raw = [] subjects_raw = [] actions_raw = [] for vals in rows: field = (vals[0] or '').strip() field_lower = field.lower() val_b = clean_value(vals[1]) if len(vals) > 1 else '' is_checked = val_b.upper() in ('ДА', 'YES', '✓', '✔', '1', 'TRUE', 'ЕСТЬ') if 'цель обработки пд' in field_lower or ('цель обработки' in field_lower and ':' in field): current_section = 'purpose' continue if 'перечисленные пдн принадлежат' in field_lower or 'принадлеж' in field_lower: current_section = 'subjects' continue if 'структура ис' in field_lower or 'структура информационной' in field_lower or 'структур.*ис' in field_lower: if val_b: is_obj['structure'] = val_b continue if 'названия программ' in field_lower or ('программ' in field_lower and 'ос взаимодейств' in field_lower): if val_b: is_obj['software'] = val_b continue if 'локальную сеть' in field_lower: is_obj['is_has_lan'] = is_checked continue if 'количество записей' in field_lower or 'количество субъект' in field_lower: if val_b: is_obj['personalDataCount'] = val_b continue if 'перечень действий с пд' in field_lower or ('действи' in field_lower and 'пд' in field_lower): current_section = 'actions' continue if 'способ обработки пд' in field_lower: if val_b: is_obj['processing_mode'] = val_b continue if 'интернет' in field_lower: is_obj['isInternet'] = is_checked continue if current_section == 'header': if 'перечень пд' in field_lower: current_section = 'pd_items' elif 'название ис' in field_lower or 'наименование ис' in field_lower or 'наименование информационной' in field_lower: is_obj['name'] = val_b log_fn(f" ИС: {val_b}") elif 'сотрудник' in field_lower and 'должност' not in field_lower: is_obj['employee'] = val_b if val_b: parts = [p.strip() for p in val_b.replace('\n', ';').split(';') if p.strip()] for part in parts: cleaned = part.strip() if ':' in cleaned and len(cleaned.split(':')[0]) < 20: cleaned = cleaned.split(':', 1)[1].strip() if cleaned and cleaned not in is_obj['usersList']: is_obj['usersList'].append(cleaned) elif 'должность сотрудника' in field_lower: is_obj['position'] = val_b elif 'наименование отдела' in field_lower: pass elif 'номер кабинета' in field_lower: is_obj['room'] = val_b elif 'значение' in field_lower or 'примечание' in field_lower: continue elif val_b and 'выбрать' not in field_lower: if not is_obj['description']: is_obj['description'] = f"{field}: {val_b}" else: is_obj['description'] += f"\n{field}: {val_b}" elif current_section == 'pd_items': if is_checked and 'выбрать' not in field_lower and field_lower != 'значение': pd_items_raw.append(field) if 'цель обработки' in field_lower: current_section = 'purpose' elif current_section == 'purpose': purpose_val = val_b or field or '' if purpose_val and 'выбрать' not in purpose_val.lower() and ':' not in purpose_val[:40]: is_obj['purpose'] = purpose_val log_fn(f" Цель: {purpose_val[:80]}") current_section = 'header' elif current_section == 'subjects': if is_checked and 'выбрать' not in field_lower: subjects_raw.append(field) if 'количество записей' in field_lower or 'действи' in field_lower: continue elif current_section == 'actions': if is_checked and 'выбрать' not in field_lower and 'защита' not in field_lower[:20]: actions_raw.append(field) if 'защита' in field_lower and 'действи' not in field_lower: current_section = 'defense' elif current_section == 'defense': defense_keywords = { 'антивирус': 'Антивирус', 'крипто': 'СКЗИ', 'межсетев': 'Межсетевой экран', 'сзи.*от несанкционирован': 'СЗИ от НСД', 'обнаружен.*вторжен': 'СОВ', 'программные.*средств.*модул': 'Доверенная загрузка', 'доверен.*загрузк': 'Доверенная загрузка', 'гарантирован': 'Доверенная загрузка', } matched_label = None for kw, label in defense_keywords.items(): if re.search(kw, field_lower): matched_label = label break if matched_label and val_b and 'выбрать' not in field_lower: is_obj['defense_tools'].append(f"{matched_label}: {val_b}") elif matched_label: pass elif 'куда' in field_lower or 'передаются' in field_lower: current_section = 'header' elif 'способ обработки' in field_lower or 'локальную сеть' in field_lower or 'интернет' in field_lower: current_section = 'header' if pd_items_raw: is_obj['personalDataList'] = pd_items_raw log_fn(f" ПДн: {len(pd_items_raw)}") if subjects_raw: is_obj['pd_subjects_list'] = subjects_raw log_fn(f" Субъектов: {len(subjects_raw)}") if actions_raw: is_obj['pd_actions'] = actions_raw log_fn(f" Действий: {len(actions_raw)}") if not is_obj['name']: is_obj['name'] = sheet_name.strip() log_fn(f" ⚠ ИС без названия — использовано имя листа: {is_obj['name']}") # Автоопределение категорий ПДн pd_cats = [] all_pd_text = ' '.join(p.lower() for p in pd_items_raw) special_kw = ['состоян.*здоров', 'национальн', 'политическ', 'религиозн', 'философ', 'судимост', 'интимн'] if any(re.search(kw, all_pd_text) for kw in special_kw): pd_cats.append('специальные') bio_kw = ['биометрическ', 'изображен.*лиц', 'голос.*человек', 'папилляр', 'дактилоскоп', 'фото.*изображен'] if any(re.search(kw, all_pd_text) for kw in bio_kw): pd_cats.append('биометрические') if pd_items_raw: pd_cats.append('иные') if pd_cats: is_obj['personalDataCategory'] = pd_cats log_fn(f" Категории ПДн: {', '.join(pd_cats)}") is_list.append(is_obj) log_fn(f" ✅ {is_obj['name']} — импортирована") if is_list: data['informationSystems'] = is_list log_fn(f"\n✅ Всего импортировано ИС: {len(is_list)}") def _import_152_employees(wb, data, settings, log_fn): """Импорт комиссии и сотрудников из листа «Сотрудники».""" sheet_name = None sheet_kw = settings.get('employees_sheet_keyword', 'сотруд') for sn in wb.sheetnames: if sheet_kw in sn.lower(): sheet_name = sn break if not sheet_name: log_fn(f"⚠ Лист по ключу '{sheet_kw}' не найден") return ws = wb[sheet_name] rows = [] for row in ws.iter_rows(min_row=1, values_only=True): vals = [clean_value(c) for c in row] if not vals or not vals[0]: continue rows.append(vals) if not rows: log_fn("⚠ Лист «Сотрудники» пуст") return commission = [] in_commission = False COMMISSION_ROLES = { 'председатель': 'Председатель комиссии', 'секретарь': 'Секретарь комиссии', 'заместитель председателя': 'Заместитель председателя комиссии', } employees = [] in_employees = False for vals in rows: col_a = (vals[0] or '').strip() col_b = clean_value(vals[1]) if len(vals) > 1 else '' col_c = clean_value(vals[2]) if len(vals) > 2 else '' col_d = clean_value(vals[3]) if len(vals) > 3 else '' col_e = clean_value(vals[4]) if len(vals) > 4 else '' a_lower = col_a.lower() if 'состав комиссии' in a_lower: in_commission = True in_employees = False log_fn(" Секция: состав комиссии") continue if 'перечень сотрудников' in a_lower: in_commission = False in_employees = True log_fn(" Секция: сотрудники с доступом к ПДн") continue if a_lower in ('№ п/п', '№', 'номер', ''): continue if in_commission: role = 'Член комиссии' # Роль ищем в колонках E (было) и D (Примечание — новый формат) for col in (col_e, col_d): if col: col_lower = col.lower() for keyword, role_name in COMMISSION_ROLES.items(): if keyword in col_lower: role = role_name break if role != 'Член комиссии': break if not col_b and not col_c: continue commission.append({ 'role': role, 'position': col_b, 'fio': col_c.replace('\n', ' '), }) if in_employees: fio = col_c if col_c else col_d if not fio: continue employee = { 'position': col_b, 'fio': fio.replace('\n', ' '), } if col_e: employee['is_list'] = [s.strip() for s in col_e.split('\n') if s.strip()] employees.append(employee) if commission: commission = [c for c in commission if c['position'] or c['fio']] data['commission'] = commission log_fn(f" Комиссия: {len(commission)} чел.") for m in commission[:3]: log_fn(f" {m['role']}: {m['position'][:40]} — {m['fio'][:40]}") if employees: data['employeesAccess'] = employees log_fn(f" Сотрудников с доступом: {len(employees)}") def _validate_company_data(data, log_fn): """Проверка обязательных полей.""" required = ['fullName', 'inn'] missing = [f for f in required if not data.get(f)] if missing: log_fn(f"⚠️ ОБЯЗАТЕЛЬНЫЕ ПОЛЯ НЕ ЗАПОЛНЕНЫ: {', '.join(missing)}") else: log_fn("✅ Все обязательные поля заполнены") inn = (data.get('inn') or '').replace(' ', '') if inn and (not inn.isdigit() or len(inn) not in (10, 12)): log_fn(f"⚠️ ИНН подозрительного формата: «{inn}»")