Игорь Надеин
5b6ab02e53
- Add main bot file with Telegram handlers - Add configuration module - Add Python dependencies - Add modules: excel_generator, moysklad_client, xml_generator, user_manager - Add .gitignore for Python, Docker, user data
514 lines
18 KiB
Python
514 lines
18 KiB
Python
"""
|
||
Валидация данных заказов МойСклад
|
||
Проверка обязательных и желательных полей, валидация форматов
|
||
"""
|
||
import re
|
||
from typing import Dict, List, Optional, Tuple
|
||
from enum import Enum
|
||
from dataclasses import dataclass
|
||
import logging
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class ValidationLevel(Enum):
|
||
"""Уровни валидации"""
|
||
VALID = "valid" # Все проверки пройдены
|
||
WARNING = "warning" # Отсутствуют желательные поля
|
||
CRITICAL = "critical" # Отсутствуют обязательные поля
|
||
|
||
|
||
@dataclass
|
||
class ValidationResult:
|
||
"""Результат валидации"""
|
||
level: ValidationLevel
|
||
field: str
|
||
message: str
|
||
value: Optional[str] = None
|
||
|
||
|
||
class OrderValidator:
|
||
"""Валидатор данных заказов"""
|
||
|
||
# НДС ставки
|
||
VAT_RATES = [0, 10, 20, "без НДС", "0%", "10%", "20%"]
|
||
|
||
def __init__(self):
|
||
self.results: List[ValidationResult] = []
|
||
|
||
def validate_inn(self, inn: str) -> bool:
|
||
"""
|
||
Валидация ИНН
|
||
|
||
ИНН юр. лица: 10 цифр
|
||
ИНН ИП: 12 цифр
|
||
Включает проверку контрольной суммы
|
||
|
||
Args:
|
||
inn: ИНН для проверки
|
||
|
||
Returns:
|
||
True если ИНН корректен
|
||
"""
|
||
if not inn or not isinstance(inn, str):
|
||
return False
|
||
|
||
inn = inn.strip()
|
||
|
||
if not inn.isdigit():
|
||
return False
|
||
|
||
if len(inn) == 10:
|
||
# ИНН юридического лица (10 цифр)
|
||
coefficients = [2, 4, 10, 3, 5, 9, 4, 6, 8]
|
||
checksum = sum(
|
||
int(inn[i]) * coefficients[i] for i in range(9)
|
||
) % 11 % 10
|
||
return checksum == int(inn[9])
|
||
|
||
elif len(inn) == 12:
|
||
# ИНН индивидуального предпринимателя (12 цифр)
|
||
# Проверка первой контрольной суммы
|
||
coefficients1 = [7, 2, 4, 10, 3, 5, 9, 4, 6, 8]
|
||
checksum1 = sum(
|
||
int(inn[i]) * coefficients1[i] for i in range(10)
|
||
) % 11 % 10
|
||
|
||
# Проверка второй контрольной суммы
|
||
coefficients2 = [3, 7, 2, 4, 10, 3, 5, 9, 4, 6, 8]
|
||
checksum2 = sum(
|
||
int(inn[i]) * coefficients2[i] for i in range(11)
|
||
) % 11 % 10
|
||
|
||
return checksum1 == int(inn[10]) and checksum2 == int(inn[11])
|
||
|
||
return False
|
||
|
||
def validate_kpp(self, kpp: str) -> bool:
|
||
"""
|
||
Валидация КПП
|
||
|
||
КПП: 9 цифр
|
||
|
||
Args:
|
||
kpp: КПП для проверки
|
||
|
||
Returns:
|
||
True если КПП корректен
|
||
"""
|
||
if not kpp or not isinstance(kpp, str):
|
||
return False
|
||
|
||
kpp = kpp.strip()
|
||
return len(kpp) == 9 and kpp.isdigit()
|
||
|
||
def validate_ogrn(self, ogrn: str) -> bool:
|
||
"""
|
||
Валидация ОГРН
|
||
|
||
ОГРН юр. лица: 13 цифр
|
||
ОГРНИП: 15 цифр
|
||
|
||
Args:
|
||
ogrn: ОГРН для проверки
|
||
|
||
Returns:
|
||
True если ОГРН корректен
|
||
"""
|
||
if not ogrn or not isinstance(ogrn, str):
|
||
return False
|
||
|
||
ogrn = ogrn.strip()
|
||
|
||
if not ogrn.isdigit():
|
||
return False
|
||
|
||
if len(ogrn) == 13:
|
||
# ОГРН юридического лица
|
||
check = int(ogrn[:12]) % 11
|
||
return (check % 10) == int(ogrn[12])
|
||
|
||
elif len(ogrn) == 15:
|
||
# ОГРНИП индивидуального предпринимателя
|
||
check = int(ogrn[:14]) % 13
|
||
return (check % 10) == int(ogrn[14])
|
||
|
||
return False
|
||
|
||
def validate_organization(self, org: Dict) -> List[ValidationResult]:
|
||
"""
|
||
Валидация данных организации (продавец)
|
||
|
||
Args:
|
||
org: Данные организации из МойСклад
|
||
|
||
Returns:
|
||
Список результатов валидации
|
||
"""
|
||
results = []
|
||
|
||
if not org:
|
||
results.append(ValidationResult(
|
||
level=ValidationLevel.CRITICAL,
|
||
field="organization",
|
||
message="Отсутствуют данные организации"
|
||
))
|
||
return results
|
||
|
||
# Обязательные поля
|
||
name = org.get("name") or org.get("legalTitle")
|
||
if not name:
|
||
results.append(ValidationResult(
|
||
level=ValidationLevel.CRITICAL,
|
||
field="organization.name",
|
||
message="Не указано наименование организации"
|
||
))
|
||
|
||
# ИНН
|
||
inn = org.get("inn")
|
||
if not inn:
|
||
results.append(ValidationResult(
|
||
level=ValidationLevel.CRITICAL,
|
||
field="organization.inn",
|
||
message="Не указан ИНН организации"
|
||
))
|
||
elif not self.validate_inn(inn):
|
||
results.append(ValidationResult(
|
||
level=ValidationLevel.CRITICAL,
|
||
field="organization.inn",
|
||
message=f"Некорректный ИНН: {inn}",
|
||
value=inn
|
||
))
|
||
|
||
# КПП (обязателен для юр. лиц)
|
||
kpp = org.get("kpp")
|
||
if inn and len(inn) == 10: # Юр. лицо
|
||
if not kpp:
|
||
results.append(ValidationResult(
|
||
level=ValidationLevel.CRITICAL,
|
||
field="organization.kpp",
|
||
message="Не указан КПП организации"
|
||
))
|
||
elif not self.validate_kpp(kpp):
|
||
results.append(ValidationResult(
|
||
level=ValidationLevel.CRITICAL,
|
||
field="organization.kpp",
|
||
message=f"Некорректный КПП: {kpp}",
|
||
value=kpp
|
||
))
|
||
|
||
# Юридический адрес
|
||
legal_address = (
|
||
org.get("legalAddress") or
|
||
org.get("actualAddress")
|
||
)
|
||
if not legal_address:
|
||
results.append(ValidationResult(
|
||
level=ValidationLevel.CRITICAL,
|
||
field="organization.legalAddress",
|
||
message="Не указан юридический адрес организации"
|
||
))
|
||
|
||
# Желательные поля
|
||
|
||
# ОГРН
|
||
ogrn = org.get("ogrn")
|
||
if not ogrn:
|
||
results.append(ValidationResult(
|
||
level=ValidationLevel.WARNING,
|
||
field="organization.ogrn",
|
||
message="Не указан ОГРН организации"
|
||
))
|
||
elif not self.validate_ogrn(ogrn):
|
||
results.append(ValidationResult(
|
||
level=ValidationLevel.WARNING,
|
||
field="organization.ogrn",
|
||
message=f"Некорректный ОГРН: {ogrn}",
|
||
value=ogrn
|
||
))
|
||
|
||
# Счета
|
||
accounts = org.get("accounts", [])
|
||
if not accounts:
|
||
results.append(ValidationResult(
|
||
level=ValidationLevel.WARNING,
|
||
field="organization.accounts",
|
||
message="Не указаны банковские реквизиты"
|
||
))
|
||
|
||
return results
|
||
|
||
def validate_counterparty(self, agent: Dict) -> List[ValidationResult]:
|
||
"""
|
||
Валидация данных контрагента (покупатель)
|
||
|
||
Args:
|
||
agent: Данные контрагента из МойСклад
|
||
|
||
Returns:
|
||
Список результатов валидации
|
||
"""
|
||
results = []
|
||
|
||
if not agent:
|
||
results.append(ValidationResult(
|
||
level=ValidationLevel.CRITICAL,
|
||
field="counterparty",
|
||
message="Отсутствуют данные контрагента"
|
||
))
|
||
return results
|
||
|
||
# Обязательные поля
|
||
name = agent.get("name") or agent.get("legalTitle")
|
||
if not name:
|
||
results.append(ValidationResult(
|
||
level=ValidationLevel.CRITICAL,
|
||
field="counterparty.name",
|
||
message="Не указано наименование контрагента"
|
||
))
|
||
|
||
# ИНН
|
||
inn = agent.get("inn")
|
||
if not inn:
|
||
results.append(ValidationResult(
|
||
level=ValidationLevel.CRITICAL,
|
||
field="counterparty.inn",
|
||
message="Не указан ИНН контрагента"
|
||
))
|
||
elif not self.validate_inn(inn):
|
||
results.append(ValidationResult(
|
||
level=ValidationLevel.CRITICAL,
|
||
field="counterparty.inn",
|
||
message=f"Некорректный ИНН: {inn}",
|
||
value=inn
|
||
))
|
||
|
||
# КПП (обязателен для юр. лиц)
|
||
kpp = agent.get("kpp")
|
||
if inn and len(inn) == 10: # Юр. лицо
|
||
if not kpp:
|
||
results.append(ValidationResult(
|
||
level=ValidationLevel.CRITICAL,
|
||
field="counterparty.kpp",
|
||
message="Не указан КПП контрагента"
|
||
))
|
||
elif not self.validate_kpp(kpp):
|
||
results.append(ValidationResult(
|
||
level=ValidationLevel.CRITICAL,
|
||
field="counterparty.kpp",
|
||
message=f"Некорректный КПП: {kpp}",
|
||
value=kpp
|
||
))
|
||
|
||
# Адрес
|
||
address = agent.get("legalAddress") or agent.get("actualAddress")
|
||
if not address:
|
||
results.append(ValidationResult(
|
||
level=ValidationLevel.CRITICAL,
|
||
field="counterparty.address",
|
||
message="Не указан адрес контрагента"
|
||
))
|
||
|
||
return results
|
||
|
||
def validate_position(
|
||
self,
|
||
position: Dict,
|
||
index: int
|
||
) -> List[ValidationResult]:
|
||
"""
|
||
Валидация позиции заказа (товар)
|
||
|
||
Args:
|
||
position: Данные позиции
|
||
index: Номер позиции
|
||
|
||
Returns:
|
||
Список результатов валидации
|
||
"""
|
||
results = []
|
||
|
||
# Наименование товара
|
||
assortment = position.get("assortment", {})
|
||
name = assortment.get("name")
|
||
if not name:
|
||
results.append(ValidationResult(
|
||
level=ValidationLevel.CRITICAL,
|
||
field=f"position[{index}].name",
|
||
message=f"Позиция {index + 1}: не указано наименование"
|
||
))
|
||
|
||
# Количество
|
||
quantity = position.get("quantity")
|
||
if not quantity or quantity <= 0:
|
||
results.append(ValidationResult(
|
||
level=ValidationLevel.CRITICAL,
|
||
field=f"position[{index}].quantity",
|
||
message=f"Позиция {index + 1}: некорректное количество"
|
||
))
|
||
|
||
# Цена
|
||
price = position.get("price")
|
||
if price is None or price < 0:
|
||
results.append(ValidationResult(
|
||
level=ValidationLevel.CRITICAL,
|
||
field=f"position[{index}].price",
|
||
message=f"Позиция {index + 1}: некорректная цена"
|
||
))
|
||
|
||
# НДС
|
||
vat = position.get("vat")
|
||
if vat is None:
|
||
results.append(ValidationResult(
|
||
level=ValidationLevel.WARNING,
|
||
field=f"position[{index}].vat",
|
||
message=f"Позиция {index + 1}: не указана ставка НДС"
|
||
))
|
||
|
||
# Единица измерения (желательно)
|
||
uom = assortment.get("uom")
|
||
if not uom:
|
||
results.append(ValidationResult(
|
||
level=ValidationLevel.WARNING,
|
||
field=f"position[{index}].uom",
|
||
message=f"Позиция {index + 1}: не указана единица измерения"
|
||
))
|
||
|
||
return results
|
||
|
||
def validate_order(self, order_data: Dict) -> List[ValidationResult]:
|
||
"""
|
||
Полная валидация заказа
|
||
|
||
Args:
|
||
order_data: Полные данные заказа (из get_full_order_data)
|
||
|
||
Returns:
|
||
Список всех результатов валидации
|
||
"""
|
||
results = []
|
||
|
||
order = order_data.get("order", {})
|
||
|
||
# Основные данные заказа
|
||
if not order.get("name"):
|
||
results.append(ValidationResult(
|
||
level=ValidationLevel.CRITICAL,
|
||
field="order.name",
|
||
message="Не указан номер заказа"
|
||
))
|
||
|
||
if not order.get("moment"):
|
||
results.append(ValidationResult(
|
||
level=ValidationLevel.CRITICAL,
|
||
field="order.moment",
|
||
message="Не указана дата заказа"
|
||
))
|
||
|
||
total_sum = order.get("sum", 0)
|
||
if total_sum <= 0:
|
||
results.append(ValidationResult(
|
||
level=ValidationLevel.CRITICAL,
|
||
field="order.sum",
|
||
message="Некорректная сумма заказа"
|
||
))
|
||
|
||
# Валидация организации
|
||
org = order_data.get("organization")
|
||
results.extend(self.validate_organization(org))
|
||
|
||
# Валидация контрагента
|
||
agent = order_data.get("counterparty")
|
||
results.extend(self.validate_counterparty(agent))
|
||
|
||
# Валидация грузополучателя (КРИТИЧНО!)
|
||
# Если грузополучатель не указан явно, используем контрагента
|
||
consignee = order_data.get("consignee")
|
||
if consignee is None:
|
||
# Используем контрагента как грузополучателя по умолчанию
|
||
consignee = order_data.get("counterparty")
|
||
if consignee is None:
|
||
results.append(ValidationResult(
|
||
level=ValidationLevel.CRITICAL,
|
||
field="order.consignee",
|
||
message="Не указан грузополучатель (обязательно для УПД)"
|
||
))
|
||
|
||
# Проверяем наличие базовых данных грузополучателя
|
||
if consignee and isinstance(consignee, dict):
|
||
if not consignee.get("name"):
|
||
results.append(ValidationResult(
|
||
level=ValidationLevel.CRITICAL,
|
||
field="consignee.name",
|
||
message="Не указано наименование грузополучателя"
|
||
))
|
||
|
||
# Валидация позиций
|
||
positions = order_data.get("positions", [])
|
||
if not positions:
|
||
results.append(ValidationResult(
|
||
level=ValidationLevel.CRITICAL,
|
||
field="order.positions",
|
||
message="Заказ не содержит товарных позиций"
|
||
))
|
||
else:
|
||
for index, position in enumerate(positions):
|
||
results.extend(self.validate_position(position, index))
|
||
|
||
return results
|
||
|
||
def get_validation_summary(
|
||
self,
|
||
results: List[ValidationResult]
|
||
) -> Tuple[ValidationLevel, str]:
|
||
"""
|
||
Получение сводки по результатам валидации
|
||
|
||
Args:
|
||
results: Список результатов валидации
|
||
|
||
Returns:
|
||
Tuple: (общий уровень, текстовая сводка)
|
||
"""
|
||
if not results:
|
||
return ValidationLevel.VALID, "✅ Все проверки пройдены успешно"
|
||
|
||
critical = [r for r in results if r.level == ValidationLevel.CRITICAL]
|
||
warnings = [r for r in results if r.level == ValidationLevel.WARNING]
|
||
|
||
if critical:
|
||
level = ValidationLevel.CRITICAL
|
||
summary = f"❌ Критических ошибок: {len(critical)}\n"
|
||
if warnings:
|
||
summary += f"⚠️ Предупреждений: {len(warnings)}\n"
|
||
summary += "\nКритические ошибки:\n"
|
||
for r in critical:
|
||
summary += f" • {r.message}\n"
|
||
elif warnings:
|
||
level = ValidationLevel.WARNING
|
||
summary = f"⚠️ Предупреждений: {len(warnings)}\n"
|
||
summary += "\nПредупреждения:\n"
|
||
for r in warnings:
|
||
summary += f" • {r.message}\n"
|
||
else:
|
||
level = ValidationLevel.VALID
|
||
summary = "✅ Все проверки пройдены"
|
||
|
||
return level, summary
|
||
|
||
|
||
# Пример использования
|
||
if __name__ == "__main__":
|
||
validator = OrderValidator()
|
||
|
||
# Тест валидации ИНН
|
||
test_inns = [
|
||
"7707083893", # Верный ИНН (10 цифр)
|
||
"500100732259", # Верный ИНН ИП (12 цифр)
|
||
"1234567890", # Неверный ИНН
|
||
]
|
||
|
||
for inn in test_inns:
|
||
result = validator.validate_inn(inn)
|
||
print(f"ИНН {inn}: {'✅ корректен' if result else '❌ некорректен'}")
|