fix: Opus review — all 5 critical bugs fixed
Deploy contracts-flask / validate (push) Successful in 0s

This commit is contained in:
2026-06-29 15:25:27 +04:00
parent 12416115ad
commit d8713561a4
3 changed files with 105 additions and 82 deletions
+42 -41
View File
@@ -15,9 +15,7 @@ import random
import string
import zipfile
import logging
import tempfile
import os
import shutil
from typing import Dict, List, Tuple, Callable, Optional
log = logging.getLogger("drhider")
@@ -29,14 +27,15 @@ log = logging.getLogger("drhider")
ENTITY_PATTERNS: Dict[str, str] = {
"phone": r'(?:\+7|8)[\s\-]?\(?\d{3}\)?[\s\-]?\d{3}[\s\-]?\d{2}[\s\-]?\d{2}',
"email": r'[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}',
"inn_ul": r'ИНН\s*\d{10}',
# 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'\d{2}\s*\d{2}\s*\d{6,7}',
"passport": r'(?:паспорт|серия\s+номер)[\s:№]*\d{2}\s*\d{2}\s*\d{6,7}',
}
# Фирмы — обнаружение по шаблону
@@ -110,39 +109,47 @@ def generate_email(original: str) -> str:
return f"{local}@{domain}"
def generate_inn10(_: str) -> str:
def generate_inn10(original: str) -> str:
prefix = re.match(r'ИНН\s*', original).group(0)
base = f"{random.randint(1,9)}{_random_digits(8)}"
return base + _checksum_inn10(base)
return prefix + base + _checksum_inn10(base)
def generate_inn12(_: str) -> str:
def generate_inn12(original: str) -> str:
prefix = re.match(r'ИНН\s*', original).group(0)
base = f"{random.randint(1,9)}{_random_digits(9)}"
return base + _checksum_inn12(base)
return prefix + base + _checksum_inn12(base)
def generate_ogrn(_: str) -> str:
def generate_ogrn(original: str) -> str:
prefix = re.match(r'ОГРН\s*', original).group(0)
base = "1" + _random_digits(11)
return base + _checksum_ogrn(base)
return prefix + base + _checksum_ogrn(base)
def generate_kpp(_: str) -> str:
return _random_digits(4) + random.choice(["01", "43", "77"]) + _random_digits(3)
def generate_kpp(original: str) -> str:
prefix = re.match(r'КПП\s*', original).group(0)
return prefix + _random_digits(4) + random.choice(["01", "43", "77"]) + _random_digits(3)
def generate_bik(_: str) -> str:
return "04" + _random_digits(7)
def generate_bik(original: str) -> str:
prefix = re.match(r'БИК\s*', original).group(0)
return prefix + "04" + _random_digits(7)
def generate_rs(_: str) -> str:
return "40702" + _random_digits(15)
def generate_rs(original: str) -> str:
prefix = re.match(r'(?:р/с|расч[её]тный\s*сч[её]т)\s*', original).group(0)
return prefix + "40702" + _random_digits(15)
def generate_ks(_: str) -> str:
return "30101" + _random_digits(15)
def generate_ks(original: str) -> str:
prefix = re.match(r'(?:к/с|корр?[еи]?спондентский\s*сч[её]т)\s*', original).group(0)
return prefix + "30101" + _random_digits(15)
def generate_passport(_: str) -> str:
return f"{random.randint(10,99)} {random.randint(10,99)} {_random_digits(6)}"
def generate_passport(original: str) -> str:
prefix = re.match(r'(?:паспорт|серия\s+номер)[\s:№]*', original, re.IGNORECASE).group(0)
return prefix + f"{random.randint(10,99)} {random.randint(10,99)} {_random_digits(6)}"
def generate_company(_: str) -> str:
@@ -177,7 +184,6 @@ class TwoPassObfuscator:
self._mapping: Dict[str, str] = {} # оригинал → замена (только для точных текстовых совпадений)
self._regex_replacements: List[Tuple[str, str, Callable]] = [] # (pattern, type, generator)
self._llm_client = llm_client
self._workdir: Optional[str] = None
def obfuscate(self, files: List[Tuple[str, bytes, str]]) -> Tuple[bytes, str]:
"""
@@ -189,14 +195,13 @@ class TwoPassObfuscator:
Returns:
(zip_bytes, mapping_csv_string)
"""
self._workdir = tempfile.mkdtemp(prefix="drhider_")
try:
# --- Распаковать ZIP-файлы ---
files = self._expand_zips(files)
# --- Распаковать ZIP-файлы ---
files = self._expand_zips(files)
try:
# --- Проход 1: сбор сущностей ---
all_texts: Dict[str, str] = {} # filename → text for LLM NER
all_docx: Dict[str, object] = {} # filename → docx Document for replacement
all_texts: Dict[str, str] = {}
all_docx: Dict[str, object] = {}
for fname, content, ctype in files:
text, doc = self._extract_text(fname, content, ctype)
@@ -205,7 +210,6 @@ class TwoPassObfuscator:
all_docx[fname] = doc
self._scan_regex(text)
# LLM NER для имён/адресов (если есть клиент)
if self._llm_client:
self._scan_llm_ner(all_texts)
@@ -214,24 +218,20 @@ class TwoPassObfuscator:
for fname, content, ctype in files:
obf_content = content
if fname in all_docx:
# Замена внутри docx
obf_content = self._replace_in_docx(all_docx[fname])
elif fname.endswith('.pdf'):
# PDF: read-only, создаём новый
obf_content = self._replace_in_text(all_texts.get(fname, ''), fname)
else:
# .doc или другой — замена по тексту
txt = all_texts.get(fname, '')
obf_content = self._replace_in_text(txt, fname)
fname = fname[:-4] + '.txt'
else:
txt = all_texts.get(fname, '')
obf_content = self._replace_in_text(txt, fname)
results.append((fname, obf_content))
# Собираем ZIP + CSV
csv_str = self._build_mapping_csv()
return self._build_zip(results, csv_str), csv_str
finally:
shutil.rmtree(self._workdir, ignore_errors=True)
self._mapping.clear()
self._regex_replacements.clear()
@@ -302,8 +302,8 @@ class TwoPassObfuscator:
text += "\n" + " | ".join(str(c) if c else "" for c in row)
elif ext == '.doc':
# .doc конвертируется в .docx ДО вызова — здесь уже docx
pass
# .doc: читаем как текст (конвертация через libreoffice не используется)
text = content.decode('utf-8', errors='replace')
else:
text = content.decode('utf-8', errors='replace')
@@ -421,10 +421,11 @@ class TwoPassObfuscator:
writer = csv.writer(buf)
writer.writerow(["тип_данных", "оригинал", "замена"])
for original, replacement in sorted(self._mapping.items()):
# Определяем тип по паттерну
etype = "text"
for t, pat in ENTITY_PATTERNS.items():
if re.match(pat, original, re.IGNORECASE):
# 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):