431 lines
20 KiB
Python
431 lines
20 KiB
Python
# -*- 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
|
||
|
||
|
||
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) > 4 and any('А' <= c <= 'Я' or c == 'Ё' for c in text):
|
||
has_lower = any('а' <= c <= 'я' or c == 'ё' for c in text)
|
||
if not has_lower:
|
||
words = text.split()
|
||
if len(words) >= 3:
|
||
text = text.title()
|
||
else:
|
||
text = text.capitalize()
|
||
return text
|
||
|
||
|
||
def import_152fz(filepath: str, log_fn: Optional[Callable] = None) -> dict:
|
||
"""Импорт данных из опросного листа Excel (152-ФЗ)."""
|
||
if openpyxl is None:
|
||
raise ImportError("openpyxl не установлен. Установите: pip install openpyxl")
|
||
|
||
if log_fn is None:
|
||
log_fn = lambda msg: None
|
||
|
||
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, log_fn)
|
||
_import_152_is_list(wb, data, log_fn)
|
||
_import_152_employees(wb, data, 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, log_fn):
|
||
"""Импорт общих сведений из листа 'Общие сведения'."""
|
||
sheet_name = None
|
||
for sn in wb.sheetnames:
|
||
if 'общие' in sn.lower():
|
||
sheet_name = sn
|
||
break
|
||
if not sheet_name:
|
||
log_fn("⚠ Лист 'Общие сведения' не найден")
|
||
return
|
||
|
||
ws = wb[sheet_name]
|
||
field_map = [
|
||
('полное наименование организации', 'fullName'),
|
||
('сокращенное наименование организации', 'shortName'),
|
||
('адрес организации \\(юридический\\)', 'addressLegal'),
|
||
('адрес организации \\(фактический\\)', 'addressActual'),
|
||
('инн', 'inn'),
|
||
('дата.*огрн', 'ogrnDate'),
|
||
('огрн', 'ogrn'),
|
||
('кпп', 'kpp'),
|
||
('основной оквэд', 'okved'),
|
||
('должность руководителя', 'chiefPosition'),
|
||
('фио руководителя', 'chiefFio'),
|
||
('должность.*защит.*информ|должность.*администрат', 'ispdnPosition'),
|
||
('фио.*защит.*информ|фио.*администрат', 'ispdnFio'),
|
||
('электронная почта.*защит|электронная почта.*ответств|электронная почта.*специалист', 'ispdnEmail'),
|
||
('телефон.*защит|телефон.*ответств|телефон.*специалист', 'ispdnPhone'),
|
||
('структур.*подраздел', 'ispdnDepartment'),
|
||
('номер.*договора|номер.*контракта', 'contractNumber'),
|
||
('дата.*договора|дата.*контракта', 'contractDate'),
|
||
('названи.*документ', 'paperDocuments'),
|
||
('место.*хранен', 'paperStorage'),
|
||
('должность.*ответственн', 'responsiblePosition'),
|
||
('фио.*ответственн', 'responsibleFio'),
|
||
('должность.*администрат.*(?:безопас|защит|пд)', 'administratorPosition'),
|
||
('фио.*администрат.*(?:безопас|защит|пд)', 'administratorFio'),
|
||
('сайт организации', 'site_name'),
|
||
('должность ответственного за сайт|должность.*отв.*сайт', 'site_responsible_position'),
|
||
('фио.*отв.*сайт|фио.*сайт', 'site_responsible_fio'),
|
||
('наименование организации.*разрабат.*сайт|наименование.*размещала.*сайт|разрабат.*сайт.*организац', 'site_service_provider'),
|
||
('инн.*разрабат.*сайт|инн.*размещала.*сайт', 'site_service_provider_inn'),
|
||
('размещение сайта|адрес расположения серверов|адрес.*сервер.*сайт', 'site_hosting'),
|
||
('адрес.*хостинг|адрес.*располож.*сервер', 'site_hosting_address'),
|
||
('электронная почта', 'email'),
|
||
('телефон', 'phone'),
|
||
]
|
||
|
||
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 ''
|
||
field = field_raw.lower()
|
||
value = clean_value(row[1]) if len(row) > 1 else ''
|
||
if not field:
|
||
continue
|
||
for keyword, key in field_map:
|
||
if re.search(keyword, field):
|
||
if value:
|
||
data[key] = value
|
||
log_fn(f" {keyword}: {value}")
|
||
break
|
||
|
||
|
||
def _import_152_is_list(wb, data, log_fn):
|
||
"""Импорт информационных систем из остальных листов."""
|
||
is_sheets = [sn for sn in wb.sheetnames if sn not in ('Общие сведения', 'Сотрудники')]
|
||
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['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 = field or val_b or ''
|
||
if purpose_val and 'выбрать' not in purpose_val.lower():
|
||
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, log_fn):
|
||
"""Импорт комиссии и сотрудников из листа 'Сотрудники'."""
|
||
sheet_name = None
|
||
for sn in wb.sheetnames:
|
||
if 'сотруд' in sn.lower():
|
||
sheet_name = sn
|
||
break
|
||
if not sheet_name:
|
||
log_fn("⚠ Лист «Сотрудники» не найден")
|
||
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 = 'Член комиссии'
|
||
if col_e:
|
||
e_lower = col_e.lower()
|
||
for keyword, role_name in COMMISSION_ROLES.items():
|
||
if keyword in e_lower:
|
||
role = role_name
|
||
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}»")
|