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" allow_direct_mapping: bool = True @dataclass(frozen=True) class StructuredGroupRule: structured_column: str one_line_columns: tuple[str, ...] component_columns: tuple[str, ...] join_with_space: bool = False MAPPING_DATA_FILE = Path(__file__).resolve().with_name("mapping_table.json") TARGET_REPORT_COLUMNS_COUNT = 204 _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 is_structured_target = _is_structured_column(column) 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) non_one_line_candidates = [ candidate for candidate in candidates if not _is_one_line_candidate(candidate) ] one_line_candidates = [ candidate for candidate in candidates if _is_one_line_candidate(candidate) ] direct_candidates = candidates allow_direct_mapping = True if is_structured_target and one_line_candidates and not non_one_line_candidates: allow_direct_mapping = False if is_structured_target and non_one_line_candidates: direct_candidates = non_one_line_candidates rules.append( MappingRule( column_name=column, candidate_keys=tuple(direct_candidates), source=( "participant" if column in participant_columns or column.startswith("ИНН") else "any" ), allow_direct_mapping=allow_direct_mapping, ) ) return tuple(rules) def _is_one_line_column(column_name: str) -> bool: normalized = column_name.lower().replace("ё", "е") return "одной строк" in normalized def _is_one_line_candidate(candidate: str) -> bool: return "строка" in candidate.lower() def _is_structured_column(column_name: str) -> bool: normalized = column_name.lower().replace("ё", "е") return "структур" in normalized or "(целый)" in normalized def _split_paths(path: str) -> tuple[str, ...]: result: list[str] = [] for chunk in path.split("|"): candidate = chunk.strip() if candidate: result.append(candidate.rstrip("/")) return tuple(result) def _common_prefix_depth(path_a: str, path_b: str) -> int: parts_a = [part for part in path_a.strip("/").split("/") if part] parts_b = [part for part in path_b.strip("/").split("/") if part] depth = 0 for left, right in zip(parts_a, parts_b): if left != right: break depth += 1 return depth def _paths_related( struct_paths: tuple[str, ...], candidate_paths: tuple[str, ...] ) -> bool: if not struct_paths or not candidate_paths: return False for struct_path in struct_paths: for candidate_path in candidate_paths: if _common_prefix_depth(struct_path, candidate_path) >= 6: return True return False def _build_structured_group_rules( mapping_data: list[dict[str, str | int]], ) -> tuple[StructuredGroupRule, ...]: by_index: dict[int, dict[str, str | int]] = { int(item["index"]): item for item in mapping_data } sorted_indexes = sorted(by_index.keys()) rules: list[StructuredGroupRule] = [] for index in sorted_indexes: item = by_index[index] structured_name = str(item.get("column_name", "")).strip() if not _is_structured_column(structured_name): continue struct_paths = _split_paths(str(item.get("xml_path", ""))) one_line_names: list[str] = [] component_names: list[str] = [] # Left-side "one line" can be immediately before structured column. left = by_index.get(index - 1) if left: left_name = str(left.get("column_name", "")).strip() if _is_one_line_column(left_name): left_paths = _split_paths(str(left.get("xml_path", ""))) if _paths_related(struct_paths, left_paths): one_line_names.append(left_name) right_boundary = next( ( candidate for candidate in sorted_indexes if candidate > index and _is_structured_column( str(by_index[candidate].get("column_name", "")).strip() ) ), None, ) block_end = right_boundary or (sorted_indexes[-1] + 1) for right_index in sorted_indexes: if right_index <= index or right_index >= block_end: continue right = by_index[right_index] right_name = str(right.get("column_name", "")).strip() right_tag = str(right.get("xml_tag", "")).strip() right_paths = _split_paths(str(right.get("xml_path", ""))) if not _paths_related(struct_paths, right_paths): if component_names: break continue if _is_one_line_column(right_name): one_line_names.append(right_name) continue if "|" in right_tag: continue component_names.append(right_name) if not component_names: continue rules.append( StructuredGroupRule( structured_column=structured_name, one_line_columns=tuple(dict.fromkeys(one_line_names)), component_columns=tuple(dict.fromkeys(component_names)), join_with_space="фио" in structured_name.lower(), ) ) 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) STRUCTURED_GROUP_RULES: tuple[StructuredGroupRule, ...] = _build_structured_group_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 not rule.allow_direct_mapping: row[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 if _is_structured_column(rule.column_name): row[rule.column_name] = _pick_structured_value( source_payload=source_payload, suffix_index=suffix_index, candidates=rule.candidate_keys, join_with_space="фио" in rule.column_name.lower(), ) else: row[rule.column_name] = _pick_value( source_payload, suffix_index, rule.candidate_keys ) _apply_structured_group_rules(row) for column in FIXED_REPORT_COLUMNS: row.setdefault(column, "") return row def _apply_structured_group_rules(row: dict[str, str]) -> None: for rule in STRUCTURED_GROUP_RULES: one_line_has_value = any( str(row.get(column_name, "")).strip() for column_name in rule.one_line_columns ) if one_line_has_value: row[rule.structured_column] = "" for component_column in rule.component_columns: row[component_column] = "" continue values = [ str(row.get(component_column, "")).strip() for component_column in rule.component_columns if str(row.get(component_column, "")).strip() ] if not values: row[rule.structured_column] = "" continue separator = " " if rule.join_with_space else ", " row[rule.structured_column] = separator.join(values) 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 "" def _pick_structured_value( *, source_payload: dict[str, str], suffix_index: dict[str, str], candidates: tuple[str, ...], join_with_space: bool, ) -> str: values: list[str] = [] for key in candidates: value = _pick_value(source_payload, suffix_index, (key,)) if value and value not in values: values.append(value) if not values: return "" separator = " " if join_with_space else ", " return separator.join(values)