Files
dokogen187/dokogen/importer.py
T
prog1764 ce1b56066a feat: переделка под ОКИИ (187-ФЗ) — объекты КИИ вместо ИС, категории 1-3, ОКВЭД, убраны ПДн/бумажные носители
- вкладка «Информационные системы» → «Объекты КИИ», убрана «Структура объекта»
- «Тип угроз»/«УЗ» → категория значимости 1/2/3/без категории
- удалены Категории ПДн, кол-во записей, перечень/субъекты/действия с ПДн + словари
- «Пользователи ИС» → «Пользователи ОКИИ», средства защиты оставлены
- вкладка «Бумажные документы» удалена полностью (GUI + логика + модели)
- ОКВЭД: поля во вкладке «Организация», словарь okved.json, выбор галочками, импорт
- нормализация регистра адреса DaData (КАПС → нормальный)
- importer: объекты КИИ (наименование объекта, ПО, СЗИ, пользователи)
2026-08-04 13:58:23 +04:00

464 lines
19 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': '', 'okvedExtra': '', 'chiefPosition': '', 'chiefFio': '',
'ispdnFio': '', 'ispdnPosition': '',
'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):
"""Импорт объектов КИИ из остальных листов (ОБЪЕКТ №1, №2, ...)."""
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': '', 'usersList': [],
'isInternet': False, 'defense_tools': [], 'software': '',
}
defense_keywords = {
'антивирус': 'Антивирус',
'крипто': 'СКЗИ',
'межсетев': 'Межсетевой экран',
'сзи.*от несанкционирован': 'СЗИ от НСД',
'обнаружен.*вторжен': 'СОВ',
'программные.*средств.*модул': 'Доверенная загрузка',
'доверен.*загрузк': 'Доверенная загрузка',
'гарантирован': 'Доверенная загрузка',
}
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:
is_obj['name'] = val_b
log_fn(f" Объект КИИ: {val_b}")
continue
if 'описание объекта' in field_lower or 'описание окки' in field_lower:
if val_b:
is_obj['description'] = val_b
continue
if 'операционная' in field_lower or 'прикладные программы' in field_lower or 'наименования программно' in field_lower:
if val_b:
is_obj['software'] = (is_obj['software'] + '; ' + val_b) if is_obj['software'] else val_b
continue
if 'интернет' in field_lower:
is_obj['isInternet'] = is_checked
continue
if 'сотрудник' in field_lower and 'должност' not in field_lower:
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)
continue
# Средства защиты
matched = False
for kw, label in defense_keywords.items():
if re.search(kw, field_lower):
if val_b and 'выбрать' not in field_lower:
is_obj['defense_tools'].append(val_b)
matched = True
break
if matched:
continue
if val_b and 'выбрать' not in field_lower and field_lower not in ('значение', 'примечание'):
# Остальные заполненные поля — в описание (адрес, сфера, тип и т.д.)
if not is_obj['description']:
is_obj['description'] = f"{field}: {val_b}"
else:
is_obj['description'] += f"\n{field}: {val_b}"
if not is_obj['name']:
is_obj['name'] = sheet_name.strip()
log_fn(f" ⚠ Объект без названия — использовано имя листа: {is_obj['name']}")
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}»")