corrections.py - main - Paff/declarant

corrections.py
539 строк23 KB
1pash1985-gif
feat: AI Ensemble - DeepSeek + Kimi parallel parsing with debates
27 июл 2026, 11:14
27 июл 2026, 11:145ea7945
# -*- coding: utf-8 -*-
"""
Система обучения — глобальные пользовательские правила (corrections.json).

Приоритет: corrections > встроенная автодетект-логика.
При каждом ручном исправлении в GUI — автоматическое добавление правила.
"""
from __future__ import annotations

import json
import uuid
from dataclasses import dataclass, field
from datetime import datetime
from pathlib import Path
from typing import Optional

# Файл по умолчанию — рядом с программой
_DEFAULT_PATH = Path(__file__).parent / 'corrections.json'


@dataclass
class Corrections:
    """In-memory представление базы правил."""

    # marking_aliases: распознанный маркинг → правильный (или null = игнорировать)
    marking_aliases: dict[str, Optional[str]] = field(default_factory=dict)

    # marking_to_column: маркинг → имя колонки europa-шаблона
    marking_to_column: dict[str, str] = field(default_factory=dict)

    # variety_to_family: сорт (upper) → семейство (ROSA, CHRYSANTHEMUM, ...)
    variety_to_family: dict[str, str] = field(default_factory=dict)

    # filename_to_marking: подстрока имени файла → маркинг
    filename_to_marking: dict[str, str] = field(default_factory=dict)

    # flower_synonyms: сокращение → полное имя (HYD → HYDRANGEA, CARN → CARNATION)
    flower_synonyms: dict[str, str] = field(default_factory=dict)

    # supplier_rules: поставщик (upper) → {country, default_flower, hs_code}
    supplier_rules: dict[str, dict] = field(default_factory=dict)

    # history: история всех корректировок (для AI-обучения и Excel-отчёта)
    history: list[dict] = field(default_factory=list)

    # output_history: правки в СФОРМИРОВАННЫХ Excel-документах
    # (europa/import/florunner/other). Отдельно, чтобы AI-мапперы могли
    # брать only свои примеры и не смешивать с recognition-правками.
    output_history: list[dict] = field(default_factory=list)

    # --- Применение правил ---

    def resolve_marking(self, raw_marking: str) -> Optional[str]:
        """Применить alias. Возвращает None если маркинг должен игнорироваться."""
        key = (raw_marking or '').strip().upper()
        if key in self.marking_aliases:
            v = self.marking_aliases[key]
            return v  # None = skip, str = замена
        return raw_marking  # без изменений

    def resolve_column(self, marking: str) -> Optional[str]:
        """Найти колонку europa для данного маркинга."""
        key = (marking or '').strip().upper()
        return self.marking_to_column.get(key)

    def resolve_family(self, variety: str) -> Optional[str]:
        """Найти семейство для сорта."""
        key = (variety or '').strip().upper()
        return self.variety_to_family.get(key)

    def resolve_marking_from_filename(self, filename: str) -> Optional[str]:
        """Найти маркинг по подстроке в имени файла."""
        fname_up = (filename or '').upper()
        for pattern, marking in self.filename_to_marking.items():
            if pattern.upper() in fname_up:
                return marking
        return None

    def resolve_flower_name(self, raw_name: str) -> str:
        """Применить выученные синонимы к имени цветка.
        
        Проверяет как точное совпадение всего имени, так и первое слово.
        Пример: 'HYD LIGHT BLUE JUMBO' → первое слово 'HYD' → 'HYDRANGEA'
        """
        key = (raw_name or '').strip().upper()
        # Точное совпадение
        if key in self.flower_synonyms:
            return self.flower_synonyms[key]
        # Первое слово как сокращение
        tokens = key.split()
        if tokens and tokens[0] in self.flower_synonyms:
            return self.flower_synonyms[tokens[0]]
        return raw_name  # без изменений

    def get_supplier_rule(self, supplier: str) -> dict:
        """Получить правила для конкретного поставщика."""
        key = (supplier or '').strip().upper()
        # Точное совпадение
        if key in self.supplier_rules:
            return self.supplier_rules[key]
        # Подстрока
        for sup_key, rules in self.supplier_rules.items():
            if sup_key in key or key in sup_key:
                return rules
        return {}

    def get_corrections_for_ai(self, max_items: int = 30) -> list[dict]:
        """Получить последние корректировки для включения в AI-промпт.
        
        Возвращает наиболее частые/недавние записи из history.
        """
        if not self.history:
            return []
        # Сортируем по частоте применения (applied_count), затем по дате
        sorted_hist = sorted(
            self.history,
            key=lambda x: (x.get('applied_count', 0), x.get('timestamp', '')),
            reverse=True,
        )
        return sorted_hist[:max_items]

    def build_rules_from_corrections(self, min_applied_count: int = 1) -> dict[str, Any]:
        """Построить индекс правил из истории корректировок.

        Возвращает dict:
          - field_exact: {field_name: {original_upper: corrected_value}}
          - text_substring: [(pattern, replacement, count), ...] — только для
            коротких однострочных значений (≤40 симв.), частота ≥ min_applied_count.
          - supplier_field: {supplier_upper: {field_name: corrected_value}}
          - filename_to_marking: уже хранится в объекте
        """
        rules: dict[str, Any] = {
            'field_exact': {},
            'text_substring': [],
            'supplier_field': {},
        }
        for h in self.history:
            count = h.get('applied_count', 1)
            if count < min_applied_count:
                continue
            field = h.get('field', '')
            original = str(h.get('original_value', '')).strip()
            corrected = str(h.get('corrected_value', '')).strip()
            supplier = str(h.get('supplier', '')).strip().upper()
            if not field or not original or not corrected or original == corrected:
                continue

            # Field exact rules
            rules['field_exact'].setdefault(field, {})
            rules['field_exact'][field][original.upper()] = corrected

            # Text substring rules: short single-line values only
            if len(original) <= 40 and '\n' not in original and original:
                rules['text_substring'].append((original, corrected, count))

            # Supplier context rules
            if supplier:
                rules['supplier_field'].setdefault(supplier, {})
                rules['supplier_field'][supplier][field] = corrected

        # Sort substring rules by frequency (desc) and length (desc) to apply
        # longer/more frequent patterns first.
        rules['text_substring'].sort(key=lambda x: (-x[2], -len(x[0])))
        return rules

    def apply_field_rules(self, rules: dict[str, Any], parsed: dict[str, Any]) -> dict[str, Any]:
        """Apply learned exact-field rules to a parsed document in-place.

        Returns the same dict (mutated).
        """
        field_exact = rules.get('field_exact') or {}
        supplier_field = rules.get('supplier_field') or {}

        supplier = str(parsed.get('supplier') or '').strip().upper()
        if supplier and supplier in supplier_field:
            for field, corrected in supplier_field[supplier].items():
                if field in parsed and parsed[field] is None:
                    parsed[field] = corrected

        for field, mapping in field_exact.items():
            current = parsed.get(field)
            if current is None:
                continue
            key = str(current).strip().upper()
            if key in mapping:
                parsed[field] = mapping[key]

        # Apply to items[] arrays if the field exists there
        for field, mapping in field_exact.items():
            for item in parsed.get('items') or []:
                if field not in item:
                    continue
                current = item[field]
                if current is None:
                    continue
                key = str(current).strip().upper()
                if key in mapping:
                    item[field] = mapping[key]

        return parsed

    # --- Мутации (из GUI) ---

    def set_marking_alias(self, raw: str, corrected: Optional[str]):
        self.marking_aliases[raw.strip().upper()] = corrected

    def set_column_mapping(self, marking: str, column: str):
        self.marking_to_column[marking.strip().upper()] = column.strip().upper()

    def set_family(self, variety: str, family: str):
        self.variety_to_family[variety.strip().upper()] = family.strip().upper()

    def set_filename_marking(self, pattern: str, marking: str):
        self.filename_to_marking[pattern.strip()] = marking.strip()

    def set_flower_synonym(self, abbreviation: str, full_name: str):
        """Добавить синоним цветка (HYD → HYDRANGEA)."""
        self.flower_synonyms[abbreviation.strip().upper()] = full_name.strip().upper()

    def set_supplier_rule(self, supplier: str, **kwargs):
        """Установить правило для поставщика.
        
        kwargs: country, default_flower, hs_code, etc.
        """
        key = supplier.strip().upper()
        if key not in self.supplier_rules:
            self.supplier_rules[key] = {}
        self.supplier_rules[key].update(kwargs)

    def add_correction(self, source_file: str, supplier: str,
                       correction_type: str, field_name: str,
                       original_value: str, corrected_value: str,
                       source_text: str = ''):
        """Записать корректировку в историю.
        
        Args:
            source_file: имя файла-источника
            supplier: поставщик
            correction_type: 'recognition' | 'document' | 'synonym'
            field_name: 'flower_name' | 'stems_qty' | 'boxes_qty' | 'hs_code' | 'country' | ...
            original_value: что было
            corrected_value: что стало
            source_text: фрагмент исходного текста из PDF/Excel (для контекста AI)
        """
        # Проверяем дубликат (обновляем applied_count)
        for h in self.history:
            if (h.get('field') == field_name and
                    h.get('original_value') == original_value and
                    h.get('corrected_value') == corrected_value):
                h['applied_count'] = h.get('applied_count', 1) + 1
                h['timestamp'] = datetime.now().isoformat()
                # Обновляем source_text если предоставлен более полный
                if source_text and len(source_text) > len(h.get('source_text', '')):
                    h['source_text'] = source_text[:500]
                return

        entry = {
            'id': str(uuid.uuid4())[:8],
            'timestamp': datetime.now().isoformat(),
            'source_file': source_file,
            'supplier': supplier,
            'type': correction_type,
            'field': field_name,
            'original_value': original_value,
            'corrected_value': corrected_value,
            'applied_count': 1,
        }
        if source_text:
            entry['source_text'] = source_text[:500]
        self.history.append(entry)

    # --- Правки сформированных документов (europa/import/florunner) ---

    def add_output_correction(self, document_type: str, sheet: str,
                              row: int, col: int,
                              original_value, corrected_value,
                              context: dict | None = None,
                              source_file: str = ''):
        """Записать правку ячейки сформированного Excel-документа.

        Args:
            document_type: 'europa' | 'import' | 'florunner' | 'other'
            sheet: имя листа Excel
            row, col: индексы ячейки (0-based) для восстановления диффа
            original_value / corrected_value: значения до/после
            context: {'marking', 'family', 'client', 'country', 'row_header',
                     'col_header', 'source_hint', ...} — область применимости
            source_file: имя output-файла (для traceability)

        Дедупликация: тот же документ+row_header+col_header+было+стало →
        инкремент applied_count и обновление timestamp.
        """
        ctx = context or {}
        row_header = ctx.get('row_header', '')
        col_header = ctx.get('col_header', '')

        for h in self.output_history:
            if (h.get('document_type') == document_type and
                    h.get('row_header') == row_header and
                    h.get('col_header') == col_header and
                    str(h.get('original_value')) == str(original_value) and
                    str(h.get('corrected_value')) == str(corrected_value)):
                h['applied_count'] = h.get('applied_count', 1) + 1
                h['timestamp'] = datetime.now().isoformat()
                # Обновляем контекст если добавились подробности
                for k, v in ctx.items():
                    if v and not h.get(k):
                        h[k] = v
                return

        entry = {
            'id': str(uuid.uuid4())[:8],
            'timestamp': datetime.now().isoformat(),
            'document_type': document_type,
            'source_file': source_file,
            'sheet': sheet,
            'row': row,
            'col': col,
            'row_header': row_header,
            'col_header': col_header,
            'original_value': original_value,
            'corrected_value': corrected_value,
            'applied_count': 1,
        }
        # Всё дополнительное из context (marking, family, client, country, …)
        for k, v in ctx.items():
            if k not in entry and v not in (None, ''):
                entry[k] = v
        self.output_history.append(entry)

        # Дополнительно: если правка явно даёт маппинг маркировка→колонка
        # или сорт→семейство — заносим в «золотые» словари для быстрого применения.
        if document_type == 'europa':
            marking = ctx.get('marking') or row_header
            column = ctx.get('col_header') or col_header
            if marking and column and marking.strip() and column.strip():
                self.marking_to_column[str(marking).strip().upper()] = str(column).strip()
            variety = ctx.get('variety') or ctx.get('flower_variety')
            family = ctx.get('family') or row_header
            if variety and family:
                self.variety_to_family[str(variety).strip().upper()] = str(family).strip().upper()

    def get_output_corrections_for_ai(self, document_type: str,
                                      max_items: int = 20) -> list[dict]:
        """Отфильтровать output_history по типу документа для few-shot AI.

        Сортировка: по applied_count (частота) DESC, затем timestamp DESC.
        """
        if not self.output_history:
            return []
        filtered = [h for h in self.output_history
                    if h.get('document_type') == document_type]
        filtered.sort(
            key=lambda x: (x.get('applied_count', 0), x.get('timestamp', '')),
            reverse=True,
        )
        return filtered[:max_items]


def load_corrections(path: Optional[str] = None) -> Corrections:
    """Загрузить из JSON. Если файл не существует — пустой объект."""
    p = Path(path) if path else _DEFAULT_PATH
    if not p.exists():
        return Corrections()
    try:
        with open(p, 'r', encoding='utf-8') as f:
            data = json.load(f)
    except (json.JSONDecodeError, OSError):
        return Corrections()
    return Corrections(
        marking_aliases=data.get('marking_aliases', {}),
        marking_to_column=data.get('marking_to_column', {}),
        variety_to_family=data.get('variety_to_family', {}),
        filename_to_marking=data.get('filename_to_marking', {}),
        flower_synonyms=data.get('flower_synonyms', {}),
        supplier_rules=data.get('supplier_rules', {}),
        history=data.get('history', []),
        output_history=data.get('output_history', []),
    )


def save_corrections(corr: Corrections, path: Optional[str] = None):
    """Сохранить в JSON (pretty-print, UTF-8)."""
    p = Path(path) if path else _DEFAULT_PATH
    data = {
        'marking_aliases': corr.marking_aliases,
        'marking_to_column': corr.marking_to_column,
        'variety_to_family': corr.variety_to_family,
        'filename_to_marking': corr.filename_to_marking,
        'flower_synonyms': corr.flower_synonyms,
        'supplier_rules': corr.supplier_rules,
        'history': corr.history,
        'output_history': corr.output_history,
    }
    with open(p, 'w', encoding='utf-8') as f:
        json.dump(data, f, ensure_ascii=False, indent=2)


def export_to_excel(corr: Corrections, path: Optional[str] = None):
    """Экспортировать corrections в Excel для ручного редактирования.
    
    Создаёт .xlsx с листами: Корректировки, Синонимы, Правила поставщиков.
    """
    from openpyxl import Workbook

    excel_path = Path(path) if path else _DEFAULT_PATH.with_suffix('.xlsx')
    wb = Workbook()

    # --- Лист 1: История корректировок ---
    ws1 = wb.active
    ws1.title = 'Корректировки'
    ws1.append(['ID', 'Дата', 'Файл', 'Поставщик', 'Тип', 'Поле',
                'Было', 'Стало', 'Кол-во применений'])
    for h in corr.history:
        ws1.append([
            h.get('id', ''),
            h.get('timestamp', ''),
            h.get('source_file', ''),
            h.get('supplier', ''),
            h.get('type', ''),
            h.get('field', ''),
            h.get('original_value', ''),
            h.get('corrected_value', ''),
            h.get('applied_count', 0),
        ])

    # --- Лист 2: Синонимы цветов ---
    ws2 = wb.create_sheet('Синонимы')
    ws2.append(['Сокращение', 'Полное название'])
    for abbr, full in corr.flower_synonyms.items():
        ws2.append([abbr, full])
    # Также variety_to_family
    ws2.append([])
    ws2.append(['--- Сорт → Семейство ---'])
    ws2.append(['Сорт', 'Семейство'])
    for var, fam in corr.variety_to_family.items():
        ws2.append([var, fam])

    # --- Лист 3: Правила поставщиков ---
    ws3 = wb.create_sheet('Поставщики')
    ws3.append(['Поставщик', 'Страна', 'Цветок по умолч.', 'HS код'])
    for sup, rules in corr.supplier_rules.items():
        ws3.append([
            sup,
            rules.get('country', ''),
            rules.get('default_flower', ''),
            rules.get('hs_code', ''),
        ])

    # --- Лист 4: Маркировки ---
    ws4 = wb.create_sheet('Маркировки')
    ws4.append(['Маркинг', '→ Колонка Europa'])
    for mark, col in corr.marking_to_column.items():
        ws4.append([mark, col])
    ws4.append([])
    ws4.append(['--- Алиасы ---'])
    ws4.append(['Исходный', '→ Замена (пусто = игнорировать)'])
    for raw, alias in corr.marking_aliases.items():
        ws4.append([raw, alias or '(игнор)'])

    wb.save(str(excel_path))
    return str(excel_path)


def import_from_excel(path: str, corr: Optional[Corrections] = None) -> Corrections:
    """Импортировать corrections из Excel.
    
    Мержит данные из Excel в существующий объект Corrections.
    """
    from openpyxl import load_workbook

    if corr is None:
        corr = Corrections()

    wb = load_workbook(path, read_only=True, data_only=True)

    # --- Лист Синонимы ---
    if 'Синонимы' in wb.sheetnames:
        ws = wb['Синонимы']
        in_synonyms = True
        for row in ws.iter_rows(min_row=2, values_only=True):
            if not row or not row[0]:
                continue
            key = str(row[0]).strip()
            if key.startswith('---'):
                in_synonyms = False
                continue
            if key in ('Сокращение', 'Сорт'):
                continue
            val = str(row[1]).strip() if row[1] else ''
            if not val:
                continue
            if in_synonyms:
                corr.flower_synonyms[key.upper()] = val.upper()
            else:
                corr.variety_to_family[key.upper()] = val.upper()

    # --- Лист Поставщики ---
    if 'Поставщики' in wb.sheetnames:
        ws = wb['Поставщики']
        for row in ws.iter_rows(min_row=2, values_only=True):
            if not row or not row[0]:
                continue
            sup = str(row[0]).strip().upper()
            rules = {}
            if row[1]:
                rules['country'] = str(row[1]).strip()
            if row[2]:
                rules['default_flower'] = str(row[2]).strip()
            if row[3]:
                rules['hs_code'] = str(row[3]).strip()
            if rules:
                corr.supplier_rules[sup] = rules

    # --- Лист Маркировки ---
    if 'Маркировки' in wb.sheetnames:
        ws = wb['Маркировки']
        in_mappings = True
        for row in ws.iter_rows(min_row=2, values_only=True):
            if not row or not row[0]:
                continue
            key = str(row[0]).strip()
            if key.startswith('---'):
                in_mappings = False
                continue
            if key in ('Маркинг', 'Исходный'):
                continue
            val = str(row[1]).strip() if row[1] else ''
            if in_mappings:
                if val:
                    corr.marking_to_column[key.upper()] = val.upper()
            else:
                if val == '(игнор)':
                    corr.marking_aliases[key.upper()] = None
                elif val:
                    corr.marking_aliases[key.upper()] = val

    wb.close()
    return corr