389 lines
13 KiB
Python
389 lines
13 KiB
Python
from __future__ import annotations
|
||
|
||
import re
|
||
import xml.etree.ElementTree as ET
|
||
from collections import Counter
|
||
from dataclasses import dataclass, field
|
||
from typing import Iterable
|
||
|
||
from .config import validate_file_name as validate_pipeline_file_name
|
||
|
||
_INDEXED_SEGMENT = re.compile(r"(?P<segment>[^.]+\[(?P<index>\d+)\])")
|
||
_EIO_SEGMENT = re.compile(
|
||
r"(?P<segment>"
|
||
r"(?:СведЕИО|БенефициарЮЛ|БенефициарФЛИП|БенефициарИНБОЮЛ)"
|
||
r"(?:\[(?P<index>\d+)\])?"
|
||
r")"
|
||
)
|
||
|
||
|
||
@dataclass
|
||
class ParticipantRow:
|
||
file_name: str
|
||
operation_index: int
|
||
record_id: str
|
||
operation_fields: dict[str, str]
|
||
participant_fields: dict[str, str]
|
||
is_continuation: bool = False
|
||
|
||
|
||
@dataclass
|
||
class OperationParseError:
|
||
operation_index: int
|
||
record_id: str
|
||
reason: str
|
||
|
||
|
||
@dataclass
|
||
class FileParseResult:
|
||
file_name: str
|
||
rows: list[ParticipantRow] = field(default_factory=list)
|
||
total_operations: int = 0
|
||
operation_errors: list[OperationParseError] = field(default_factory=list)
|
||
fatal_error: str | None = None
|
||
|
||
@property
|
||
def is_fatal(self) -> bool:
|
||
return self.fatal_error is not None
|
||
|
||
|
||
def validate_file_name(file_name: str) -> bool:
|
||
return validate_pipeline_file_name(file_name)
|
||
|
||
|
||
def parse_xml_content(file_name: str, xml_content: bytes | str) -> FileParseResult:
|
||
result = FileParseResult(file_name=file_name)
|
||
|
||
if not validate_file_name(file_name):
|
||
result.fatal_error = "Некорректное имя XML-файла"
|
||
return result
|
||
|
||
try:
|
||
root = ET.fromstring(xml_content)
|
||
except ET.ParseError as exc:
|
||
result.fatal_error = f"Ошибка XML: {exc}"
|
||
return result
|
||
|
||
try:
|
||
operations = _get_operations(root)
|
||
common_fields = _extract_common_fields(root)
|
||
except ValueError as exc:
|
||
result.fatal_error = str(exc)
|
||
return result
|
||
|
||
for index, operation in enumerate(operations, start=1):
|
||
result.total_operations += 1
|
||
record_id = f"ОПЕРАЦИЯ_{index}"
|
||
try:
|
||
operation_fields = _extract_direct_fields(
|
||
operation, excluded_tags={"УчастникОп"}
|
||
)
|
||
operation_fields = {**common_fields, **operation_fields}
|
||
operation_fields = _collapse_indexed_fields(operation_fields)
|
||
record_id = _extract_record_id(operation_fields, index)
|
||
operation_rows = [operation_fields]
|
||
participants = _find_children_by_name(operation, "УчастникОп")
|
||
|
||
if participants:
|
||
for participant in participants:
|
||
participant_fields = _extract_direct_fields(participant)
|
||
participant_rows = _expand_indexed_fields(participant_fields)
|
||
result.rows.extend(
|
||
_build_expanded_rows(
|
||
file_name=file_name,
|
||
operation_index=index,
|
||
record_id=record_id,
|
||
operation_rows=operation_rows[:1],
|
||
participant_rows=participant_rows,
|
||
)
|
||
)
|
||
else:
|
||
result.rows.extend(
|
||
_build_expanded_rows(
|
||
file_name=file_name,
|
||
operation_index=index,
|
||
record_id=record_id,
|
||
operation_rows=operation_rows,
|
||
participant_rows=[{}],
|
||
)
|
||
)
|
||
except Exception as exc: # noqa: BLE001
|
||
result.operation_errors.append(
|
||
OperationParseError(
|
||
operation_index=index,
|
||
record_id=record_id,
|
||
reason=f"Ошибка обработки операции: {exc}",
|
||
)
|
||
)
|
||
|
||
return result
|
||
|
||
|
||
def _get_operations(root: ET.Element) -> list[ET.Element]:
|
||
if _normalize_tag(root.tag) != "СообщОперКО":
|
||
raise ValueError("Некорректный корневой элемент XML (ожидался СообщОперКО)")
|
||
|
||
inform_part = _find_child_by_name(root, "ИнформЧасть")
|
||
if inform_part is None:
|
||
raise ValueError("В XML отсутствует элемент ИнформЧасть")
|
||
|
||
details = _find_child_by_name(inform_part, "СведКО")
|
||
if details is None:
|
||
raise ValueError("В XML отсутствует элемент СведКО")
|
||
|
||
return _find_children_by_name(details, "Операция")
|
||
|
||
|
||
def _extract_direct_fields(
|
||
element: ET.Element,
|
||
excluded_tags: Iterable[str] | None = None,
|
||
) -> dict[str, str]:
|
||
excluded = set(excluded_tags or ())
|
||
fields: dict[str, str] = {}
|
||
children = [child for child in element if _normalize_tag(child.tag) not in excluded]
|
||
counts = Counter(_normalize_tag(child.tag) for child in children)
|
||
structural_repeats = _find_structural_repeat_tags(children, counts)
|
||
positions: Counter[str] = Counter()
|
||
|
||
for child in children:
|
||
tag = _normalize_tag(child.tag)
|
||
position = positions[tag]
|
||
positions[tag] += 1
|
||
path = f"{tag}[{position}]" if tag in structural_repeats else tag
|
||
_collect_leaf_fields(child, path, fields)
|
||
|
||
return fields
|
||
|
||
|
||
def _build_expanded_rows(
|
||
*,
|
||
file_name: str,
|
||
operation_index: int,
|
||
record_id: str,
|
||
operation_rows: list[dict[str, str]],
|
||
participant_rows: list[dict[str, str]],
|
||
) -> list[ParticipantRow]:
|
||
"""Строит основную и независимые continuation-строки без декартова умножения."""
|
||
rows = [
|
||
ParticipantRow(
|
||
file_name=file_name,
|
||
operation_index=operation_index,
|
||
record_id=record_id,
|
||
operation_fields=operation_rows[0],
|
||
participant_fields=participant_rows[0],
|
||
)
|
||
]
|
||
rows.extend(
|
||
ParticipantRow(
|
||
file_name=file_name,
|
||
operation_index=operation_index,
|
||
record_id=record_id,
|
||
operation_fields=operation_fields,
|
||
participant_fields={},
|
||
is_continuation=True,
|
||
)
|
||
for operation_fields in operation_rows[1:]
|
||
)
|
||
rows.extend(
|
||
ParticipantRow(
|
||
file_name=file_name,
|
||
operation_index=operation_index,
|
||
record_id=record_id,
|
||
operation_fields=operation_rows[0],
|
||
participant_fields=participant_fields,
|
||
is_continuation=True,
|
||
)
|
||
for participant_fields in participant_rows[1:]
|
||
)
|
||
return rows
|
||
|
||
|
||
def _collapse_indexed_fields(fields: dict[str, str]) -> dict[str, str]:
|
||
"""Сохраняет все операционные повторы в одной строке без потери значений."""
|
||
collapsed: dict[str, str] = {}
|
||
for key, value in fields.items():
|
||
deindexed_key = re.sub(r"\[\d+\]", "", key)
|
||
_append_value(collapsed, deindexed_key, value)
|
||
return collapsed
|
||
|
||
|
||
def _expand_indexed_fields(
|
||
fields: dict[str, str],
|
||
*,
|
||
group_eio: bool = True,
|
||
) -> list[dict[str, str]]:
|
||
"""Разворачивает индексированные sibling-блоки в аддитивные наборы полей."""
|
||
base_fields: dict[str, str] = {}
|
||
families: dict[str, dict[int, dict[str, str]]] = {}
|
||
eio_occurrences: dict[str, int] = {}
|
||
|
||
for key, value in fields.items():
|
||
eio_match = _EIO_SEGMENT.search(key) if group_eio else None
|
||
match = eio_match or _INDEXED_SEGMENT.search(key)
|
||
if match is None:
|
||
base_fields[key] = value
|
||
continue
|
||
|
||
indexed_segment = match.group("segment")
|
||
family_segment = re.sub(r"\[\d+\]$", "", indexed_segment)
|
||
if eio_match is not None:
|
||
occurrence_key = f"{key[: match.start()]}{indexed_segment}"
|
||
occurrence = eio_occurrences.setdefault(
|
||
occurrence_key,
|
||
len(eio_occurrences),
|
||
)
|
||
family = "__eio_or_beneficiary__"
|
||
else:
|
||
family = f"{key[: match.start()]}{family_segment}"
|
||
occurrence = int(match.group("index"))
|
||
deindexed_key = f"{key[: match.start()]}{family_segment}{key[match.end() :]}"
|
||
families.setdefault(family, {}).setdefault(occurrence, {})[
|
||
deindexed_key
|
||
] = value
|
||
|
||
if not families:
|
||
return [base_fields]
|
||
|
||
primary = dict(base_fields)
|
||
continuations: list[dict[str, str]] = []
|
||
for occurrences in families.values():
|
||
for position, occurrence in enumerate(sorted(occurrences)):
|
||
expanded_occurrence = _expand_indexed_fields(
|
||
occurrences[occurrence],
|
||
group_eio=False,
|
||
)
|
||
if position == 0:
|
||
primary.update(expanded_occurrence[0])
|
||
continuations.extend(expanded_occurrence[1:])
|
||
else:
|
||
continuations.extend(expanded_occurrence)
|
||
|
||
return [primary, *continuations]
|
||
|
||
|
||
def _find_structural_repeat_tags(
|
||
children: list[ET.Element],
|
||
counts: Counter[str],
|
||
) -> set[str]:
|
||
"""Возвращает повторяемые теги структурных, а не скалярных XML-узлов."""
|
||
first_by_tag: dict[str, ET.Element] = {}
|
||
for child in children:
|
||
first_by_tag.setdefault(_normalize_tag(child.tag), child)
|
||
return {
|
||
tag
|
||
for tag, first_child in first_by_tag.items()
|
||
if counts[tag] > 1 and (len(first_child) > 0 or bool(first_child.attrib))
|
||
}
|
||
|
||
|
||
def _extract_record_id(operation_fields: dict[str, str], operation_index: int) -> str:
|
||
possible_keys = (
|
||
"ИдентификаторЗаписи",
|
||
"НомерЗаписи",
|
||
"ИдЗаписи",
|
||
"ИдЗап",
|
||
"ИдОпер",
|
||
"ID",
|
||
)
|
||
for key in possible_keys:
|
||
if operation_fields.get(key):
|
||
return operation_fields[key]
|
||
return f"ОПЕРАЦИЯ_{operation_index}"
|
||
|
||
|
||
def _normalize_tag(tag: str) -> str:
|
||
if "}" in tag:
|
||
return tag.split("}", maxsplit=1)[1]
|
||
return tag
|
||
|
||
|
||
def _find_child_by_name(parent: ET.Element, child_name: str) -> ET.Element | None:
|
||
for child in parent:
|
||
if _normalize_tag(child.tag) == child_name:
|
||
return child
|
||
return None
|
||
|
||
|
||
def _find_children_by_name(parent: ET.Element, child_name: str) -> list[ET.Element]:
|
||
return [child for child in parent if _normalize_tag(child.tag) == child_name]
|
||
|
||
|
||
def _extract_common_fields(root: ET.Element) -> dict[str, str]:
|
||
common: dict[str, str] = {}
|
||
|
||
service_part = _find_child_by_name(root, "СлужЧасть")
|
||
if service_part is not None:
|
||
common.update(_extract_direct_fields(service_part))
|
||
|
||
inform_part = _find_child_by_name(root, "ИнформЧасть")
|
||
if inform_part is None:
|
||
return common
|
||
|
||
info_bank = _find_child_by_name(inform_part, "ИнфБанк")
|
||
if info_bank is not None:
|
||
common.update(_extract_direct_fields(info_bank))
|
||
|
||
details = _find_child_by_name(inform_part, "СведКО")
|
||
if details is None:
|
||
return common
|
||
|
||
common.update(
|
||
_extract_direct_fields(details, excluded_tags={"Операция", "ИнфФилиал"})
|
||
)
|
||
branch_info = _find_child_by_name(details, "ИнфФилиал")
|
||
if branch_info is not None:
|
||
common.update(_extract_direct_fields(branch_info))
|
||
|
||
return common
|
||
|
||
|
||
def _collect_leaf_fields(
|
||
element: ET.Element, current_path: str, fields: dict[str, str]
|
||
) -> None:
|
||
_collect_attributes(element, current_path, fields)
|
||
children = list(element)
|
||
if not children:
|
||
text = (element.text or "").strip()
|
||
if text or not element.attrib:
|
||
_append_value(fields, current_path, text)
|
||
return
|
||
|
||
counts = Counter(_normalize_tag(child.tag) for child in children)
|
||
structural_repeats = _find_structural_repeat_tags(children, counts)
|
||
positions: Counter[str] = Counter()
|
||
for child in children:
|
||
child_name = _normalize_tag(child.tag)
|
||
position = positions[child_name]
|
||
positions[child_name] += 1
|
||
child_path = (
|
||
f"{child_name}[{position}]"
|
||
if child_name in structural_repeats
|
||
else child_name
|
||
)
|
||
_collect_leaf_fields(child, f"{current_path}.{child_path}", fields)
|
||
|
||
|
||
def _collect_attributes(
|
||
element: ET.Element, current_path: str, fields: dict[str, str]
|
||
) -> None:
|
||
for raw_name, raw_value in element.attrib.items():
|
||
attr_name = _normalize_tag(raw_name)
|
||
_append_value(fields, f"{current_path}.{attr_name}", raw_value.strip())
|
||
|
||
|
||
def _append_value(fields: dict[str, str], key: str, value: str) -> None:
|
||
if key not in fields:
|
||
fields[key] = value
|
||
return
|
||
|
||
existing = fields[key]
|
||
if not value:
|
||
return
|
||
if not existing:
|
||
fields[key] = value
|
||
return
|
||
if value in existing.split("; "):
|
||
return
|
||
|
||
fields[key] = f"{existing}; {value}"
|