# precheck/scanner.py from typing import Dict, List, Optional, Any, Tuple from dataclasses import dataclass, field from .registry import FIELD_REGISTRY, FieldDefinition @dataclass class FieldStatus: status: str # "FOUND", "NOT_FOUND", "UNCERTAIN" value: Any = None confidence: float = 1.0 message: str = "" @dataclass class ScanReport: field_statuses: Dict[str, FieldStatus] = field(default_factory=dict) extracted_values: Dict[str, Any] = field(default_factory=dict) overall_confidence: float = 0.0 missing_required: List[str] = field(default_factory=list) class PreCheckScanner: """ Сканирует текст письма (и вложений) на наличие ключевых полей, используя детерминированные экстракторы из реестра. """ def __init__(self, registry: List[FieldDefinition] = None): self.registry = registry or FIELD_REGISTRY def scan(self, context_text: str, sources: Optional[List[Dict]] = None, shipping_type_name: Optional[str] = None) -> ScanReport: """ Основной метод сканирования. :param context_text: полный текст письма и вложений :param sources: исходные письма (для возможной привязки к ID) :param shipping_type_name: имя типа перевозки для учёта дополнительных обязательных полей :return: ScanReport """ report = ScanReport() found_count = 0 total_required = 0 missing_required = [] for field_def in self.registry: # Проверяем, является ли поле обязательным для данного типа перевозки (упрощённо: если required_by_default) # В будущем можно учесть shipping_type.mandatory_fields if field_def.required_by_default: total_required += 1 # Если нет экстрактора, считаем поле не найденным if field_def.extractor is None: status = FieldStatus(status="NOT_FOUND", confidence=0.0, message="Нет детерминированного экстрактора") else: try: # Вызываем экстрактор с текстом value = field_def.extractor(context_text, None) if value is not None and (field_def.validator is None or field_def.validator(value)): status = FieldStatus(status="FOUND", value=value, confidence=0.9) report.extracted_values[field_def.name] = value found_count += 1 else: # Если есть значение, но не прошло валидацию - UNCERTAIN, иначе NOT_FOUND if value is not None: status = FieldStatus(status="UNCERTAIN", value=value, confidence=0.4, message="Значение не прошло валидацию") else: status = FieldStatus(status="NOT_FOUND", confidence=0.0) except Exception as e: status = FieldStatus(status="NOT_FOUND", confidence=0.0, message=str(e)) report.field_statuses[field_def.name] = status if field_def.required_by_default and status.status != "FOUND": missing_required.append(field_def.name) report.missing_required = missing_required total_fields = len(self.registry) report.overall_confidence = found_count / total_fields if total_fields > 0 else 0.0 # Учёт зависимостей: если dimension найдены, попробуем вычислить total_volume_cbm if "dimensions" in report.extracted_values and "total_volume_cbm" not in report.extracted_values: dims = report.extracted_values["dimensions"] try: vol = sum( (d.get("length_cm", 0) * d.get("width_cm", 0) * d.get("height_cm", 0)) / 1_000_000 for d in dims ) if vol > 0: report.extracted_values["total_volume_cbm"] = round(vol, 6) # обновим статус if "total_volume_cbm" in report.field_statuses: report.field_statuses["total_volume_cbm"] = FieldStatus( status="FOUND", value=vol, confidence=0.7, message="Вычислено из габаритов" ) # если было в missing_required, убрать if "total_volume_cbm" in report.missing_required: report.missing_required.remove("total_volume_cbm") except Exception: pass return report