Files
dokogenMU/dokogen/importer.py
T
prog1764 da4946cb59 feat: ребрендинг под 187-ФЗ (КИИ)
- Все упоминания 152-ФЗ заменены на 187-ФЗ (заголовок окна, справка,
  импорт, переменные, docstring'и, spec)
- import_152fz -> import_187fz, direction '152fz' -> '187fz'
- autosave_152 -> autosave_187
- Опросный_лист_152_ФЗ.xlsx -> Опросный_лист_187_ФЗ.xlsx
- Промпт ИИ: 'юрист по защите персональных данных (152-ФЗ)' -> 187-ФЗ
2026-08-03 23:25:18 +04:00

561 lines
25 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
# -*- coding: utf-8 -*-
"""DokoGen — Модуль импорта из Excel (опросный лист 187-ФЗ)"""
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_187fz(filepath: str, log_fn: Optional[Callable] = None, settings_path: Optional[str] = None) -> dict:
"""Импорт данных из опросного листа Excel (187-ФЗ).
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_187_common(wb, data, settings, log_fn)
_import_187_is_list(wb, data, settings, log_fn)
_import_187_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_187_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_187_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_187_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}»")