diff --git a/app/pipeline/mapping.py b/app/pipeline/mapping.py index 3b9eb58..a23311b 100644 --- a/app/pipeline/mapping.py +++ b/app/pipeline/mapping.py @@ -95,6 +95,7 @@ def _build_rules(mapping_data: list[dict[str, str | int]]) -> tuple[MappingRule, "ФИО ( в одной строке)", } 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() @@ -133,7 +134,12 @@ def _build_rules(mapping_data: list[dict[str, str | int]]) -> tuple[MappingRule, candidate_keys=tuple(direct_candidates), source=( "participant" - if column in participant_columns or column.startswith("ИНН") + if ( + column in participant_columns + or column.startswith("ИНН") + or index in range(128, 140) + or index in range(180, 193) + ) else "any" ), allow_direct_mapping=allow_direct_mapping, @@ -338,6 +344,7 @@ def build_fixed_row( _apply_structured_group_rules(row) _apply_conditional_rules_1_74(row) _apply_conditional_rules_75_123(row) + _apply_conditional_rules_124_203(row) for column in FIXED_REPORT_COLUMNS: row.setdefault(column, "") return row @@ -551,3 +558,49 @@ def _apply_conditional_rules_75_123(row: dict[str, str]) -> None: # 105. Аналогично для наименования иностранного банка. if _row_value(row, 104) == "0" and not _row_value(row, 105): _set_row_value(row, 105, "Информация отсутствует") + + +def _allow_one_line_address(row: dict[str, 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[str, str]) -> None: + participant_type = _row_value(row, 109) + + # 124. Для ФЛ и ИП показатель отсутствует. + if participant_type in {"2", "3"}: + _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, "") diff --git a/app/pipeline/report.py b/app/pipeline/report.py index 2a8e90e..8acc0db 100644 --- a/app/pipeline/report.py +++ b/app/pipeline/report.py @@ -1,5 +1,6 @@ from __future__ import annotations +import re import shutil from dataclasses import dataclass from datetime import date, datetime @@ -60,11 +61,11 @@ TOP_GROUP_HEADERS: tuple[GroupHeader, ...] = ( ), GroupHeader( start_col=87, - end_col=121, + end_col=126, title="Участник операции (базовые сведения)", ), GroupHeader( - start_col=122, + start_col=127, end_col=154, title="Участник операции (идентификация, документы, адреса)", ), @@ -148,6 +149,12 @@ class StreamingReportWriter: self.sheet.write_datetime( self._next_data_row, index, value, cell_format ) + elif self._is_text_identifier_column(column_name): + self.sheet.write_string( + self._next_data_row, + index, + self._normalize_identifier_string(value), + ) elif cell_format is not None: self.sheet.write(self._next_data_row, index, value, cell_format) else: @@ -209,6 +216,18 @@ class StreamingReportWriter: return raw_value, None return raw_value, default_format + def _is_text_identifier_column(self, column_name: str) -> bool: + keywords = ("ИНН", "ОГРН", "КПП", "БИК", "СВИФТ") + return any(keyword in column_name for keyword in keywords) + + def _normalize_identifier_string(self, value: object) -> str: + if value is None: + return "" + text = str(value).strip() + if re.fullmatch(r"\d+\.0+", text): + return text.split(".", maxsplit=1)[0] + return text + def _parse_date_value(self, value: object) -> datetime | None: if isinstance(value, datetime): return value diff --git a/tests/unit/test_mapping.py b/tests/unit/test_mapping.py index 88e3a38..1097fe3 100644 --- a/tests/unit/test_mapping.py +++ b/tests/unit/test_mapping.py @@ -436,3 +436,115 @@ def test_patronymic_supports_typo_tag_ocht() -> None: }, ) assert row["Отчество"] == "Каримович" + + +def test_col124_cleared_for_physical_person_and_ip() -> None: + row_fl = build_fixed_row( + file_name="f.xml", + record_id="R1", + operation_index=1, + operation_fields={"ТипФЛЧастнаяПрактика": "1"}, + participant_fields={"ТипУчастника": "2"}, + ) + assert row_fl["Признак филиала: / Тип ФЛЧП"] == "" + + row_ip = build_fixed_row( + file_name="f.xml", + record_id="R1", + operation_index=1, + operation_fields={"ТипФЛЧастнаяПрактика": "1"}, + participant_fields={"ТипУчастника": "3"}, + ) + assert row_ip["Признак филиала: / Тип ФЛЧП"] == "" + + +def test_col125_snils_cleared_for_ip() -> None: + row = build_fixed_row( + file_name="f.xml", + record_id="R1", + operation_index=1, + operation_fields={}, + participant_fields={"ТипУчастника": "3", "СНИЛСФЛ": "123-456-789 00"}, + ) + assert row["ОМС/СНИЛС/Номер телефона"] == "" + + +def test_one_line_address_columns_cleared_when_participant_is_resident_client() -> None: + row = build_fixed_row( + file_name="f.xml", + record_id="R1", + operation_index=1, + operation_fields={ + "ПризнУчастника": "1", + "ПризнКлиент": "1", + "АдресСтрока": "г. Москва, ул. Тестовая, д. 1", + }, + participant_fields={}, + ) + assert row["Адрес в одной строке"] == "" + assert row["Место государственной регистрации ( в одной строке)"] == "" + assert ( + row["Место государственной регистрации ЕИО/Бенефициара (одной строкой)"] == "" + ) + + +def test_okato_cleared_when_country_code_is_not_russia() -> None: + row = build_fixed_row( + file_name="f.xml", + record_id="R1", + operation_index=1, + operation_fields={ + "КодОКСМ": "840", + "КодСубъектаПоОКАТО": "45", + }, + participant_fields={}, + ) + assert row["Код ОКАТО"] == "" + assert row["ОКАТО"] == "" + + +def test_col163_ogrn_eio_benef_cleared_for_physical_person() -> None: + row = build_fixed_row( + file_name="f.xml", + record_id="R1", + operation_index=1, + operation_fields={"ОГРНИП": "123456789012345"}, + participant_fields={"ТипУчастника": "2"}, + ) + assert row["ОГРН ЕИО/Бенефициара"] == "" + + +def test_dul_sign_for_legal_entity_does_not_use_operation_doc_fields() -> None: + row = build_fixed_row( + file_name="f.xml", + record_id="R1", + operation_index=1, + operation_fields={"ДокУдЛичн": "451099", "ТипУчастника": "1"}, + participant_fields={}, + ) + assert row["Признак ДУЛ"] == "" + + +def test_dul_columns_128_134_keep_expected_positions() -> None: + row = build_fixed_row( + file_name="f.xml", + record_id="R1", + operation_index=1, + operation_fields={}, + participant_fields={ + "ВидДокКод": "21", + "СерияДок": "0104", + "НомДок": "476635", + "КемВыданДок": "ОВД", + "КодПодр": "222-022", + "ДатВыдачиДок": "23/08/2005", + "ДокУдЛичн": "1", + }, + ) + assert row["ДУЛ (структура полностью)"] == "1, 21, 0104, 476635, ОВД, 222-022" + assert row["Признак ДУЛ"] == "1" + assert row["Код вида документа/Наименование"] == "21" + assert row["Серия"] == "0104" + assert row["Номер"] == "476635" + assert row["Орган, выдавший документ"] == "ОВД" + assert row["КП"] == "222-022" diff --git a/tests/unit/test_report.py b/tests/unit/test_report.py index ce1bc64..d1aa911 100644 --- a/tests/unit/test_report.py +++ b/tests/unit/test_report.py @@ -110,6 +110,31 @@ def test_write_report_keeps_invalid_date_as_text_without_date_format( assert cell.number_format != "DD.MM.YYYY" +def test_write_report_writes_inn_like_values_as_text(tmp_path: Path) -> None: + destination = tmp_path / "report.xlsx" + rows = [ + ReportRow( + file_name="f.xml", + record_id="RID", + operation_index=1, + operation_fields={}, + participant_fields={"ИННЮЛ": 2820000210.0}, + ) + ] + + write_report(rows, destination) + + workbook = load_workbook(destination) + sheet = workbook["Отчет"] + inn_col_idx = next( + index + for index, name in enumerate(FIXED_REPORT_COLUMNS, start=1) + if "ИНН" in name + ) + cell = sheet.cell(row=3, column=inn_col_idx) + assert cell.value == "2820000210" + + def _column_letter(index: int) -> str: value = index result = ""