# -*- 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 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): """Код ОКВЭД → «код - название». 1) Ищем в словаре dictionaries/okved.json (строки вида «27.32.2 Название»). 2) Если не нашли — запрашиваем DaData suggest okved2 (если есть токен). 3) Иначе возвращаем код как есть. """ code = (code or '').strip() if not code: return '' # 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 - ...» / «27.32.2 ...» if item.split()[0].rstrip('.') == code.rstrip('.'): # нормализуем: «код - название» rest = item[len(item.split()[0]):].strip().lstrip('- ').strip() if rest: return f"{code} - {rest}" 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, '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 return code 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 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 = [] for ok in (d.get('okveds') or []): code = (ok or {}).get('code') or '' 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 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 = "(перечень не предоставлен — используй знания об актуальном перечне БДУ ФСТЭК)" system = ( "Ты — эксперт ФСТЭК по безопасности критической информационной инфраструктуры " "(187-ФЗ, приказ ФСТЭК № 236). На основе характеристик объекта КИИ выбери из " "предоставленного перечня АКТУАЛЬНЫЕ угрозы безопасности информации и верни " "ТОЛЬКО JSON: " '{"threats": ["УБИ.xxx Название угрозы", ...]} — от 10 до 40 угроз, ' "без пояснений и комментариев. Не выдумывай угрозы, которых нет в перечне; " "используй полные строки из перечня (код + название) как они есть." ) user = ( "Характеристики объекта КИИ:\n" f"{json.dumps(security_data, ensure_ascii=False)}\n\n" "Актуальный перечень угроз БДУ ФСТЭК:\n" f"{threat_list_text}" ) 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()] return threats_out, None def suggest_object_params_ai(object_data, api_key, model='deepseek-chat', base_url=None, typical_objects=None): """ИИ предлагает характеристики объекта КИИ: назначение, сферу, тип, архитектуру, типовой отраслевой объект — как рекомендацию оператору. object_data — dict: name, description, purpose (что известно об объекте). typical_objects — перечень типовых отраслевых объектов (из 360-р) для выбора. Возвращает (dict, ошибка): dict с полями purpose / sphere / object_type / architecture / typical_object / обоснование. """ if not api_key: return None, "Не задан API-ключ ИИ (Файл → Настройки → API)" if typical_objects: to_text = "\n".join(f"{i+1}. {t}" for i, t in enumerate(typical_objects[:400])) else: to_text = "(перечень не предоставлен)" system = ( "Ты — эксперт ФСТЭК по категорированию объектов критической информационной инфраструктуры " "(187-ФЗ, ПП РФ № 127, Распоряжение ПРФ № 360-р, приказ ФСТЭК № 236). " "По описанию объекта КИИ предложи его характеристики. Верни ТОЛЬКО JSON: " '{"назначение": "короткое назначение объекта", ' '"сфера": "одна из сфер: Здравоохранение, Наука, Транспорт, Связь, Энергетика, ' 'Государственная регистрация прав, Банковская сфера и финансовый рынок, ' 'Топливно-энергетический комплекс, Атомная энергетика, Оборонная промышленность, ' 'Ракетно-космическая промышленность, Горнодобывающая промышленность, ' 'Металлургическая промышленность, Химическая промышленность, Иная сфера деятельности", ' '"тип": "Информационная система | Автоматизированные системы управления | ' 'Информационно-телекоммуникационная сеть", ' '"архитектура": "одно из: клиент-серверная система, автономная (standalone), ' 'распределенная система, иное", ' '"типовой_объект": "полная строка из предоставленного перечня, наиболее подходящая, ' 'или пустая строка", ' '"обоснование": "почему выбраны такие характеристики (1-2 предложения)"} ' "Не выдумывай типовой объект — бери только из перечня." ) user = ( "Информация об объекте КИИ:\n" f"{json.dumps(object_data, ensure_ascii=False)}\n\n" "Перечень типовых отраслевых объектов КИИ (Распоряжение ПРФ № 360-р):\n" f"{to_text}" ) 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)" system = ( "Ты — эксперт по безопасности критической информационной инфраструктуры (187-ФЗ, " "приказ ФСТЭК России от 25.12.2017 № 239 «Об утверждении Требований по обеспечению " "безопасности значимых объектов КИИ» и приказ № 236). Проверь заполненные сведения " "об объекте КИИ и дай рекомендации. Верни ТОЛЬКО 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 = ( "Ты — юрист по защите персональных данных (187-ФЗ). Проверь реквизиты организации. " "Верни ТОЛЬКО 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