marking_utils.py - main - Paff/declarant

marking_utils.py
239 строк9 KB
1pash1985-gif
fix(europa): resolve marking/column mismatches vs manager template PL_727
21 июл 2026, 20:36
21 июл 2026, 20:363f25b50
"""Утилиты нормализации маркировок клиентов (client codes).

Маркировка — короткий код клиента-получателя внутри партии
(B-AQUA, B-AIGUL, K-BO, ARSK, NEFT, MJ, RASUL и т.п.).

ЭТО НЕ:
  - название компании-получателя (EFLOWERS B.V., BIFLORICA)
  - название поставщика/фермы (GLOW FLOWERS SAS, LA CONCHITA)
  - адрес или страна

Правила нормализации:
  1. Строку компании (с юр. суффиксом B.V., S.A.S. и т.п.) отбрасываем.
  2. Всегда UPPER, strip, схлопывание внутренних пробелов.
  3. Префикс `B-` факультативен: `AQUA` эквивалентно `B-AQUA`.
  4. Возможен парсинг из имени папки: "1561 AQUA" → ("1561", "AQUA").
  5. Возможен парсинг из имени файла: "BESTFLORA-2_b-AQUA_23_Jun_2026.xlsx"
     → маркировка "B-AQUA".
"""
from __future__ import annotations

import re
from typing import Optional


# Юр. суффиксы компаний. Если строка их содержит — это НЕ маркировка.
_COMPANY_SUFFIXES = re.compile(
    r'\b('
    r'B\.?V\.?|N\.?V\.?|'                       # BV, NV (Netherlands)
    r'S\.?A\.?S\.?|S\.?A\.?|S\.?L\.?|S\.?R\.?L\.?|S\.?R\.?O\.?|'  # SAS, SA, SL, SRL, SRO
    r'INC\.?|LTD\.?|LLC|LLP|'                   # Anglo
    r'GMBH|AG|KG|OHG|'                          # Germany
    r'CO\.?\s*LTD|CORP\.?|PLC|'
    r'OY|AB'                                    # Nordic
    r')\b',
    re.IGNORECASE
)

# Слова, характерные для названий компаний (даже без юр. суффикса)
_COMPANY_WORDS = re.compile(
    r'\b(FLOWERS?|FLORA|FLORAL|FARMS?|PLANTATION|EXPORT|IMPORT|'
    r'GROUP|COMPANY|TRADING|LOGISTICS?|CARGO|SHIPPING)\b',
    re.IGNORECASE
)


def is_client_name(s: str) -> bool:
    """True, если строка похожа на название компании, а не на маркировку."""
    if not s:
        return False
    s = s.strip()
    if len(s) > 30:
        # Слишком длинное — вряд ли маркировка (обычно до 15 символов)
        return True
    if _COMPANY_SUFFIXES.search(s):
        return True
    # Наличие "компанийного" слова + пробел + приличная длина
    if _COMPANY_WORDS.search(s) and ' ' in s and len(s) > 10:
        return True
    return False


def normalize_marking(raw: Optional[str]) -> str:
    """Привести маркировку к каноническому виду.
    Возвращает '' если невалидна (клиент/пустая)."""
    if not raw:
        return ''
    s = str(raw).strip()
    if not s or is_client_name(s):
        return ''
    # Схлопнуть пробелы, UPPER, убрать хвостовую пунктуацию
    s = re.sub(r'\s+', ' ', s).upper()
    s = s.strip(' .-,;:/\\')
    if not s:
        return ''
    # Ограничение по длине после нормализации — жёстко отсеиваем «предложения»
    if len(s) > 25:
        return ''
    return s


def strip_b_prefix(marking: str) -> str:
    """Убрать префикс `B-` для сравнения (`B-AQUA` → `AQUA`)."""
    if not marking:
        return ''
    m = marking.strip().upper()
    if m.startswith('B-'):
        return m[2:]
    return m


def markings_equivalent(a: str, b: str) -> bool:
    """True если a и b — одна маркировка (`AQUA` == `B-AQUA`)."""
    a = normalize_marking(a)
    b = normalize_marking(b)
    if not a or not b:
        return False
    if a == b:
        return True
    return strip_b_prefix(a) == strip_b_prefix(b)


# --- Парсинг из имени папки / файла ------------------------------------------

# Соответствует: "1561 AQUA", "1561 B-FIT", "4351 BOSS", "444 KZE",
# "1561_AQUA", "1561-KAZAN"
_FOLDER_FLIGHT_MARK_RE = re.compile(
    r'^\s*(?P<flight>\d{2,5})[\s_\-]+(?P<mark>[A-Za-z][\w.\-]{0,24})'
)

# В имени файла между _ или - может быть "b-AQUA" (или "B-AIGUL") с любым регистром
_FILENAME_MARK_RE = re.compile(
    r'(?<![A-Za-z0-9])(B-[A-Za-z][A-Za-z0-9]{1,15})(?![A-Za-z0-9])',
    re.IGNORECASE
)


def parse_folder_marking(folder_name: str) -> tuple[Optional[str], Optional[str]]:
    """Разобрать имя папки на (flight, marking).

    Примеры:
      '1561 AQUA'      → ('1561', 'AQUA')
      '1561 B-FIT'     → ('1561', 'B-FIT')
      '4351 NERO 5кор' → ('4351', 'NERO')
      'AIGUL'          → (None, 'AIGUL')
      '444'            → ('444', None)
    Не найдено → (None, None).
    """
    if not folder_name:
        return None, None
    name = folder_name.strip()

    m = _FOLDER_FLIGHT_MARK_RE.match(name)
    if m:
        flight = m.group('flight')
        mark = normalize_marking(m.group('mark'))
        return (flight or None), (mark or None)

    # Только маркировка без flight
    m2 = re.match(r'^([A-Za-z][\w.\-]{1,24})\s*$', name)
    if m2:
        mark = normalize_marking(m2.group(1))
        return None, (mark or None)

    # Только flight без маркировки
    m3 = re.match(r'^(\d{2,5})\s*$', name)
    if m3:
        return m3.group(1), None

    return None, None


def parse_filename_marking(filename: str) -> Optional[str]:
    """Найти маркировку в имени файла.

    Примеры:
      'BESTFLORA-2_b-AQUA_23_Jun_2026.xlsx' → 'B-AQUA'
      'Invoice 309282 23-06-2026 ISHAKOVA.xls' → None (нет `B-` префикса)
      'BESTFLORA-2_b-METELAGA98_23_Jun_2026.xlsx' → 'B-METELAGA98'
    """
    if not filename:
        return None
    m = _FILENAME_MARK_RE.search(filename)
    if not m:
        return None
    mark = normalize_marking(m.group(1))
    return mark or None


# --- Резолвер: применить приоритеты и corrections ---------------------------

def resolve_marking(
    ai_marking: Optional[str] = None,
    item_markings: Optional[list[str]] = None,
    filename: Optional[str] = None,
    folder_hint: Optional[str] = None,
    user_hint: Optional[str] = None,
    corrections=None,
) -> str:
    """Собрать финальную маркировку по приоритету источников.

    Порядок (сверху = сильнее):
      1) user_hint            — то, что явно указал пользователь при загрузке
      2) folder_hint          — из имени папки (напр. "1561 AQUA")
      3) filename             — паттерн `B-XXX` в имени файла
      4) item_markings        — маркировки в строках товара (single value)
      5) ai_marking           — маркировка из шапки AI-JSON (последняя надежда)

    Если ни один источник не даёт валидной маркировки — возвращаем ''.
    corrections — объект `corrections.Corrections`; применяет алиасы к
    победившей маркировке.
    """
    # 1) явное указание пользователя
    m = normalize_marking(user_hint)
    if m:
        return _apply_corrections(m, corrections)

    # 2) папка
    _, folder_mark = parse_folder_marking(folder_hint or '')
    if folder_mark:
        return _apply_corrections(folder_mark, corrections)

    # 3) имя файла
    fn_mark = parse_filename_marking(filename or '')
    if fn_mark:
        return _apply_corrections(fn_mark, corrections)

    # 4) маркировки на позициях
    if item_markings:
        clean = [normalize_marking(x) for x in item_markings]
        clean = [x for x in clean if x]
        if clean:
            # Если все одинаковые (или эквивалентные) — берём первую
            first = clean[0]
            if all(markings_equivalent(x, first) for x in clean):
                return _apply_corrections(first, corrections)
            # Иначе — самая частая
            from collections import Counter
            most = Counter(clean).most_common(1)[0][0]
            return _apply_corrections(most, corrections)

    # 5) маркировка из шапки AI-JSON
    m = normalize_marking(ai_marking)
    if m:
        return _apply_corrections(m, corrections)

    return ''


def _apply_corrections(marking: str, corrections) -> str:
    """Применить aliases из corrections.json (если объект передан)."""
    if not marking or corrections is None:
        return marking
    try:
        resolved = corrections.resolve_marking(marking)
    except Exception:
        return marking
    if resolved is None:
        return ''  # заблокировано пользователем в правилах
    return normalize_marking(resolved) or marking