Files
dokogen187/dokogen/importer.py
T
prog1764 ed71343d92 fix: ОКВЭД импорт/DaData/словарь+интернет, доменное имя, вкладка ОКИИ → вложенные вкладки, словарики, СЗИ разбивка, адрес отдельно
- ОКВЭД: resolve_okved (словарь okved.json + DaData suggest okved2), применяется при импорте и DaData; галочки «ОКВЭД осн./доп.» в «что применять»
- DaData: показ и подстановка доп. ОКВЭД
- доменное имя: поле во вкладке «Организация» + импорт
- вкладка «Объекты КИИ»: сверху Наименование/Описание/Категория, остальное вкладками: Основное (назначение, сфера, тип, архитектура, типовой отраслевой объект — со словариками и выпадающими списками), Субъект, Сети и ПО, Безопасность (236), СЗИ, Пользователи ОКИИ
- новые словари: spheres.json, object_types.json, architectures.json, typical_objects.json
- адрес объекта — в отдельное поле (не в описание)
- СЗИ: разбивка слипшихся названий (Kaspersky...)
2026-08-04 15:33:36 +04:00

622 lines
29 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, ...).
Заполняет поля 236 приказа ФСТЭК: сфера, назначение, тип, архитектура,
сети электросвязи, ПО, СЗИ, сведения о субъекте.
"""
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': '', 'address': '', 'usersList': [],
'isInternet': False, 'defense_tools': [], 'software': '',
# 236 приказ
'sphere': '', 'purpose': '', 'object_type': '', 'architecture': '',
'typical_object': '',
'subject_name': '', 'subject_address': '',
'subject_chief_position': '', 'subject_chief_fio': '',
'subject_security_position': '', 'subject_security_fio': '',
'subject_department': '', 'subject_department_chief': '',
'subject_phone': '', 'subject_email': '',
'subject_inn': '', 'subject_kpp': '',
'net_category': '', 'net_operator': '', 'net_purpose': '', 'net_method': '',
'hw_sw': '', 'os_sw': '', 'app_sw': '',
'attacker_category': '', 'threats': '', 'incidents': '',
'criteria_values': '', 'criteria_justification': '',
'org_measures': '', 'tech_measures': '',
}
defense_keywords = {
'антивирус': 'Антивирус',
'крипто': 'СКЗИ',
'межсетев': 'Межсетевой экран',
'сзи.*от несанкционирован': 'СЗИ от НСД',
'обнаружен.*вторжен': 'СОВ',
'программные.*средств.*модул': 'Доверенная загрузка',
'доверен.*загрузк': 'Доверенная загрузка',
'гарантирован': 'Доверенная загрузка',
}
def _set(field_lower, key, val):
"""Записать значение, если ключевое слово найдено в вопросе."""
if val and 'выбрать' not in field_lower:
is_obj[key] = val
return True
return False
def _split_entries(text):
"""Разбивает слипшиеся названия СЗИ: по переносам строк, ';' и по границе
'цифра + заглавная буква' (напр. '15.1.0.22239Kaspersky' → два названия)."""
if not text:
return []
parts = re.split(r'[\n;]+', text)
out = []
for p in parts:
p = p.strip()
if not p:
continue
# граница: цифра/точка → заглавная буква без пробела
sub = re.split(r'(?<=[0-9.])(?=[А-ЯЁA-Z])', p)
for s in sub:
s = s.strip()
if s and s not in out:
out.append(s)
return out
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 or 'местонахождение объекта' in field_lower:
is_obj['address'] = val_b
log_fn(f" Адрес объекта: {val_b[:60]}")
continue
# --- Описание ---
if 'описание окки' in field_lower or 'описание объекта' in field_lower:
if val_b:
is_obj['description'] = val_b
continue
# --- Сфера деятельности ---
if 'сфера' in field_lower and 'деятельност' in field_lower:
if _set(field_lower, 'sphere', val_b):
log_fn(f" Сфера: {val_b[:60]}")
continue
# --- Назначение объекта ---
if 'назначение объекта' in field_lower:
if _set(field_lower, 'purpose', val_b):
log_fn(f" Назначение: {val_b[:60]}")
continue
# --- Тип объекта ---
if 'тип объекта' in field_lower:
if _set(field_lower, 'object_type', val_b):
log_fn(f" Тип объекта: {val_b[:60]}")
continue
# --- Архитектура ---
if 'архитектура' in field_lower:
if _set(field_lower, 'architecture', val_b):
log_fn(f" Архитектура: {val_b[:60]}")
continue
# --- Типовой отраслевой объект ---
if 'типового отраслевого' in field_lower or 'типовой отраслевой' in field_lower:
if _set(field_lower, 'typical_object', val_b):
log_fn(f" Типовой объект: {val_b[:60]}")
continue
# --- Субъект ---
if 'наименование организации эксплуатирующ' in field_lower or ('наименование субъекта' in field_lower):
if _set(field_lower, 'subject_name', val_b):
log_fn(f" Субъект: {val_b[:60]}")
continue
if 'адрес местонахождения' in field_lower:
if _set(field_lower, 'subject_address', val_b):
log_fn(f" Адрес субъекта: {val_b[:60]}")
continue
if 'инн лица, эксплуатирующего' in field_lower or 'инн субъекта' in field_lower:
if _set(field_lower, 'subject_inn', val_b):
log_fn(f" ИНН субъекта: {val_b[:60]}")
continue
if 'кпп его обособленных' in field_lower or 'кпп субъекта' in field_lower:
if _set(field_lower, 'subject_kpp', val_b):
log_fn(f" КПП субъекта: {val_b[:60]}")
continue
if 'элемент (компонент) объекта, который эксплуатируется' in field_lower:
if _set(field_lower, 'subject_department', val_b):
log_fn(f" Элемент объекта: {val_b[:60]}")
continue
# --- Сети электросвязи ---
if 'категория сети электросвязи' in field_lower:
if _set(field_lower, 'net_category', val_b):
log_fn(f" Категория сети: {val_b[:60]}")
continue
if 'наименование провайдера' in field_lower or 'оператор связи' in field_lower:
if _set(field_lower, 'net_operator', val_b):
log_fn(f" Оператор: {val_b[:60]}")
continue
if 'цель взаимодействия' in field_lower:
if _set(field_lower, 'net_purpose', val_b):
log_fn(f" Цель взаимодействия: {val_b[:60]}")
continue
if 'способ взаимодействия' in field_lower:
if _set(field_lower, 'net_method', val_b):
log_fn(f" Способ взаимодействия: {val_b[:60]}")
continue
# --- ПО и аппаратные средства ---
if 'наименования программно-аппаратных средств' in field_lower:
if _set(field_lower, 'hw_sw', val_b):
log_fn(f" ПО/аппаратные средства: {val_b[:60]}")
continue
if 'операционная' in field_lower and 'система' in field_lower:
if _set(field_lower, 'os_sw', val_b):
log_fn(f" ОС: {val_b[:60]}")
continue
if 'прикладные программы' in field_lower:
if _set(field_lower, 'app_sw', val_b):
log_fn(f" Прикладные программы: {val_b[:60]}")
continue
if 'общесистемн' in field_lower:
if _set(field_lower, 'os_sw', val_b):
log_fn(f" Общесистемное ПО: {val_b[:60]}")
continue
# --- Безопасность / 236 приказ ---
if 'категория нарушителя' in field_lower:
if _set(field_lower, 'attacker_category', val_b):
log_fn(f" Категория нарушителя: {val_b[:60]}")
continue
if 'основные угрозы' in field_lower or 'угрозы безопасности' in field_lower:
if _set(field_lower, 'threats', val_b):
log_fn(f" Угрозы: {val_b[:60]}")
continue
if 'компьютерных инцидентов' in field_lower or 'инцидент' in field_lower:
if _set(field_lower, 'incidents', val_b):
log_fn(f" Инциденты: {val_b[:60]}")
continue
if 'показателей критериев' in field_lower or 'показатели критериев' in field_lower:
if _set(field_lower, 'criteria_values', val_b):
log_fn(f" Показатели: {val_b[:60]}")
continue
if 'обоснование полученных' in field_lower or 'обоснование значений' in field_lower:
if _set(field_lower, 'criteria_justification', val_b):
log_fn(f" Обоснование: {val_b[:60]}")
continue
if 'организационные меры' in field_lower:
if _set(field_lower, 'org_measures', val_b):
log_fn(f" Орг. меры: {val_b[:60]}")
continue
if 'технические меры' in field_lower:
if _set(field_lower, 'tech_measures', val_b):
log_fn(f" Тех. меры: {val_b[:60]}")
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:
for tool in _split_entries(val_b):
if tool not in is_obj['defense_tools']:
is_obj['defense_tools'].append(tool)
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}»")