SFM/app/pipeline/mapping.py

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_EIO_ADDRESS_ONE_LINE,
    COL_EIO_INN,
    COL_EIO_REG_PLACE_ONE_LINE,
    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_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,
    EMPLOYEE_ABSENT_CODE,
    EXPORT_SUBSIDIARY_COMMENT_CODES,
    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,
            operation_fields=operation_fields,
            participant_fields=participant_fields,
        )
    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 _pick_direct_payload_value(payload: dict[str, str], field_name: str) -> str:
    """Возвращает только прямое поле текущего XML-блока без suffix-поиска."""
    for key, value in payload.items():
        if value and _normalize_indexed_path(key) == field_name:
            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,
    operation_fields: dict[str, str],
    participant_fields: dict[str, str],
) -> 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_fields, participant_fields)
    _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_fields: dict[str, str],
    participant_fields: dict[str, str],
) -> None:
    """Правило 34: объединяет допустимые комментарии операции и участника."""
    operation_comment = _pick_direct_payload_value(operation_fields, "Коммент")
    participant_comment = ""
    if codes & EXPORT_SUBSIDIARY_COMMENT_CODES:
        participant_comment = _pick_direct_payload_value(
            participant_fields, "КомментУчастник"
        )

    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 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, "")