from __future__ import annotations import json import re from dataclasses import dataclass from pathlib import Path @dataclass(frozen=True) class MappingRule: column_name: str candidate_keys: tuple[str, ...] source: str = "any" MAPPING_DATA_FILE = Path(__file__).resolve().with_name("mapping_table.json") TARGET_REPORT_COLUMNS_COUNT = 203 _MANUAL_COLUMN_NAME_FIXES: dict[int, str] = { 1: "Имя XML файла", 178: "Место государственной регистрации ЕИО/Бенефициара (структура полностью)", 183: "Номер", } _INVALID_TAGS = {"", "-", "Источник", "путь", "подразумеваются", "тэга", "нашла"} def _load_mapping_data() -> list[dict[str, str | int]]: raw = json.loads(MAPPING_DATA_FILE.read_text(encoding="utf-8")) if not isinstance(raw, list): return [] result: list[dict[str, str | int]] = [] for item in raw: if not isinstance(item, dict): continue idx = int(item.get("index", 0)) if not 1 <= idx <= TARGET_REPORT_COLUMNS_COUNT: continue column_name = str(item.get("column_name", "")).strip() xml_tag = str(item.get("xml_tag", "")).strip() xml_path = str(item.get("xml_path", "")).strip() if idx in _MANUAL_COLUMN_NAME_FIXES: column_name = _MANUAL_COLUMN_NAME_FIXES[idx] column_name = _cleanup_column_name(column_name) result.append( { "index": idx, "column_name": column_name, "xml_tag": xml_tag, "xml_path": xml_path, } ) return sorted(result, key=lambda row: int(row["index"])) def _cleanup_column_name(value: str) -> str: cleaned = re.sub(r"\s+", " ", value).strip() cleaned = re.sub(r"\s+не\s+нашла.*$", "", cleaned, flags=re.IGNORECASE) return cleaned.strip(" -") def _build_fixed_columns(mapping_data: list[dict[str, str | int]]) -> tuple[str, ...]: columns: list[str] = [] for row in mapping_data: column = str(row.get("column_name", "")).strip() columns.append(column or f"Колонка {row['index']}") if len(columns) < TARGET_REPORT_COLUMNS_COUNT: for index in range(len(columns) + 1, TARGET_REPORT_COLUMNS_COUNT + 1): columns.append(f"Колонка {index}") return tuple(columns[:TARGET_REPORT_COLUMNS_COUNT]) def _build_rules(mapping_data: list[dict[str, str | int]]) -> tuple[MappingRule, ...]: rules: list[MappingRule] = [] participant_columns = { "Признак резидента участника", "Тип участника", "ИНН участника", "Наименование участника", "Счет участника", } for row in mapping_data: column = str(row.get("column_name", "")).strip() tag = str(row.get("xml_tag", "")).strip() path = str(row.get("xml_path", "")).strip() if not column or tag in _INVALID_TAGS: continue candidates = list(_extract_tag_candidate_keys(tag)) if column.startswith("ИНН"): candidates = ["ИННФЛИП", "ИННФЛ", "ИННЮЛ", "ИНН", *candidates] if column == "Тип участника": candidates = ["Тип", "ТипУчастника", *candidates] if path: for path_tag in _extract_path_candidate_keys(path): if path_tag not in candidates: candidates.append(path_tag) rules.append( MappingRule( column_name=column, candidate_keys=tuple(candidates), source=( "participant" if column in participant_columns or column.startswith("ИНН") else "any" ), ) ) return tuple(rules) def _extract_tag_candidate_keys(tag: str) -> tuple[str, ...]: candidates: list[str] = [] for raw_tag in re.split(r"[|,]", tag): cleaned_tag = raw_tag.strip() if not cleaned_tag or cleaned_tag in _INVALID_TAGS: continue if cleaned_tag not in candidates: candidates.append(cleaned_tag) return tuple(candidates) def _extract_path_candidate_keys(path: str) -> tuple[str, ...]: candidates: list[str] = [] for raw_path in re.findall(r"/[^\s|]+", path): parts = [part for part in raw_path.strip("/").split("/") if part] if not parts: continue dotted_path = ".".join(parts) if dotted_path and dotted_path not in candidates: candidates.append(dotted_path) leaf_tag = parts[-1] if leaf_tag and leaf_tag not in _INVALID_TAGS and leaf_tag not in candidates: candidates.append(leaf_tag) return tuple(candidates) _MAPPING_DATA = _load_mapping_data() FIXED_REPORT_COLUMNS: tuple[str, ...] = _build_fixed_columns(_MAPPING_DATA) REPORT_MAPPING_RULES: tuple[MappingRule, ...] = _build_rules(_MAPPING_DATA) def build_fixed_row( *, file_name: str, record_id: str, operation_index: int, operation_fields: dict[str, str], participant_fields: dict[str, str], ) -> dict[str, str]: row: dict[str, str] = { "Имя XML файла": file_name, "Идентификатор записи": record_id, "Уникальный номер операции": str(operation_index), } merged = _merge_fields(operation_fields, participant_fields) merged_suffix_index = _build_suffix_index(merged) participant_suffix_index = _build_suffix_index(participant_fields) for rule in REPORT_MAPPING_RULES: if row.get(rule.column_name): continue if rule.source == "participant": source_payload = participant_fields suffix_index = participant_suffix_index else: source_payload = merged suffix_index = merged_suffix_index row[rule.column_name] = _pick_value( source_payload, suffix_index, rule.candidate_keys ) for column in FIXED_REPORT_COLUMNS: row.setdefault(column, "") return row def _merge_fields( operation_fields: dict[str, str], participant_fields: dict[str, str], ) -> dict[str, str]: merged: dict[str, str] = {} merged.update(operation_fields) merged.update(participant_fields) return merged def _build_suffix_index(payload: dict[str, str]) -> dict[str, str]: index: dict[str, str] = {} for key, value in payload.items(): if not value: continue short = key.rsplit(".", maxsplit=1)[-1] index.setdefault(short, value) return index def _pick_value( payload: dict[str, str], suffix_index: dict[str, str], candidates: tuple[str, ...], ) -> str: for key in candidates: if key in payload and payload[key]: return payload[key] short = key.rsplit(".", maxsplit=1)[-1] if short in payload and payload[short]: return payload[short] if short in suffix_index: return suffix_index[short] return ""