558 lines
26 KiB
Python
558 lines
26 KiB
Python
"""
|
||
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)
|