from __future__ import annotations import json import re from dataclasses import dataclass from pathlib import Path @dataclass(frozen=True) class MappingRule: column_index: int column_name: str candidate_keys: tuple[str, ...] source: str = "any" allow_direct_mapping: bool = True allow_short_lookup: bool = True @dataclass(frozen=True) class StructuredGroupRule: structured_column: int one_line_columns: tuple[int, ...] component_columns: tuple[int, ...] join_with_space: bool = False MAPPING_DATA_FILE = Path(__file__).resolve().with_name("mapping_table.json") TARGET_REPORT_COLUMNS_COUNT = 204 _MANUAL_COLUMN_NAME_FIXES: dict[int, str] = { 1: "Имя XML файла", 178: "Место государственной регистрации ЕИО/Бенефициара (одной строкой)", 183: "Номер", } _INVALID_TAGS = {"", "-", "Источник", "путь", "подразумеваются", "тэга", "нашла"} def _load_mapping_data() -> list[dict[str, str | int]]: raw = json.loads(MAPPING_DATA_FILE.read_text(encoding="utf-8")) if not isinstance(raw, list): return [] result: list[dict[str, str | int]] = [] for item in raw: if not isinstance(item, dict): continue idx = int(item.get("index", 0)) if not 1 <= idx <= TARGET_REPORT_COLUMNS_COUNT: continue column_name = str(item.get("column_name", "")).strip() xml_tag = str(item.get("xml_tag", "")).strip() xml_path = str(item.get("xml_path", "")).strip() if idx in _MANUAL_COLUMN_NAME_FIXES: column_name = _MANUAL_COLUMN_NAME_FIXES[idx] column_name = _cleanup_column_name(column_name) result.append( { "index": idx, "column_name": column_name, "xml_tag": xml_tag, "xml_path": xml_path, } ) return sorted(result, key=lambda row: int(row["index"])) def _cleanup_column_name(value: str) -> str: cleaned = re.sub(r"\s+", " ", value).strip() cleaned = re.sub(r"\s+не\s+нашла.*$", "", cleaned, flags=re.IGNORECASE) return cleaned.strip(" -") def _build_fixed_columns(mapping_data: list[dict[str, str | int]]) -> tuple[str, ...]: columns: list[str] = [] for row in mapping_data: column = str(row.get("column_name", "")).strip() columns.append(column or f"Колонка {row['index']}") if len(columns) < TARGET_REPORT_COLUMNS_COUNT: for index in range(len(columns) + 1, TARGET_REPORT_COLUMNS_COUNT + 1): columns.append(f"Колонка {index}") return tuple(columns[:TARGET_REPORT_COLUMNS_COUNT]) def _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: list[dict[str, str | int]]) -> tuple[MappingRule, ...]: rules: list[MappingRule] = [] participant_columns = { "Признак резидента участника", "Тип участника", "ИНН участника", "Наименование участника", "Счет участника", "Фамилия", "Имя", "Отчество", "ФИО (по структуре полностью)", "ФИО (в одной строке)", "ФИО ( в одной строке)", "Код страны гражданства", "Признак принадлежности к публичным лицам", } for row in mapping_data: index = int(row.get("index", 0)) column = str(row.get("column_name", "")).strip() tag = str(row.get("xml_tag", "")).strip() path = str(row.get("xml_path", "")).strip() if not column or tag in _INVALID_TAGS: continue is_structured_target = _is_structured_column(column) or index == 153 candidates = list(_extract_tag_candidate_keys(tag)) if column.startswith("ИНН") and index < 154: 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) allow_short_lookup = True if column == "БИК банка плательщика": candidates = list(_bank_bik_candidate_keys("СведБанкПлательщик")) allow_short_lookup = False elif column == "БИК банка получателя": candidates = list(_bank_bik_candidate_keys("СведБанкПолучатель")) allow_short_lookup = False elif column == "Код страны гражданства": candidates = [ candidate for candidate in [*path_candidates, *candidates] if candidate.endswith("СведФЛИП.КодОКСМ") ] allow_short_lookup = False elif index == 119: candidates = [ candidate for candidate in [*path_candidates, *candidates] if candidate.endswith("ИдентификацияФЛ") ] allow_short_lookup = False elif index == 123: candidates = [ candidate for candidate in [*path_candidates, *candidates] if candidate.endswith("ДатаРегЮЛ") or candidate.endswith("ДатаРождения") ] allow_short_lookup = False elif index == 152: candidates = [ "УчастникЮЛ.СведЮЛ.АдрРегЮЛ.АдресСтрока", "СообщОперКО.ИнформЧасть.СведКО.Операция.УчастникОП.УчастникЮЛ.СведЮЛ.АдрРегЮЛ.АдресСтрока", ] allow_short_lookup = False elif index == 153: base_paths = ( "УчастникЮЛ.СведЮЛ.АдрРегЮЛ", "СообщОперКО.ИнформЧасть.СведКО.Операция.УчастникОП.УчастникЮЛ.СведЮЛ.АдрРегЮЛ", ) address_tags = ( "Индекс", "КодОКСМ", "КодСубъектаПоОКАТО", "Район", "Пункт", "Улица", "Дом", "Корп", "Оф", ) candidates = [ f"{base_path}.{tag}" for base_path in base_paths for tag in address_tags ] allow_short_lookup = False elif index == 202: candidates = [ candidate for candidate in [*path_candidates, *candidates] if candidate.endswith("ИННЭмитентЦП") ] allow_short_lookup = False elif 49 <= index <= 101 or 155 <= index <= 204: allow_short_lookup = False 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) ] one_line_candidates = [ candidate for candidate in candidates if _is_one_line_candidate(candidate) ] direct_candidates = candidates allow_direct_mapping = True if is_structured_target and one_line_candidates and not non_one_line_candidates: allow_direct_mapping = False if is_structured_target and non_one_line_candidates: direct_candidates = non_one_line_candidates if 194 <= index <= 204 or index in {41, 43}: source = "operation" elif 155 <= index <= 193: source = "participant" elif index in {119, 122, 124}: source = "participant" else: source = ( "participant" if ( column in participant_columns or column.startswith("ИНН") or index in range(128, 140) ) else "any" ) rules.append( MappingRule( column_index=index, column_name=column, candidate_keys=tuple(direct_candidates), source=source, allow_direct_mapping=allow_direct_mapping, allow_short_lookup=allow_short_lookup, ) ) return tuple(rules) def _is_one_line_column(column_name: str) -> bool: normalized = column_name.lower().replace("ё", "е") return "одной строк" in normalized def _is_one_line_candidate(candidate: str) -> bool: return "строка" in candidate.lower() def _is_structured_column(column_name: str) -> bool: normalized = column_name.lower().replace("ё", "е") return "структур" in normalized or "(целый)" in normalized def _split_paths(path: str) -> tuple[str, ...]: result: list[str] = [] for chunk in path.split("|"): candidate = chunk.strip() if candidate: result.append(candidate.rstrip("/")) return tuple(result) def _common_prefix_depth(path_a: str, path_b: str) -> int: parts_a = [part for part in path_a.strip("/").split("/") if part] parts_b = [part for part in path_b.strip("/").split("/") if part] depth = 0 for left, right in zip(parts_a, parts_b): if left != right: break depth += 1 return depth def _paths_related( struct_paths: tuple[str, ...], candidate_paths: tuple[str, ...] ) -> bool: if not struct_paths or not candidate_paths: return False for struct_path in struct_paths: for candidate_path in candidate_paths: if _common_prefix_depth(struct_path, candidate_path) >= 6: return True return False def _is_path_under_any_base(path: str, base_paths: tuple[str, ...]) -> bool: for base_path in base_paths: if path == base_path or path.startswith(f"{base_path}/"): return True return False def _build_structured_group_rules( mapping_data: list[dict[str, str | int]], ) -> tuple[StructuredGroupRule, ...]: by_index: dict[int, dict[str, str | int]] = { int(item["index"]): item for item in mapping_data } sorted_indexes = sorted(by_index.keys()) rules: list[StructuredGroupRule] = [] for index in sorted_indexes: item = by_index[index] structured_name = str(item.get("column_name", "")).strip() if not _is_structured_column(structured_name): continue struct_paths = _split_paths(str(item.get("xml_path", ""))) struct_base_paths = tuple( path.rsplit("/", maxsplit=1)[0] for path in struct_paths if "/" in path ) is_address_structured = "адрес" in structured_name.lower().replace("ё", "е") one_line_indexes: list[int] = [] component_indexes: list[int] = [] # Поле "в одной строке" может располагаться слева от структурного. left = by_index.get(index - 1) if left: left_name = str(left.get("column_name", "")).strip() if _is_one_line_column(left_name): left_paths = _split_paths(str(left.get("xml_path", ""))) if _paths_related(struct_paths, left_paths): one_line_indexes.append(index - 1) right_boundary = next( ( candidate for candidate in sorted_indexes if candidate > index and _is_structured_column( str(by_index[candidate].get("column_name", "")).strip() ) ), None, ) block_end = right_boundary or (sorted_indexes[-1] + 1) for right_index in sorted_indexes: if right_index <= index or right_index >= block_end: continue right = by_index[right_index] right_name = str(right.get("column_name", "")).strip() right_tag = str(right.get("xml_tag", "")).strip() right_paths = _split_paths(str(right.get("xml_path", ""))) if is_address_structured: is_related = any( _is_path_under_any_base(candidate_path, struct_base_paths) for candidate_path in right_paths ) else: is_related = _paths_related(struct_paths, right_paths) if not is_related: if component_indexes: break continue if _is_one_line_column(right_name): one_line_indexes.append(right_index) if component_indexes: break continue if "|" in right_tag: continue component_indexes.append(right_index) if not component_indexes: continue rules.append( StructuredGroupRule( structured_column=index, one_line_columns=tuple(dict.fromkeys(one_line_indexes)), component_columns=tuple(dict.fromkeys(component_indexes)), join_with_space="фио" in structured_name.lower(), ) ) return tuple(rules) def _extract_tag_candidate_keys(tag: str) -> tuple[str, ...]: candidates: list[str] = [] for raw_tag in re.split(r"[|,]", tag): cleaned_tag = raw_tag.strip() if not cleaned_tag or cleaned_tag in _INVALID_TAGS: continue if cleaned_tag not in candidates: candidates.append(cleaned_tag) return tuple(candidates) def _extract_path_candidate_keys(path: str) -> tuple[str, ...]: candidates: list[str] = [] for raw_path in re.findall(r"/[^\s|]+", path): parts = [part for part in raw_path.strip("/").split("/") if part] if not parts: continue 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 = _load_mapping_data() FIXED_REPORT_COLUMNS: tuple[str, ...] = _build_fixed_columns(_MAPPING_DATA) REPORT_MAPPING_RULES: tuple[MappingRule, ...] = _build_rules(_MAPPING_DATA) STRUCTURED_GROUP_RULES: tuple[StructuredGroupRule, ...] = _build_structured_group_rules( _MAPPING_DATA ) _ACCOUNT_PLACEHOLDER = "00000000000000000000" _CURRENCY_OPERATION_CODES = frozenset(str(code) for code in range(6101, 6127)) 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], ) -> dict[int, str]: row: dict[int, str] = { 1: file_name, 7: record_id, } merged = _merge_fields(operation_fields, participant_fields) merged_suffix_index = _build_suffix_index(merged) operation_suffix_index = _build_suffix_index(operation_fields) participant_suffix_index = _build_suffix_index(participant_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 = participant_fields suffix_index = participant_suffix_index elif rule.source == "operation": source_payload = operation_fields suffix_index = operation_suffix_index else: source_payload = merged suffix_index = merged_suffix_index if _is_structured_column(rule.column_name) or rule.column_index == 153: 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 if rule.column_index == 153 else True ), ) else: row[rule.column_index] = _pick_value( source_payload, suffix_index, rule.candidate_keys, allow_short_lookup=rule.allow_short_lookup, ) _apply_participant_identity_rules(row, participant_fields) _apply_structured_group_rules(row) _apply_conditional_rules_1_74(row) _apply_conditional_rules_75_123(row) _apply_conditional_rules_124_203(row) if not has_eio_block: for index in range(155, 194): _set_row_value(row, index, "") if not has_cp_block: for index in range(194, 205): _set_row_value(row, index, "") for index in range(1, TARGET_REPORT_COLUMNS_COUNT + 1): row.setdefault(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], ) -> 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, ) by_name: dict[str, str] = {} for index, column_name in enumerate(FIXED_REPORT_COLUMNS, start=1): by_name.setdefault(column_name, by_index.get(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( _row_value(row, column_index).strip() for column_index in rule.one_line_columns ) if one_line_has_value: _set_row_value(row, rule.structured_column, "") for component_column_index in rule.component_columns: _set_row_value(row, component_column_index, "") continue values = [ _row_value(row, component_column_index).strip() for component_column_index in rule.component_columns if _row_value(row, component_column_index).strip() ] if not values: _set_row_value(row, rule.structured_column, "") continue separator = " " if rule.join_with_space else ", " _set_row_value(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(key.startswith(prefix) or nested_marker in key 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(): if value and (key == scoped_path or key.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(_row_value(row, 109)) 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_value(row, 119, identification) else: _set_row_value(row, 119, "") if has_legal_entity or participant_type == "1": 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_value(row, 122, value) def _has_any_prefixed_key(payload: dict[str, str], prefixes: tuple[str, ...]) -> bool: return any(any(key.startswith(prefix) for prefix in prefixes) for key in payload) 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 key in payload and payload[key]: return payload[key] if not allow_short_lookup: continue short = key.rsplit(".", maxsplit=1)[-1] if short in payload and payload[short]: return payload[short] if short in suffix_index: return suffix_index[short] return "" def _pick_structured_value( *, source_payload: dict[str, str], suffix_index: dict[str, str], candidates: tuple[str, ...], join_with_space: bool, 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 _column(index: int) -> str: return FIXED_REPORT_COLUMNS[index - 1] def _row_value(row: dict[int, str], index: int) -> str: return str(row.get(index, "")).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_value(row: dict[int, str], index: int, value: str) -> None: row[index] = value def _extract_numeric_codes(value: str) -> set[str]: return {match for match in re.findall(r"\d+", value)} def _operation_and_extra_codes(row: dict[int, str]) -> set[str]: op_code = _row_value(row, 19) extra_codes = _row_value(row, 20) return _extract_numeric_codes(op_code) | _extract_numeric_codes(extra_codes) def _apply_conditional_rules_1_74(row: dict[int, str]) -> None: codes = _operation_and_extra_codes(row) op_type_code = _row_value(row, 14) transfer_type = _row_value(row, 35) operator_type = _row_value(row, 37) # 10. Допустимые значения: 0..5, иначе очищаем. if _row_value(row, 10) not in {"0", "1", "2", "3", "4", "5"}: _set_row_value(row, 10, "") # 11. Заполняется только для приостановлений и валидных кодов основания. if _row_value(row, 11) not in {"1", "2", "3"}: _set_row_value(row, 11, "") # 15. Заполняется только для допустимых значений признака ЭСП. if _row_value(row, 15) not in {"1", "2", "3", "4"}: _set_row_value(row, 15, "") # 21. Коды необычной операции заполняются только при признаке 6001. if "6001" not in codes: _set_row_value(row, 21, "") # 22-24. Для операций с цифровыми правами (код 8) не заполняем. if op_type_code == "8": _set_row_value(row, 22, "") _set_row_value(row, 23, "") _set_row_value(row, 24, "") # 26. Сумма продаваемой валюты выводится только при конверсии и не равна нулю. conversion_amount = _row_value(row, 26) normalized_amount = conversion_amount.replace(",", ".") if not _row_value(row, 25) or re.fullmatch(r"[+-]?0+(?:\.0+)?", normalized_amount): _set_row_value(row, 26, "") # 27. Признак VO относится только к валютным операциям 6101-6126. if not codes & _CURRENCY_OPERATION_CODES: _set_row_value(row, 27, "") # 28. Идентификатор подозрительной деятельности только для 6001. if "6001" not in codes: _set_row_value(row, 28, "") # 30. Наименование драгметалла только для кода C99. if _row_value(row, 29).upper() != "C99": _set_row_value(row, 30, "") # 31. Предмет операции только для кодов 5020..5023. if not {"5020", "5021", "5022", "5023"} & codes: _set_row_value(row, 31, "") # 36. Код территории заполняется только для признака 5016. if "5016" not in codes: _set_row_value(row, 36, "") # 38. Счет плательщика обязателен для видов перевода 1/2/3/10 с заглушкой. if transfer_type in {"1", "2", "3", "10"}: if not _row_value(row, 38): _set_row_value(row, 38, _ACCOUNT_PLACEHOLDER) elif transfer_type in {"12", "13", "14"}: pass else: _set_row_value(row, 38, "") # 39 / 47. Идентификаторы ЭСП отсутствуют для признака операции 9. if op_type_code == "9": _set_row_value(row, 39, "") _set_row_value(row, 47, "") # 40. Банк плательщика заполняется только при типе оператора 2 или 4. if operator_type not in {"2", "4"}: _set_row_value(row, 40, "") # 42/43. Реквизиты банка получателя только при типе оператора 1 или 4. if operator_type not in {"1", "4"}: _set_row_value(row, 42, "") _set_row_value(row, 43, "") # 44. Корсчет банка плательщика с заглушкой там, где это применимо. if operator_type in {"3", "5"} or transfer_type in {"12", "13"}: _set_row_value(row, 44, "") elif not _row_value(row, 44): _set_row_value(row, 44, _ACCOUNT_PLACEHOLDER) # 45. Корсчет банка получателя с заглушкой там, где это применимо. if operator_type in {"3", "5"} or transfer_type in {"10", "11"}: _set_row_value(row, 45, "") elif not _row_value(row, 45): _set_row_value(row, 45, _ACCOUNT_PLACEHOLDER) # 46. Счет получателя с заглушкой для обязательных видов перевода. if transfer_type in {"1", "4", "7", "12"}: if not _row_value(row, 46): _set_row_value(row, 46, _ACCOUNT_PLACEHOLDER) elif transfer_type in {"10", "11", "14"}: pass else: _set_row_value(row, 46, "") # 62. Статус перевода заполняется только для видов 2/5/8. if transfer_type not in {"2", "5", "8"}: _set_row_value(row, 62, "") # Для безналичных операций не заполняем блоки приема/выдачи наличных. if transfer_type in {"1", "2", "3", "4", "7", "10", "11", "12", "13", "14"}: for column_index in (50, 61, 65): _set_row_value(row, column_index, "") def _apply_conditional_rules_75_123(row: dict[int, str]) -> None: codes = _operation_and_extra_codes(row) # 96. Код территории заполняется только для признака 5016 (дубль 36). if "5016" not in codes: _set_row_value(row, 96, "") # 103. Если операция без участия сотрудника и данных нет -> подставляем заглушку. if _row_value(row, 104) == "0" and not _row_value(row, 103): _set_row_value(row, 103, "Информация отсутствует") # 105. Аналогично для наименования иностранного банка. if _row_value(row, 104) == "0" and not _row_value(row, 105): _set_row_value(row, 105, "Информация отсутствует") transfer_type = _row_value(row, 35) if transfer_type in {"1", "2", "3", "4", "7", "10", "11", "12", "13", "14"}: for column_index in (76, 84, 95, 99, 101): _set_row_value(row, column_index, "") participant_type = _normalize_code(_row_value(row, 109)) # 119. Для ЮЛ и ФЛ поле ППЦР не заполняется. if participant_type in {"1", "2"}: _set_row_value(row, 119, "") # 122. Для ФЛ идентификация не подставляется вместо отсутствующего КПП. if participant_type == "2": _set_row_value(row, 122, "") def _allow_one_line_address(row: dict[int, str]) -> bool: resident_flag = _row_value(row, 110) client_flag = _row_value(row, 111) 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 _apply_conditional_rules_124_203(row: dict[int, str]) -> None: participant_type = _normalize_code(_row_value(row, 109)) # 124. Для ФЛ, ИП и ФЛЧП показатель отсутствует. if participant_type in {"2", "3", "4"}: _set_row_value(row, 124, "") # 125. Для ИП СНИЛС не выводится. if participant_type == "3": insurance_value = _row_value(row, 125) if re.fullmatch(r"\d{3}-?\d{3}-?\d{3}\s?\d{2}", insurance_value): _set_row_value(row, 125, "") # 140/151/177. Однострочные адреса только для нерезидента, # неопределенного резидентства или резидента-не клиента. if not _allow_one_line_address(row): _set_row_value(row, 141, "") _set_row_value(row, 152, "") _set_row_value(row, 178, "") # Правило КодОКСМ/КодОКАТО в адресных блоках: # ОКАТО заполняется только при российском коде страны 643. okato_country_pairs = ( (52, 53), (67, 68), (86, 87), (144, 145), (170, 171), ) for country_col, okato_col in okato_country_pairs: if _row_value(row, country_col) != "643": _set_row_value(row, okato_col, "") # 162. Для ФЛ показатель отсутствует. if participant_type == "2": _set_row_value(row, 163, "")