excel_writer.py - main - Paff/declarant

excel_writer.py
543 строки22 KB
1pash1985-gif
Декларант - веб-версия для Timeweb Cloud
15 июл 2026, 09:36
15 июл 2026, 09:36a6b08b8
"""
Модуль для работы с Excel таблицей (XLSM).
Чтение структуры таблицы и запись данных о коробках.
"""
import os
from typing import Optional
from openpyxl import load_workbook
from openpyxl.workbook import Workbook
from openpyxl.worksheet.worksheet import Worksheet


class ExcelTableManager:
    """Менеджер для работы с Excel таблицей."""

    def __init__(self):
        self.workbook: Optional[Workbook] = None
        self.worksheet: Optional[Worksheet] = None
        self.file_path: str = ""
        
        # Кэшированные данные структуры
        self._flower_rows: dict[str, int] = {}  # flower_name -> row_number (для записи)
        self._client_cols: dict[str, int] = {}  # client_name -> col_number
        self._header_row: int = 1
        self._flower_col: int = 1
        
        # Полный упорядоченный список строк (включая дубликаты)
        self._flower_list: list[dict] = []
        # flower_list[i] = {name_en, name_ru, nl, code, bg_color, kvo_base, row}

    def load_table(self, file_path: str) -> bool:
        """
        Загрузить Excel таблицу.
        
        Args:
            file_path: Путь к XLSM файлу
            
        Returns:
            True если успешно загружен
        """
        try:
            self.file_path = file_path
            # keep_vba=True для сохранения макросов в XLSM
            self.workbook = load_workbook(file_path, keep_vba=True)
            self.worksheet = self.workbook.active
            self._parse_structure()
            return True
        except Exception as e:
            print(f"Ошибка загрузки таблицы: {e}")
            return False

    def _parse_structure(self):
        """Распарсить структуру таблицы (найти строки цветов и колонки клиентов)."""
        if not self.worksheet:
            return
        
        self._flower_rows = {}
        self._client_cols = {}
        self._flower_list = []
        
        # Найти строку заголовка (где имена клиентов)
        self._header_row = self._find_header_row()
        
        # Найти колонку K-VO (колонт E = col 5 в оригинале)
        kvo_col = None
        
        # Найти колонки клиентов в строке заголовка
        if self._header_row > 0:
            for col in range(1, self.worksheet.max_column + 1):
                cell_value = self.worksheet.cell(row=self._header_row, column=col).value
                if cell_value and isinstance(cell_value, str):
                    cell_value = cell_value.strip().upper()
                    if cell_value in ["K-VO", "К-ВО"]:
                        kvo_col = col
                    elif cell_value and cell_value not in ["NL", ""]:
                        self._client_cols[cell_value] = col
        
        # Найти строки с названиями цветов
        for row in range(self._header_row + 1, self.worksheet.max_row + 1):
            # Колонка A (1): английское название
            cell_en = self.worksheet.cell(row=row, column=1).value
            if cell_en and isinstance(cell_en, str):
                flower_en = cell_en.strip()
                if flower_en:
                    # Колонка B (2): русское название
                    cell_ru = self.worksheet.cell(row=row, column=2).value
                    flower_ru = ""
                    if cell_ru and isinstance(cell_ru, str):
                        flower_ru = cell_ru.strip()
                    
                    # Колонка C (3): страна
                    cell_nl = self.worksheet.cell(row=row, column=3).value
                    nl_value = ""
                    if cell_nl and isinstance(cell_nl, str):
                        nl_value = cell_nl.strip()
                    
                    # Колонка D (4): код товара
                    cell_code = self.worksheet.cell(row=row, column=4).value
                    code = ""
                    if cell_code:
                        code = str(cell_code).strip()
                    
                    # Колонка E (5) или kvo_col: К-ВО из базы
                    kvo_base = 0
                    kvo_check_col = kvo_col if kvo_col else 5
                    cell_kvo = self.worksheet.cell(row=row, column=kvo_check_col).value
                    if cell_kvo:
                        try:
                            kvo_base = int(cell_kvo)
                        except (ValueError, TypeError):
                            kvo_base = 0
                    
                    # Цвет фона ячейки
                    bg_color = self._get_cell_bg_color(row, 1)
                    
                    entry = {
                        "name_en": flower_en,
                        "name_ru": flower_ru,
                        "nl": nl_value,
                        "code": code,
                        "bg_color": bg_color,
                        "kvo_base": kvo_base,
                        "row": row
                    }
                    self._flower_list.append(entry)
                    
                    # Также сохранить в dict для быстрого поиска (последняя строка)
                    self._flower_rows[flower_en.upper()] = row
        
        print(f"Структура: {len(self._flower_list)} строк ({len(self._flower_rows)} уникальных), {len(self._client_cols)} клиентов")
        print(f"Клиенты: {list(self._client_cols.keys())}")

    def _get_cell_bg_color(self, row: int, col: int) -> str:
        """Получить цвет фона ячейки в формате #RRGGBB."""
        if not self.worksheet:
            return "#FFFFFF"
        
        cell = self.worksheet.cell(row=row, column=col)
        fill = cell.fill
        
        # Нет заливки — белый фон
        if not fill or fill.patternType is None or fill.patternType == "none":
            return "#FFFFFF"
        
        if not fill.fgColor:
            return "#FFFFFF"
        
        fg = fill.fgColor
        
        # Прямой RGB
        if fg.type == "rgb" and fg.rgb:
            rgb = str(fg.rgb)
            if len(rgb) == 8:  # AARRGGBB
                hex_color = rgb[2:]
                if hex_color == "000000":  # Чёрный = нет заливки
                    return "#FFFFFF"
                return f"#{hex_color}"
            elif len(rgb) == 6:
                return f"#{rgb}"
        
        # Theme color — преобразовать в RGB
        if fg.type == "theme" and fg.theme is not None:
            # Стандартные цвета темы Office
            theme_colors = [
                "FFFFFF",  # 0: Background 1 (white)
                "000000",  # 1: Text 1 (black)
                "EEECE1",  # 2: Background 2
                "1F497D",  # 3: Accent 1 (dark blue)
                "4F81BD",  # 4: Accent 2 (blue)
                "C0504D",  # 5: Accent 3 (red)
                "9BBB59",  # 6: Accent 4 (green)
                "8064A2",  # 7: Accent 5 (purple)
                "4BACC6",  # 8: Accent 6 (teal)
                "F79646",  # 9: Accent 7 (orange)
            ]
            
            if fg.theme < len(theme_colors):
                base_hex = theme_colors[fg.theme]
                # Применить tint (упрощённо)
                tint = fg.tint if fg.tint else 0
                if tint != 0:
                    base_hex = self._apply_tint(base_hex, tint)
                return f"#{base_hex}"
        
        return "#FFFFFF"

    @staticmethod
    def _apply_tint(hex_color: str, tint: float) -> str:
        """Применить tint к hex цвету. Упрощённая версия."""
        try:
            r = int(hex_color[0:2], 16)
            g = int(hex_color[2:4], 16)
            b = int(hex_color[4:6], 16)
            
            if tint < 0:
                # Затемнить
                factor = 1 + tint
                r = int(r * factor)
                g = int(g * factor)
                b = int(b * factor)
            else:
                # Осветлить
                r = int(r + (255 - r) * tint)
                g = int(g + (255 - g) * tint)
                b = int(b + (255 - b) * tint)
            
            r = max(0, min(255, r))
            g = max(0, min(255, g))
            b = max(0, min(255, b))
            return f"{r:02X}{g:02X}{b:02X}"
        except (ValueError, IndexError):
            return hex_color

    def get_flowers_full_data(self) -> dict:
        """Получить расширенные данные о цветах (устаревший метод, используйте get_flowers_ordered_list)."""
        # Для обратной совместимости — возвращает dict
        result = {}
        for entry in self._flower_list:
            result[entry["name_en"].upper()] = entry
        return result

    def get_flowers_ordered_list(self) -> list[dict]:
        """Получить упорядоченный список цветов (включая дубликаты)."""
        return self._flower_list

    def _find_header_row(self) -> int:
        """Найти строку заголовка (где имена клиентов)."""
        if not self.worksheet:
            return 1
        
        # Ищем строку с "K-VO" или известными именами клиентов
        for row in range(1, min(20, self.worksheet.max_row + 1)):
            for col in range(1, min(20, self.worksheet.max_column + 1)):
                cell_value = self.worksheet.cell(row=row, column=col).value
                if cell_value:
                    cell_str = str(cell_value).strip().upper()
                    if cell_str in ["K-VO", "К-ВО", "AIRAT", "CLOV", "ARSK"]:
                        return row
        
        # По умолчанию первая строка
        return 1

    def get_flowers_list(self) -> list[str]:
        """Получить список всех цветов из таблицы."""
        return list(self._flower_rows.keys())

    def get_clients_list(self) -> list[str]:
        """Получить список всех клиентов из таблицы."""
        return list(self._client_cols.keys())

    def find_flower_row(self, flower_name: str) -> Optional[int]:
        """
        Найти номер строки для цветка.
        
        Args:
            flower_name: Название цветка
            
        Returns:
            Номер строки или None
        """
        # Ищем точное совпадение
        flower_upper = flower_name.upper().strip()
        if flower_upper in self._flower_rows:
            return self._flower_rows[flower_upper]
        
        # Ищем частичное совпадение
        for name, row in self._flower_rows.items():
            if flower_upper in name or name in flower_upper:
                return row
        
        return None

    def find_client_column(self, client_name: str) -> Optional[int]:
        """
        Найти номер колонки для клиента.
        
        Args:
            client_name: Имя клиента
            
        Returns:
            Номер колонки или None
        """
        client_upper = client_name.upper().strip()
        if client_upper in self._client_cols:
            return self._client_cols[client_upper]
        
        # Ищем частичное совпадение
        for name, col in self._client_cols.items():
            if client_upper in name or name in client_upper:
                return col
        
        return None

    def get_current_value(self, flower_name: str, client_name: str) -> Optional[int]:
        """
        Получить текущее значение в ячейке.
        
        Args:
            flower_name: Название цветка
            client_name: Имя клиента
            
        Returns:
            Текущее значение или None
        """
        row = self.find_flower_row(flower_name)
        col = self.find_client_column(client_name)
        
        if row and col:
            value = self.worksheet.cell(row=row, column=col).value
            if value is None:
                return 0
            try:
                return int(value)
            except (ValueError, TypeError):
                return 0
        return None

    def set_boxes_count(self, flower_name: str, client_name: str, boxes_count: int, 
                        add_to_existing: bool = True) -> bool:
        """
        Установить количество коробок в ячейку.
        
        Args:
            flower_name: Название цветка
            client_name: Имя клиента
            boxes_count: Количество коробок
            add_to_existing: Если True, добавляет к существующему значению
            
        Returns:
            True если успешно
        """
        row = self.find_flower_row(flower_name)
        col = self.find_client_column(client_name)
        
        if not row or not col:
            print(f"Не найдена ячейка: {flower_name} x {client_name}")
            return False
        
        current_value = 0
        if add_to_existing:
            existing = self.worksheet.cell(row=row, column=col).value
            if existing:
                try:
                    current_value = int(existing)
                except (ValueError, TypeError):
                    current_value = 0
        
        new_value = current_value + boxes_count
        self.worksheet.cell(row=row, column=col).value = new_value
        
        # Обновить K-VO (сумму)
        self._update_kvo(row)
        
        return True

    def clear_all_boxes(self):
        """Очистить все значения коробок в таблице (обнулить все ячейки клиентов)."""
        if not self.worksheet:
            return
        
        for row in self._flower_rows.values():
            for col in self._client_cols.values():
                self.worksheet.cell(row=row, column=col).value = 0
            # Обновить K-VO для каждой строки
            self._update_kvo(row)
        
        print(f"Очищены все коробки в таблице ({len(self._flower_rows)} цветов x {len(self._client_cols)} клиентов)")

    def fill_from_session(self, session_data: list) -> int:
        """
        Заполнить таблицу данными из сессии.
        
        Args:
            session_data: Список словарей [{flower_name, client_name, boxes_count}]
            
        Returns:
            Количество успешно записанных ячеек
        """
        success = 0
        for entry in session_data:
            flower = entry.get("flower_name", "")
            client = entry.get("client_name", "")
            boxes = entry.get("boxes_count", 0)
            
            if boxes > 0 and self.set_boxes_count(flower, client, boxes, add_to_existing=False):
                success += 1
        
        return success

    def write_session_to_base(self, session_data: list) -> tuple:
        """
        Записать данные сессии в базовую таблицу.
        Сначала очищает все коробки, затем заполняет из сессии.
        
        Args:
            session_data: Список словарей [{flower_name, client_name, boxes_count}]
            
        Returns:
            (success_count, error_count)
        """
        # Очистить таблицу
        self.clear_all_boxes()
        
        # Заполнить из сессии
        success = self.fill_from_session(session_data)
        errors = len(session_data) - success
        
        return (success, errors)

    def _update_kvo(self, row: int):
        """Обновить колонку K-VO (сумма по строке)."""
        if not self.worksheet:
            return
        
        # Найти колонку K-VO
        kvo_col = None
        for col in range(1, self.worksheet.max_column + 1):
            cell_value = self.worksheet.cell(row=self._header_row, column=col).value
            if cell_value and str(cell_value).strip().upper() in ["K-VO", "К-ВО"]:
                kvo_col = col
                break
        
        if kvo_col:
            # Посчитать сумму по всем клиентским колонкам
            total = 0
            for col in self._client_cols.values():
                value = self.worksheet.cell(row=row, column=col).value
                if value:
                    try:
                        total += int(value)
                    except (ValueError, TypeError):
                        pass
            self.worksheet.cell(row=row, column=kvo_col).value = total

    def save(self, file_path: Optional[str] = None) -> bool:
        """
        Сохранить таблицу.
        
        Args:
            file_path: Путь для сохранения (если None, сохраняет в исходный файл)
            
        Returns:
            True если успешно
        """
        try:
            save_path = file_path or self.file_path
            if self.workbook:
                self.workbook.save(save_path)
                return True
            return False
        except Exception as e:
            print(f"Ошибка сохранения: {e}")
            return False

    def preview_results(self, results: list[dict]) -> str:
        """
        Создать текстовый предпросмотр результатов перед записью.
        
        Args:
            results: Список словарей с результатами
            
        Returns:
            Текстовое представление
        """
        lines = []
        lines.append("=" * 60)
        lines.append("Предпросмотр записи в таблицу:")
        lines.append("=" * 60)
        
        for res in results:
            client = res.get("client", "")
            flower = res.get("flower", "")
            boxes = res.get("boxes", 0)
            current = self.get_current_value(flower, client)
            
            row = self.find_flower_row(flower)
            col = self.find_client_column(client)
            
            status = "✅" if row and col else "❌"
            current_str = str(current) if current is not None else "N/A"
            
            lines.append(f"{status} {client:15} | {flower:20} | +{boxes:3} | было: {current_str}")
        
        lines.append("=" * 60)
        return "\n".join(lines)


class ResultAggregator:
    """Агрегатор результатов парсинга для записи в таблицу."""

    @staticmethod
    def aggregate_by_client_flower(scan_results: dict) -> list[dict]:
        """
        Агрегировать результаты парсинга по клиент+цветок.
        
        Args:
            scan_results: Результаты от FolderScanner {client: [InvoiceData]}
            
        Returns:
            Список словарей [{client, flower, boxes, source, file_paths}]
        """
        aggregated = []
        
        for client_name, invoices in scan_results.items():
            # Собираем все позиции по всем инвойсам клиента
            flower_boxes: dict[str, int] = {}
            flower_stems: dict[str, int] = {}
            source_folders = set()
            file_paths = []  # Список путей к файлам
            
            for invoice in invoices:
                source_folders.add(os.path.dirname(invoice.file_path))
                file_paths.append(invoice.file_path)  # Сохраняем путь к файлу
                
                for item in invoice.items:
                    # Используем сопоставленное имя или исходное
                    flower_name = item.matched_flower or item.raw_name
                    if flower_name:
                        if flower_name not in flower_boxes:
                            flower_boxes[flower_name] = 0
                            flower_stems[flower_name] = 0
                        flower_boxes[flower_name] += item.boxes_qty
                        flower_stems[flower_name] += item.stems_qty
            
            # Если нет позиций, используем total_boxes
            if not flower_boxes and any(inv.total_boxes > 0 for inv in invoices):
                total = sum(inv.total_boxes for inv in invoices)
                total_s = sum(inv.total_stems for inv in invoices)
                flower_boxes["НЕИЗВЕСТНО"] = total
                flower_stems["НЕИЗВЕСТНО"] = total_s
            
            # Создаем записи
            for flower_name, boxes in flower_boxes.items():
                aggregated.append({
                    "client": client_name,
                    "flower": flower_name,
                    "boxes": boxes,
                    "stems": flower_stems.get(flower_name, 0),
                    "source": ", ".join(source_folders),
                    "file_paths": file_paths,  # Добавляем пути к файлам
                    "is_matched": flower_name != "НЕИЗВЕСТНО"
                })
        
        return aggregated