src/components/data_validator.py - master - alsoalgo/CTF

src/components/data_validator.py
373 строки15 KB
alsoalgo
some improvements
15 дек 2025, 17:15
15 дек 2025, 17:155bd30a7
"""
Валидатор данных для проверки корректности извлеченных данных из документов.
"""
import re
from datetime import datetime
from typing import Dict, List, Tuple, Optional, Any
import pandas as pd


class DataValidator:
    """
    Класс для валидации данных из документов CTF.
    Проверяет наличие обязательных полей, форматы данных и бизнес-правила.
    """
    
    def __init__(self):
        # Обязательные поля для CTF заявки
        self.required_fields = [
            'Supplier',
            'Off-taker',
            'Product',
            'Amount, USD',
            'Currency'
        ]
        
        # Обязательные документы для CTF сделки
        self.required_documents = [
            'contract',  # Контракт
            'invoice',  # Счет-фактура
        ]
        
        # Опциональные документы
        self.optional_documents = [
            'insurance',  # Страховой полис
            'forwarder',  # Документы экспедитора
        ]
    
    def validate_dataframe(self, df: pd.DataFrame) -> Dict[str, Any]:
        """
        Валидирует DataFrame с данными заявки.
        
        :param df: DataFrame с данными заявки
        :return: Словарь с результатами валидации
        """
        errors = []
        warnings = []
        
        if df is None or df.empty:
            return {
                'valid': False,
                'errors': ['DataFrame is empty or None'],
                'warnings': []
            }
        
        # Проверяем наличие обязательных полей (новая структура с PURCHASE и SALE)
        if 'Field' not in df.columns:
            return {
                'valid': False,
                'errors': ['DataFrame must have "Field" column'],
                'warnings': []
            }
        
        # Проверяем наличие колонок PURCHASE и SALE (новая структура) или Value (старая структура)
        has_new_structure = 'PURCHASE' in df.columns and 'SALE' in df.columns
        has_old_structure = 'Value' in df.columns
        
        if not has_new_structure and not has_old_structure:
            return {
                'valid': False,
                'errors': ['DataFrame must have "PURCHASE" and "SALE" columns (or "Value" for old structure)'],
                'warnings': []
            }
        
        # Создаем словарь полей для удобства
        # Для новой структуры проверяем значения в обеих колонках
        fields_dict = {}
        for _, row in df.iterrows():
            field_name = str(row.get('Field', '')).strip()
            if field_name:
                if has_new_structure:
                    # Для новой структуры берем значение из PURCHASE или SALE в зависимости от поля
                    purchase_value = str(row.get('PURCHASE', '')).strip()
                    sale_value = str(row.get('SALE', '')).strip()
                    
                    # Определяем, из какой колонки брать значение
                    if field_name == 'Supplier' or field_name == 'Country of Origin':
                        fields_dict[field_name] = purchase_value
                    elif field_name == 'Off-taker' or field_name == 'Country of Destination':
                        fields_dict[field_name] = sale_value
                    else:
                        # Для общих полей берем из PURCHASE (или SALE, если PURCHASE пусто)
                        fields_dict[field_name] = purchase_value if purchase_value else sale_value
                else:
                    # Старая структура
                    field_value = str(row.get('Value', '')).strip()
                    fields_dict[field_name] = field_value
        
        # Проверяем обязательные поля
        for field in self.required_fields:
            if field not in fields_dict or not fields_dict[field] or fields_dict[field] == '':
                errors.append(f"Required field '{field}' is missing or empty")
        
        # Валидируем форматы данных
        format_errors = self._validate_formats(fields_dict)
        errors.extend(format_errors)
        
        # Проверяем бизнес-правила
        business_errors, business_warnings = self._validate_business_rules(fields_dict)
        errors.extend(business_errors)
        warnings.extend(business_warnings)
        
        return {
            'valid': len(errors) == 0,
            'errors': errors,
            'warnings': warnings,
            'fields_checked': len(fields_dict)
        }
    
    def _validate_formats(self, fields_dict: Dict[str, str]) -> List[str]:
        """
        Проверяет форматы данных.
        
        :param fields_dict: Словарь полей и значений
        :return: Список ошибок формата
        """
        errors = []
        
        # Проверка суммы
        if 'Amount, USD' in fields_dict:
            amount = fields_dict['Amount, USD']
            if amount:
                try:
                    amount_float = float(amount.replace(',', '').replace(' ', ''))
                    if amount_float <= 0:
                        errors.append(f"Amount must be positive, got: {amount}")
                except ValueError:
                    errors.append(f"Invalid amount format: {amount}")
        
        # Проверка количества
        if 'Quantity, MT' in fields_dict:
            quantity = fields_dict['Quantity, MT']
            if quantity:
                try:
                    qty_float = float(quantity.replace(',', '').replace(' ', ''))
                    if qty_float <= 0:
                        errors.append(f"Quantity must be positive, got: {quantity}")
                except ValueError:
                    errors.append(f"Invalid quantity format: {quantity}")
        
        # Проверка дат
        date_fields = ['Contract Date', 'Delivery Date']
        for field in date_fields:
            if field in fields_dict and fields_dict[field]:
                date_str = fields_dict[field]
                if not self._is_valid_date(date_str):
                    errors.append(f"Invalid date format in '{field}': {date_str}")
        
        # Проверка валюты
        if 'Currency' in fields_dict:
            currency = fields_dict.get('Currency', '').upper()
            valid_currencies = ['USD', 'EUR', 'GBP', 'RUB']
            if currency and currency not in valid_currencies:
                errors.append(f"Unsupported currency: {currency}. Supported: {', '.join(valid_currencies)}")
        
        return errors
    
    def _is_valid_date(self, date_str: str) -> bool:
        """
        Проверяет, является ли строка валидной датой.
        
        :param date_str: Строка с датой
        :return: True если дата валидна
        """
        if not date_str or date_str.strip() == '':
            return False
        
        # Пробуем различные форматы дат
        date_formats = [
            '%Y-%m-%d',
            '%d.%m.%Y',
            '%d/%m/%Y',
            '%Y.%m.%d',
            '%d-%m-%Y'
        ]
        
        for fmt in date_formats:
            try:
                datetime.strptime(date_str.strip(), fmt)
                return True
            except ValueError:
                continue
        
        return False
    
    def _validate_business_rules(self, fields_dict: Dict[str, str]) -> Tuple[List[str], List[str]]:
        """
        Проверяет бизнес-правила CTF.
        
        :param fields_dict: Словарь полей и значений
        :return: (список ошибок, список предупреждений)
        """
        errors = []
        warnings = []
        
        # Проверка: сумма должна быть положительной
        if 'Amount, USD' in fields_dict and fields_dict['Amount, USD']:
            try:
                amount = float(fields_dict['Amount, USD'].replace(',', '').replace(' ', ''))
                if amount > 1000000:
                    warnings.append(f"Very large amount: {amount:,.2f} USD")
            except (ValueError, AttributeError):
                pass
        
        # Проверка: дата поставки должна быть после даты контракта
        if 'Contract Date' in fields_dict and 'Delivery Date' in fields_dict:
            contract_date = self._parse_date(fields_dict['Contract Date'])
            delivery_date = self._parse_date(fields_dict['Delivery Date'])
            
            if contract_date and delivery_date:
                if delivery_date < contract_date:
                    errors.append("Delivery date cannot be before contract date")
        
        # Проверка: поставщик и покупатель не должны совпадать
        if 'Supplier' in fields_dict and 'Off-taker' in fields_dict:
            supplier = fields_dict['Supplier'].strip().upper()
            off_taker = fields_dict['Off-taker'].strip().upper()
            if supplier and off_taker and supplier == off_taker:
                warnings.append("Supplier and Off-taker are the same")
        
        return errors, warnings
    
    def _parse_date(self, date_str: str) -> Optional[datetime]:
        """
        Парсит дату из строки.
        
        :param date_str: Строка с датой
        :return: Объект datetime или None
        """
        if not date_str or date_str.strip() == '':
            return None
        
        date_formats = [
            '%Y-%m-%d',
            '%d.%m.%Y',
            '%d/%m/%Y',
            '%Y.%m.%d',
            '%d-%m-%Y'
        ]
        
        for fmt in date_formats:
            try:
                return datetime.strptime(date_str.strip(), fmt)
            except ValueError:
                continue
        
        return None
    
    def validate_documents(self, document_types: List[str]) -> Dict[str, Any]:
        """
        Проверяет наличие обязательных документов.
        
        :param document_types: Список типов документов (например, ['invoice', 'contract'])
        :return: Словарь с результатами проверки
        """
        missing_required = []
        found_optional = []
        
        document_types_lower = [doc.lower() for doc in document_types]
        
        # Проверяем обязательные документы
        for req_doc in self.required_documents:
            if req_doc not in document_types_lower:
                missing_required.append(req_doc)
        
        # Проверяем опциональные документы
        for opt_doc in self.optional_documents:
            if opt_doc in document_types_lower:
                found_optional.append(opt_doc)
        
        return {
            'all_required_present': len(missing_required) == 0,
            'missing_required': missing_required,
            'found_optional': found_optional,
            'total_documents': len(document_types)
        }
    
    def validate_document_consistency(
        self, 
        documents_data: Dict[str, pd.DataFrame]
    ) -> Dict[str, Any]:
        """
        Проверяет согласованность данных между документами.
        
        :param documents_data: Словарь где ключ - тип документа, значение - DataFrame
        :return: Словарь с результатами проверки
        """
        inconsistencies = []
        
        # Извлекаем ключевые поля из каждого документа
        extracted_data = {}
        
        for doc_type, df in documents_data.items():
            if df is None or df.empty:
                continue
            
            # Пытаемся извлечь основные поля
            doc_data = {}
            
            # Ищем supplier
            for col in df.columns:
                col_lower = str(col).lower()
                if 'supplier' in col_lower and len(df) > 0:
                    doc_data['supplier'] = str(df.iloc[0][col]).strip()
                if 'off-taker' in col_lower or 'buyer' in col_lower or 'customer' in col_lower:
                    if len(df) > 0:
                        doc_data['off_taker'] = str(df.iloc[0][col]).strip()
                if 'amount' in col_lower or 'total' in col_lower:
                    if len(df) > 0:
                        try:
                            doc_data['amount'] = float(str(df.iloc[0][col]).replace(',', '').replace(' ', ''))
                        except (ValueError, AttributeError):
                            pass
            
            if doc_data:
                extracted_data[doc_type] = doc_data
        
        # Сравниваем данные между документами
        if len(extracted_data) < 2:
            return {
                'consistent': True,
                'inconsistencies': [],
                'message': 'Not enough documents to compare'
            }
        
        # Сравниваем supplier
        suppliers = [data.get('supplier', '').upper() for data in extracted_data.values() if data.get('supplier')]
        if len(set(suppliers)) > 1:
            inconsistencies.append({
                'field': 'Supplier',
                'values': list(set(suppliers)),
                'message': 'Supplier names differ between documents'
            })
        
        # Сравниваем off_taker
        off_takers = [data.get('off_taker', '').upper() for data in extracted_data.values() if data.get('off_taker')]
        if len(set(off_takers)) > 1:
            inconsistencies.append({
                'field': 'Off-taker',
                'values': list(set(off_takers)),
                'message': 'Off-taker names differ between documents'
            })
        
        # Сравниваем суммы (с допуском 1%)
        amounts = [data.get('amount') for data in extracted_data.values() if data.get('amount')]
        if len(amounts) > 1:
            amounts = [a for a in amounts if a is not None]
            if amounts:
                max_amount = max(amounts)
                min_amount = min(amounts)
                if max_amount > 0:
                    diff_percent = ((max_amount - min_amount) / max_amount) * 100
                    if diff_percent > 1:  # Разница больше 1%
                        inconsistencies.append({
                            'field': 'Amount',
                            'values': amounts,
                            'message': f'Amounts differ by {diff_percent:.2f}%'
                        })
        
        return {
            'consistent': len(inconsistencies) == 0,
            'inconsistencies': inconsistencies,
            'documents_compared': len(extracted_data)
        }