Фаза 2: site/app.py → blueprint'ы (main, health, api), старый drhider.py удалён
Deploy drhider / validate (push) Waiting to run

This commit is contained in:
2026-07-12 08:34:06 +04:00
parent 2cf4e08100
commit 51bc1a79e7
7 changed files with 182 additions and 632 deletions
+2 -1
View File
@@ -1,4 +1,5 @@
НЕ СПЕШИ ! не ошибайся. НЕ ПРЕДПОЛАГАЙ !
не надо ДОГАДОК !
не надо самостоятельно что либо "УЛуЧШАТЬ" - никаких изменений рабочего кода без прямого разрешения
есть сомнения - лучше остановись и спроси
есть сомнения - лучше остановись и спроси
ВСЁ записывай в хистори ! что планировал что сделал в чём ошибся и тд
-557
View File
@@ -1,557 +0,0 @@
"""
DrHider — обфускация документов (двухпроходная, в памяти, без БД).
Проход 1: собрать все сущности из всех файлов → глобальный словарь замен.
Проход 2: применить замены → собрать ZIP с обфусцированными файлами + mapping.csv.
Согласованность: одна и та же сущность во всех файлах → одно и то же фиктивное значение.
"""
DRHIDER_VERSION = "1.3"
import io
import csv
import re
import random
import string
import zipfile
import logging
import os
from typing import Dict, List, Tuple, Callable, Optional
log = logging.getLogger("drhider")
# ═══════════════════════════════════════════
# Regex-паттерны для обнаружения сущностей
# ═══════════════════════════════════════════
ENTITY_PATTERNS: Dict[str, str] = {
"phone": r'(?<!\d)(?:\+7|8)[\s\-]?\(?\d{3}\)?[\s\-]?\d{3}[\s\-]?\d{2}[\s\-]?\d{2}(?!\d)',
"email": r'\b[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}\b',
# 12-значный ИНН ДО 10-значного (иначе 12-значный матчится как 10)
"inn_fl": r'ИНН\s*\d{12}',
"inn_ul": r'ИНН\s*\d{10}',
"ogrn": r'ОГРН\s*\d{13}',
"kpp": r'КПП\s*\d{9}',
"bik": r'БИК\s*\d{9}',
"rs": r'(?:р/с|расч[её]тный\s*сч[её]т)\s*\d{20}',
"ks": r'(?:к/с|кор/сч|корр?[еи]?спондентский\s*сч[её]т)\s*\d{20}',
"passport": r'(?:паспорт|серия\s+номер)[\s:№]*\d{2}\s*\d{2}\s*\d{6,7}',
}
# Фирмы — обнаружение по шаблону
COMPANY_PATTERN = re.compile(
r'(?:ООО|ЗАО|ОАО|АО|ПАО|ИП|ТОО)\s+(?:«[^»]+»|"[^"]+"|\u201c[^\u201d]+\u201d|[А-ЯA-Z][\w\-\.]{2,40})',
re.IGNORECASE
)
# ФИО — Фамилия + инициалы или полное имя (только кириллица)
PERSON_PATTERN = re.compile(
r'\b[А-Я][а-я]+\s+[А-Я]\.[А-Я]\.'
r'|\b[А-Я][а-я]+\s+[А-Я][а-я]+\s+[А-Я][а-я]+'
)
# ═══════════════════════════════════════════
# Генераторы фиктивных значений
# ═══════════════════════════════════════════
# Словари русских имён
RU_SURNAMES = ["Иванов", "Смирнов", "Кузнецов", "Попов", "Васильев", "Петров",
"Соколов", "Михайлов", "Новиков", "Фёдоров", "Морозов", "Волков", "Алексеев",
"Лебедев", "Семёнов", "Егоров", "Павлов", "Козлов", "Степанов", "Николаев"]
RU_NAMES = ["Александр", "Дмитрий", "Сергей", "Андрей", "Алексей", "Максим",
"Евгений", "Иван", "Михаил", "Николай", "Владимир", "Павел", "Виктор", "Олег"]
RU_PATRONYMICS = ["Александрович", "Дмитриевич", "Сергеевич", "Андреевич",
"Алексеевич", "Иванович", "Михайлович", "Николаевич", "Владимирович",
"Павлович", "Викторович", "Олегович", "Евгеньевич", "Максимович"]
RU_CITIES = ["Москва", "Санкт-Петербург", "Новосибирск", "Екатеринбург",
"Казань", "Нижний Новгород", "Челябинск", "Самара", "Омск", "Ростов-на-Дону",
"Уфа", "Красноярск", "Воронеж", "Пермь", "Волгоград"]
RU_STREETS = ["Ленина", "Мира", "Пушкина", "Гагарина", "Советская",
"Кирова", "Октябрьская", "Молодёжная", "Садовая", "Центральная"]
FAKE_DOMAINS = ["example.ru", "mail.test", "company.local", "org.example.ru"]
def _checksum_inn10(inn: str) -> str:
"""Контрольная сумма для 10-значного ИНН."""
coeffs = [2, 4, 10, 3, 5, 9, 4, 6, 8]
s = sum(int(inn[i]) * coeffs[i] for i in range(9))
return str((s % 11) % 10)
def _checksum_inn12(inn: str) -> str:
"""Контрольные суммы для 12-значного ИНН (первые 10 цифр)."""
c1 = [7, 2, 4, 10, 3, 5, 9, 4, 6, 8]
c2 = [3, 7, 2, 4, 10, 3, 5, 9, 4, 6, 8]
s1 = sum(int(inn[i]) * c1[i] for i in range(10))
n1 = (s1 % 11) % 10
inn2 = inn + str(n1)
s2 = sum(int(inn2[i]) * c2[i] for i in range(11))
n2 = (s2 % 11) % 10
return str(n1) + str(n2)
def _checksum_ogrn(ogrn: str) -> str:
"""Контрольная сумма для ОГРН (12 цифр → остаток от деления на 11)."""
s = int(ogrn) % 11
return str(s % 10)
def _random_digits(n: int) -> str:
return ''.join(random.choice(string.digits) for _ in range(n))
def _random_letters(n: int) -> str:
return ''.join(random.choice(string.ascii_lowercase) for _ in range(n))
def generate_phone(_: str) -> str:
code = random.choice(["495", "499", "812", "383", "343"])
return f"+7 ({code}) {_random_digits(3)}-{_random_digits(2)}-{_random_digits(2)}"
def generate_email(original: str) -> str:
"""Генерирует email с тем же форматом."""
domain = random.choice(FAKE_DOMAINS)
local = _random_letters(random.randint(5, 10))
return f"{local}@{domain}"
def generate_inn10(original: str) -> str:
prefix = re.match(r'ИНН\s*', original, re.IGNORECASE).group(0)
base = f"{random.randint(1,9)}{_random_digits(8)}"
return prefix + base + _checksum_inn10(base)
def generate_inn12(original: str) -> str:
prefix = re.match(r'ИНН\s*', original, re.IGNORECASE).group(0)
base = f"{random.randint(1,9)}{_random_digits(9)}"
return prefix + base + _checksum_inn12(base)
def generate_ogrn(original: str) -> str:
prefix = re.match(r'ОГРН\s*', original, re.IGNORECASE).group(0)
base = "1" + _random_digits(11)
return prefix + base + _checksum_ogrn(base)
def generate_kpp(original: str) -> str:
prefix = re.match(r'КПП\s*', original, re.IGNORECASE).group(0)
return prefix + _random_digits(4) + random.choice(["01", "43", "77"]) + _random_digits(3)
def generate_bik(original: str) -> str:
prefix = re.match(r'БИК\s*', original, re.IGNORECASE).group(0)
return prefix + "04" + _random_digits(7)
def generate_rs(original: str) -> str:
prefix = re.match(r'(?:р/с|расч[её]тный\s*сч[её]т)\s*', original, re.IGNORECASE).group(0)
return prefix + "40702" + _random_digits(15)
def generate_ks(original: str) -> str:
prefix = re.match(r'(?:к/с|кор/сч|корр?[еи]?спондентский\s*сч[её]т)\s*', original, re.IGNORECASE).group(0)
return prefix + "30101" + _random_digits(15)
def generate_passport(original: str) -> str:
m = re.match(r'(?:паспорт|серия\s+номер)[\s:№]*', original, re.IGNORECASE)
prefix = m.group(0) if m else ""
return prefix + f"{random.randint(10,99)} {random.randint(10,99)} {_random_digits(6)}"
def generate_company(_: str) -> str:
forms = ["ООО", "ЗАО", "АО"]
nouns = ["Технология", "Прогресс", "Гарант", "Стандарт", "Импульс",
"Вектор", "Сфера", "Альянс", "Синтез", "Меридиан", "Спектр", "Формат"]
return f'{random.choice(forms)} «{random.choice(nouns)}»'
def generate_person(_: str) -> str:
s = random.choice(RU_SURNAMES)
n = random.choice(RU_NAMES)
p = random.choice(RU_PATRONYMICS)
return f"{s} {n[0]}.{p[0]}."
def generate_address(_: str) -> str:
city = random.choice(RU_CITIES)
street = random.choice(RU_STREETS)
house = random.randint(1, 200)
return f"{city}, ул. {street}, д. {house}"
# ═══════════════════════════════════════════
# Основной класс
# ═══════════════════════════════════════════
class TwoPassObfuscator:
"""Двухпроходный обфускатор: сбор сущностей → замена."""
def __init__(self, llm_client=None):
self._mapping: Dict[str, str] = {} # оригинал → замена (только для точных текстовых совпадений)
self._regex_replacements: List[Tuple[str, str, Callable]] = [] # (pattern, type, generator)
self._sorted_keys: List[str] = [] # ключи mapping, отсортированные по длине (убывание)
self._llm_client = llm_client
def obfuscate(self, files: List[Tuple[str, bytes, str]]) -> Tuple[bytes, str]:
"""
Главная точка входа.
Args:
files: [(original_filename, content_bytes, content_type), ...]
Returns:
(zip_bytes, mapping_csv_string)
"""
# --- Распаковать ZIP-файлы ---
files = self._expand_zips(files)
# --- Конвертировать PDF → DOCX ---
files = self._convert_pdfs_to_docx(files)
try:
# --- Проход 1: сбор сущностей ---
all_texts: Dict[str, str] = {}
all_docx: Dict[str, object] = {}
for fname, content, ctype in files:
text, doc = self._extract_text(fname, content, ctype)
all_texts[fname] = text
if doc is not None:
all_docx[fname] = doc
if text and text != "[DOC binary — not parsed]":
self._scan_regex(text)
if self._llm_client:
self._scan_llm_ner(all_texts)
# Предсортировать ключи один раз для _apply_replacements
self._sorted_keys = sorted(self._mapping.keys(), key=len, reverse=True)
# --- Проход 2: замена ---
results = []
for fname, content, ctype in files:
obf_content = content
if fname.endswith('.doc'):
# .doc — бинарный, оставляем как есть
pass
elif fname in all_docx:
obf_content = self._replace_in_docx(all_docx[fname])
else:
txt = all_texts.get(fname, '')
obf_content = self._replace_in_text(txt, fname)
results.append((fname, obf_content))
csv_str = self._build_mapping_csv()
return self._build_zip(results, csv_str), csv_str
finally:
self._mapping.clear()
self._regex_replacements.clear()
self._sorted_keys.clear()
def _convert_pdfs_to_docx(self, files: List[Tuple[str, bytes, str]]) -> List[Tuple[str, bytes, str]]:
"""Конвертировать PDF в DOCX через pdfplumber. При совпадении имён — _из_pdf."""
import pdfplumber
from docx import Document as DocxDocument
result = []
existing_names = {f[0] for f in files}
for fname, content, ctype in files:
if not fname.lower().endswith('.pdf'):
result.append((fname, content, ctype))
continue
try:
doc = DocxDocument()
with pdfplumber.open(io.BytesIO(content)) as pdf:
for page in pdf.pages:
tables = page.extract_tables()
for table in tables:
if table:
rows = [[str(c or "").strip() for c in (row or [])] for row in table]
rows = [r for r in rows if any(r)]
if rows:
t = doc.add_table(rows=len(rows), cols=len(rows[0]))
t.style = 'Table Grid'
for ri, row in enumerate(rows):
for ci, cell_text in enumerate(row):
t.rows[ri].cells[ci].text = cell_text
text = page.extract_text()
if text:
for line in text.split('\n'):
line = line.strip()
if line:
doc.add_paragraph(line)
buf = io.BytesIO()
doc.save(buf)
new_name = fname[:-4] + '.docx'
if new_name in existing_names:
new_name = fname[:-4] + '_из_pdf.docx'
existing_names.add(new_name)
result.append((new_name, buf.getvalue(), ctype))
except Exception as e:
log.warning("PDF→DOCX error for %s: %s", fname, e)
result.append((fname, content, ctype))
return result
def _expand_zips(self, files: List[Tuple[str, bytes, str]]) -> List[Tuple[str, bytes, str]]:
"""Распаковать ZIP-файлы, заменив их содержимым. Остальные файлы — как есть.
Защита от ZIP-бомб: ratio + накопительный размер (как в compare/unzip.py)."""
result = []
for fname, content, ctype in files:
if fname.lower().endswith('.zip'):
try:
with zipfile.ZipFile(io.BytesIO(content)) as zf:
if len(zf.infolist()) > 500:
log.warning("ZIP too many files, skipping: %s", fname)
result.append((fname, content, ctype))
continue
total_uncompressed = 0
for info in zf.infolist():
if info.is_dir():
continue
# ZIP bomb: ratio check
if info.compress_size > 0:
ratio = info.file_size / info.compress_size
if ratio > 100:
log.warning("ZIP bomb ratio %.0f:1, skipping: %s", ratio, fname)
result.append((fname, content, ctype))
break
# cp437 → utf8 (как в compare/unzip.py)
name = info.filename
try:
name = name.encode("cp437").decode("utf-8", errors="replace")
except (UnicodeDecodeError, UnicodeEncodeError):
pass
# Убрать path traversal
name = os.path.basename(name)
if not name or name.endswith("/") or ".." in name or "/" in name or "\\" in name:
continue
inner_data = zf.read(info)
total_uncompressed += len(inner_data)
if total_uncompressed > 500 * 1024 * 1024: # 500 MB
log.warning("ZIP uncompressed limit exceeded, stopping: %s", fname)
break
result.append((name, inner_data, ""))
except Exception as e:
log.warning("Failed to expand ZIP %s: %s", fname, e)
result.append((fname, content, ctype))
else:
result.append((fname, content, ctype))
return result
# --- Проход 1: обнаружение ---
def _extract_text(self, fname: str, content: bytes, ctype: str) -> Tuple[str, Optional[object]]:
"""Извлечь текст из файла. Возвращает (text, docx_document_or_None)."""
doc = None
text = ""
ext = os.path.splitext(fname)[1].lower()
if ext == '.docx':
try:
from docx import Document
except ImportError:
text = content.decode('utf-8', errors='replace')
return text, None
doc = Document(io.BytesIO(content))
text = "\n".join(p.text for p in doc.paragraphs)
for table in doc.tables:
for row in table.rows:
text += "\n" + " | ".join(cell.text for cell in row.cells)
elif ext == '.pdf':
try:
import pdfplumber
except ImportError:
text = content.decode('utf-8', errors='replace')
return text, None
with pdfplumber.open(io.BytesIO(content)) as pdf:
for page in pdf.pages:
t = page.extract_text()
if t:
text += t + "\n"
for table in page.extract_tables():
for row in table:
text += "\n" + " | ".join(str(c) if c else "" for c in row)
elif ext == '.doc':
# .doc — бинарный формат, без libreoffice не парсим
# Пропускаем без изменений (не обфусцируем)
text = "[DOC binary — not parsed]"
return text, None
else:
text = content.decode('utf-8', errors='replace')
return text, doc
def _scan_regex(self, text: str):
"""Сканировать текст regex-паттернами, заполнить словарь замен."""
for entity_type, pattern in ENTITY_PATTERNS.items():
for match in re.finditer(pattern, text, re.IGNORECASE | re.MULTILINE):
original = match.group(0).strip()
if original and original not in self._mapping:
generator = ENTITY_GENERATORS.get(entity_type, lambda x: "XXX")
self._mapping[original] = generator(original)
# Компании (кроме НУБЕС — это Исполнитель)
for match in COMPANY_PATTERN.finditer(text):
original = match.group(0).strip()
if original and original not in self._mapping:
if re.search(r'НУБЕС|NUBES', original, re.IGNORECASE):
continue # Исполнитель — не заменяем
self._mapping[original] = generate_company(original)
# ФИО (Фамилия И.О.)
for match in PERSON_PATTERN.finditer(text):
original = match.group(0).strip()
if original and original not in self._mapping:
self._mapping[original] = generate_person(original)
def _scan_llm_ner(self, all_texts: Dict[str, str]):
"""LLM NER для обнаружения имён и адресов во всех файлах."""
combined = "\n\n---FILE---\n\n".join(
f"FILE: {fname}\n{t[:3000]}" for fname, t in all_texts.items()
)
prompt = (
"Ты — система обнаружения персональных данных в документах. "
"Найди ВСЕ следующие сущности в тексте ниже:\n\n"
"1. ФИО (полные и сокращённые — 'Иванов И.И.', 'Петров А.С.')\n"
"2. Названия компаний-контрагентов (не 'НУБЕС')\n"
"3. Почтовые адреса\n"
"4. Паспортные данные\n\n"
"Формат ответа — JSON-массив:\n"
'[{"type": "person"|"company"|"address"|"passport", "value": "найденный текст"}]\n\n'
f"Текст:\n{combined[:8000]}"
)
try:
raw = self._llm_client.complete(prompt)
import json
# Игнорируем markdown-обёртку
raw = raw.strip()
if raw.startswith("```"):
raw = raw.split("\n", 1)[1]
if raw.endswith("```"):
raw = raw[:-3]
entities = json.loads(raw)
for ent in entities:
val = ent.get("value", "").strip()
if val and val not in self._mapping:
if ent.get("type") == "person":
self._mapping[val] = generate_person(val)
elif ent.get("type") == "company":
self._mapping[val] = generate_company(val)
elif ent.get("type") == "address":
self._mapping[val] = generate_address(val)
elif ent.get("type") == "passport":
self._mapping[val] = generate_passport(val)
except Exception as e:
log.warning("LLM NER failed: %s", e)
# --- Проход 2: замена ---
def _replace_in_docx(self, doc) -> bytes:
"""Заменить сущности в docx. Склеиваем runs → заменяем → пишем в первый run, очищаем остальные."""
for para in doc.paragraphs:
if not para.runs:
continue
full_text = "".join(run.text for run in para.runs)
replaced = self._apply_replacements(full_text)
if replaced != full_text:
para.runs[0].text = replaced
for run in para.runs[1:]:
run.text = ""
for table in doc.tables:
for row in table.rows:
for cell in row.cells:
for para in cell.paragraphs:
if not para.runs:
continue
full_text = "".join(run.text for run in para.runs)
replaced = self._apply_replacements(full_text)
if replaced != full_text:
para.runs[0].text = replaced
for run in para.runs[1:]:
run.text = ""
buf = io.BytesIO()
doc.save(buf)
return buf.getvalue()
def _replace_in_text(self, text: str, fname: str) -> bytes:
"""Заменить сущности в plain text (для PDF и прочих)."""
replaced = self._apply_replacements(text)
# Для PDF пока отдаём текст (MVP — без сохранения форматирования PDF)
return replaced.encode('utf-8')
def _apply_replacements(self, text: str) -> str:
"""Применить все замены из словаря mapping к строке. Сначала длинные, потом короткие."""
result = text
for original in self._sorted_keys:
replacement = self._mapping[original]
# Границы слова — если сущность начинается и заканчивается на \w
if original and original[0].isalnum() and original[-1].isalnum():
pattern = r'(?<!\w)' + re.escape(original) + r'(?!\w)'
else:
pattern = re.escape(original)
result = re.sub(pattern, replacement, result)
return result
# --- Сборка выдачи ---
def _build_zip(self, files: List[Tuple[str, bytes]], mapping_csv: str = "") -> bytes:
"""Собрать ZIP с обфусцированными файлами + mapping.csv."""
buf = io.BytesIO()
with zipfile.ZipFile(buf, 'w', zipfile.ZIP_DEFLATED) as zf:
for fname, content in files:
info = zipfile.ZipInfo(fname)
info.flag_bits |= 0x800 # UTF-8 filename
zf.writestr(info, content)
if mapping_csv:
zf.writestr("mapping.csv", '\ufeff'.encode('utf-8') + mapping_csv.encode('utf-8'))
return buf.getvalue()
def _build_mapping_csv(self) -> str:
"""Собрать mapping.csv."""
buf = io.StringIO()
writer = csv.writer(buf)
writer.writerow(["тип_данных", "оригинал", "замена"])
for original, replacement in sorted(self._mapping.items()):
etype = "text"
# inn_fl ДО inn_ul (12 цифр vs 10)
for t in ["phone", "email", "inn_fl", "inn_ul", "ogrn", "kpp", "bik", "rs", "ks", "passport"]:
pat = ENTITY_PATTERNS.get(t, "")
if pat and re.match(pat, original, re.IGNORECASE):
etype = t
break
if COMPANY_PATTERN.match(original):
etype = "company"
writer.writerow([etype, original, replacement])
return buf.getvalue()
# ═══════════════════════════════════════════
# Маппинг entity_type → генератор
# ═══════════════════════════════════════════
ENTITY_GENERATORS: Dict[str, Callable] = {
"phone": generate_phone,
"email": generate_email,
"inn_ul": generate_inn10,
"inn_fl": generate_inn12,
"ogrn": generate_ogrn,
"kpp": generate_kpp,
"bik": generate_bik,
"rs": generate_rs,
"ks": generate_ks,
"passport": generate_passport,
}
def obfuscate_files(files: List[Tuple[str, bytes, str]], llm_client=None) -> Tuple[bytes, str]:
"""Удобная функция: обфусцировать список файлов → (zip_bytes, csv_string)."""
obf = TwoPassObfuscator(llm_client=llm_client)
return obf.obfuscate(files)
+37 -74
View File
@@ -1,87 +1,50 @@
"""
DrHider — Managed Flask приложение на платформе Штурвал.
Приложение создаётся фабрикой create_app().
Штурвал запускает Flask самостоятельно (без Dockerfile/gunicorn).
Роуты разнесены по blueprint'ам:
- main_bp: GET / — веб-интерфейс
- health_bp: GET /health — проверка живости
- api_bp: POST /api/drhider — обфускация документов
"""
import os
import io
import sys
import json
import zipfile
from flask import Flask, render_template, request, send_file, jsonify
from flask import Flask
# Добавляем корень в sys.path чтобы import drhider работал
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
# Добавляем корень проекта в sys.path для импорта пакета drhider
_sys_path_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
if _sys_path_root not in sys.path:
sys.path.insert(0, _sys_path_root)
from drhider import obfuscate_files
app = Flask(__name__)
VERSION = "1.0.0"
MAX_BODY = 200 * 1024 * 1024 # 200 MB
# Версия приложения (меняется при изменениях)
VERSION = "2.0.0"
class LLMClient:
"""LLM-клиент для drhider (обнаружение компаний/ФИО)."""
def create_app():
"""Создать и настроить Flask-приложение.
def __init__(self):
import httpx
self._httpx = httpx
Returns:
Экземпляр Flask с зарегистрированными blueprint'ами.
"""
app = Flask(__name__)
def complete(self, prompt: str) -> str:
key = os.environ.get("LLM_KEY") or os.environ.get("LLM_API_KEY", "")
r = self._httpx.post(
os.environ.get("LLM_URL", "https://api.aillm.ru/v1/chat/completions"),
json={
"model": os.environ.get("LLM_MODEL", "gpt-oss-120b"),
"messages": [{"role": "user", "content": prompt}],
"max_tokens": 8000,
"temperature": 0.1,
},
headers={
"Authorization": f"Bearer {key}",
"Content-Type": "application/json",
},
timeout=120,
)
r.raise_for_status()
return r.json()["choices"][0]["message"]["content"]
# Конфигурация
app.config["VERSION"] = VERSION
app.config["MAX_CONTENT_LENGTH"] = 200 * 1024 * 1024 # 200 MB
# Регистрируем blueprint'ы
from routes import register_routes
register_routes(app)
return app
@app.route("/")
def index():
return render_template("index.html", version=VERSION)
@app.route("/health")
def health():
return jsonify({"ok": True, "version": VERSION})
@app.route("/api/drhider", methods=["POST"])
def api_drhider():
"""Обфускация: multipart/form-data с файлами → ZIP."""
uploaded = request.files.getlist("files")
if not uploaded:
return jsonify({"ok": False, "error": "Нет файлов"}), 400
files = []
for f in uploaded:
if f.filename:
files.append((f.filename, f.read(), f.mimetype or ""))
if not files:
return jsonify({"ok": False, "error": "Нет файлов"}), 400
try:
llm = LLMClient()
zip_data, _csv = obfuscate_files(files, llm_client=llm)
return send_file(
io.BytesIO(zip_data),
mimetype="application/zip",
as_attachment=True,
download_name="drhider_output.zip",
)
except Exception as e:
import traceback
traceback.print_exc()
return jsonify({"ok": False, "error": str(e)}), 500
# Экземпляр приложения — точка входа для Штурвала
app = create_app()
+20
View File
@@ -0,0 +1,20 @@
"""
Регистрация всех blueprint'ов приложения.
Импортируется из site/app.py при создании Flask-приложения.
"""
from .main_bp import main_bp
from .health_bp import health_bp
from .api_bp import api_bp
def register_routes(app):
"""Зарегистрировать все blueprint'ы на Flask-приложении.
Args:
app: Экземпляр Flask-приложения
"""
app.register_blueprint(main_bp)
app.register_blueprint(health_bp)
app.register_blueprint(api_bp)
+73
View File
@@ -0,0 +1,73 @@
"""
Blueprint: API обфускации (POST /api/drhider).
Принимает multipart/form-data с файлами, возвращает ZIP-архив
с обфусцированными документами.
"""
import io
import traceback
from flask import Blueprint, request, send_file, jsonify
from drhider import obfuscate_files, LLMClient
# ═══════════════════════════════════════════════════════════════════════════
# Blueprint: API DrHider
# ═══════════════════════════════════════════════════════════════════════════
api_bp = Blueprint("api", __name__, url_prefix="/api")
# Максимальный размер тела запроса: 200 MB
MAX_BODY = 200 * 1024 * 1024
@api_bp.route("/drhider", methods=["POST"])
def drhider():
"""Обфускация документов.
Принимает multipart/form-data с полем 'files' (один или несколько файлов).
Возвращает ZIP-архив с обфусцированными документами + mapping.csv.
Поддерживаемые форматы:
.docx, .pdf, .txt, .doc (бинарный — без изменений)
.zip (распаковывается, содержимое обфусцируется)
Returns:
200: ZIP-архив (application/zip)
400: {"ok": false, "error": "..."}
500: {"ok": false, "error": "..."}
"""
# ── Получаем файлы из запроса ──
uploaded = request.files.getlist("files")
if not uploaded:
return jsonify({"ok": False, "error": "Нет файлов"}), 400
# Формируем список для obfuscate_files: [(name, bytes, mimetype), ...]
files = []
for f in uploaded:
if f.filename:
files.append((f.filename, f.read(), f.mimetype or ""))
if not files:
return jsonify({"ok": False, "error": "Нет файлов"}), 400
# ── Обфускация ──
try:
# Создаём LLM-клиент для NER-сканирования
llm = LLMClient()
# Вызываем ядро обфускации
zip_data, _csv = obfuscate_files(files, llm_client=llm)
# Отдаём ZIP-архив
return send_file(
io.BytesIO(zip_data),
mimetype="application/zip",
as_attachment=True,
download_name="drhider_output.zip",
)
except Exception as e:
# Логируем полный трейсбек на сервере
traceback.print_exc()
return jsonify({"ok": False, "error": str(e)}), 500
+27
View File
@@ -0,0 +1,27 @@
"""
Blueprint: health-check (GET /health).
Используется платформой Штурвал для проверки живости приложения.
"""
from flask import Blueprint, jsonify, current_app
# ═══════════════════════════════════════════════════════════════════════════
# Blueprint: health check
# ═══════════════════════════════════════════════════════════════════════════
health_bp = Blueprint("health", __name__)
@health_bp.route("/health")
def health():
"""Эндпоинт проверки живости.
Возвращает JSON с версией приложения.
Используется платформой для readiness/liveness probes.
Returns:
{"ok": true, "version": "X.Y.Z"}
"""
version = current_app.config.get("VERSION", "0.0.0")
return jsonify({"ok": True, "version": version})
+23
View File
@@ -0,0 +1,23 @@
"""
Blueprint: главная страница (GET /).
Отдаёт HTML-интерфейс DrHider.
"""
from flask import Blueprint, render_template, current_app
# ═══════════════════════════════════════════════════════════════════════════
# Blueprint: главная страница
# ═══════════════════════════════════════════════════════════════════════════
main_bp = Blueprint("main", __name__)
@main_bp.route("/")
def index():
"""Главная страница — веб-интерфейс DrHider.
Передаёт в шаблон текущую версию приложения.
"""
version = current_app.config.get("VERSION", "0.0.0")
return render_template("index.html", version=version)