reference_loader.py - main - Paff/declarant

reference_loader.py
265 строк11 KB
1pash1985-gif
Декларант - веб-версия для Timeweb Cloud
15 июл 2026, 09:36
15 июл 2026, 09:36a6b08b8
# -*- coding: utf-8 -*-
"""
Загрузчик справочников из мастер-xlsx рейса (например 03.07.2026-700.xlsx).

Читает три листа:
- "Загрузка"       — определяет позиции столбцов и первую строку данных
- "Лист1 (16)"     — маркировка → клиент/эл.почта/телефон
- "Фермы"           — FARMS → HANDLERS (короткий код фермы → название хендлера)

Используется автозаполнителем ведомости.
"""
from __future__ import annotations
import os
import re
from dataclasses import dataclass, field
from typing import Optional

import openpyxl


def _norm_key(s: str) -> str:
    """Нормализовать ключ справочника: upper + strip + убрать двойные пробелы."""
    if s is None:
        return ""
    s = str(s).strip().upper()
    s = re.sub(r'\s+', ' ', s)
    return s


def _norm_marking_variants(s: str) -> list[str]:
    """Сгенерировать варианты маркировки для fuzzy-поиска.

    Пример:
      "B-BOOM55" → ["B-BOOM55", "BBOOM55", "BOOM55", "B BOOM55"]
      "BESTFLORA-B-ACR.30" → ["BESTFLORA-B-ACR.30", "B-ACR.30", "ACR.30", …]
    """
    key = _norm_key(s)
    if not key:
        return []
    variants = {key}
    # без дефисов/точек/пробелов
    variants.add(re.sub(r'[\s\-.]', '', key))
    # с заменой дефисов на пробелы и наоборот
    variants.add(key.replace('-', ' '))
    variants.add(key.replace(' ', '-'))
    # без префикса "B-" / "B " / "BESTFLORA-"
    for prefix in ('BESTFLORA-', 'B-', 'B '):
        if key.startswith(prefix):
            variants.add(key[len(prefix):])
    return [v for v in variants if v]


@dataclass
class ClientRef:
    """Одна запись справочника "Лист1 (16)"."""
    marking: str        # исходная маркировка
    client: str         # ФИО/название клиента
    email: str = ""
    phone: str = ""


@dataclass
class FarmRef:
    """Одна запись справочника "Фермы"."""
    farm: str           # короткий код фермы
    handler: str        # название хендлера


@dataclass
class ManifestColumns:
    """Позиции колонок в листе "Загрузка"."""
    firm: int = 1         # A
    boxes: int = 2        # B
    trolleys: int = 3     # C
    marking: int = 4      # D
    readress: int = 5     # E
    client: int = 6       # F
    phone: int = 9        # I
    # PART 1 / PART 2 sub-table
    part_mawb: int = 1     # A
    part_eta: int = 2      # B
    part_bxs: int = 3      # C
    part_full: int = 4     # D
    part_weight: int = 5   # E
    part_charg: int = 6    # F
    part_origin: int = 7   # G
    part_refs: int = 8     # H
    part_client: int = 9   # I
    part_precool: int = 10 # J


@dataclass
class ReferenceData:
    """Полные справочники, прочитанные из мастер-xlsx."""
    markings: dict[str, ClientRef] = field(default_factory=dict)   # key = norm variant
    farms: dict[str, FarmRef] = field(default_factory=dict)         # key = norm farm code
    handlers_set: set[str] = field(default_factory=set)             # для быстрой проверки firm
    columns: ManifestColumns = field(default_factory=ManifestColumns)
    default_readress: str = "Казань"

    def lookup_client(self, marking: str) -> Optional[ClientRef]:
        """Найти клиента по маркировке (пробует несколько нормализаций)."""
        if not marking:
            return None
        for v in _norm_marking_variants(marking):
            hit = self.markings.get(v)
            if hit:
                return hit
        return None

    def lookup_handler(self, farm_code: str) -> Optional[str]:
        """Определить название хендлера по коду фермы."""
        if not farm_code:
            return None
        key = _norm_key(farm_code)
        rec = self.farms.get(key)
        return rec.handler if rec else None


def load_references(master_xlsx: str) -> ReferenceData:
    """Загрузить все справочники из мастер-xlsx рейса.

    Источники mapping-а `marking → client`:
      1) лист "Загрузка": столбцы D (маркировка), F (клиент), I (телефон) — реальные
         примеры этого/предыдущего рейса. Точность 100%, приоритет высший.
      2) лист "Лист1 (16)": накопленный общий словарь, приоритет ниже.
    """
    if not os.path.exists(master_xlsx):
        raise FileNotFoundError(f"Master xlsx not found: {master_xlsx}")

    wb = openpyxl.load_workbook(master_xlsx, data_only=True, read_only=True)
    ref = ReferenceData()

    # Найти листы по фрагменту имени (регистр/пробелы могут различаться)
    sh_zag = _find_sheet(wb, ['Загрузка', 'загрузка', 'Zagruzka'])
    sh_lst = _find_sheet(wb, ['Лист1', 'клиент', 'client'])
    sh_farm = _find_sheet(wb, ['Фермы', 'фермы', 'Farms', 'farms'])

    # Сначала — низкоприоритетный общий словарь
    if sh_lst:
        ref.markings.update(_load_markings(wb[sh_lst]))
    # Затем — высокоприоритетный «этот рейс» (перезатирает пересечения)
    if sh_zag:
        loading_map = _load_markings_from_loading(wb[sh_zag])
        ref.markings.update(loading_map)
    if sh_farm:
        ref.farms, ref.handlers_set = _load_farms(wb[sh_farm])

    wb.close()
    return ref


def _find_sheet(wb, candidates: list[str]) -> Optional[str]:
    for name in wb.sheetnames:
        for c in candidates:
            if c.lower() in name.lower():
                return name
    return None


def _load_markings(ws) -> dict[str, ClientRef]:
    """Прочитать лист 'Лист1 (16)' → dict[norm_key → ClientRef]."""
    result: dict[str, ClientRef] = {}
    # Первая строка — заголовок (маркировка | клиент | эл почта | телефон)
    rows = ws.iter_rows(min_row=2, values_only=True)
    for row in rows:
        if not row or not row[0]:
            continue
        marking = str(row[0]).strip()
        if not marking:
            continue
        client = str(row[1]).strip() if len(row) > 1 and row[1] else ""
        email = str(row[2]).strip() if len(row) > 2 and row[2] else ""
        phone = str(row[3]).strip() if len(row) > 3 and row[3] else ""
        rec = ClientRef(marking=marking, client=client, email=email, phone=phone)
        # Регистрируем ВСЕ варианты нормализации → одна запись
        for v in _norm_marking_variants(marking):
            # Не перезаписываем более точную запись менее точной
            if v not in result:
                result[v] = rec
    return result


def _load_farms(ws) -> tuple[dict[str, FarmRef], set[str]]:
    """Прочитать лист 'Фермы' → dict[farm_code → FarmRef] + set(handlers)."""
    result: dict[str, FarmRef] = {}
    handlers: set[str] = set()
    rows = ws.iter_rows(min_row=2, values_only=True)
    for row in rows:
        if not row:
            continue
        farm = str(row[0]).strip() if row[0] else ""
        handler = str(row[1]).strip() if len(row) > 1 and row[1] else ""
        # Строки-разделители алфавита (одна буква в колонке A) пропускаем
        if len(farm) <= 1 and not handler:
            continue
        if farm:
            key = _norm_key(farm)
            if key not in result:
                result[key] = FarmRef(farm=farm, handler=handler)
        if handler:
            handlers.add(_norm_key(handler))
    return result, handlers


def _load_markings_from_loading(ws) -> dict[str, ClientRef]:
    """Прочитать столбцы D (маркировка) / F (клиент) / I (телефон) с листа "Загрузка".

    Строит mapping реальных примеров текущего/предыдущего рейса. После встречи
    заголовка PART 1 переключается на подтаблицу AWB: H (Order/Refs) → I (Client).
    """
    result: dict[str, ClientRef] = {}
    mode = 'main'   # main | part
    for row in ws.iter_rows(min_row=2, values_only=True):
        if not row:
            continue
        # Определяем переключатели режимов
        first_cell = str(row[0]).strip() if len(row) > 0 and row[0] else ""
        upper = first_cell.upper()
        if upper.startswith('MAWB'):
            # это заголовок PART-таблицы — со следующей строки читаем как part
            mode = 'part'
            continue
        if upper.startswith('PART '):
            mode = 'part'
            continue
        if upper.startswith(('TOTAL', 'IP HANDLERS')):
            continue

        if mode == 'main':
            marking = str(row[3]).strip() if len(row) > 3 and row[3] else ""
            client = str(row[5]).strip() if len(row) > 5 and row[5] else ""
            phone = str(row[8]).strip() if len(row) > 8 and row[8] else ""
        else:  # part
            marking = str(row[7]).strip() if len(row) > 7 and row[7] else ""
            client = str(row[8]).strip() if len(row) > 8 and row[8] else ""
            phone = ""
        if not marking:
            continue
        rec = ClientRef(marking=marking, client=client, phone=phone)
        for v in _norm_marking_variants(marking):
            # Приоритет: главная ведомость выше PART; не затираем уже добавленное из main
            if mode == 'main' or v not in result:
                result[v] = rec
    return result


if __name__ == '__main__':
    # Отладочный запуск
    import sys
    try:
        sys.stdout.reconfigure(encoding='utf-8')
    except Exception:
        pass
    fp = sys.argv[1] if len(sys.argv) > 1 else \
        r'c:\Users\0\Desktop\Задача по логистике\03.07.2026\MTL 700\03.07.2026-700.xlsx'
    ref = load_references(fp)
    print(f'Marking справочник: {len(ref.markings)} записей')
    print(f'Farm справочник:    {len(ref.farms)} записей')
    print(f'Handlers:           {len(ref.handlers_set)} уникальных')
    # Проверка
    for test in ['IZUM', 'BUTA', 'B-BOOM55', 'PATRON', 'MAX', 'FSFL01-CHELNY',
                 'TJ PRADO', 'SUN FLORA', 'ALEX KAZAN']:
        hit = ref.lookup_client(test)
        print(f'  {test:20s} -> {hit.client if hit else "MISS"}')