""" Валидация данных заказов МойСклад Проверка обязательных и желательных полей, валидация форматов """ 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"цена равна нулю" )) # Проверка суммы строки (если есть цена и количество) 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 [ ...]") 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()