1025 lines
48 KiB
Python
1025 lines
48 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""DokoGen — проверка данных через внешние сервисы.
|
||
|
||
Провайдеры:
|
||
- Морфер 3.0 (morpher.ru) — склонение ФИО/названий по падежам. Бесплатно, без ключа.
|
||
- DaData (dadata.ru) — проверка организации по ИНН/ОГРН (наименование, адрес, КПП).
|
||
Нужен API-ключ (бесплатный тариф ~10 000 запросов/день).
|
||
- ИИ (OpenAI-совместимый) — ИИ-проверка склонений и реквизитов (DeepSeek, ChatGPT и др.).
|
||
Нужен API-ключ и модель. Данные ИИ ВСЕГДА помечаются дисклеймером.
|
||
|
||
Каждая функция возвращает (result, error):
|
||
- result: dict с полями {source, source_url, data, disclaimer, apply}
|
||
- error: строка ошибки или None
|
||
"""
|
||
|
||
import json
|
||
|
||
def _load_prompt(name):
|
||
"""Загружает промт из dictionaries/prompts/<name>.json (system/rules/response_format)."""
|
||
import os as _os
|
||
p = _os.path.join(_os.path.dirname(_os.path.abspath(__file__)),
|
||
'dictionaries', 'prompts', name + '.json')
|
||
try:
|
||
with open(p, encoding='utf-8') as f:
|
||
return json.load(f)
|
||
except Exception:
|
||
return {}
|
||
|
||
import re
|
||
from datetime import datetime, timedelta
|
||
import urllib.parse
|
||
import urllib.request
|
||
import urllib.error
|
||
|
||
MORPHER_URL = "https://ws3.morpher.ru/russian/declension"
|
||
DADATA_URL = "https://suggestions.dadata.ru/suggestions/api/4_1/rs/findById/party"
|
||
DEFAULT_AI_URL = "https://api.deepseek.com/chat/completions"
|
||
|
||
CASES = ['И', 'Р', 'Д', 'В', 'Т', 'П']
|
||
CASE_NAMES = {
|
||
'И': 'Именительный', 'Р': 'Родительный', 'Д': 'Дательный',
|
||
'В': 'Винительный', 'Т': 'Творительный', 'П': 'Предложный',
|
||
}
|
||
|
||
|
||
def _http_json(url, data=None, headers=None, timeout=30):
|
||
"""HTTP-запрос, возвращает (json, error_str)."""
|
||
body = None
|
||
if data is not None:
|
||
body = json.dumps(data).encode('utf-8')
|
||
req = urllib.request.Request(url, data=body, headers=headers or {})
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=timeout) as resp:
|
||
raw = resp.read().decode('utf-8')
|
||
return json.loads(raw), None
|
||
except urllib.error.HTTPError as e:
|
||
try:
|
||
err_body = e.read().decode('utf-8')
|
||
except Exception:
|
||
err_body = ''
|
||
return None, f"HTTP {e.code}: {err_body[:300]}"
|
||
except Exception as e:
|
||
return None, str(e)
|
||
|
||
|
||
# ============================================================
|
||
# НОРМАЛИЗАЦИЯ ТЕКСТА (КАПС → нормальный регистр, сокращения адреса)
|
||
# ============================================================
|
||
_CAPS_ABBR = {
|
||
'АРМ', 'СКУД', 'СЗИ', 'СКЗИ', 'СОВ', 'ЛВС', 'ИНН', 'КПП', 'ОГРН', 'ЦОД',
|
||
'МБУ', 'МКУ', 'ООО', 'СНИЛС', 'СБИС', 'ЕМСЭД', 'ФИО', 'ИСПДН', 'ККТ',
|
||
}
|
||
|
||
_ADDRESS_ABBR = {
|
||
'Г': 'г.', 'УЛ': 'ул.', 'Д': 'д.', 'КВ': 'кв.', 'ПР': 'пр.',
|
||
'ПЕР': 'пер.', 'ПЛ': 'пл.', 'ОБЛ': 'обл.', 'ПОС': 'пос.',
|
||
'СТ': 'ст.', 'Ш': 'ш.', 'Р-Н': 'р-н.', 'Б-Р': 'б-р.', 'МКР': 'мкр.',
|
||
'НАБ': 'наб.', 'ТУП': 'туп.', 'ПР-Д': 'пр-д.', 'КРП': 'крп.',
|
||
'Г.': 'г.', 'УЛ.': 'ул.', 'Д.': 'д.', 'Р-Н.': 'р-н.',
|
||
'Г.О': 'г.о.', 'Г.О.': 'г.о.', 'ГО': 'г.о.', 'М.О': 'м.о.', 'М.О.': 'м.о.',
|
||
}
|
||
|
||
# Слова, которые в адресах пишутся со строчной буквы (кроме случаев,
|
||
# когда стоят первым словом после точки — «г. Электроугли» и т.п.)
|
||
_ADDRESS_LOWERCASE = {
|
||
'область', 'край', 'республика', 'округ', 'район', 'город', 'поселок',
|
||
'посёлок', 'деревня', 'село', 'станица', 'хутор', 'аул', 'улица',
|
||
'проспект', 'переулок', 'бульвар', 'шоссе', 'набережная', 'площадь',
|
||
'проезд', 'тупик', 'аллея', 'линия', 'дом', 'строение', 'корпус',
|
||
'квартира', 'офис', 'помещение', 'этаж', 'комната', 'участок',
|
||
}
|
||
|
||
|
||
def _normalize_text(text):
|
||
"""КАПС → нормальный регистр по словам (сохраняя аббревиатуры и числа)."""
|
||
if not text:
|
||
return ''
|
||
words = str(text).split()
|
||
out = []
|
||
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:
|
||
out.append(w)
|
||
continue
|
||
if any('а' <= c <= 'я' or c == 'ё' for c in core):
|
||
out.append(w)
|
||
continue
|
||
if not any('А' <= c <= 'Я' or c == 'Ё' for c in core):
|
||
out.append(w)
|
||
continue
|
||
letters = [c for c in core if c.isalpha()]
|
||
if len(letters) <= 2:
|
||
out.append(w)
|
||
continue
|
||
if core.upper() in _CAPS_ABBR:
|
||
out.append(w)
|
||
continue
|
||
out.append(pre + core[0].upper() + core[1:].lower() + post)
|
||
return ' '.join(out)
|
||
|
||
|
||
def normalize_address(text):
|
||
"""«Г Крымск» → «г. Крымск», «Р-Н» → «р-н.», «МОСКОВСКАЯ ОБЛАСТЬ» → «Московская область».
|
||
|
||
Приводит адрес к нормальному регистру: каждое слово с заглавной буквы,
|
||
адресные сокращения (г., ул., д., обл., г.о.) и административные термины
|
||
(область, район, город, дом и т.п.) — со строчной.
|
||
"""
|
||
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)
|
||
res = ''.join(out)
|
||
# КАПС → нормальный регистр по словам
|
||
res = _normalize_text(res)
|
||
# Адресные термины — со строчной буквы (сохраняя пунктуацию)
|
||
words = res.split(' ')
|
||
fixed = []
|
||
for w in words:
|
||
m = re.match(r'^([^\w]*)([\w\-]+)([^\w]*)$', w, re.UNICODE)
|
||
if m:
|
||
pre, core, post = m.group(1), m.group(2), m.group(3)
|
||
if core.lower() in _ADDRESS_LOWERCASE:
|
||
fixed.append(pre + core.lower() + post)
|
||
continue
|
||
fixed.append(w)
|
||
return ' '.join(fixed)
|
||
|
||
|
||
def resolve_okved(code, dadata_token=None):
|
||
"""Код ОКВЭД → «код название» (без дефиса), например «27.32.2 Производство силовых кабелей».
|
||
|
||
Принимает и голый код («27.32.2»), и строку с названием
|
||
(«27.32.2 - Производство силовых кабелей» / «27.32.2 Производство силовых кабелей»).
|
||
1) Ищем в словаре dictionaries/okved.json (строки вида «27.32.2 Название»).
|
||
2) Если не нашли — запрашиваем DaData suggest okved2 (если есть токен).
|
||
3) Иначе возвращаем нормализованную строку «код название» как есть.
|
||
"""
|
||
code = (code or '').strip()
|
||
if not code:
|
||
return ''
|
||
# Извлекаем код из входной строки: «27.32.2 - Производство силовых кабелей» → «27.32.2»
|
||
m = re.match(r'^\s*(\d+(?:\.\d+)*)', code)
|
||
code_only = m.group(1) if m else code.split()[0].rstrip('.')
|
||
# 1) Словарь
|
||
try:
|
||
import os, json
|
||
dict_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'dictionaries', 'okved.json')
|
||
with open(dict_path, encoding='utf-8') as f:
|
||
items = json.load(f)
|
||
for item in items:
|
||
item = str(item).strip()
|
||
# совпадение по коду: «27.32.2» или «27.32.2 Название»
|
||
if item.split()[0].rstrip('.') == code_only.rstrip('.'):
|
||
return item
|
||
except Exception:
|
||
pass
|
||
# 2) DaData suggest okved2
|
||
if dadata_token:
|
||
try:
|
||
js, err = _http_json('https://suggestions.dadata.ru/suggestions/api/4_1/rs/suggest/okved2',
|
||
data={'query': code_only, 'count': 1},
|
||
headers={'Content-Type': 'application/json',
|
||
'Authorization': f'Token {dadata_token}'})
|
||
if not err and js:
|
||
sugg = (js.get('suggestions') or [])
|
||
if sugg and sugg[0].get('value'):
|
||
return sugg[0]['value']
|
||
except Exception:
|
||
pass
|
||
# 3) Нормализуем вход: «27.32.2 - Название» → «27.32.2 Название» (без дефиса)
|
||
rest = code[len(m.group(1)):].strip() if m else ''
|
||
rest = rest.lstrip('- ').strip()
|
||
if rest:
|
||
return f"{code_only} {rest}"
|
||
return code_only
|
||
|
||
|
||
def _fmt_date(value):
|
||
"""Дата: '2005-02-09' → '09.02.2005'; миллисекунды 1107907200000 → дата."""
|
||
if not value:
|
||
return ''
|
||
# миллисекунды с эпохи (DaData иногда отдаёт так)
|
||
if isinstance(value, (int, float)) and value > 10 ** 10:
|
||
try:
|
||
dt = datetime(1970, 1, 1) + timedelta(milliseconds=value)
|
||
return dt.strftime('%d.%m.%Y')
|
||
except Exception:
|
||
pass
|
||
s = str(value).strip()
|
||
parts = s.split('-')
|
||
if len(parts) == 3 and all(p.isdigit() for p in parts):
|
||
return f"{parts[2]}.{parts[1]}.{parts[0]}"
|
||
return s
|
||
|
||
|
||
# ============================================================
|
||
# МОРФЕР 3.0 — склонение (бесплатно, без ключа)
|
||
# ============================================================
|
||
def check_declension_morpher(text):
|
||
"""Склонение слова/ФИО/названия через Морфер 3.0."""
|
||
if not text or not text.strip():
|
||
return None, "Пустое значение для склонения"
|
||
url = MORPHER_URL + '?' + urllib.parse.urlencode({'s': text.strip(), 'format': 'json'})
|
||
js, err = _http_json(url)
|
||
if err:
|
||
return None, err
|
||
if not isinstance(js, dict):
|
||
return None, "Неожиданный ответ Морфера"
|
||
|
||
decl = {}
|
||
for c in CASES:
|
||
val = js.get(c) or js.get(c.lower()) or ''
|
||
if isinstance(val, str) and val:
|
||
decl[c] = val
|
||
|
||
if not decl:
|
||
return None, "Морфер не смог просклонять (возможно, это не слово/ФИО)"
|
||
|
||
# Морфер не возвращает именительный падеж (он равен исходному слову) —
|
||
# добавляем его сами, иначе поле «Именительный» остаётся пустым
|
||
if 'И' not in decl:
|
||
decl['И'] = text.strip()
|
||
|
||
return {
|
||
'source': 'Морфер 3.0',
|
||
'source_url': 'https://morpher.ru',
|
||
'disclaimer': '',
|
||
'kind': 'declension',
|
||
'original': text.strip(),
|
||
'data': decl,
|
||
'apply': {'declension': decl},
|
||
}, None
|
||
|
||
|
||
# ============================================================
|
||
# DADATA — проверка организации по ИНН (нужен API-ключ)
|
||
# ============================================================
|
||
def get_company_finance_dadata(inn, token):
|
||
"""Доходы организации по годам через DaData (findById/party → finance/finance_history).
|
||
|
||
Возвращает (dict, ошибка): dict вида
|
||
{'finance': {...}, 'history': [{'year': 2023, 'income': 123456, ...}, ...]}
|
||
finance — показатели за последний отчётный год;
|
||
history — история по годам (если доступна в тарифе DaData).
|
||
"""
|
||
if not token:
|
||
return None, "Не задан API-ключ DaData (Файл → Настройки → API)"
|
||
inn = (inn or '').replace(' ', '')
|
||
if not inn or not inn.isdigit():
|
||
return None, "Введите ИНН для загрузки доходов"
|
||
headers = {
|
||
'Content-Type': 'application/json',
|
||
'Authorization': f'Token {token}',
|
||
}
|
||
js, err = _http_json(DADATA_URL, data={'query': inn}, headers=headers)
|
||
if err:
|
||
return None, err
|
||
suggestions = (js or {}).get('suggestions') or []
|
||
if not suggestions:
|
||
return None, f"DaData: организация с ИНН {inn} не найдена"
|
||
d = suggestions[0].get('data') or {}
|
||
return {
|
||
'finance': d.get('finance') or {},
|
||
'history': d.get('finance_history') or [],
|
||
'inn': d.get('inn', inn),
|
||
'name': (d.get('name') or {}).get('short_with_opf') or suggestions[0].get('value', ''),
|
||
}, None
|
||
|
||
|
||
def check_company_dadata(inn, token):
|
||
"""Проверка организации по ИНН/ОГРН через DaData.
|
||
|
||
Данные нормализуются: регистр (КАПС → обычный), адрес (г., ул., д.),
|
||
дата ОГРН (в т.ч. из миллисекунд).
|
||
"""
|
||
if not token:
|
||
return None, "Не задан API-ключ DaData (Файл → Настройки → API)"
|
||
inn = (inn or '').replace(' ', '')
|
||
if not inn or not inn.isdigit():
|
||
return None, "Введите ИНН или ОГРН для проверки"
|
||
headers = {
|
||
'Content-Type': 'application/json',
|
||
'Authorization': f'Token {token}',
|
||
}
|
||
js, err = _http_json(DADATA_URL, data={'query': inn}, headers=headers)
|
||
if err:
|
||
return None, err
|
||
suggestions = (js or {}).get('suggestions') or []
|
||
if not suggestions:
|
||
return None, f"DaData: организация с ИНН/ОГРН {inn} не найдена"
|
||
s = suggestions[0]
|
||
d = s.get('data') or {}
|
||
name = d.get('name') or {}
|
||
address = d.get('address') or {}
|
||
management = d.get('management') or {}
|
||
|
||
full_name = _normalize_text(name.get('full_with_opf') or s.get('value', ''))
|
||
short_name = _normalize_text(name.get('short_with_opf') or name.get('short') or '')
|
||
addr = normalize_address(address.get('value', ''))
|
||
# DaData отдаёт индекс отдельным полем postal_code — добавляем его
|
||
# в начало адреса, если его там ещё нет
|
||
postal = (address.get('data') or {}).get('postal_code') or address.get('postal_code') or ''
|
||
postal = str(postal).strip()
|
||
if postal and postal.isdigit():
|
||
addr_first = (addr or '').split(',')[0].strip()
|
||
if not addr_first.startswith(postal):
|
||
addr = f"{postal}, {addr}" if addr else postal
|
||
chief_position = _normalize_text(management.get('post', ''))
|
||
chief_fio = _normalize_text(management.get('name', ''))
|
||
ogrn_date = _fmt_date(d.get('ogrn_date') or '')
|
||
|
||
# ОКВЭД: основной (okved) и дополнительные (okveds — список словарей с 'code')
|
||
okved = d.get('okved', '') or ''
|
||
okved_extra = []
|
||
raw_okveds = d.get('okveds') or []
|
||
if isinstance(raw_okveds, str):
|
||
# DaData иногда отдаёт доп. ОКВЭД одной строкой через запятую
|
||
raw_okveds = [x.strip() for x in raw_okveds.split(',') if x.strip()]
|
||
for ok in raw_okveds:
|
||
if isinstance(ok, dict):
|
||
code = (ok.get('code') or '').strip()
|
||
else:
|
||
code = str(ok or '').strip()
|
||
if code:
|
||
okved_extra.append(code)
|
||
|
||
return {
|
||
'source': 'DaData',
|
||
'source_url': 'https://dadata.ru',
|
||
'disclaimer': '',
|
||
'kind': 'company',
|
||
'query': inn,
|
||
'data': {
|
||
'full_name': full_name,
|
||
'short_name': short_name,
|
||
'address': addr,
|
||
'kpp': d.get('kpp', ''),
|
||
'ogrn': d.get('ogrn', ''),
|
||
'ogrn_date': ogrn_date,
|
||
'chief_position': chief_position,
|
||
'chief_fio': chief_fio,
|
||
'inn': d.get('inn', inn),
|
||
'okved': okved,
|
||
'okved_extra': okved_extra,
|
||
},
|
||
'apply': {'company': {
|
||
'full_name': full_name,
|
||
'short_name': short_name,
|
||
'address': addr,
|
||
'kpp': d.get('kpp', ''),
|
||
'ogrn': d.get('ogrn', ''),
|
||
'ogrn_date': ogrn_date,
|
||
'chief_position': chief_position,
|
||
'chief_fio': chief_fio,
|
||
'okved': okved,
|
||
'okved_extra': okved_extra,
|
||
}},
|
||
}, None
|
||
|
||
|
||
# ============================================================
|
||
# ИИ (OpenAI-совместимый: DeepSeek, ChatGPT и др.)
|
||
# ============================================================
|
||
AI_DISCLAIMER = ("⚠ Сгенерировано ИИ. Информацию рекомендуется перепроверить "
|
||
"по официальным источникам.")
|
||
|
||
|
||
def _ai_json(api_key, model, base_url, system, user, timeout=60):
|
||
"""Запрос к OpenAI-совместимому API (DeepSeek, ChatGPT и т.п.)."""
|
||
if not api_key:
|
||
return None, "Не задан API-ключ ИИ (Файл → Настройки → API)"
|
||
url = (base_url or DEFAULT_AI_URL).rstrip('/')
|
||
if not url.endswith('/chat/completions'):
|
||
url += '/chat/completions'
|
||
headers = {
|
||
'Content-Type': 'application/json',
|
||
'Authorization': f'Bearer {api_key}',
|
||
}
|
||
payload = {
|
||
'model': model or 'deepseek-chat',
|
||
'messages': [
|
||
{'role': 'system', 'content': system},
|
||
{'role': 'user', 'content': user},
|
||
],
|
||
'temperature': 0.2,
|
||
'response_format': {'type': 'json_object'},
|
||
}
|
||
js, err = _http_json(url, data=payload, headers=headers, timeout=timeout)
|
||
if err:
|
||
return None, err
|
||
try:
|
||
content = js['choices'][0]['message']['content']
|
||
return json.loads(content), None
|
||
except Exception:
|
||
return None, "ИИ вернул неожиданный ответ"
|
||
|
||
|
||
def check_declension_ai(text, api_key, model='deepseek-chat', base_url=None):
|
||
"""ИИ-рецензия склонений: проверяет падежи и предлагает исправления."""
|
||
if not text or not text.strip():
|
||
return None, "Пустое значение для склонения"
|
||
system = (
|
||
"Ты — эксперт по русскому языку. Склоняй слово/ФИО/название организации "
|
||
"по падежам. Верни ТОЛЬКО JSON: "
|
||
'{"И": "...", "Р": "...", "Д": "...", "В": "...", "Т": "...", "П": "...", "комментарий": "..."}'
|
||
)
|
||
user = f"Просклоняй: {text.strip()}"
|
||
js, err = _ai_json(api_key, model, base_url, system, user)
|
||
if err:
|
||
return None, err
|
||
|
||
decl = {}
|
||
for c in CASES:
|
||
val = js.get(c) or js.get(c.lower()) or ''
|
||
if isinstance(val, str) and val:
|
||
decl[c] = val
|
||
comment = js.get('комментарий', '') or ''
|
||
|
||
if not decl:
|
||
return None, "ИИ не смог просклонять"
|
||
|
||
# Именительный падеж равен исходному слову — подставляем, если ИИ его не вернул
|
||
if 'И' not in decl:
|
||
decl['И'] = text.strip()
|
||
|
||
return {
|
||
'source': 'ИИ',
|
||
'source_url': 'openai-совместимый API',
|
||
'disclaimer': AI_DISCLAIMER,
|
||
'kind': 'declension',
|
||
'original': text.strip(),
|
||
'data': decl,
|
||
'comment': comment,
|
||
'apply': {'declension': decl},
|
||
}, None
|
||
|
||
|
||
# Поля организации для ИИ-проверки (для читаемого вывода)
|
||
COMPANY_FIELDS = [
|
||
('full_name', 'Полное наименование'),
|
||
('short_name', 'Краткое наименование'),
|
||
('address', 'Юридический адрес'),
|
||
('inn', 'ИНН'),
|
||
('kpp', 'КПП'),
|
||
('ogrn', 'ОГРН'),
|
||
('ogrn_date', 'Дата ОГРН'),
|
||
('chief_position', 'Руководитель (должность)'),
|
||
('chief_fio', 'Руководитель (ФИО)'),
|
||
]
|
||
|
||
# Поля для сравнения: (ключ, подпись, тип)
|
||
# type: text — регистр/пунктуация игнорируются; digits — любое отличие = расхождение
|
||
_COMPARE_FIELDS = [
|
||
('full_name', 'Наименование', 'text'),
|
||
('short_name', 'Краткое наименование', 'text'),
|
||
('address', 'Юридический адрес', 'text'),
|
||
('inn', 'ИНН', 'digits'),
|
||
('kpp', 'КПП', 'digits'),
|
||
('ogrn', 'ОГРН', 'digits'),
|
||
('ogrn_date', 'Дата ОГРН', 'text'),
|
||
('chief_position', 'Руководитель (должность)', 'text'),
|
||
('chief_fio', 'Руководитель (ФИО)', 'text'),
|
||
]
|
||
|
||
|
||
def _norm_compare(s):
|
||
"""Нормализация для сравнения: регистр, ё→е, пробелы/точки/дефисы/кавычки убираются."""
|
||
s = (s or '').lower().replace('ё', 'е')
|
||
s = re.sub(r'[\s.,;:()«»"\'\-—–]+', '', s)
|
||
return s
|
||
|
||
|
||
def _norm_digits(s):
|
||
"""Только цифры."""
|
||
return re.sub(r'\D', '', s or '')
|
||
|
||
|
||
def compare_company_values(old_data, new_data):
|
||
"""Сравнивает старые и новые значения реквизитов организации.
|
||
|
||
Возвращает список расхождений: [(подпись, было, стало), ...].
|
||
- text: разница в регистре/пунктуации (ИВАНОВ И.И. vs Иванов И.И.) НЕ считается;
|
||
- digits (ИНН/КПП/ОГРН): любое отличие цифр = расхождение.
|
||
"""
|
||
diffs = []
|
||
for key, label, ftype in _COMPARE_FIELDS:
|
||
old = (old_data or {}).get(key) or ''
|
||
new = (new_data or {}).get(key) or ''
|
||
if not old or not new:
|
||
continue
|
||
if ftype == 'digits':
|
||
if _norm_digits(old) != _norm_digits(new):
|
||
diffs.append((label, old, new))
|
||
else:
|
||
if _norm_compare(old) != _norm_compare(new):
|
||
diffs.append((label, old, new))
|
||
return diffs
|
||
|
||
|
||
def _ubi_code(text):
|
||
"""Извлекает код УБИ из строки: «УБИ.069 Название» → '069'."""
|
||
m = re.search(r'уби\.?\s*(\d+)', str(text), re.IGNORECASE)
|
||
return m.group(1) if m else None
|
||
|
||
|
||
def _norm_threat_text(text):
|
||
"""Нормализация названия угрозы для сравнения: без ведущего номера, лишних пробелов."""
|
||
s = re.sub(r'^\d+[.)]?\s*', '', str(text or '').strip())
|
||
s = re.sub(r'\s+', ' ', s).strip()
|
||
return s.lower()
|
||
|
||
|
||
def normalize_threats(text):
|
||
"""Склеить переносы внутри названий угроз.
|
||
|
||
В опросниках встречается «УБИ.209 Угроза … памяти\r\nядра процессора» —
|
||
перенос строки внутри одного названия. Строка, не начинающаяся с «УБИ.»,
|
||
считается продолжением предыдущей угрозы.
|
||
"""
|
||
if not text:
|
||
return ''
|
||
lines = [ln.strip() for ln in str(text).split('\n') if ln.strip()]
|
||
out = []
|
||
for ln in lines:
|
||
if re.match(r'^уби\.?\s*\d+', ln, re.IGNORECASE):
|
||
out.append(ln)
|
||
elif out:
|
||
out[-1] = out[-1] + ' ' + ln
|
||
else:
|
||
out.append(ln)
|
||
return '\n'.join(out)
|
||
|
||
|
||
def split_okved_extra(value):
|
||
"""Разбить okvedExtra (строка или список строк с переносами) на записи «код название».
|
||
|
||
В опросниках доп. ОКВЭД часто идут склеенными:
|
||
«25.93 — Производство изделий из проволоки\nцепей и пружин 32.12.1 — …»
|
||
Здесь «цепей и пружин» — продолжение названия 25.93, а 32.12.1 — новый код.
|
||
Нарезаем по кодам ОКВЭД (двузначный код с точками в начале записи).
|
||
"""
|
||
if isinstance(value, (list, tuple)):
|
||
text = ' '.join(str(x) for x in value)
|
||
else:
|
||
text = str(value or '')
|
||
text = text.replace('\r', ' ').replace('\n', ' ')
|
||
# Нарезаем по началу кода ОКВЭД: «25.93», «32.12.1», «43.2» …
|
||
# Перед кодом не должно быть цифры или точки (чтобы не разрезать «25.93» на «25» и «93»)
|
||
parts = re.split(r'(?<!\d)(?<!\.)(?=\d{2}(?:\.\d+)*\b)', text)
|
||
out = []
|
||
for p in parts:
|
||
p = p.strip().strip('-—– ').strip()
|
||
if p and re.match(r'^\d{2}(?:\.\d+)*', p):
|
||
out.append(p)
|
||
return out
|
||
|
||
|
||
def filter_threats_by_dict(lines, threats_full=None):
|
||
"""Приводит строки угроз (от ИИ или оператора) к каноническому виду по словарю УБИ.
|
||
|
||
Для каждой строки:
|
||
- есть код УБИ → берём полную запись «УБИ.xxx Название» из словаря;
|
||
- кода нет → ищем по названию (точное совпадение или вхождение);
|
||
- не нашлось в словаре → отбрасываем (мусор от ИИ).
|
||
|
||
Возвращает уникальные канонические строки, отсортированные по коду УБИ.
|
||
"""
|
||
if threats_full is None:
|
||
from .threat_rules import load_threats
|
||
threats_full = load_threats()
|
||
by_code = {}
|
||
by_norm = {}
|
||
for t in threats_full:
|
||
t = str(t).strip()
|
||
code = _ubi_code(t)
|
||
if code:
|
||
by_code.setdefault(code, t)
|
||
by_norm.setdefault(_norm_threat_text(t), t)
|
||
out = []
|
||
seen = set()
|
||
for line in lines or []:
|
||
line = str(line or '').strip()
|
||
if not line:
|
||
continue
|
||
code = _ubi_code(line)
|
||
canon = None
|
||
if code and code in by_code:
|
||
canon = by_code[code]
|
||
else:
|
||
n = _norm_threat_text(line)
|
||
if n in by_norm:
|
||
canon = by_norm[n]
|
||
else:
|
||
for k, v in by_norm.items():
|
||
if n and (n in k or k in n):
|
||
canon = v
|
||
break
|
||
if canon and canon not in seen:
|
||
seen.add(canon)
|
||
out.append(canon)
|
||
out.sort(key=lambda t: (int(_ubi_code(t)) if _ubi_code(t) else 10 ** 9, t))
|
||
return out
|
||
|
||
|
||
def suggest_threats_ai(security_data, api_key, model='deepseek-chat', base_url=None,
|
||
threats=None):
|
||
"""ИИ предлагает актуальные угрозы БИ для объекта КИИ по его характеристикам.
|
||
|
||
security_data — dict: name, description, object_type, sphere, architecture,
|
||
attacker_category, has_internet (True/False), etc.
|
||
threats — актуальный перечень угроз БДУ ФСТЭК (список строк «УБИ.xxx …»).
|
||
Передаётся в промт целиком, чтобы ИИ выбирал из свежего списка,
|
||
а не из своего устаревшего.
|
||
|
||
Возвращает (список_угроз, ошибка).
|
||
"""
|
||
if not api_key:
|
||
return None, "Не задан API-ключ ИИ (Файл → Настройки → API)"
|
||
|
||
if threats:
|
||
threat_list_text = "\n".join(f"{i+1}. {t}" for i, t in enumerate(threats))
|
||
else:
|
||
threat_list_text = "(перечень не предоставлен — используй знания об актуальном перечне БДУ ФСТЭК)"
|
||
|
||
pr = _load_prompt('suggest_threats')
|
||
system = pr.get('system', (
|
||
"Ты — эксперт ФСТЭК по безопасности критической информационной инфраструктуры "
|
||
"(187-ФЗ, приказ ФСТЭК № 236). На основе характеристик объекта КИИ выбери из "
|
||
"предоставленного перечня АКТУАЛЬНЫЕ угрозы безопасности информации."
|
||
))
|
||
rules = pr.get('rules', [])
|
||
resp = pr.get('response_format',
|
||
'Верни ТОЛЬКО JSON: {"threats": ["УБИ.xxx Название угрозы", ...]} '
|
||
'— от 10 до 40 угроз, без пояснений.')
|
||
user = (
|
||
"Характеристики объекта КИИ:\n"
|
||
f"{json.dumps(security_data, ensure_ascii=False)}\n\n"
|
||
"Актуальный перечень угроз БДУ ФСТЭК (выбирай ТОЛЬКО из него, "
|
||
"используй полные строки как они есть):\n"
|
||
f"{threat_list_text}\n\n"
|
||
+ ("Правила:\n" + "\n".join(f"{i+1}. {r}" for i, r in enumerate(rules)) + "\n\n" if rules else "")
|
||
+ resp
|
||
)
|
||
js, err = _ai_json(api_key, model, base_url, system, user)
|
||
if err:
|
||
return None, err
|
||
threats_out = js.get('threats') or []
|
||
if isinstance(threats_out, str):
|
||
threats_out = [threats_out]
|
||
threats_out = [str(t).strip() for t in threats_out if str(t).strip()]
|
||
# Жёсткий фильтр по словарю: мусор отбрасываем, «УБИ.069» дополняем
|
||
# полным названием из актуального перечня (если он передан).
|
||
threats_out = filter_threats_by_dict(threats_out, threats)
|
||
return threats_out, None
|
||
|
||
|
||
def check_threats_ai(security_data, current_threats, api_key, model='deepseek-chat', base_url=None):
|
||
"""ИИ проверяет уже подобранный список угроз БИ для объекта КИИ.
|
||
|
||
В отличие от suggest_threats_ai (подбор заново), эта функция проверяет
|
||
текущий список: какие угрозы лишние, каких не хватает, и даёт комментарий.
|
||
|
||
Возвращает (dict, ошибка):
|
||
{"лишние": [...], "недостающие": [...], "комментарий": "..."}
|
||
"""
|
||
if not api_key:
|
||
return None, "Не задан API-ключ ИИ (Файл → Настройки → API)"
|
||
if not current_threats:
|
||
return None, "Список угроз пуст — сначала подберите угрозы (программой или ИИ)."
|
||
|
||
system = (
|
||
"Ты — эксперт ФСТЭК по безопасности критической информационной инфраструктуры "
|
||
"(187-ФЗ, приказ ФСТЭК № 236). Проверь, корректно ли подобран перечень угроз "
|
||
"безопасности информации для объекта КИИ."
|
||
)
|
||
user = (
|
||
"Характеристики объекта КИИ:\n"
|
||
f"{json.dumps(security_data, ensure_ascii=False)}\n\n"
|
||
"Текущий перечень угроз (подобран программой или оператором):\n"
|
||
+ "\n".join(f"{i+1}. {t}" for i, t in enumerate(current_threats))
|
||
+ "\n\nВерни ТОЛЬКО JSON: "
|
||
'{"лишние": ["угрозы, которые НЕ актуальны для этого объекта"], '
|
||
'"недостающие": ["актуальные угрозы, которых в списке не хватает"], '
|
||
'"комментарий": "краткий вывод по перечню"}'
|
||
)
|
||
js, err = _ai_json(api_key, model, base_url, system, user)
|
||
if err:
|
||
return None, err
|
||
# «Недостающие»/«лишние» — тоже через словарь: мусор отбрасываем,
|
||
# неполные строки («УБИ.069») дополняем полными названиями.
|
||
missing = filter_threats_by_dict(js.get('недостающие') or [])
|
||
extra = filter_threats_by_dict(js.get('лишние') or [])
|
||
comment = js.get('комментарий') or ''
|
||
return {'лишние': extra, 'недостающие': missing, 'комментарий': comment}, None
|
||
|
||
|
||
def suggest_object_params_ai(object_data, api_key, model='deepseek-chat', base_url=None,
|
||
typical_objects=None, spheres=None, object_types=None,
|
||
architectures=None):
|
||
"""ИИ предлагает характеристики объекта КИИ: назначение, сферу, тип, архитектуру,
|
||
типовой отраслевой объект — как рекомендацию оператору.
|
||
|
||
object_data — dict: name, description, purpose (что известно об объекте).
|
||
typical_objects — перечень типовых отраслевых объектов (из 360-р) для выбора;
|
||
spheres / object_types / architectures — полные списки допустимых значений.
|
||
Промт берётся из dictionaries/prompts/suggest_object_params.json.
|
||
|
||
Возвращает (dict, ошибка): dict с полями
|
||
назначение / сфера / тип / архитектура / типовой_объект / обоснование.
|
||
"""
|
||
if not api_key:
|
||
return None, "Не задан API-ключ ИИ (Файл → Настройки → API)"
|
||
|
||
# Загрузка промта из папки dictionaries/prompts
|
||
import os as _os
|
||
prompt_path = _os.path.join(_os.path.dirname(_os.path.abspath(__file__)),
|
||
'dictionaries', 'prompts', 'suggest_object_params.json')
|
||
try:
|
||
with open(prompt_path, encoding='utf-8') as _f:
|
||
prompt = json.load(_f)
|
||
except Exception:
|
||
prompt = {}
|
||
system = prompt.get('system', 'Ты — эксперт ФСТЭК по категорированию объектов КИИ.')
|
||
rules = prompt.get('rules', [])
|
||
response_format = prompt.get('response_format',
|
||
'Верни ТОЛЬКО JSON: {"назначение": "...", "сфера": "...", '
|
||
'"тип": "...", "архитектура": "...", "типовой_объект": "...", '
|
||
'"обоснование": "..."}')
|
||
|
||
# Полные списки допустимых значений
|
||
def _fmt_list(name, items):
|
||
if not items:
|
||
return f"{name}: (список не предоставлен)"
|
||
return f"{name}:\n" + "\n".join(f"- {i}" for i in items)
|
||
|
||
lists_text = "\n\n".join([
|
||
_fmt_list("Сферы деятельности (выбирай ТОЛЬКО из них)", spheres),
|
||
_fmt_list("Типы объектов (выбирай ТОЛЬКО из них)", object_types),
|
||
_fmt_list("Архитектуры (выбирай ТОЛЬКО из них)", architectures),
|
||
])
|
||
if typical_objects:
|
||
to_text = "\n".join(f"{i+1}. {t}" for i, t in enumerate(typical_objects[:400]))
|
||
else:
|
||
to_text = "(перечень не предоставлен)"
|
||
|
||
rules_text = "\n".join(f"{i+1}. {r}" for i, r in enumerate(rules))
|
||
|
||
user = (
|
||
"Информация об объекте КИИ:\n"
|
||
f"{json.dumps(object_data, ensure_ascii=False)}\n\n"
|
||
f"{lists_text}\n\n"
|
||
"Перечень типовых отраслевых объектов КИИ (Распоряжение ПРФ № 360-р, "
|
||
"выбирай ТОЛЬКО из него, копируй строку дословно):\n"
|
||
f"{to_text}\n\n"
|
||
"Правила:\n"
|
||
f"{rules_text}\n\n"
|
||
f"{response_format}"
|
||
)
|
||
js, err = _ai_json(api_key, model, base_url, system, user)
|
||
if err:
|
||
return None, err
|
||
return {
|
||
'назначение': str(js.get('назначение', '') or '').strip(),
|
||
'сфера': str(js.get('сфера', '') or '').strip(),
|
||
'тип': str(js.get('тип', '') or '').strip(),
|
||
'архитектура': str(js.get('архитектура', '') or '').strip(),
|
||
'типовой_объект': str(js.get('типовой_объект', '') or '').strip(),
|
||
'обоснование': str(js.get('обоснование', '') or '').strip(),
|
||
}, None
|
||
|
||
|
||
def check_security_ai(security_data, api_key, model='deepseek-chat', base_url=None):
|
||
"""ИИ-проверка заполненной вкладки «Безопасность» объекта КИИ (236 приказ).
|
||
|
||
Возвращает (результат, ошибка): результат — dict с 'оценка', 'замечания', 'предложения'.
|
||
"""
|
||
if not api_key:
|
||
return None, "Не задан API-ключ ИИ (Файл → Настройки → API)"
|
||
pr = _load_prompt('check_security')
|
||
system = pr.get('system', (
|
||
"Ты — эксперт по безопасности критической информационной инфраструктуры (187-ФЗ, "
|
||
"приказ ФСТЭК России от 25.12.2017 № 239 и приказ № 236). Проверь заполненные "
|
||
"сведения об объекте КИИ и дай рекомендации."
|
||
))
|
||
system = system + " " + pr.get('response_format',
|
||
'Верни ТОЛЬКО JSON: {"оценка": "ok|warning|error", "замечания": ["..."], '
|
||
'"предложения": {"attacker_category": "...", "threats": "...", "incidents": "...", '
|
||
'"criteria_values": "...", "criteria_justification": "...", '
|
||
'"org_measures": "...", "tech_measures": "..."}} '
|
||
'В «предложениях» укажи, что добавить/исправить по каждому разделу, иначе пустые строки.')
|
||
user = json.dumps(security_data, ensure_ascii=False)
|
||
js, err = _ai_json(api_key, model, base_url, system, user)
|
||
if err:
|
||
return None, err
|
||
notes = js.get('замечания') or []
|
||
if isinstance(notes, str):
|
||
notes = [notes]
|
||
proposals = js.get('предложения') or {}
|
||
if not isinstance(proposals, dict):
|
||
proposals = {}
|
||
return {
|
||
'source': 'ИИ',
|
||
'source_url': 'openai-совместимый API',
|
||
'disclaimer': AI_DISCLAIMER,
|
||
'kind': 'security',
|
||
'data': {
|
||
'rating': js.get('оценка', 'warning'),
|
||
'notes': notes,
|
||
'proposals': proposals,
|
||
},
|
||
}, None
|
||
|
||
|
||
def check_company_ai(company_data, api_key, model='deepseek-chat', base_url=None):
|
||
"""ИИ-проверка реквизитов организации.
|
||
|
||
Возвращает оценку, замечания, предложения и полный «снимок» данных
|
||
(что ИИ увидел и как оценил каждое поле) — чтобы пользователь видел
|
||
не только «ОК», но и информацию по организации.
|
||
"""
|
||
if not api_key:
|
||
return None, "Не задан API-ключ ИИ (Файл → Настройки → API)"
|
||
system = (
|
||
"Ты — специалист по проверке реквизитов организации (ИНН/КПП/ОГРН). "
|
||
"Проверь реквизиты организации. "
|
||
"Верни ТОЛЬКО JSON: "
|
||
'{"оценка": "ok|warning|error", '
|
||
'"замечания": ["..."], '
|
||
'"предложения": {"full_name": "...", "short_name": "...", "address": "...", '
|
||
'"inn": "...", "kpp": "...", "ogrn": "...", "ogrn_date": "...", '
|
||
'"chief_position": "...", "chief_fio": "..."}, '
|
||
'"поля": {"full_name": {"статус": "ok|пусто|ошибка", "комментарий": "..."}, ...}} '
|
||
"В «предложениях» указывай исправленные значения только если уверен, иначе пустые строки. "
|
||
"В «полях» дай оценку каждого поля (ok/пусто/ошибка) и комментарий."
|
||
)
|
||
user = json.dumps(company_data, ensure_ascii=False)
|
||
js, err = _ai_json(api_key, model, base_url, system, user)
|
||
if err:
|
||
return None, err
|
||
|
||
notes = js.get('замечания') or []
|
||
if isinstance(notes, str):
|
||
notes = [notes]
|
||
proposals = js.get('предложения') or {}
|
||
if not isinstance(proposals, dict):
|
||
proposals = {}
|
||
fields = js.get('поля') or {}
|
||
if not isinstance(fields, dict):
|
||
fields = {}
|
||
|
||
return {
|
||
'source': 'ИИ',
|
||
'source_url': 'openai-совместимый API',
|
||
'disclaimer': AI_DISCLAIMER,
|
||
'kind': 'company',
|
||
'data': {
|
||
'rating': js.get('оценка', 'warning'),
|
||
'notes': notes,
|
||
'proposals': proposals,
|
||
'fields': fields,
|
||
},
|
||
'apply': {'company': {k: v for k, v in proposals.items() if v}},
|
||
}, None
|
||
|
||
|
||
def suggest_defense_ai(defense_data, api_key, model='deepseek-chat', base_url=None):
|
||
"""ИИ-рекомендации по средствам защиты информации (СЗИ).
|
||
|
||
По категории значимости, угрозам и текущему списку СЗИ ИИ подсказывает,
|
||
чего не хватает (например, антивирус, СКЗИ, SIEM и т.п.).
|
||
Возвращает (результат, ошибка).
|
||
"""
|
||
if not api_key:
|
||
return None, "Не задан API-ключ ИИ (Файл → Настройки → API)"
|
||
pr = _load_prompt('suggest_defense')
|
||
system = pr.get('system', (
|
||
"Ты — эксперт по безопасности КИИ (187-ФЗ, приказ ФСТЭК № 239). "
|
||
"По категории значимости объекта и актуальным угрозам определи, каких "
|
||
"средств защиты информации (СЗИ) не хватает, и дай рекомендации."
|
||
))
|
||
system = system + " " + pr.get('response_format',
|
||
'Верни ТОЛЬКО JSON: {"рекомендации": [{"сзи": "...", "зачем": "..."}], '
|
||
'"общие_советы": ["..."]}')
|
||
user = json.dumps(defense_data, ensure_ascii=False)
|
||
js, err = _ai_json(api_key, model, base_url, system, user)
|
||
if err:
|
||
return None, err
|
||
recs = js.get('рекомендации') or []
|
||
if not isinstance(recs, list):
|
||
recs = []
|
||
tips = js.get('общие_советы') or []
|
||
if isinstance(tips, str):
|
||
tips = [tips]
|
||
return {
|
||
'source': 'ИИ',
|
||
'source_url': 'openai-совместимый API',
|
||
'disclaimer': AI_DISCLAIMER,
|
||
'kind': 'defense',
|
||
'data': {
|
||
'рекомендации': recs,
|
||
'общие_советы': tips,
|
||
},
|
||
}, None
|
||
|
||
|
||
def check_criteria_ai(criteria_data, api_key, model='deepseek-chat', base_url=None):
|
||
"""ИИ-проверка категорий значимости по ПП РФ № 127.
|
||
|
||
По данным объекта КИИ (назначение, сфера, категории по показателям,
|
||
обоснования) ИИ проверяет корректность присвоенных категорий.
|
||
Возвращает (результат, ошибка).
|
||
"""
|
||
if not api_key:
|
||
return None, "Не задан API-ключ ИИ (Файл → Настройки → API)"
|
||
pr = _load_prompt('check_criteria')
|
||
system = pr.get('system', (
|
||
"Ты — эксперт по категорированию объектов КИИ (187-ФЗ, ПП РФ № 127). "
|
||
"Проверь присвоенные категории значимости по показателям критериев."
|
||
))
|
||
system = system + " " + pr.get('response_format',
|
||
'Верни ТОЛЬКО JSON: {"оценка": "ok|warning|error", '
|
||
'"замечания": [{"показатель": "...", "категория": "...", "комментарий": "..."}], '
|
||
'"предложения": {"criteria_values": "...", "criteria_justification": "..."}, '
|
||
'"итоговая_категория": "I|II|III|—"}')
|
||
user = json.dumps(criteria_data, ensure_ascii=False)
|
||
js, err = _ai_json(api_key, model, base_url, system, user)
|
||
if err:
|
||
return None, err
|
||
notes = js.get('замечания') or []
|
||
if isinstance(notes, str):
|
||
notes = [notes]
|
||
proposals = js.get('предложения') or {}
|
||
if not isinstance(proposals, dict):
|
||
proposals = {}
|
||
return {
|
||
'source': 'ИИ',
|
||
'source_url': 'openai-совместимый API',
|
||
'disclaimer': AI_DISCLAIMER,
|
||
'kind': 'criteria',
|
||
'data': {
|
||
'rating': js.get('оценка', 'warning'),
|
||
'notes': notes,
|
||
'proposals': proposals,
|
||
'final_category': js.get('итоговая_категория', ''),
|
||
},
|
||
}, None
|
||
|
||
|
||
def check_object_partial(blocks, api_key, model='deepseek-chat', base_url=None, workdir=None):
|
||
"""ИИ-проверка объекта КИИ по частям.
|
||
|
||
blocks: список [{'name': 'Реквизиты организации', 'prompt': 'check_company', 'data': {...}}, ...]
|
||
Для каждого блока: данные пишутся во временный файл (вход), вызывается ИИ,
|
||
ответ пишется во временный файл (выход). Возвращает (results, errors):
|
||
results — [{name, ok, text, files: {in: путь, out: путь}}].
|
||
"""
|
||
import tempfile
|
||
import os
|
||
if not api_key:
|
||
return None, "Не задан API-ключ ИИ (Файл → Настройки → API)"
|
||
if workdir is None:
|
||
workdir = tempfile.mkdtemp(prefix='dokogen_ai_')
|
||
results = []
|
||
for i, blk in enumerate(blocks, 1):
|
||
name = blk.get('name', f'Блок {i}')
|
||
prompt_name = blk.get('prompt', 'check_security')
|
||
data = blk.get('data', {})
|
||
f_in = os.path.join(workdir, f'block_{i}_input.json')
|
||
f_out = os.path.join(workdir, f'block_{i}_output.json')
|
||
try:
|
||
with open(f_in, 'w', encoding='utf-8') as f:
|
||
json.dump(data, f, ensure_ascii=False, indent=2)
|
||
except Exception:
|
||
pass
|
||
pr = _load_prompt(prompt_name)
|
||
system = pr.get('system', 'Ты — эксперт по КИИ (187-ФЗ). Проверь данные и дай рекомендации.')
|
||
system = system + ' ' + pr.get('response_format', 'Верни ТОЛЬКО JSON с результатом проверки.')
|
||
user = json.dumps(data, ensure_ascii=False)
|
||
js, err = _ai_json(api_key, model, base_url, system, user)
|
||
if err:
|
||
results.append({'name': name, 'ok': False, 'text': f'Ошибка: {err}', 'files': {'in': f_in, 'out': f_out}})
|
||
continue
|
||
try:
|
||
with open(f_out, 'w', encoding='utf-8') as f:
|
||
json.dump(js, f, ensure_ascii=False, indent=2)
|
||
except Exception:
|
||
pass
|
||
text = json.dumps(js, ensure_ascii=False, indent=2)
|
||
results.append({'name': name, 'ok': True, 'text': text, 'files': {'in': f_in, 'out': f_out}})
|
||
return results, None
|