diff --git a/app/pipeline/mapping.py b/app/pipeline/mapping.py index 38f0cb1..8b54749 100644 --- a/app/pipeline/mapping.py +++ b/app/pipeline/mapping.py @@ -11,6 +11,15 @@ class MappingRule: column_name: str candidate_keys: tuple[str, ...] source: str = "any" + allow_direct_mapping: bool = True + + +@dataclass(frozen=True) +class StructuredGroupRule: + structured_column: str + one_line_columns: tuple[str, ...] + component_columns: tuple[str, ...] + join_with_space: bool = False MAPPING_DATA_FILE = Path(__file__).resolve().with_name("mapping_table.json") @@ -18,7 +27,7 @@ TARGET_REPORT_COLUMNS_COUNT = 204 _MANUAL_COLUMN_NAME_FIXES: dict[int, str] = { 1: "Имя XML файла", - 178: "Место государственной регистрации ЕИО/Бенефициара (структура полностью)", + 178: "Место государственной регистрации ЕИО/Бенефициара (одной строкой)", 183: "Номер", } @@ -78,6 +87,12 @@ def _build_rules(mapping_data: list[dict[str, str | int]]) -> tuple[MappingRule, "ИНН участника", "Наименование участника", "Счет участника", + "Фамилия", + "Имя", + "Отчество", + "ФИО (по структуре полностью)", + "ФИО (в одной строке)", + "ФИО ( в одной строке)", } for row in mapping_data: column = str(row.get("column_name", "")).strip() @@ -85,6 +100,7 @@ def _build_rules(mapping_data: list[dict[str, str | int]]) -> tuple[MappingRule, path = str(row.get("xml_path", "")).strip() if not column or tag in _INVALID_TAGS: continue + is_structured_target = _is_structured_column(column) candidates = list(_extract_tag_candidate_keys(tag)) if column.startswith("ИНН"): candidates = ["ИННФЛИП", "ИННФЛ", "ИННЮЛ", "ИНН", *candidates] @@ -94,15 +110,149 @@ def _build_rules(mapping_data: list[dict[str, str | int]]) -> tuple[MappingRule, for path_tag in _extract_path_candidate_keys(path): if path_tag not in candidates: candidates.append(path_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 rules.append( MappingRule( column_name=column, - candidate_keys=tuple(candidates), + candidate_keys=tuple(direct_candidates), source=( "participant" if column in participant_columns or column.startswith("ИНН") else "any" ), + allow_direct_mapping=allow_direct_mapping, + ) + ) + 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 _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", ""))) + one_line_names: list[str] = [] + component_names: list[str] = [] + + # Left-side "one line" can be immediately before structured column. + 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_names.append(left_name) + + 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 not _paths_related(struct_paths, right_paths): + if component_names: + break + continue + if _is_one_line_column(right_name): + one_line_names.append(right_name) + continue + if "|" in right_tag: + continue + component_names.append(right_name) + + if not component_names: + continue + + rules.append( + StructuredGroupRule( + structured_column=structured_name, + one_line_columns=tuple(dict.fromkeys(one_line_names)), + component_columns=tuple(dict.fromkeys(component_names)), + join_with_space="фио" in structured_name.lower(), ) ) return tuple(rules) @@ -137,6 +287,9 @@ def _extract_path_candidate_keys(path: str) -> tuple[str, ...]: _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 +) def build_fixed_row( @@ -158,20 +311,57 @@ def build_fixed_row( for rule in REPORT_MAPPING_RULES: if row.get(rule.column_name): continue + if not rule.allow_direct_mapping: + row[rule.column_name] = "" + continue if rule.source == "participant": source_payload = participant_fields suffix_index = participant_suffix_index else: source_payload = merged suffix_index = merged_suffix_index - row[rule.column_name] = _pick_value( - source_payload, suffix_index, rule.candidate_keys - ) + if _is_structured_column(rule.column_name): + row[rule.column_name] = _pick_structured_value( + source_payload=source_payload, + suffix_index=suffix_index, + candidates=rule.candidate_keys, + join_with_space="фио" in rule.column_name.lower(), + ) + else: + row[rule.column_name] = _pick_value( + source_payload, suffix_index, rule.candidate_keys + ) + _apply_structured_group_rules(row) for column in FIXED_REPORT_COLUMNS: row.setdefault(column, "") return row +def _apply_structured_group_rules(row: dict[str, str]) -> None: + for rule in STRUCTURED_GROUP_RULES: + one_line_has_value = any( + str(row.get(column_name, "")).strip() + for column_name in rule.one_line_columns + ) + if one_line_has_value: + row[rule.structured_column] = "" + for component_column in rule.component_columns: + row[component_column] = "" + continue + + values = [ + str(row.get(component_column, "")).strip() + for component_column in rule.component_columns + if str(row.get(component_column, "")).strip() + ] + if not values: + row[rule.structured_column] = "" + continue + + separator = " " if rule.join_with_space else ", " + row[rule.structured_column] = separator.join(values) + + def _merge_fields( operation_fields: dict[str, str], participant_fields: dict[str, str], @@ -206,3 +396,21 @@ def _pick_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, +) -> str: + values: list[str] = [] + for key in candidates: + value = _pick_value(source_payload, suffix_index, (key,)) + if value and value not in values: + values.append(value) + if not values: + return "" + separator = " " if join_with_space else ", " + return separator.join(values) diff --git a/app/pipeline/report.py b/app/pipeline/report.py index 45cc655..7a42955 100644 --- a/app/pipeline/report.py +++ b/app/pipeline/report.py @@ -1,12 +1,12 @@ from __future__ import annotations +import shutil from dataclasses import dataclass from pathlib import Path +from tempfile import NamedTemporaryFile from typing import Iterable -from openpyxl import Workbook -from openpyxl.cell import WriteOnlyCell -from openpyxl.styles import Alignment, Font, PatternFill +import xlsxwriter from .mapping import FIXED_REPORT_COLUMNS, build_fixed_row @@ -20,14 +20,110 @@ class ReportRow: participant_fields: dict[str, str] +@dataclass(frozen=True) +class GroupHeader: + start_col: int + end_col: int + title: str + + +# Grouped upper header from the template report for readability. +TOP_GROUP_HEADERS: tuple[GroupHeader, ...] = ( + GroupHeader( + start_col=1, + end_col=4, + title="Информация о КО, передающей сведения в уполномоченный орган", + ), + GroupHeader(start_col=5, end_col=8, title="Сведения об уполномоченном лице"), + GroupHeader( + start_col=11, + end_col=13, + title="Информация о филиале КО, передающем сведения в уполномоченный орган", + ), + GroupHeader(start_col=14, end_col=36, title="Информация об операции"), + GroupHeader( + start_col=37, + end_col=48, + title="Сведения о переводах денежных средств, в том числе электронных денежных средств", + ), + GroupHeader( + start_col=49, + end_col=53, + title="Сведения о месте приема наличных денежных средств", + ), + GroupHeader( + start_col=55, + end_col=59, + title="Сведения о месте выдачи наличных денежных средств", + ), + GroupHeader( + start_col=60, + end_col=64, + title="Сведения о месте авторизации ЭСП", + ), + GroupHeader( + start_col=67, + end_col=69, + title=( + "Сведения о внесении наличных денежных средств на свой банковский счет " + "или о получении наличных денежных средств со своего банковского счета " + "у одного оператора по переводу денежных средств" + ), + ), + GroupHeader( + start_col=70, + end_col=74, + title="Сведения о месте приема (выдачи) наличных денежных средств", + ), + GroupHeader( + start_col=76, + end_col=80, + title="Сведения о месте совершения операции с денежными средствами в наличной форме", + ), + GroupHeader(start_col=87, end_col=104, title="Сведения об участниках операции"), + GroupHeader(start_col=105, end_col=118, title="Сведения ЕИО"), +) + + class StreamingReportWriter: def __init__(self) -> None: self.columns = list(FIXED_REPORT_COLUMNS) + temp_file = NamedTemporaryFile(suffix=".xlsx", delete=False) + self._temp_report_path = Path(temp_file.name) + temp_file.close() + self.workbook = xlsxwriter.Workbook( + str(self._temp_report_path), + {"constant_memory": True}, + ) + self._is_closed = False + self.sheet = self.workbook.add_worksheet("Отчет") + self._header_format = self.workbook.add_format( + { + "bold": True, + "align": "center", + "valign": "vcenter", + "text_wrap": True, + "bg_color": "#E2F0D9", + } + ) + self._group_format = self.workbook.add_format( + { + "bold": True, + "align": "center", + "valign": "vcenter", + "text_wrap": True, + "bg_color": "#FFFFFF", + } + ) + self._date_format = self.workbook.add_format({"num_format": "DD.MM.YYYY"}) + self._sum_format = self.workbook.add_format({"num_format": "#,##0.00"}) self._column_formats = [self._resolve_format(column) for column in self.columns] - self.workbook = Workbook(write_only=True) - self.sheet = self.workbook.create_sheet(title="Отчет") - self.sheet.freeze_panes = "A2" - self.sheet.append(self._build_header_row()) + self._build_group_header_row() + self._build_header_row() + # Freeze both header rows. + self.sheet.freeze_panes(2, 0) + self.sheet.autofilter(1, 0, 1, len(self.columns) - 1) + self._next_data_row = 2 def append_row(self, row: ReportRow) -> None: data = build_fixed_row( @@ -40,41 +136,66 @@ class StreamingReportWriter: values: list[object] = [] for index, column_name in enumerate(self.columns): raw_value = data.get(column_name, "") - number_format = self._column_formats[index] - if number_format is None: - values.append(raw_value) - continue - cell = WriteOnlyCell(self.sheet, value=raw_value) - cell.number_format = number_format - values.append(cell) - self.sheet.append(values) + values.append(raw_value) + for index, value in enumerate(values): + cell_format = self._column_formats[index] + if cell_format is not None: + self.sheet.write(self._next_data_row, index, value, cell_format) + else: + self.sheet.write(self._next_data_row, index, value) + self._next_data_row += 1 def save(self, destination: Path) -> None: destination.parent.mkdir(parents=True, exist_ok=True) - self.workbook.save(destination) + try: + if not self._is_closed: + self.workbook.close() + self._is_closed = True + shutil.copyfile(self._temp_report_path, destination) + finally: + self._cleanup_temp_file() - def _build_header_row(self) -> list[WriteOnlyCell]: - fill = PatternFill(start_color="E2F0D9", end_color="E2F0D9", fill_type="solid") - result: list[WriteOnlyCell] = [] - for name in self.columns: - cell = WriteOnlyCell(self.sheet, value=name) - cell.font = Font(bold=True) - cell.alignment = Alignment( - horizontal="center", - vertical="center", - wrap_text=True, + def _build_group_header_row(self) -> None: + for header in TOP_GROUP_HEADERS: + if header.start_col > len(self.columns): + continue + first_col = header.start_col - 1 + last_col = min(header.end_col, len(self.columns)) - 1 + self.sheet.merge_range( + 0, + first_col, + 0, + last_col, + header.title, + self._group_format, ) - cell.fill = fill - result.append(cell) - return result - def _resolve_format(self, column_name: str) -> str | None: + def _build_header_row(self) -> None: + for index, name in enumerate(self.columns): + self.sheet.write(1, index, name, self._header_format) + + def _resolve_format(self, column_name: str): if "Дата" in column_name: - return "DD.MM.YYYY" + return self._date_format if "Сумм" in column_name or "Сумма" in column_name: - return "#,##0.00" + return self._sum_format return None + def _cleanup_temp_file(self) -> None: + try: + self._temp_report_path.unlink(missing_ok=True) + except Exception: # noqa: BLE001 + pass + + def __del__(self) -> None: + if not getattr(self, "_is_closed", True): + try: + self.workbook.close() + except Exception: # noqa: BLE001 + pass + self._is_closed = True + self._cleanup_temp_file() + def write_report(rows: Iterable[ReportRow], destination: Path) -> None: writer = StreamingReportWriter() diff --git a/requirements.txt b/requirements.txt index 5676b34..044e41e 100644 --- a/requirements.txt +++ b/requirements.txt @@ -6,5 +6,6 @@ black==24.4.2 isort==5.13.2 pre-commit==3.7.1 openpyxl +xlsxwriter==3.2.9 pytest smbprotocol==1.15.0 diff --git a/tests/unit/test_mapping.py b/tests/unit/test_mapping.py index 8138483..c359b99 100644 --- a/tests/unit/test_mapping.py +++ b/tests/unit/test_mapping.py @@ -150,3 +150,147 @@ def test_build_fixed_row_skips_empty_suffix_values_in_index() -> None: }, ) assert row[_column_with("ИНН ")] == "222222222222" + + +def test_structured_address_is_built_from_right_columns() -> None: + row = build_fixed_row( + file_name="f.xml", + record_id="R1", + operation_index=1, + operation_fields={}, + participant_fields={ + "Индекс": "674500", + "КодОКСМ": "643", + "Пункт": "Москва", + "Улица": "Тверская", + "Дом": "1", + }, + ) + assert row["Адрес по структуре (целый)"] == "674500, 643, Москва, Тверская, 1" + + +def test_one_line_address_clears_structured_and_component_columns() -> None: + row = build_fixed_row( + file_name="f.xml", + record_id="R1", + operation_index=1, + operation_fields={}, + participant_fields={ + "АдресСтрока": "г. Москва, Тверская, 1", + "Индекс": "674500", + "КодОКСМ": "643", + "Улица": "Тверская", + "Дом": "1", + }, + ) + assert row["Адрес в одной строке"] == "г. Москва, Тверская, 1" + assert row["Адрес по структуре (целый)"] == "" + assert row["Индекс"] == "" + assert row["Код страны"] == "" + assert row["Улица"] == "" + assert row["Дом"] == "" + + +def test_fio_columns_prefer_participant_fields_over_operation_common_fields() -> None: + row = build_fixed_row( + file_name="f.xml", + record_id="R1", + operation_index=1, + operation_fields={ + "Фам": "Кузнецов", + "Имя": "Сергей", + "Отч": "Владимирович", + }, + participant_fields={ + "УчастникФЛИП.СведФЛИП.ФИОФЛИП.Фам": "Азимов", + "УчастникФЛИП.СведФЛИП.ФИОФЛИП.Имя": "Ислом", + "УчастникФЛИП.СведФЛИП.ФИОФЛИП.Отч": "Каримович", + }, + ) + assert row["Фамилия"] == "Азимов" + assert row["Имя"] == "Ислом" + assert row["Отчество"] == "Каримович" + assert row["ФИО (по структуре полностью)"] == "Азимов Ислом Каримович" + + +def test_one_line_fio_clears_structured_and_component_fio_columns() -> None: + row = build_fixed_row( + file_name="f.xml", + record_id="R1", + operation_index=1, + operation_fields={}, + participant_fields={ + "ФИОСтрока": "Азимов Ислом Каримович", + "Фам": "Азимов", + "Имя": "Ислом", + "Отч": "Каримович", + }, + ) + assert row["ФИО (в одной строке)"] == "Азимов Ислом Каримович" + assert row["ФИО (по структуре полностью)"] == "" + assert row["Фамилия"] == "" + assert row["Имя"] == "" + assert row["Отчество"] == "" + assert row["Наименование ЮЛ/ФИО ( по структуре полный)"] == "" + + +def test_structured_column_is_not_filled_directly_from_single_line_tag() -> None: + row = build_fixed_row( + file_name="f.xml", + record_id="R1", + operation_index=1, + operation_fields={}, + participant_fields={ + "АдресСтрока": "г. Москва, ул. Пример, д. 1", + }, + ) + assert row["Адрес в одной строке"] == "г. Москва, ул. Пример, д. 1" + assert ( + row[ + "Адрес ЮЛ/ Адрес места жительства (места нахождения) " + "ЕИО/Бенефициара (по структуре полный)" + ] + == "" + ) + + +def test_structured_column_without_component_block_joins_candidate_values() -> None: + row = build_fixed_row( + file_name="f.xml", + record_id="R1", + operation_index=1, + operation_fields={}, + participant_fields={ + "Индекс": "121467", + "КодОКСМ": "643", + "Пункт": "Москва", + "Улица": "Молодогвардейская", + "Дом": "8", + "Корп": "1", + "Оф": "28", + }, + ) + assert ( + row["Место государственной регистрации ЕИО/Бенефициара (структура полностью)"] + == "121467, 643, Москва, Молодогвардейская, 8, 1, 28" + ) + + +def test_structured_column_uses_non_one_line_candidate_when_mixed_tags() -> None: + row = build_fixed_row( + file_name="f.xml", + record_id="R1", + operation_index=1, + operation_fields={}, + participant_fields={ + "АдрУчредитель": "г. Москва, ул. Пушкина, д. 1", + }, + ) + assert row["Адрес в одной строке"] == "" + assert ( + row[ + "Адрес ЮЛ/ Адрес места жительства (места нахождения) " + "ЕИО/Бенефициара (по структуре полный)" + ] + == "г. Москва, ул. Пушкина, д. 1" + ) diff --git a/tests/unit/test_report.py b/tests/unit/test_report.py index 759090a..f00f35c 100644 --- a/tests/unit/test_report.py +++ b/tests/unit/test_report.py @@ -5,7 +5,7 @@ from pathlib import Path from openpyxl import load_workbook from app.pipeline.mapping import FIXED_REPORT_COLUMNS -from app.pipeline.report import ReportRow, write_report +from app.pipeline.report import TOP_GROUP_HEADERS, ReportRow, write_report def test_write_report_streaming_preserves_header_and_formats(tmp_path: Path) -> None: @@ -24,10 +24,11 @@ def test_write_report_streaming_preserves_header_and_formats(tmp_path: Path) -> workbook = load_workbook(destination) sheet = workbook["Отчет"] - headers = [cell.value for cell in sheet[1]] + headers = [cell.value for cell in sheet[2]] assert headers == list(FIXED_REPORT_COLUMNS) + assert sheet.freeze_panes == "A3" row_values = [ - sheet.cell(row=2, column=index).value + sheet.cell(row=3, column=index).value for index in range(1, len(FIXED_REPORT_COLUMNS) + 1) ] assert "SKO115FZ_01_123456789_20260616_X00001.xml" in row_values @@ -41,7 +42,7 @@ def test_write_report_streaming_preserves_header_and_formats(tmp_path: Path) -> None, ) if date_col_idx is not None: - assert sheet.cell(row=2, column=date_col_idx).number_format == "DD.MM.YYYY" + assert sheet.cell(row=3, column=date_col_idx).number_format == "DD.MM.YYYY" sum_col_idx = next( ( @@ -52,4 +53,34 @@ def test_write_report_streaming_preserves_header_and_formats(tmp_path: Path) -> None, ) if sum_col_idx is not None: - assert sheet.cell(row=2, column=sum_col_idx).number_format == "#,##0.00" + assert sheet.cell(row=3, column=sum_col_idx).number_format == "#,##0.00" + + +def test_write_report_contains_top_group_header_row(tmp_path: Path) -> None: + destination = tmp_path / "report.xlsx" + write_report([], destination) + + workbook = load_workbook(destination) + sheet = workbook["Отчет"] + assert sheet.cell(row=1, column=1).value == ( + "Информация о КО, передающей сведения в уполномоченный орган" + ) + merged_ranges = {str(rng) for rng in sheet.merged_cells.ranges} + expected_ranges = set() + for header in TOP_GROUP_HEADERS: + if header.start_col > len(FIXED_REPORT_COLUMNS): + continue + end_col = min(header.end_col, len(FIXED_REPORT_COLUMNS)) + expected_ranges.add( + f"{_column_letter(header.start_col)}1:{_column_letter(end_col)}1" + ) + assert merged_ranges == expected_ranges + + +def _column_letter(index: int) -> str: + value = index + result = "" + while value: + value, remainder = divmod(value - 1, 26) + result = chr(65 + remainder) + result + return result