Исходный код DokoGen (генератор документов 152-ФЗ)

dokogen/importer.py

@@ -0,0 +1,560 @@
+# -*- 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
+
+
+# Известные аббревиатуры, которые сохраняются КАПСОМ при нормализации
+_CAPS_ABBR = {
+    'АРМ', 'СКУД', 'СЗИ', 'СКЗИ', 'СОВ', 'ЛВС', 'ИНН', 'КПП', 'ОГРН', 'ЦОД',
+    'МБУ', 'МКУ', 'ООО', 'СНИЛС', 'СБИС', 'ЕМСЭД', 'ФИО', 'ИСПДН', 'ККТ',
+}
+
+
+def _normalize_caps(text: str) -> str:
+    """Приводит КАПС-текст к нормальному регистру по словам.
+
+    Сохраняет:
+    - аббревиатуры из белого списка (АРМ, СКУД, ООО, ИНН и т.п.)
+    - инициалы и короткие служебные слова (И., Г., УЛ., №1)
+    - числа и номера (350000, №1)
+    - слова, уже содержащие строчные буквы
+    Длинные слова целиком заглавными (ДИРЕКТОР, КРАСНОДАР, ПЕТРОВ) → «Директор», «Краснодар».
+    """
+    words = text.split()
+    result = []
+    for w in words:
+        # отделяем пунктуацию по краям слова
+        pre = ''
+        post = ''
+        core = w
+        while core and not core[0].isalnum():
+            pre += core[0]
+            core = core[1:]
+        while core and not core[-1].isalnum():
+            post = core[-1] + post
+            core = core[:-1]
+        if not core:
+            result.append(w)
+            continue
+        # в слове уже есть строчные буквы — не трогаем
+        if any('а' <= c <= 'я' or c == 'ё' for c in core):
+            result.append(w)
+            continue
+        # нет заглавных кириллических букв (числа, латиница и т.п.) — не трогаем
+        if not any('А' <= c <= 'Я' or c == 'Ё' for c in core):
+            result.append(w)
+            continue
+        letters = [c for c in core if c.isalpha()]
+        # инициалы и служебные слова (1-2 буквы: И., Г., УЛ., ООО — но ООО в белом списке)
+        if len(letters) <= 2:
+            result.append(w)
+            continue
+        # известная аббревиатура — сохраняем как есть
+        if core.upper() in _CAPS_ABBR:
+            result.append(w)
+            continue
+        # обычное длинное слово: первая заглавная, остальные строчные
+        normalized = core[0].upper() + core[1:].lower()
+        result.append(pre + normalized + post)
+    return ' '.join(result)
+
+
+_ADDRESS_ABBR = {
+    # нормальные сокращения для адресов: «Г» → «г.», «Р-Н» → «р-н.» и т.п.
+    'Г': 'г.', 'УЛ': 'ул.', 'Д': 'д.', 'КВ': 'кв.', 'ПР': 'пр.',
+    'ПЕР': 'пер.', 'ПЛ': 'пл.', 'ОБЛ': 'обл.', 'ПОС': 'пос.',
+    'СТ': 'ст.', 'Ш': 'ш.', 'Р-Н': 'р-н.', 'Б-Р': 'б-р.', 'МКР': 'мкр.',
+    'НАБ': 'наб.', 'ТУП': 'туп.', 'ПР-Д': 'пр-д.', 'КРП': 'крп.',
+    'Г.': 'г.', 'УЛ.': 'ул.', 'Д.': 'д.', 'Р-Н.': 'р-н.',
+}
+
+
+def normalize_address(text: str) -> str:
+    """Приводит адрес к нормальному виду: «Г Крымск» → «г. Крымск»,
+    «Р-Н Крымский» → «р-н. Крымский», «УЛ Карла» → «ул. Карла»."""
+    if not text:
+        return ''
+    parts = re.split(r'(\s+|[,\;])', text)
+    out = []
+    for p in parts:
+        if p.strip() and p.strip().upper() in _ADDRESS_ABBR:
+            out.append(_ADDRESS_ABBR[p.strip().upper()])
+        else:
+            out.append(p)
+    return ''.join(out)
+
+
+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())
+    # Если весь текст КАПСОМ (нет строчных букв) — нормализуем по словам
+    if len(text) > 2 and any('А' <= c <= 'Я' or c == 'Ё' for c in text):
+        has_lower = any('а' <= c <= 'я' or c == 'ё' for c in text)
+        if not has_lower:
+            text = _normalize_caps(text)
+    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]
+
+    # Нормализация адресов: «Г Крымск» → «г. Крымск», «Р-Н» → «р-н.» и т.п.
+    for key in ('addressLegal', 'addressActual'):
+        if data.get(key):
+            data[key] = normalize_address(data[key])
+
+    _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:
+                if val_b:
+                    is_obj['structure'] = val_b
+                elif ':' in field:
+                    # Значение может быть записано в той же ячейке после двоеточия
+                    tail = field.split(':', 1)[1].strip()
+                    if tail and 'выбрать' not in tail.lower():
+                        is_obj['structure'] = tail
+                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 or 'описание информационной' in field_lower:
+                    if val_b:
+                        is_obj['description'] = val_b
+                        log_fn(f"  Описание: {val_b[:60]}")
+                    continue
+                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:
+                    # Без префикса категории: просто название средства (Kaspersky, КриптоПро)
+                    is_obj['defense_tools'].append(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}»")