prog1764
7886edc0af
- Профили: 152-ФЗ, 187-ФЗ, 117-ФЗ, МУ, КЗИ, ПЗИ - Сканирование Excel: что импортируется и куда - Галочки: включить/выключить поля, добавить свои привязки - Сохранение настроек по типу опросника - Выгрузка данных в JSON (export/export_<тип>.json) - Тесты: отключение полей, пользовательские привязки, секции, ИС
351 lines
14 KiB
Python
351 lines
14 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""ImportMapper — ядро: профили опросников, сканирование Excel, извлечение данных."""
|
||
|
||
import os
|
||
import re
|
||
import json
|
||
from datetime import datetime, timedelta
|
||
|
||
try:
|
||
import openpyxl
|
||
except ImportError:
|
||
openpyxl = None
|
||
|
||
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
|
||
PROFILES_DIR = os.path.join(BASE_DIR, 'profiles')
|
||
SETTINGS_DIR = os.path.join(BASE_DIR, 'settings')
|
||
EXPORT_DIR = os.path.join(BASE_DIR, 'export')
|
||
|
||
|
||
# ============================================================
|
||
# НОРМАЛИЗАЦИЯ ЗНАЧЕНИЙ
|
||
# ============================================================
|
||
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()
|
||
text = text.title() if len(words) >= 3 else text.capitalize()
|
||
return text
|
||
|
||
|
||
def is_noise_label(val: str) -> bool:
|
||
"""Служебные строки опросника, которые импортировать не нужно."""
|
||
noise = {'№ п/п', '№', 'номер', 'п/п', '№пп', '№ пп', 'значение', 'примечание',
|
||
'вопрос', 'ответ', 'выбрать', 'да', 'нет', 'нет данных', 'поле'}
|
||
if val.strip() in noise:
|
||
return True
|
||
if re.fullmatch(r'\d+', val.strip()):
|
||
return True
|
||
return False
|
||
|
||
|
||
# ============================================================
|
||
# ПРОФИЛИ ОПРОСНИКОВ
|
||
# ============================================================
|
||
def list_profiles():
|
||
"""Список доступных профилей опросников."""
|
||
result = []
|
||
if not os.path.isdir(PROFILES_DIR):
|
||
return result
|
||
for fn in sorted(os.listdir(PROFILES_DIR)):
|
||
if fn.endswith('.json'):
|
||
try:
|
||
with open(os.path.join(PROFILES_DIR, fn), encoding='utf-8') as f:
|
||
profile = json.load(f)
|
||
result.append(profile)
|
||
except Exception:
|
||
pass
|
||
return result
|
||
|
||
|
||
def load_profile(profile_id):
|
||
for p in list_profiles():
|
||
if p.get('id') == profile_id:
|
||
return p
|
||
return None
|
||
|
||
|
||
def settings_path(profile_id):
|
||
os.makedirs(SETTINGS_DIR, exist_ok=True)
|
||
return os.path.join(SETTINGS_DIR, f'settings_{profile_id}.json')
|
||
|
||
|
||
def load_settings(profile_id):
|
||
try:
|
||
with open(settings_path(profile_id), encoding='utf-8') as f:
|
||
return json.load(f)
|
||
except Exception:
|
||
return {'disabled': [], 'custom': []}
|
||
|
||
|
||
def save_settings(profile_id, settings):
|
||
with open(settings_path(profile_id), 'w', encoding='utf-8') as f:
|
||
json.dump(settings, f, ensure_ascii=False, indent=2)
|
||
|
||
|
||
def is_disabled(settings, key):
|
||
return key in settings.get('disabled', [])
|
||
|
||
|
||
# ============================================================
|
||
# СКАНИРОВАНИЕ EXCEL
|
||
# ============================================================
|
||
def scan_excel(filepath, profile):
|
||
"""Сканирует Excel по правилам профиля.
|
||
|
||
Возвращает (detected, unmapped, error):
|
||
detected — пары (заголовок → поле), которые импортируются
|
||
unmapped — найденные заголовки без привязки
|
||
"""
|
||
if not openpyxl:
|
||
return [], [], 'openpyxl не установлен. Установите: pip install openpyxl'
|
||
|
||
rules = profile.get('rules', [])
|
||
common_sheet = profile.get('common_sheet', 'общие')
|
||
employee_sheet = profile.get('employee_sheet', 'сотруд')
|
||
|
||
detected = []
|
||
unmapped = []
|
||
|
||
try:
|
||
wb = openpyxl.load_workbook(filepath, data_only=True, read_only=False)
|
||
except Exception as e:
|
||
return [], [], f'Не удалось открыть файл: {e}'
|
||
|
||
try:
|
||
for sheet_name in wb.sheetnames:
|
||
sheet_lower = sheet_name.lower()
|
||
if common_sheet and common_sheet.lower() in sheet_lower:
|
||
sheet_type = 'common'
|
||
elif employee_sheet and employee_sheet.lower() in sheet_lower:
|
||
sheet_type = 'employees'
|
||
else:
|
||
sheet_type = 'is'
|
||
|
||
ws = wb[sheet_name]
|
||
for row in ws.iter_rows(min_row=1, values_only=True):
|
||
label = str(row[0]).strip() if row and row[0] is not None else ''
|
||
value = str(row[1]).strip() if len(row) > 1 and row[1] is not None else ''
|
||
if not label:
|
||
continue
|
||
|
||
label_lower = label.lower()
|
||
mapped = False
|
||
for rule in rules:
|
||
pattern = rule.get('pattern', '')
|
||
if pattern and re.search(pattern, label_lower):
|
||
detected.append({
|
||
'field': rule.get('field', ''),
|
||
'excel_text': label,
|
||
'value': value[:60],
|
||
'sheet': sheet_name,
|
||
'desc': rule.get('desc', rule.get('field', '')),
|
||
})
|
||
mapped = True
|
||
break
|
||
|
||
if not mapped:
|
||
if is_noise_label(label_lower):
|
||
continue
|
||
unmapped.append({
|
||
'excel_text': label,
|
||
'sheet': sheet_name,
|
||
'suggested': _suggest_field(label, sheet_type, profile),
|
||
})
|
||
finally:
|
||
try:
|
||
wb.close()
|
||
except Exception:
|
||
pass
|
||
|
||
return detected, unmapped, ''
|
||
|
||
|
||
def _suggest_field(value, sheet_type, profile):
|
||
"""Пытается предположить, какому полю соответствует заголовок."""
|
||
val = value.lower().strip()
|
||
if is_noise_label(val):
|
||
return ''
|
||
|
||
for rule in profile.get('rules', []):
|
||
pattern = rule.get('pattern', '')
|
||
if pattern and re.search(pattern, val):
|
||
return rule.get('field', '')
|
||
|
||
keywords = {
|
||
'fullName': ['полное наименование', 'наименование организации', 'полное название'],
|
||
'shortName': ['сокращенное наименование', 'краткое наименование', 'сокращ'],
|
||
'addressLegal': ['адрес', 'место нахождения'],
|
||
'chiefFio': ['руководител', 'глава', 'директор'],
|
||
'inn': ['инн', 'идентификационный номер'],
|
||
'ogrn': ['огрн', 'основной государствен'],
|
||
'kpp': ['кпп'],
|
||
'phone': ['телефон', 'контактный'],
|
||
'email': ['почта', 'email', 'e-mail'],
|
||
'is_name': ['наименование ис', 'название ис', 'наименование информационной'],
|
||
'is_software': ['программное обеспечение', 'программный'],
|
||
}
|
||
for field, kws in keywords.items():
|
||
for kw in kws:
|
||
if kw in val:
|
||
return field
|
||
|
||
return 'unknown_' + value[:20].replace(' ', '_').lower()
|
||
|
||
|
||
# ============================================================
|
||
# ИЗВЛЕЧЕНИЕ ДАННЫХ (ПО НАСТРОЙКАМ РЕДАКТОРА)
|
||
# ============================================================
|
||
def extract_data(filepath, profile, settings):
|
||
"""Извлекает данные из опросника с учётом настроек редактора.
|
||
|
||
Возвращает dict: {'fields': {...}, 'sections': {...}, 'is_list': [...]}
|
||
"""
|
||
if not openpyxl:
|
||
return {'error': 'openpyxl не установлен'}
|
||
|
||
disabled = set(settings.get('disabled', []))
|
||
custom = settings.get('custom', [])
|
||
rules = profile.get('rules', [])
|
||
common_sheet = profile.get('common_sheet', 'общие')
|
||
employee_sheet = profile.get('employee_sheet', 'сотруд')
|
||
|
||
result = {'fields': {}, 'sections': {}, 'is_list': []}
|
||
|
||
# Все секции профиля всегда присутствуют в выгрузке (пустые, если отключены)
|
||
for sec_id in profile.get('sections', {}):
|
||
result['sections'][sec_id] = []
|
||
|
||
try:
|
||
wb = openpyxl.load_workbook(filepath, data_only=True, read_only=False)
|
||
except Exception as e:
|
||
return {'error': str(e)}
|
||
|
||
try:
|
||
# --- Общие сведения ---
|
||
common_sn = None
|
||
for sn in wb.sheetnames:
|
||
if common_sheet and common_sheet.lower() in sn.lower():
|
||
common_sn = sn
|
||
break
|
||
if common_sn:
|
||
ws = wb[common_sn]
|
||
for row in ws.iter_rows(min_row=1, values_only=True):
|
||
label = clean_value(row[0]) if row and row[0] is not None else ''
|
||
value = clean_value(row[1]) if len(row) > 1 else ''
|
||
if not label:
|
||
continue
|
||
label_lower = label.lower()
|
||
for rule in rules:
|
||
pattern = rule.get('pattern', '')
|
||
field = rule.get('field', '')
|
||
if pattern and re.search(pattern, label_lower):
|
||
if not is_disabled(settings, field) and value:
|
||
result['fields'][field] = value
|
||
break
|
||
|
||
# --- Секции (комиссия, сотрудники) ---
|
||
sections = profile.get('sections', {})
|
||
emp_sn = None
|
||
for sn in wb.sheetnames:
|
||
if employee_sheet and employee_sheet.lower() in sn.lower():
|
||
emp_sn = sn
|
||
break
|
||
if emp_sn and sections:
|
||
ws = wb[emp_sn]
|
||
current = None
|
||
for row in ws.iter_rows(min_row=1, values_only=True):
|
||
col_a = clean_value(row[0]) if row and row[0] is not None else ''
|
||
a_lower = col_a.lower()
|
||
header_hit = None
|
||
for sec_id, sec_cfg in sections.items():
|
||
if is_disabled(settings, sec_id):
|
||
continue
|
||
if re.search(sec_cfg.get('header', ''), a_lower):
|
||
header_hit = sec_id
|
||
break
|
||
if header_hit:
|
||
current = header_hit
|
||
result['sections'].setdefault(current, [])
|
||
continue
|
||
if current:
|
||
# Заголовок таблицы внутри секции — пропускаем
|
||
if a_lower in ('№ п/п', '№', 'номер', 'п/п', '№пп', '№ пп'):
|
||
continue
|
||
row_data = [clean_value(c) if c is not None else ''
|
||
for c in row[:6]]
|
||
if any(row_data):
|
||
# Номер строки в первой колонке не нужен
|
||
if re.fullmatch(r'\d+', row_data[0].strip()):
|
||
row_data[0] = ''
|
||
result['sections'][current].append(row_data)
|
||
|
||
# --- Информационные системы (листы, отличные от общих/сотрудников) ---
|
||
if profile.get('is_enabled'):
|
||
for sn in wb.sheetnames:
|
||
if common_sheet and common_sheet.lower() in sn.lower():
|
||
continue
|
||
if employee_sheet and employee_sheet.lower() in sn.lower():
|
||
continue
|
||
ws = wb[sn]
|
||
is_obj = {'name': '', 'rows': []}
|
||
for row in ws.iter_rows(min_row=1, values_only=True):
|
||
vals = [clean_value(c) for c in row]
|
||
label = (vals[0] or '').strip() if vals else ''
|
||
if not label:
|
||
continue
|
||
is_obj['rows'].append(vals)
|
||
label_lower = label.lower()
|
||
if re.search(r'название ис|наименование ис|наименование информационной',
|
||
label_lower) and not is_disabled(settings, 'is_name'):
|
||
if len(vals) > 1:
|
||
is_obj['name'] = vals[1]
|
||
if is_obj['rows']:
|
||
if not is_obj['name']:
|
||
is_obj['name'] = sn.strip()
|
||
result['is_list'].append(is_obj)
|
||
|
||
# --- Пользовательские привязки ---
|
||
for cm in custom:
|
||
excel_text = (cm.get('excel_text') or '').strip().lower()
|
||
sheet = (cm.get('sheet') or '').strip()
|
||
field = (cm.get('field') or '').strip()
|
||
if not excel_text or not field:
|
||
continue
|
||
target_sheets = [sheet] if sheet in wb.sheetnames else wb.sheetnames
|
||
for sn in target_sheets:
|
||
ws = wb[sn]
|
||
for row in ws.iter_rows(min_row=1, values_only=True):
|
||
label = clean_value(row[0]) if row and row[0] is not None else ''
|
||
if label.lower().strip() == excel_text:
|
||
value = clean_value(row[1]) if len(row) > 1 else ''
|
||
if value:
|
||
result['fields'][field] = value
|
||
break
|
||
finally:
|
||
try:
|
||
wb.close()
|
||
except Exception:
|
||
pass
|
||
|
||
return result
|