417 lines
14 KiB
Python
417 lines
14 KiB
Python
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)
|