815 lines
35 KiB
Python
815 lines
35 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")
|
||
company_type = agent.get("companyType", "legal")
|
||
|
||
if not inn:
|
||
# Для физических лиц (не ИП) ИНН не обязателен
|
||
if company_type == "individual":
|
||
results.append(ValidationResult(
|
||
level=ValidationLevel.WARNING,
|
||
field="counterparty.inn",
|
||
message="Не указан ИНН контрагента (физическое лицо)"
|
||
))
|
||
else:
|
||
# Для юр. лиц и ИП ИНН обязателен
|
||
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.WARNING,
|
||
field="counterparty.address",
|
||
message="Не указан адрес контрагента"
|
||
))
|
||
|
||
return results
|
||
|
||
def validate_position(
|
||
self,
|
||
position: Dict,
|
||
index: int
|
||
) -> List[ValidationResult]:
|
||
"""
|
||
Валидация позиции заказа (товар)
|
||
|
||
Args:
|
||
position: Данные позиции
|
||
index: Номер позиции
|
||
|
||
Returns:
|
||
Список результатов валидации
|
||
"""
|
||
results = []
|
||
pos_num = index + 1 # Человекочитаемый номер позиции
|
||
|
||
# Наименование товара
|
||
assortment = position.get("assortment", {})
|
||
name = assortment.get("name")
|
||
if not name:
|
||
results.append(ValidationResult(
|
||
level=ValidationLevel.CRITICAL,
|
||
field=f"position[{index}].name",
|
||
message=f"Позиция {pos_num}: не указано наименование товара"
|
||
))
|
||
|
||
# Количество
|
||
quantity = position.get("quantity")
|
||
if quantity is None:
|
||
results.append(ValidationResult(
|
||
level=ValidationLevel.CRITICAL,
|
||
field=f"position[{index}].quantity",
|
||
message=f"Позиция {pos_num} ({name or 'без названия'}): не указано количество"
|
||
))
|
||
elif quantity <= 0:
|
||
results.append(ValidationResult(
|
||
level=ValidationLevel.CRITICAL,
|
||
field=f"position[{index}].quantity",
|
||
message=f"Позиция {pos_num} ({name or 'без названия'}): "
|
||
f"некорректное количество {quantity}"
|
||
))
|
||
|
||
# Цена (КРИТИЧНО: цена должна быть > 0)
|
||
price = position.get("price")
|
||
if price is None:
|
||
results.append(ValidationResult(
|
||
level=ValidationLevel.CRITICAL,
|
||
field=f"position[{index}].price",
|
||
message=f"Позиция {pos_num} ({name or 'без названия'}): не указана цена"
|
||
))
|
||
elif price < 0:
|
||
results.append(ValidationResult(
|
||
level=ValidationLevel.CRITICAL,
|
||
field=f"position[{index}].price",
|
||
message=f"Позиция {pos_num} ({name or 'без названия'}): "
|
||
f"отрицательная цена {price / 100:.2f} руб"
|
||
))
|
||
elif price == 0:
|
||
# КРИТИЧНО: нулевая цена недопустима для коммерческих документов
|
||
results.append(ValidationResult(
|
||
level=ValidationLevel.CRITICAL,
|
||
field=f"position[{index}].price",
|
||
message=f"Позиция {pos_num} ({name or 'без названия'}): "
|
||
f"цена равна нулю (недопустимо для УПД/ТОРГ-12)"
|
||
))
|
||
|
||
# Проверка суммы строки (если есть цена и количество)
|
||
if price is not None and quantity is not None and price > 0 and quantity > 0:
|
||
line_total = (price / 100) * quantity # Переводим копейки в рубли
|
||
if line_total == 0:
|
||
results.append(ValidationResult(
|
||
level=ValidationLevel.CRITICAL,
|
||
field=f"position[{index}].total",
|
||
message=f"Позиция {pos_num} ({name or 'без названия'}): "
|
||
f"сумма строки равна нулю"
|
||
))
|
||
|
||
# НДС
|
||
vat = position.get("vat")
|
||
if vat is None:
|
||
results.append(ValidationResult(
|
||
level=ValidationLevel.WARNING,
|
||
field=f"position[{index}].vat",
|
||
message=f"Позиция {pos_num} ({name or 'без названия'}): "
|
||
f"не указана ставка НДС (будет использовано значение по умолчанию)"
|
||
))
|
||
|
||
# Единица измерения (желательно)
|
||
uom = assortment.get("uom")
|
||
if not uom:
|
||
results.append(ValidationResult(
|
||
level=ValidationLevel.WARNING,
|
||
field=f"position[{index}].uom",
|
||
message=f"Позиция {pos_num} ({name or 'без названия'}): "
|
||
f"не указана единица измерения (будет использовано 'шт')"
|
||
))
|
||
|
||
# Код товара (желательно для учёта)
|
||
code = assortment.get("code")
|
||
if not code:
|
||
results.append(ValidationResult(
|
||
level=ValidationLevel.WARNING,
|
||
field=f"position[{index}].code",
|
||
message=f"Позиция {pos_num} ({name or 'без названия'}): "
|
||
f"не указан код товара"
|
||
))
|
||
|
||
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="Не указано наименование грузополучателя"
|
||
))
|
||
|
||
# КРИТИЧНО: адрес грузополучателя обязателен для УПД
|
||
consignee_address = (
|
||
consignee.get("deliveryAddress") or
|
||
consignee.get("actualAddress") or
|
||
consignee.get("legalAddress")
|
||
)
|
||
if not consignee_address:
|
||
# Фолбэк: используем адрес отгрузки из документа, если он есть
|
||
shipment_addr = (
|
||
(order.get("shipmentAddressFull") or {}).get("addInfo")
|
||
or order.get("shipmentAddress")
|
||
)
|
||
if shipment_addr:
|
||
consignee_address = shipment_addr # используется для логики
|
||
else:
|
||
results.append(ValidationResult(
|
||
level=ValidationLevel.CRITICAL,
|
||
field="consignee.address",
|
||
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
|
||
|
||
|
||
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
|
||
import sys
|
||
import re
|
||
import math
|
||
from decimal import Decimal, InvalidOperation
|
||
import xml.etree.ElementTree as ET
|
||
from collections import defaultdict
|
||
from typing import List, Tuple, Optional
|
||
|
||
GUID_RE = re.compile(r"^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$")
|
||
|
||
# НДС: базовые ставки, которые точно поддерживаются большинством конфигураций РФ
|
||
KNOWN_VAT_RATES = {Decimal("0"), Decimal("5"), Decimal("7"), Decimal("10"), Decimal("18"), Decimal("20")}
|
||
|
||
def localname(tag: str) -> str:
|
||
"""Вернуть локальное имя узла без пространства имён."""
|
||
return tag.split('}', 1)[-1]
|
||
|
||
def text_of(el: Optional[ET.Element]) -> str:
|
||
return (el.text or "").strip() if el is not None else ""
|
||
|
||
def dec(value: str) -> Optional[Decimal]:
|
||
if value is None:
|
||
return None
|
||
v = value.replace(" ", "").replace("\u00A0", "").replace(",", ".")
|
||
try:
|
||
return Decimal(v)
|
||
except (InvalidOperation, ValueError):
|
||
return None
|
||
|
||
def findall_by_localname(root: ET.Element, name: str) -> List[ET.Element]:
|
||
"""Поиск по локальному имени вне зависимости от ns."""
|
||
return [e for e in root.iter() if localname(e.tag) == name]
|
||
|
||
def parent_path(el: ET.Element, parent_map: dict, max_len: int = 6) -> str:
|
||
"""Короткая «дорожка» для сообщений об ошибках."""
|
||
parts = []
|
||
cur = el
|
||
for _ in range(max_len):
|
||
parts.append(localname(cur.tag))
|
||
cur = get_parent(cur, parent_map)
|
||
if cur is None:
|
||
break
|
||
return "/".join(reversed(parts))
|
||
|
||
def attach_parents(root: ET.Element):
|
||
"""Проставить ссылки на родителей, чтобы собирать человекочитаемые пути."""
|
||
# Используем словарь вместо прямого setattr
|
||
parent_map = {}
|
||
for p in root.iter():
|
||
for c in list(p):
|
||
parent_map[id(c)] = p
|
||
return parent_map
|
||
|
||
def get_parent(el: ET.Element, parent_map: dict) -> Optional[ET.Element]:
|
||
"""Получить родителя элемента из карты."""
|
||
return parent_map.get(id(el))
|
||
|
||
class Issue:
|
||
def __init__(self, level: str, where: str, msg: str):
|
||
self.level = level # CRITICAL | WARNING
|
||
self.where = where
|
||
self.msg = msg
|
||
def __str__(self):
|
||
return f"[{self.level}] {self.where}: {self.msg}"
|
||
|
||
def validate_guid(s: str) -> bool:
|
||
return bool(GUID_RE.match(s))
|
||
|
||
def is_dirty_link(s: str) -> bool:
|
||
"""Посторонние хвосты в значении ссылки (например '?expand=...')."""
|
||
return "?" in s or "http://" in s.lower() or "https://" in s.lower()
|
||
|
||
def validate_file(path: str) -> Tuple[List[Issue], List[Issue]]:
|
||
critical: List[Issue] = []
|
||
warnings: List[Issue] = []
|
||
|
||
try:
|
||
tree = ET.parse(path)
|
||
except ET.ParseError as e:
|
||
return [Issue("CRITICAL", path, f"XML не разобрался: {e}")], []
|
||
|
||
root = tree.getroot()
|
||
parent_map = attach_parents(root)
|
||
|
||
# --- 1) Общие проверки корня/версии/типа сообщения
|
||
root_name = localname(root.tag)
|
||
if root_name.lower() not in {"enterprisedata", "package", "пакет", "сообщение"}:
|
||
warnings.append(Issue("WARNING", root_name, "Корневой элемент не похож на EnterpriseData (проверка наименования)"))
|
||
|
||
# --- 2) Проверка всех <Ссылка>
|
||
for link in findall_by_localname(root, "Ссылка"):
|
||
value = text_of(link)
|
||
where = parent_path(link, parent_map)
|
||
if value == "":
|
||
# для некоторых классификаторов допускается пустая ссылка (если есть коды) — проверим ниже
|
||
parent = get_parent(link, parent_map)
|
||
pname = localname(parent.tag) if parent is not None else ""
|
||
if pname in {"Номенклатура", "Контрагент", "Организация", "ДоговорКонтрагента"}:
|
||
critical.append(Issue("CRITICAL", where, "Пустая ссылка у обязательного объектного реквизита"))
|
||
else:
|
||
warnings.append(Issue("WARNING", where, "Пустая ссылка"))
|
||
continue
|
||
|
||
if is_dirty_link(value):
|
||
critical.append(Issue("CRITICAL", where, f"Ссылка содержит посторонние хвосты/URL: «{value}»"))
|
||
|
||
if not validate_guid(value):
|
||
critical.append(Issue("CRITICAL", where, f"Некорректный GUID: «{value}»"))
|
||
|
||
# --- 3) Документ: номер/дата/ключевые ссылки
|
||
docs = [e for e in root.iter() if localname(e.tag) in {"Документ", "Document"}]
|
||
if not docs:
|
||
warnings.append(Issue("WARNING", root_name, "Не найден узел «Документ» (валидация продолжится по общим правилам)"))
|
||
for doc in docs:
|
||
num = text_of(next((c for c in doc if localname(c.tag) == "Номер"), None))
|
||
date = text_of(next((c for c in doc if localname(c.tag) == "Дата"), None))
|
||
if num.strip() == "":
|
||
critical.append(Issue("CRITICAL", parent_path(doc, parent_map), "Не заполнен номер документа"))
|
||
elif num != num.strip():
|
||
warnings.append(Issue("WARNING", parent_path(doc, parent_map), f"Номер документа содержит ведущие/хвостовые пробелы: «{num}»"))
|
||
|
||
if date.strip() == "":
|
||
critical.append(Issue("CRITICAL", parent_path(doc, parent_map), "Не заполнена дата документа"))
|
||
|
||
# ключевые ссылки шапки, если присутствуют
|
||
for key_name in ("Организация", "Контрагент", "ДоговорКонтрагента"):
|
||
node = next((c for c in doc if localname(c.tag) == key_name), None)
|
||
if node is not None:
|
||
ref = text_of(next((c for c in node if localname(c.tag) == "Ссылка"), None))
|
||
if ref == "":
|
||
critical.append(Issue("CRITICAL", f"{parent_path(node, parent_map)}", f"Пустая ссылка у «{key_name}»"))
|
||
elif not validate_guid(ref):
|
||
critical.append(Issue("CRITICAL", f"{parent_path(node, parent_map)}", f"Некорректный GUID у «{key_name}»: «{ref}»"))
|
||
|
||
# --- 4) Табличные строки: номенклатура/ЕИ/арифметика/НДС
|
||
# Ищем элементы, у которых есть дочерняя «Номенклатура»
|
||
for el in root.iter():
|
||
if any(localname(c.tag) == "Номенклатура" for c in list(el)):
|
||
row = el
|
||
ppath = parent_path(row, parent_map)
|
||
|
||
# Номенклатура -> Ссылка
|
||
nom = next((c for c in row if localname(c.tag) == "Номенклатура"), None)
|
||
if nom is not None:
|
||
ref = text_of(next((c for c in nom if localname(c.tag) == "Ссылка"), None))
|
||
if ref == "":
|
||
critical.append(Issue("CRITICAL", ppath + "/Номенклатура", "Пустая ссылка номенклатуры"))
|
||
elif not validate_guid(ref):
|
||
critical.append(Issue("CRITICAL", ppath + "/Номенклатура", f"Некорректный GUID номенклатуры: «{ref}»"))
|
||
elif is_dirty_link(ref):
|
||
critical.append(Issue("CRITICAL", ppath + "/Номенклатура", f"Ссылка номенклатуры содержит хвосты/URL: «{ref}»"))
|
||
|
||
# ЕИ: допускается пустая ссылка, если указан код ОКЕИ (например, 796)
|
||
uom = next((c for c in row if localname(c.tag) in {"ЕдиницаИзмерения", "ЕИ"}), None)
|
||
if uom is not None:
|
||
uom_ref = text_of(next((c for c in uom if localname(c.tag) == "Ссылка"), None))
|
||
uom_code = text_of(next((c for c in uom if localname(c.tag) in {"Код", "КодОКЕИ"}), None))
|
||
if uom_ref == "" and uom_code == "":
|
||
warnings.append(Issue("WARNING", ppath + "/ЕдиницаИзмерения", "Пустая ссылка ЕИ и отсутствует код ОКЕИ (например, 796 — «шт»)"))
|
||
|
||
# Кол-во/Цена/Сумма
|
||
qty = dec(text_of(next((c for c in row if localname(c.tag) in {"Количество", "Количествo", "Qty"}), None)))
|
||
price = dec(text_of(next((c for c in row if localname(c.tag) in {"Цена", "Price"}), None)))
|
||
amount = dec(text_of(next((c for c in row if localname(c.tag) in {"Сумма", "Amount"}), None)))
|
||
vat_rate = dec(text_of(next((c for c in row if localname(c.tag) in {"СтавкаНДС", "VAT"}), None)))
|
||
|
||
if qty is not None and qty < 0:
|
||
warnings.append(Issue("WARNING", ppath, f"Отрицательное количество: {qty}"))
|
||
if price is not None and price < 0:
|
||
warnings.append(Issue("WARNING", ppath, f"Отрицательная цена: {price}"))
|
||
if amount is not None and amount < 0:
|
||
warnings.append(Issue("WARNING", ppath, f"Отрицательная сумма: {amount}"))
|
||
|
||
# Простейшая арифметика (если НДС включён в цену — точного флага может не быть; делаем мягкую проверку)
|
||
if qty is not None and price is not None and amount is not None:
|
||
calc = (qty * price).quantize(Decimal("0.01"))
|
||
if amount.quantize(Decimal("0.01")) != calc:
|
||
warnings.append(Issue("WARNING", ppath, f"Сумма строки ({amount}) не равна Кол-во*Цена ({calc}) — проверьте флаг «Цена/Сумма включает НДС» и округления"))
|
||
|
||
# НДС ставка: неизвестная — как возможная проблема совместимости
|
||
if vat_rate is not None and vat_rate not in KNOWN_VAT_RATES:
|
||
warnings.append(Issue("WARNING", ppath + "/СтавкаНДС", f"Нетипичная ставка НДС {vat_rate}% — проверьте поддержку в вашей конфигурации"))
|
||
|
||
# --- 5) Валюта/страна (коды должны быть валидными)
|
||
for cur in findall_by_localname(root, "Валюта"):
|
||
code3 = text_of(next((c for c in cur if localname(c.tag) in {"КодБуквенный", "Код"}), None))
|
||
num = text_of(next((c for c in cur if localname(c.tag) in {"КодЧисловой", "КодISO"}), None))
|
||
if not code3 and not num:
|
||
warnings.append(Issue("WARNING", parent_path(cur, parent_map), "Валюта без буквенного/числового ISO-кода (например, RUB/643)"))
|
||
|
||
for country in findall_by_localname(root, "Страна"):
|
||
ccode = text_of(next((c for c in country if localname(c.tag) in {"Код", "КодЧисловой"}), None))
|
||
if not ccode:
|
||
warnings.append(Issue("WARNING", parent_path(country, parent_map), "Страна без кода (для РФ обычно 643)"))
|
||
|
||
# --- 6) Загрязнённые номера документов (пробелы, табы и т.п.) уже отмечены; добавим проверку длины
|
||
for num in findall_by_localname(root, "Номер"):
|
||
n = num.text or ""
|
||
if len(n) > 50:
|
||
warnings.append(Issue("WARNING", parent_path(num, parent_map), f"Очень длинный номер документа ({len(n)} символов)"))
|
||
|
||
return critical, warnings
|
||
|
||
|
||
def main():
|
||
if len(sys.argv) < 2:
|
||
print("Использование: python ed_validator.py <file.xml> [<file2.xml> ...]")
|
||
sys.exit(2)
|
||
|
||
exit_code = 0
|
||
for path in sys.argv[1:]:
|
||
crit, warn = validate_file(path)
|
||
print(f"\n=== Проверка файла: {path} ===")
|
||
if not crit and not warn:
|
||
print("Проблем не найдено.")
|
||
continue
|
||
|
||
if crit:
|
||
print("\nКРИТИЧЕСКИЕ ОШИБКИ:")
|
||
for i in crit:
|
||
print(" -", i)
|
||
exit_code = 1
|
||
if warn:
|
||
print("\nВОЗМОЖНЫЕ ПРОБЛЕМЫ:")
|
||
for i in warn:
|
||
print(" -", i)
|
||
|
||
sys.exit(exit_code)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|