from __future__ import annotations import re from dataclasses import dataclass from .column_constants import ( ACCOUNT_PLACEHOLDER, BLOCK_CP, BLOCK_EIO, CASHLESS_CLEARED_COLUMNS, CASHLESS_TRANSFER_TYPES, COL_ADDRESS_ONE_LINE, COL_ADDRESS_STRUCTURED, COL_AMOUNT_CURRENCY, COL_AMOUNT_RUB, COL_BIRTH_REG_DATE, COL_BRANCH_SIGN, COL_CARD_HOLDER_INFO, COL_CLIENT_SIGN, COL_COUNTRY_CODE, COL_CP_ISSUER_INN, COL_CURRENCY, COL_EIO_ADDRESS_ONE_LINE, COL_EIO_INN, COL_EIO_REG_PLACE_ONE_LINE, COL_EIO_REG_PLACE_STRUCTURED, COL_EMPLOYEE_SIGN, COL_ESP_SIGN, COL_EXTRA_CODES, COL_EXTRA_INFO, COL_FILE_NAME, COL_FOREIGN_BANK_NAME, COL_FTR_SIGN, COL_IDENT_FL, COL_INN, COL_INSURANCE_NUMBER, COL_ITEM_TYPE, COL_KPP, COL_METAL_CODE, COL_METAL_NAME, COL_OPERATION_CODE, COL_OPERATION_DOCUMENTS, COL_OPERATION_SIGN, COL_OPERATOR_TYPE, COL_PARTICIPANT_TYPE, COL_PAYER_ACCOUNT, COL_PAYER_BANK_ACCOUNT, COL_PAYER_BANK_BIK, COL_PAYER_BANK_NAME, COL_PAYER_ESP, COL_RECIPIENT_ACCOUNT, COL_RECIPIENT_BANK_ACCOUNT, COL_RECIPIENT_BANK_BIK, COL_RECIPIENT_BANK_NAME, COL_RECIPIENT_ESP, COL_RECORD_ID, COL_REG_PLACE_ONE_LINE, COL_REG_PLACE_STRUCTURED, COL_RESIDENT_SIGN, COL_SALE_AMOUNT, COL_SALE_CURRENCY, COL_STAY_DOCUMENT_TYPE, COL_SUSPENSION_BASIS, COL_SUSPICIOUS_ID, COL_TERRITORY_CODE, COL_TERRITORY_CODE_CARD, COL_TRANSFER_STATUS, COL_TRANSFER_TYPE, COL_UNUSUAL_CODES, EMPLOYEE_ABSENT_CODE, EXPORT_SUBSIDIARY_COMMENT_CODES, OKATO_COUNTRY_PAIRS, PARTICIPANT_ADDRESS_COMPONENT_COLUMNS, PARTICIPANT_TYPE_FL, PARTICIPANT_TYPE_FLCHP, PARTICIPANT_TYPE_IP, PARTICIPANT_TYPE_UL, PRECIOUS_ITEM_CODES, PRECIOUS_METAL_CODE, RUSSIA_COUNTRY_CODE, SUSPICIOUS_CODE, TERRITORY_CODE, VALID_ESP_SIGNS, VALID_FTR_SIGNS, VALID_SUSPENSION_BASES, ) from .column_registry import COLUMNS, ColumnMeta @dataclass(frozen=True) class _CompiledCandidate: raw: str normalized: str short: str strict_scoped: bool @dataclass(frozen=True) class MappingRule: column_index: int column_id: str column_name: str compiled_candidates: tuple[_CompiledCandidate, ...] source: str = "any" allow_direct_mapping: bool = True allow_short_lookup: bool = True structured_value: bool = False @dataclass(frozen=True, slots=True) class _LookupEntry: raw_path: str normalized_path: str value: str @dataclass(frozen=True, slots=True) class _PayloadLookup: values: dict[str, str] entries: tuple[_LookupEntry, ...] normalized_paths: tuple[str, ...] normalized_exact_first: dict[str, str] normalized_suffix_first: dict[str, str] raw_short_first: dict[str, str] @dataclass(frozen=True) class StructuredGroupRule: structured_column: int one_line_columns: tuple[int, ...] component_columns: tuple[int, ...] join_with_space: bool = False _INVALID_TAGS = {"", "-", "Источник", "путь", "подразумеваются", "тэга", "нашла"} _INDEXED_PATH_SEGMENT = re.compile(r"\[\d+\]") _OPERATION_DOCUMENT_FIELDS = ( "КодДок", "ИноеНаимДок", "ДатаДок", "НомДок", "СодДок", ) _OPERATION_DOCUMENT_FIELD = re.compile( r"(?:^|\.)ОснованиеОп(?:\[(?P\d+)\])?\." r"(?PКодДок|ИноеНаимДок|ДатаДок|НомДок|СодДок)$" ) _EIO_BLOCK_NAMES = ( "СведЕИО", "БенефициарЮЛ", "БенефициарФЛИП", "БенефициарИНБОЮЛ", ) _PARTICIPANT_ADDRESS_BASE_PATHS = ( "УчастникЮЛ.СведЮЛ.АдрРегЮЛ", "УчастникФЛИП.СведФЛИП.АдрРег", "УчастникИНБОЮЛ.СведИНБОЮЛ.МестоДеятельностьИНБОЮЛ", ) _ONE_LINE_STRUCTURED_PAIRS = ( (COL_REG_PLACE_ONE_LINE, COL_REG_PLACE_STRUCTURED), (COL_EIO_REG_PLACE_ONE_LINE, COL_EIO_REG_PLACE_STRUCTURED), ) def _bank_bik_candidate_keys(bank_block: str) -> tuple[str, ...]: candidates: list[str] = [] for transfers_block in ("СведенияПереводыДС", "СвединияПереводыДС"): path = ( "/СообщОперКО/ИнформЧасть/СведКО/Операция/" f"{transfers_block}/{bank_block}/БИККО" ) for candidate in _extract_path_candidate_keys(path): if ( candidate.endswith(f"{bank_block}.БИККО") and candidate not in candidates ): candidates.append(candidate) return tuple(candidates) def _compile_candidates(candidates: tuple[str, ...]) -> tuple[_CompiledCandidate, ...]: normalized_candidates = tuple( _INDEXED_PATH_SEGMENT.sub("", candidate) for candidate in candidates ) return tuple( _CompiledCandidate( raw=candidate, normalized=normalized, short=candidate.rsplit(".", maxsplit=1)[-1], strict_scoped="." in normalized and not any( other != normalized and other.endswith(f".{normalized}") for other in normalized_candidates ), ) for candidate, normalized in zip(candidates, normalized_candidates) ) def _build_rules(mapping_data: tuple[ColumnMeta, ...]) -> tuple[MappingRule, ...]: rules: list[MappingRule] = [] for meta in mapping_data: index = meta.index column = meta.name tag = meta.xml_tag path = meta.xml_path if not column or tag in _INVALID_TAGS: continue candidates = list(_extract_tag_candidate_keys(tag)) if index == COL_INN: candidates = ["ИННФЛИП", "ИННФЛ", "ИННЮЛ", "ИНН", *candidates] if column == "Тип участника": candidates = ["Тип", "ТипУчастника", *candidates] if column == "Отчество": # В некоторых выгрузках встречается опечатка тега "Очт". candidates = ["Отч", "Очт", *candidates] path_candidates = list(_extract_path_candidate_keys(path)) if path else [] if path: for path_tag in path_candidates: if path_tag not in candidates: candidates.append(path_tag) if index == COL_PAYER_BANK_BIK: candidates = list(_bank_bik_candidate_keys("СведБанкПлательщик")) elif index == COL_RECIPIENT_BANK_BIK: candidates = list(_bank_bik_candidate_keys("СведБанкПолучатель")) elif index == COL_COUNTRY_CODE: candidates = [ candidate for candidate in [*path_candidates, *candidates] if candidate.endswith("СведФЛИП.КодОКСМ") ] elif index == COL_IDENT_FL: candidates = [ candidate for candidate in [*path_candidates, *candidates] if candidate.endswith("ИдентификацияФЛ") ] elif index == COL_BIRTH_REG_DATE: candidates = [ candidate for candidate in [*path_candidates, *candidates] if candidate.endswith(("ДатаРегЮЛ", "ДатаРождения")) ] elif index == COL_STAY_DOCUMENT_TYPE: candidates = [ candidate for candidate in path_candidates if candidate.endswith("УчастникФЛИП.СведФЛИП.СведДокПреб.ВидДокКод") ] elif ( index == COL_ADDRESS_STRUCTURED or index in PARTICIPANT_ADDRESS_COMPONENT_COLUMNS ): candidates = [ f"{base_path}.{address_tag}" for base_path in _PARTICIPANT_ADDRESS_BASE_PATHS for address_tag in _extract_tag_candidate_keys(tag) ] elif index == COL_REG_PLACE_ONE_LINE: candidates = [ "УчастникЮЛ.СведЮЛ.АдрРегЮЛ.АдресСтрока", "СообщОперКО.ИнформЧасть.СведКО.Операция.УчастникОП.УчастникЮЛ.СведЮЛ.АдрРегЮЛ.АдресСтрока", ] elif index == COL_REG_PLACE_STRUCTURED: base_paths = ( "УчастникЮЛ.СведЮЛ.АдрРегЮЛ", "СообщОперКО.ИнформЧасть.СведКО.Операция.УчастникОП.УчастникЮЛ.СведЮЛ.АдрРегЮЛ", ) address_tags = ( "Индекс", "КодОКСМ", "КодСубъектаПоОКАТО", "Район", "Пункт", "Улица", "Дом", "Корп", "Оф", ) candidates = [ f"{base_path}.{tag}" for base_path in base_paths for tag in address_tags ] elif index == COL_CP_ISSUER_INN: candidates = [ candidate for candidate in [*path_candidates, *candidates] if candidate.endswith("ИННЭмитентЦП") ] if not candidates: candidates = list(_extract_tag_candidate_keys(tag)) non_one_line_candidates = [ candidate for candidate in candidates if not _is_one_line_candidate(candidate) ] direct_candidates = candidates if meta.structured_value and non_one_line_candidates: direct_candidates = non_one_line_candidates candidate_keys = tuple(direct_candidates) rules.append( MappingRule( column_index=index, column_id=meta.column_id, column_name=column, compiled_candidates=_compile_candidates(candidate_keys), source=meta.source_scope, allow_direct_mapping=meta.allow_direct_mapping, allow_short_lookup=meta.allow_short_lookup, structured_value=meta.structured_value, ) ) return tuple(rules) def _is_one_line_candidate(candidate: str) -> bool: return "строка" in candidate.lower() def _build_structured_group_rules( mapping_data: tuple[ColumnMeta, ...], ) -> tuple[StructuredGroupRule, ...]: grouped: dict[str, list[ColumnMeta]] = {} for meta in mapping_data: if meta.structured_group: grouped.setdefault(meta.structured_group, []).append(meta) rules: list[StructuredGroupRule] = [] for group_id, members in grouped.items(): aggregates = [ member for member in members if member.structured_role == "aggregate" ] if len(aggregates) != 1: raise ValueError( f"Структурная группа {group_id!r} должна иметь один aggregate" ) aggregate = aggregates[0] one_line_columns = tuple( member.index for member in sorted( (item for item in members if item.structured_role == "one_line"), key=lambda item: item.structured_order, ) ) component_columns = tuple( member.index for member in sorted( (item for item in members if item.structured_role == "component"), key=lambda item: item.structured_order, ) ) if not component_columns: raise ValueError(f"Структурная группа {group_id!r} не содержит components") rules.append( StructuredGroupRule( structured_column=aggregate.index, one_line_columns=one_line_columns, component_columns=component_columns, join_with_space=aggregate.join_with_space, ) ) return tuple(sorted(rules, key=lambda rule: rule.structured_column)) 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 for start_index in range(len(parts)): dotted_path = ".".join(parts[start_index:]) if dotted_path and dotted_path not in candidates: candidates.append(dotted_path) return tuple(candidates) _MAPPING_DATA = COLUMNS.ordered_columns() FIXED_REPORT_COLUMNS: tuple[str, ...] = COLUMNS.all_columns() REPORT_MAPPING_RULES: tuple[MappingRule, ...] = _build_rules(_MAPPING_DATA) STRUCTURED_GROUP_RULES: tuple[StructuredGroupRule, ...] = _build_structured_group_rules( _MAPPING_DATA ) def build_fixed_row_by_index( *, file_name: str, record_id: str, operation_index: int, operation_fields: dict[str, str], participant_fields: dict[str, str], is_continuation: bool = False, ) -> dict[int, str]: row: dict[int, str] = { COL_FILE_NAME: file_name, COL_RECORD_ID: record_id, } operation_lookup = _build_payload_lookup(operation_fields) participant_lookup = _build_payload_lookup(participant_fields) participant_base_lookup, participant_eio_lookup = _partition_participant_lookup( participant_lookup ) merged_lookup = _build_payload_lookup( _merge_fields(operation_fields, participant_fields) ) has_eio_block = _lookup_has_any_prefixed_key( participant_lookup, ( "УчастникЮЛ.СведЕИО.", "УчастникЮЛ.БенефициарЮЛ.", "УчастникФЛИП.БенефициарФЛИП.", "УчастникИНБОЮЛ.БенефициарИНБОЮЛ.", "УчастникИНБОЮЛ.СведИНБОЮЛ.Учредитель.", "СведЕИО.", "БенефициарЮЛ.", "БенефициарФЛИП.", "БенефициарИНБОЮЛ.", ), ) has_cp_block = _lookup_has_any_prefixed_key( operation_lookup, ( "СведЦП.", "СообщОперКО.ИнформЧасть.СведКО.Операция.СведЦП.", ), ) for rule in REPORT_MAPPING_RULES: if rule.column_index in row: continue if not rule.allow_direct_mapping: row[rule.column_index] = "" continue if rule.source == "participant": source_lookup = ( participant_eio_lookup if rule.column_index in BLOCK_EIO else participant_base_lookup ) elif rule.source == "operation": source_lookup = operation_lookup else: source_lookup = merged_lookup allow_short_lookup = rule.allow_short_lookup or rule.column_index in BLOCK_EIO if rule.structured_value: row[rule.column_index] = _pick_structured_value( lookup=source_lookup, candidates=rule.compiled_candidates, join_with_space="фио" in rule.column_name.lower(), allow_short_lookup=allow_short_lookup, ) else: row[rule.column_index] = _pick_value( source_lookup, rule.compiled_candidates, allow_short_lookup=allow_short_lookup, ) _set( row, COL_OPERATION_DOCUMENTS, _format_operation_documents(operation_lookup), ) if not is_continuation: _apply_participant_identity_rules(row, participant_lookup) # Сначала убираем запрещённые однострочные адреса, чтобы структурная # группа не очистила их компоненты до проверки резидентства. _rule_address_one_line(row) _apply_structured_group_rules(row) _apply_one_line_structured_pairs(row) if not is_continuation: _apply_business_rules( row, has_eio_block=has_eio_block, has_cp_block=has_cp_block, operation_lookup=operation_lookup, participant_lookup=participant_lookup, ) else: _apply_continuation_validation_rules(row) for meta in COLUMNS.ordered_columns(): row.setdefault(meta.index, "") return row def build_fixed_row( *, file_name: str, record_id: str, operation_index: int, operation_fields: dict[str, str], participant_fields: dict[str, str], is_continuation: bool = False, ) -> dict[str, str]: by_index = build_fixed_row_by_index( file_name=file_name, record_id=record_id, operation_index=operation_index, operation_fields=operation_fields, participant_fields=participant_fields, is_continuation=is_continuation, ) by_name: dict[str, str] = {} for meta in COLUMNS.ordered_columns(): by_name.setdefault(meta.name, by_index.get(meta.index, "")) return by_name def _apply_structured_group_rules(row: dict[int, str]) -> None: for rule in STRUCTURED_GROUP_RULES: one_line_has_value = any( _get(row, column_index).strip() for column_index in rule.one_line_columns ) if one_line_has_value: _set(row, rule.structured_column, "") for component_column_index in rule.component_columns: _set(row, component_column_index, "") continue values = [ _get(row, component_column_index).strip() for component_column_index in rule.component_columns if _get(row, component_column_index).strip() ] if not values: continue separator = " " if rule.join_with_space else ", " _set(row, rule.structured_column, separator.join(values)) def _apply_one_line_structured_pairs(row: dict[int, str]) -> None: """Оставляет только однострочное значение, если заполнены обе версии поля.""" for one_line_column, structured_column in _ONE_LINE_STRUCTURED_PAIRS: if _get(row, one_line_column): _set(row, structured_column, "") 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 _normalize_indexed_path(path: str) -> str: """Убирает индексы повторов, сохраняя логическую структуру XML-пути.""" return _INDEXED_PATH_SEGMENT.sub("", path) def _build_payload_lookup(payload: dict[str, str]) -> _PayloadLookup: entries = tuple( _LookupEntry( raw_path=key, normalized_path=_normalize_indexed_path(key), value=value, ) for key, value in payload.items() ) return _build_payload_lookup_from_entries(entries) def _build_payload_lookup_from_entries( entries: tuple[_LookupEntry, ...], ) -> _PayloadLookup: values = {entry.raw_path: entry.value for entry in entries} normalized_exact_first: dict[str, str] = {} normalized_suffix_first: dict[str, str] = {} raw_short_first: dict[str, str] = {} for entry in entries: if not entry.value: continue normalized_exact_first.setdefault(entry.normalized_path, entry.value) path_parts = entry.normalized_path.split(".") for start_index in range(len(path_parts)): suffix = ".".join(path_parts[start_index:]) normalized_suffix_first.setdefault(suffix, entry.value) raw_short = entry.raw_path.rsplit(".", maxsplit=1)[-1] raw_short_first.setdefault(raw_short, entry.value) return _PayloadLookup( values=values, entries=entries, normalized_paths=tuple(entry.normalized_path for entry in entries), normalized_exact_first=normalized_exact_first, normalized_suffix_first=normalized_suffix_first, raw_short_first=raw_short_first, ) def _is_normalized_eio_path(normalized_path: str) -> bool: surrounded_path = f".{normalized_path}." return ".СведИНБОЮЛ.Учредитель." in surrounded_path or any( f".{block_name}." in surrounded_path for block_name in _EIO_BLOCK_NAMES ) def _partition_participant_lookup( participant_lookup: _PayloadLookup, ) -> tuple[_PayloadLookup, _PayloadLookup]: base_entries: list[_LookupEntry] = [] eio_entries: list[_LookupEntry] = [] for entry in participant_lookup.entries: target = ( eio_entries if _is_normalized_eio_path(entry.normalized_path) else base_entries ) target.append(entry) return ( _build_payload_lookup_from_entries(tuple(base_entries)), _build_payload_lookup_from_entries(tuple(eio_entries)), ) def _lookup_has_payload_block(lookup: _PayloadLookup, block_name: str) -> bool: prefix = f"{block_name}." nested_marker = f".{block_name}." return any( normalized_path.startswith(prefix) or nested_marker in normalized_path for normalized_path in lookup.normalized_paths ) def _lookup_pick_scoped_value(lookup: _PayloadLookup, scoped_path: str) -> str: value = lookup.normalized_suffix_first.get(scoped_path, "") return str(value).strip() if value else "" def _lookup_pick_direct_value(lookup: _PayloadLookup, field_name: str) -> str: """Возвращает только прямое поле текущего XML-блока без suffix-поиска.""" value = lookup.normalized_exact_first.get(field_name, "") return str(value).strip() if value else "" def _format_operation_documents(lookup: _PayloadLookup) -> str: indexed_documents: dict[int, dict[str, str]] = {} single_document: dict[str, str] = {} for entry in lookup.entries: match = _OPERATION_DOCUMENT_FIELD.search(entry.raw_path) if match is None: continue field_name = match.group("field") value = str(entry.value).strip() document_index = match.group("index") if document_index is None: single_document[field_name] = value continue indexed_documents.setdefault(int(document_index), {})[field_name] = value documents = ( [indexed_documents[index] for index in sorted(indexed_documents)] if indexed_documents else [single_document] if single_document else [] ) formatted_documents = [] for position, document in enumerate(documents, start=1): fields = [ document.get(field_name, "") for field_name in _OPERATION_DOCUMENT_FIELDS ] formatted_documents.append(f"{position}. {', '.join(fields)}") return "; ".join(formatted_documents) def _apply_participant_identity_rules( row: dict[int, str], participant_lookup: _PayloadLookup ) -> None: participant_type = _normalize_code(_get(row, COL_PARTICIPANT_TYPE)) has_legal_entity = _lookup_has_payload_block(participant_lookup, "УчастникЮЛ") has_physical_person = _lookup_has_payload_block(participant_lookup, "УчастникФЛИП") has_foreign_structure = _lookup_has_payload_block( participant_lookup, "УчастникИНБОЮЛ" ) if has_physical_person and not has_legal_entity: identification = _lookup_pick_scoped_value( participant_lookup, "УчастникФЛИП.ИдентификацияФЛ" ) _set(row, COL_IDENT_FL, identification) else: _set(row, COL_IDENT_FL, "") if has_legal_entity or participant_type == PARTICIPANT_TYPE_UL: value = _lookup_pick_scoped_value(participant_lookup, "УчастникЮЛ.СведЮЛ.КППЮЛ") elif has_foreign_structure: value = _lookup_pick_scoped_value( participant_lookup, "УчастникИНБОЮЛ.СведИНБОЮЛ.ПризнакОргФормаИНБОЮЛ", ) elif has_physical_person: value = _lookup_pick_scoped_value( participant_lookup, "УчастникФЛИП.ИдентификацияФЛ" ) else: value = "" _set(row, COL_KPP, value) def _lookup_has_any_prefixed_key( lookup: _PayloadLookup, prefixes: tuple[str, ...] ) -> bool: return any( any(normalized_path.startswith(prefix) for prefix in prefixes) for normalized_path in lookup.normalized_paths ) def _pick_value( lookup: _PayloadLookup, candidates: tuple[_CompiledCandidate, ...], *, allow_short_lookup: bool = True, ) -> str: for candidate in candidates: if not allow_short_lookup and not candidate.strict_scoped: continue if value := lookup.values.get(candidate.raw): return value if value := lookup.normalized_suffix_first.get(candidate.normalized): return value if not allow_short_lookup: for entry in lookup.entries: if ( entry.value and "." in entry.normalized_path and candidate.normalized.endswith(f".{entry.normalized_path}") ): return entry.value continue if value := lookup.values.get(candidate.short): return value if value := lookup.raw_short_first.get(candidate.short): return value return "" def _pick_structured_value( *, lookup: _PayloadLookup, candidates: tuple[_CompiledCandidate, ...], join_with_space: bool, allow_short_lookup: bool = True, ) -> str: values: list[str] = [] for candidate in candidates: value = _pick_value( lookup, (candidate,), allow_short_lookup=allow_short_lookup, ) if value and value not in values: values.append(value) if not values: return "" separator = " " if join_with_space else ", " return separator.join(values) def _get(row: dict[int, str], column: int) -> str: return str(row.get(column, "")).strip() def _normalize_code(value: str) -> str: normalized = value.strip() if re.fullmatch(r"\d+", normalized): return normalized.lstrip("0") or "0" return normalized def _set(row: dict[int, str], column: int, value: str) -> None: row[column] = value def _clear_columns(row: dict[int, str], columns: tuple[int, ...]) -> None: for column in columns: _set(row, column, "") def _extract_numeric_codes(value: str) -> set[str]: return {match for match in re.findall(r"\d+", value)} def _operation_codes(row: dict[int, str]) -> set[str]: op_code = _get(row, COL_OPERATION_CODE) extra_codes = _get(row, COL_EXTRA_CODES) return _extract_numeric_codes(op_code) | _extract_numeric_codes(extra_codes) def _apply_business_rules( row: dict[int, str], *, has_eio_block: bool, has_cp_block: bool, operation_lookup: _PayloadLookup, participant_lookup: _PayloadLookup, ) -> None: codes = _operation_codes(row) operation_sign = _get(row, COL_OPERATION_SIGN) transfer_type = _get(row, COL_TRANSFER_TYPE) operator_type = _get(row, COL_OPERATOR_TYPE) participant_type = _normalize_code(_get(row, COL_PARTICIPANT_TYPE)) _rule_ftr_sign(row) _rule_suspension_basis(row) _rule_esp_sign(row) _rule_unusual_codes(row, codes) _rule_digital_rights_currency(row, operation_sign) _rule_sale_currency(row) _rule_extra_info(row, codes, operation_lookup, participant_lookup) _rule_suspicious_activity(row, codes) _rule_metal_name(row) _rule_item_type(row, codes) _rule_territory_codes(row, codes) _rule_payer_account(row, transfer_type) _rule_esp_identifiers(row, operation_sign) _rule_payer_bank(row, operator_type) _rule_recipient_bank(row, operator_type) _rule_correspondent_accounts(row, operator_type, transfer_type) _rule_recipient_account(row, transfer_type) _rule_transfer_status(row, transfer_type) _rule_cashless_clears_cash_columns(row, transfer_type) _rule_card_without_employee(row) _rule_participant_identity(row, participant_type) _rule_branch_sign(row, participant_type) _rule_snils_for_ip(row, participant_type) _rule_address_one_line(row) _rule_okato_by_country(row) _rule_eio_inn_for_physical_person(row, participant_type) if not has_eio_block: _clear_columns(row, BLOCK_EIO) if not has_cp_block: _clear_columns(row, BLOCK_CP) def _apply_continuation_validation_rules(row: dict[int, str]) -> None: """Проверяет локальные коды повтора без подстановки значений основной строки.""" _rule_ftr_sign(row) _rule_suspension_basis(row) _rule_esp_sign(row) def _rule_ftr_sign(row: dict[int, str]) -> None: """Правило 10: оставляем только допустимые значения признака ФТР.""" if _get(row, COL_FTR_SIGN) not in VALID_FTR_SIGNS: _set(row, COL_FTR_SIGN, "") def _rule_suspension_basis(row: dict[int, str]) -> None: """Правило 11: основание приостановления должно быть допустимым.""" if _get(row, COL_SUSPENSION_BASIS) not in VALID_SUSPENSION_BASES: _set(row, COL_SUSPENSION_BASIS, "") def _rule_esp_sign(row: dict[int, str]) -> None: """Правило 15: признак ЭСП должен быть допустимым.""" if _get(row, COL_ESP_SIGN) not in VALID_ESP_SIGNS: _set(row, COL_ESP_SIGN, "") def _rule_unusual_codes(row: dict[int, str], codes: set[str]) -> None: """Правило 21: коды необычной операции относятся только к 6001.""" if SUSPICIOUS_CODE not in codes: _set(row, COL_UNUSUAL_CODES, "") def _rule_digital_rights_currency(row: dict[int, str], operation_sign: str) -> None: """Правила 22-24: для цифровых прав валютные суммы не выводятся.""" if operation_sign == "8": _set(row, COL_CURRENCY, "") _set(row, COL_AMOUNT_CURRENCY, "") _set(row, COL_AMOUNT_RUB, "") def _rule_sale_currency(row: dict[int, str]) -> None: """Правило 26: нулевая сумма или отсутствие конверсии дают пустое поле.""" conversion_amount = _get(row, COL_SALE_AMOUNT) normalized_amount = conversion_amount.replace(",", ".") if not _get(row, COL_SALE_CURRENCY) or re.fullmatch( r"[+-]?0+(?:\.0+)?", normalized_amount ): _set(row, COL_SALE_AMOUNT, "") def _rule_extra_info( row: dict[int, str], codes: set[str], operation_lookup: _PayloadLookup, participant_lookup: _PayloadLookup, ) -> None: """Правило 34: объединяет допустимые комментарии операции и участника.""" operation_comment = _lookup_pick_direct_value(operation_lookup, "Коммент") participant_comment = "" if codes & EXPORT_SUBSIDIARY_COMMENT_CODES: participant_comment = _lookup_pick_direct_value( participant_lookup, "КомментУчастник" ) values = list( dict.fromkeys( value for value in (operation_comment, participant_comment) if value ) ) _set(row, COL_EXTRA_INFO, "; ".join(values)) def _rule_suspicious_activity(row: dict[int, str], codes: set[str]) -> None: """Правило 28: идентификатор подозрительной деятельности только для 6001.""" if SUSPICIOUS_CODE not in codes: _set(row, COL_SUSPICIOUS_ID, "") def _rule_metal_name(row: dict[int, str]) -> None: """Правило 30: наименование металла только для кода C99.""" if _get(row, COL_METAL_CODE).upper() != PRECIOUS_METAL_CODE: _set(row, COL_METAL_NAME, "") def _rule_item_type(row: dict[int, str], codes: set[str]) -> None: """Правило 31: предмет операции только для кодов 5020-5023.""" if not codes & PRECIOUS_ITEM_CODES: _set(row, COL_ITEM_TYPE, "") def _rule_territory_codes(row: dict[int, str], codes: set[str]) -> None: """Правила 36 и 96: код территории только для операции 5016.""" if TERRITORY_CODE not in codes: _set(row, COL_TERRITORY_CODE, "") _set(row, COL_TERRITORY_CODE_CARD, "") def _rule_payer_account(row: dict[int, str], transfer_type: str) -> None: """Правило 38: счёт плательщика зависит от вида перевода.""" if transfer_type in {"1", "2", "3", "10"}: if not _get(row, COL_PAYER_ACCOUNT): _set(row, COL_PAYER_ACCOUNT, ACCOUNT_PLACEHOLDER) elif transfer_type not in {"12", "13", "14"}: _set(row, COL_PAYER_ACCOUNT, "") def _rule_esp_identifiers(row: dict[int, str], operation_sign: str) -> None: """Правила 39 и 47: идентификаторы ЭСП отсутствуют для признака 9.""" if operation_sign == "9": _set(row, COL_PAYER_ESP, "") _set(row, COL_RECIPIENT_ESP, "") def _rule_payer_bank(row: dict[int, str], operator_type: str) -> None: """Правило 40: банк плательщика только для типов оператора 2 и 4.""" if operator_type not in {"2", "4"}: _set(row, COL_PAYER_BANK_NAME, "") def _rule_recipient_bank(row: dict[int, str], operator_type: str) -> None: """Правила 42-43: банк получателя только для типов оператора 1 и 4.""" if operator_type not in {"1", "4"}: _set(row, COL_RECIPIENT_BANK_NAME, "") _set(row, COL_RECIPIENT_BANK_BIK, "") def _rule_correspondent_accounts( row: dict[int, str], operator_type: str, transfer_type: str ) -> None: """Правила 44-45: корсчета зависят от оператора и вида перевода.""" if not operator_type and not transfer_type: _set(row, COL_PAYER_BANK_ACCOUNT, "") _set(row, COL_RECIPIENT_BANK_ACCOUNT, "") return if operator_type in {"3", "5"} or transfer_type in {"12", "13"}: _set(row, COL_PAYER_BANK_ACCOUNT, "") elif not _get(row, COL_PAYER_BANK_ACCOUNT): _set(row, COL_PAYER_BANK_ACCOUNT, ACCOUNT_PLACEHOLDER) if operator_type in {"3", "5"} or transfer_type in {"10", "11"}: _set(row, COL_RECIPIENT_BANK_ACCOUNT, "") elif not _get(row, COL_RECIPIENT_BANK_ACCOUNT): _set(row, COL_RECIPIENT_BANK_ACCOUNT, ACCOUNT_PLACEHOLDER) def _rule_recipient_account(row: dict[int, str], transfer_type: str) -> None: """Правило 46: счёт получателя зависит от вида перевода.""" if transfer_type in {"1", "4", "7", "12"}: if not _get(row, COL_RECIPIENT_ACCOUNT): _set(row, COL_RECIPIENT_ACCOUNT, ACCOUNT_PLACEHOLDER) elif transfer_type not in {"10", "11", "14"}: _set(row, COL_RECIPIENT_ACCOUNT, "") def _rule_transfer_status(row: dict[int, str], transfer_type: str) -> None: """Правило 62: статус перевода только для видов 2, 5 и 8.""" if transfer_type not in {"2", "5", "8"}: _set(row, COL_TRANSFER_STATUS, "") def _rule_cashless_clears_cash_columns(row: dict[int, str], transfer_type: str) -> None: """Для безналичных переводов очищаем согласованные наличные поля.""" if transfer_type in CASHLESS_TRANSFER_TYPES: _clear_columns(row, CASHLESS_CLEARED_COLUMNS) def _rule_card_without_employee(row: dict[int, str]) -> None: """Правила 103 и 105: подставляем заглушку при отсутствии сотрудника.""" if _get(row, COL_EMPLOYEE_SIGN) != EMPLOYEE_ABSENT_CODE: return if not _get(row, COL_CARD_HOLDER_INFO): _set(row, COL_CARD_HOLDER_INFO, "Информация отсутствует") if not _get(row, COL_FOREIGN_BANK_NAME): _set(row, COL_FOREIGN_BANK_NAME, "Информация отсутствует") def _rule_participant_identity(row: dict[int, str], participant_type: str) -> None: """Правила 119 и 122: идентификация и КПП зависят от типа участника.""" if participant_type in {PARTICIPANT_TYPE_UL, PARTICIPANT_TYPE_FL}: _set(row, COL_IDENT_FL, "") if participant_type == PARTICIPANT_TYPE_FL: _set(row, COL_KPP, "") def _rule_branch_sign(row: dict[int, str], participant_type: str) -> None: """Правило 124: показатель отсутствует для ФЛ, ИП и ФЛЧП.""" if participant_type in { PARTICIPANT_TYPE_FL, PARTICIPANT_TYPE_IP, PARTICIPANT_TYPE_FLCHP, }: _set(row, COL_BRANCH_SIGN, "") def _rule_snils_for_ip(row: dict[int, str], participant_type: str) -> None: """Правило 125: СНИЛС не выводится для ИП.""" if participant_type != PARTICIPANT_TYPE_IP: return insurance_value = _get(row, COL_INSURANCE_NUMBER) if re.fullmatch(r"\d{3}-?\d{3}-?\d{3}\s?\d{2}", insurance_value): _set(row, COL_INSURANCE_NUMBER, "") def _allow_one_line_address(row: dict[int, str]) -> bool: resident_flag = _get(row, COL_RESIDENT_SIGN) client_flag = _get(row, COL_CLIENT_SIGN) if not resident_flag and not client_flag: return True return resident_flag in {"0", "9"} or (resident_flag == "1" and client_flag == "0") def _rule_address_one_line(row: dict[int, str]) -> None: """Правила 141, 152, 167 и 178: однострочные адреса только для нерезидентов.""" if not _allow_one_line_address(row): _set(row, COL_ADDRESS_ONE_LINE, "") _set(row, COL_REG_PLACE_ONE_LINE, "") _set(row, COL_EIO_ADDRESS_ONE_LINE, "") _set(row, COL_EIO_REG_PLACE_ONE_LINE, "") def _rule_okato_by_country(row: dict[int, str]) -> None: """ОКАТО адресного блока заполняется только для России.""" for country_column, okato_column in OKATO_COUNTRY_PAIRS: if _get(row, country_column) != RUSSIA_COUNTRY_CODE: _set(row, okato_column, "") def _rule_eio_inn_for_physical_person( row: dict[int, str], participant_type: str ) -> None: """Правило 163: показатель ЕИО отсутствует для ФЛ.""" if participant_type == PARTICIPANT_TYPE_FL: _set(row, COL_EIO_INN, "")