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_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_CURRENCY_SIGN, COL_EIO_ADDRESS_ONE_LINE, COL_EIO_INN, COL_EIO_REG_PLACE_ONE_LINE, COL_EMPLOYEE_SIGN, COL_ESP_SIGN, COL_EXTRA_CODES, 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_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_SUSPENSION_BASIS, COL_SUSPICIOUS_ID, COL_TERRITORY_CODE, COL_TERRITORY_CODE_CARD, COL_TRANSFER_STATUS, COL_TRANSFER_TYPE, COL_UNUSUAL_CODES, CURRENCY_OPERATION_CODES, EMPLOYEE_ABSENT_CODE, OKATO_COUNTRY_PAIRS, 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 MappingRule: column_index: int column_id: str column_name: str candidate_keys: tuple[str, ...] source: str = "any" allow_direct_mapping: bool = True allow_short_lookup: bool = True structured_value: bool = False @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+\]") _EIO_BLOCK_NAMES = ( "СведЕИО", "БенефициарЮЛ", "БенефициарФЛИП", "БенефициарИНБОЮЛ", ) 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 _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_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 rules.append( MappingRule( column_index=index, column_id=meta.column_id, column_name=column, candidate_keys=tuple(direct_candidates), 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, } merged = _merge_fields(operation_fields, participant_fields) merged_suffix_index = _build_suffix_index(merged) operation_suffix_index = _build_suffix_index(operation_fields) has_eio_block = _has_any_prefixed_key( participant_fields, ( "УчастникЮЛ.СведЕИО.", "УчастникЮЛ.БенефициарЮЛ.", "УчастникФЛИП.БенефициарФЛИП.", "УчастникИНБОЮЛ.БенефициарИНБОЮЛ.", "УчастникИНБОЮЛ.СведИНБОЮЛ.Учредитель.", "СведЕИО.", "БенефициарЮЛ.", "БенефициарФЛИП.", "БенефициарИНБОЮЛ.", ), ) has_cp_block = _has_any_prefixed_key( operation_fields, ( "СведЦП.", "СообщОперКО.ИнформЧасть.СведКО.Операция.СведЦП.", ), ) 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_payload = _scope_participant_payload( participant_fields, eio_only=rule.column_index in BLOCK_EIO, ) suffix_index = _build_suffix_index(source_payload) elif rule.source == "operation": source_payload = operation_fields suffix_index = operation_suffix_index else: source_payload = merged suffix_index = merged_suffix_index if rule.structured_value: row[rule.column_index] = _pick_structured_value( source_payload=source_payload, suffix_index=suffix_index, candidates=rule.candidate_keys, join_with_space="фио" in rule.column_name.lower(), allow_short_lookup=rule.allow_short_lookup, ) else: row[rule.column_index] = _pick_value( source_payload, suffix_index, rule.candidate_keys, allow_short_lookup=rule.allow_short_lookup, ) if not is_continuation: _apply_participant_identity_rules(row, participant_fields) # Сначала убираем запрещённые однострочные адреса, чтобы структурная # группа не очистила их компоненты до проверки резидентства. _rule_address_one_line(row) _apply_structured_group_rules(row) if not is_continuation: _apply_business_rules( row, has_eio_block=has_eio_block, has_cp_block=has_cp_block, ) 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 _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 _has_payload_block(payload: dict[str, str], block_name: str) -> bool: prefix = f"{block_name}." nested_marker = f".{block_name}." return any( (normalized := _normalize_indexed_path(key)).startswith(prefix) or nested_marker in normalized for key in payload ) def _pick_scoped_payload_value(payload: dict[str, str], scoped_path: str) -> str: nested_suffix = f".{scoped_path}" for key, value in payload.items(): normalized = _normalize_indexed_path(key) if value and (normalized == scoped_path or normalized.endswith(nested_suffix)): return str(value).strip() return "" def _apply_participant_identity_rules( row: dict[int, str], participant_fields: dict[str, str] ) -> None: participant_type = _normalize_code(_get(row, COL_PARTICIPANT_TYPE)) has_legal_entity = _has_payload_block(participant_fields, "УчастникЮЛ") has_physical_person = _has_payload_block(participant_fields, "УчастникФЛИП") has_foreign_structure = _has_payload_block(participant_fields, "УчастникИНБОЮЛ") if has_physical_person and not has_legal_entity: identification = _pick_scoped_payload_value( participant_fields, "УчастникФЛИП.ИдентификацияФЛ" ) _set(row, COL_IDENT_FL, identification) else: _set(row, COL_IDENT_FL, "") if has_legal_entity or participant_type == PARTICIPANT_TYPE_UL: value = _pick_scoped_payload_value( participant_fields, "УчастникЮЛ.СведЮЛ.КППЮЛ" ) elif has_foreign_structure: value = _pick_scoped_payload_value( participant_fields, "УчастникИНБОЮЛ.СведИНБОЮЛ.ПризнакОргФормаИНБОЮЛ", ) elif has_physical_person: value = _pick_scoped_payload_value( participant_fields, "УчастникФЛИП.ИдентификацияФЛ" ) else: value = "" _set(row, COL_KPP, value) def _has_any_prefixed_key(payload: dict[str, str], prefixes: tuple[str, ...]) -> bool: return any( any(_normalize_indexed_path(key).startswith(prefix) for prefix in prefixes) for key in payload ) def _normalize_indexed_path(path: str) -> str: """Убирает индексы повторов, сохраняя логическую структуру XML-пути.""" return _INDEXED_PATH_SEGMENT.sub("", path) def _is_eio_path(path: str) -> bool: normalized = _normalize_indexed_path(path) return ".СведИНБОЮЛ.Учредитель." in f".{normalized}." or any( f".{block_name}." in f".{normalized}." for block_name in _EIO_BLOCK_NAMES ) def _scope_participant_payload( payload: dict[str, str], *, eio_only: bool, ) -> dict[str, str]: """Изолирует поля участника от вложенного блока ЕИО/бенефициара.""" return { key: value for key, value in payload.items() if _is_eio_path(key) is eio_only } 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, ...], *, allow_short_lookup: bool = True, ) -> str: for key in candidates: if value := payload.get(key): return value normalized_key = _normalize_indexed_path(key) for payload_key, value in payload.items(): normalized_payload_key = _normalize_indexed_path(payload_key) if value and ( normalized_payload_key == normalized_key or normalized_payload_key.endswith(f".{normalized_key}") ): return value if not allow_short_lookup: continue short = key.rsplit(".", maxsplit=1)[-1] if value := payload.get(short): return value 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, allow_short_lookup: bool = True, ) -> str: values: list[str] = [] for key in candidates: value = _pick_value( source_payload, suffix_index, (key,), 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, ) -> 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_currency_sign(row, codes) _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_currency_sign(row: dict[int, str], codes: set[str]) -> None: """Правило 27: признак VO только для валютных операций 6101-6126.""" if not codes & CURRENCY_OPERATION_CODES: _set(row, COL_CURRENCY_SIGN, "") 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 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, "")