---

utils/helpers/lds_status_report_xlsx_utils.py
"""
Утилиты для разбора xlsx-отчёта о режиме работы СОУ.
"""

from __future__ import annotations

import re
import tempfile
from dataclasses import dataclass, field
from datetime import datetime, time, timedelta
from pathlib import Path
from typing import Dict, List, Optional, Tuple

import allure
from openpyxl import load_workbook
from openpyxl.worksheet.worksheet import Worksheet

from constants.test_constants import BaseTN3Constants as TestConst
from constants.test_constants import ExportLdsStatusReportConstants as LdsReportConst
from constants.test_constants import ExportReportConstants as ReportConst
from utils.helpers.report_xlsx_utils import (
    ReportTitleInfo,
    _stringify_cell,
    attach_report_file_to_allure,
    build_column_cells,
    is_xlsx_extension,
    is_xlsx_file_bytes,
    normalize_report_period_naive,
    parse_report_datetime,
    report_period_comparison_bounds,
)
from utils.helpers.ws_test_utils import localize_as_moscow


@dataclass
class LdsStatusReportSectionRow:
    """Строка участка с длительностями режимов СОУ."""

    row_index: int
    section_name: str
    cells: Dict[str, str] = field(default_factory=dict)

    @property
    def mode_durations_seconds(self) -> Dict[str, int]:
        return {
            column_name: parse_duration_seconds(self.cells.get(column_name)) or 0
            for column_name in LdsReportConst.MODE_DURATION_COLUMNS
        }

    @property
    def modes_sum_seconds(self) -> int:
        return sum(self.mode_durations_seconds.values())


@dataclass
class LdsStatusReportParsed:
    """Разобранный отчёт о режиме работы СОУ."""

    title_info: ReportTitleInfo
    column_headers: List[str]
    section_rows: List[LdsStatusReportSectionRow]
    total_duration_seconds: Optional[int] = None
    total_duration_raw: str = ""
    total_label_row_index: Optional[int] = None


def parse_duration_seconds(value: object) -> Optional[int]:
    """Парсит длительность из ячейки (H:MM:SS, MM:SS или time/timedelta из Excel)."""
    if value is None:
        return None
    if isinstance(value, timedelta):
        return int(value.total_seconds())
    if isinstance(value, time):
        return (
            value.hour * TestConst.SECONDS_PER_HOUR
            + value.minute * TestConst.SEC_PER_MIN
            + value.second
        )
    if isinstance(value, datetime):
        return (
            value.hour * TestConst.SECONDS_PER_HOUR
            + value.minute * TestConst.SEC_PER_MIN
            + value.second
        )

    duration_text = _stringify_cell(value).strip()
    if not duration_text:
        return None

    parts = duration_text.split(":")
    try:
        if len(parts) == LdsReportConst.DURATION_PARTS_COUNT_H_MM_SS:
            hours, minutes, seconds = (int(part) for part in parts)
            return (
                hours * TestConst.SECONDS_PER_HOUR
                + minutes * TestConst.SEC_PER_MIN
                + seconds
            )
        if len(parts) == LdsReportConst.DURATION_PARTS_COUNT_MM_SS:
            minutes, seconds = (int(part) for part in parts)
            return minutes * TestConst.SEC_PER_MIN + seconds
    except ValueError:
        return None
    return None


def is_duration_cell_filled(value: object) -> bool:
    """Ячейка с длительностью заполнена (допускается 0:00:00)."""
    return parse_duration_seconds(value) is not None


def format_duration_seconds(total_seconds: int) -> str:
    hours, remainder = divmod(total_seconds, TestConst.SECONDS_PER_HOUR)
    minutes, seconds = divmod(remainder, TestConst.SEC_PER_MIN)
    return f"{hours}:{minutes:02d}:{seconds:02d}"


def parse_lds_status_report_title(title_raw: object) -> ReportTitleInfo:
    title_str = _stringify_cell(title_raw)
    match = re.search(LdsReportConst.REPORT_HEADER_PERIOD_PATTERN, title_str)
    if match is None:
        return ReportTitleInfo(raw_title=title_str)

    return ReportTitleInfo(
        raw_title=title_str,
        period_start=parse_report_datetime(match.group("period_start")),
        period_end=parse_report_datetime(match.group("period_end")),
    )


def build_lds_status_report_file_name(
    tu_description: str,
    period_start: datetime,
    period_end: datetime,
) -> str:
    start_text = normalize_report_period_naive(period_start).strftime(ReportConst.REPORT_FILE_NAME_DATETIME_FORMAT)
    end_text = normalize_report_period_naive(period_end).strftime(ReportConst.REPORT_FILE_NAME_DATETIME_FORMAT)
    return (
        f"{LdsReportConst.LDS_STATUS_REPORT_NAME_PART}. {tu_description} {start_text} - {end_text}"
        f"{ReportConst.XLSX_EXTENSION}"
    )


def parse_period_from_lds_status_report_file_name(file_name: str) -> Tuple[Optional[datetime], Optional[datetime]]:
    match = re.search(LdsReportConst.REPORT_FILE_NAME_PERIOD_PATTERN, file_name.strip(), re.IGNORECASE)
    if match is None:
        return None, None

    parse_format = ReportConst.REPORT_FILE_NAME_DATETIME_FORMAT.replace("_", ":")

    def _parse_part(value: str) -> Optional[datetime]:
        try:
            return datetime.strptime(value.replace("_", ":"), parse_format)
        except ValueError:
            return None

    return _parse_part(match.group("period_start")), _parse_part(match.group("period_end"))


def get_lds_status_report_column_headers(worksheet: Worksheet) -> List[str]:
    headers: List[str] = []
    column_index = 1
    while True:
        cell_value = worksheet.cell(row=LdsReportConst.REPORT_COLUMN_HEADERS_ROW, column=column_index).value
        if cell_value is None or not str(cell_value).strip():
            break
        headers.append(_stringify_cell(cell_value).strip())
        column_index += 1
    return headers


def _find_total_work_duration(worksheet: Worksheet) -> Tuple[Optional[int], str, Optional[int]]:
    for row_index, row_values in enumerate(
        worksheet.iter_rows(min_row=LdsReportConst.REPORT_DATA_FIRST_ROW, values_only=True),
        start=LdsReportConst.REPORT_DATA_FIRST_ROW,
    ):
        for column_index, cell_value in enumerate(row_values):
            if cell_value is None:
                continue
            cell_text = _stringify_cell(cell_value).strip()
            if LdsReportConst.TOTAL_WORK_DURATION_LABEL not in cell_text:
                continue

            duration_candidates = []
            if column_index + 1 < len(row_values):
                duration_candidates.append(row_values[column_index + 1])
            if row_index + 1 <= worksheet.max_row:
                duration_candidates.append(
                    worksheet.cell(row=row_index + 1, column=column_index + 1).value
                )

            for candidate in duration_candidates:
                duration_seconds = parse_duration_seconds(candidate)
                if duration_seconds is not None:
                    return duration_seconds, _stringify_cell(candidate).strip(), row_index

            return None, "", row_index

    return None, "", None


def parse_lds_status_report_worksheet(
    worksheet: Worksheet,
    expected_section_names: List[str],
) -> LdsStatusReportParsed:
    headers = get_lds_status_report_column_headers(worksheet)
    title_info = parse_lds_status_report_title(
        worksheet.cell(row=LdsReportConst.REPORT_TITLE_ROW, column=1).value
    )
    total_duration_seconds, total_duration_raw, total_label_row_index = _find_total_work_duration(worksheet)

    section_rows: List[LdsStatusReportSectionRow] = []
    expected_names_lower = {name.lower() for name in expected_section_names}

    for row_index, row_values in enumerate(
        worksheet.iter_rows(
            min_row=LdsReportConst.REPORT_DATA_FIRST_ROW,
            max_col=len(headers) if headers else 5,
            values_only=True,
        ),
        start=LdsReportConst.REPORT_DATA_FIRST_ROW,
    ):
        if total_label_row_index is not None and row_index >= total_label_row_index:
            break

        cells = build_column_cells(row_values, headers)
        section_name = cells.get(LdsReportConst.COL_SECTION, "").strip()
        if not section_name:
            continue
        if section_name.lower() not in expected_names_lower:
            continue

        section_rows.append(
            LdsStatusReportSectionRow(
                row_index=row_index,
                section_name=section_name,
                cells=cells,
            )
        )

    return LdsStatusReportParsed(
        title_info=title_info,
        column_headers=headers,
        section_rows=section_rows,
        total_duration_seconds=total_duration_seconds,
        total_duration_raw=total_duration_raw,
        total_label_row_index=total_label_row_index,
    )


def save_lds_status_report_bytes_to_temp_file(file_bytes: bytes) -> Optional[Path]:
    try:
        with tempfile.NamedTemporaryFile(
            suffix=ReportConst.XLSX_EXTENSION,
            prefix="lds_status_report_",
            delete=False,
        ) as temp_file:
            temp_file.write(file_bytes)
            return Path(temp_file.name)
    except OSError:
        return None


def load_lds_status_report_worksheet(file_path: Path) -> Optional[Worksheet]:
    if not file_path.exists():
        return None
    try:
        workbook = load_workbook(filename=str(file_path), read_only=True, data_only=True)
    except Exception:
        return None
    sheet_names = workbook.sheetnames
    if not sheet_names:
        return None
    return workbook[sheet_names[ReportConst.DEFAULT_SHEET_INDEX]]


def format_section_rows_for_allure(section_rows: List[LdsStatusReportSectionRow]) -> str:
    lines = []
    for row in section_rows:
        durations_text = ", ".join(
            f"{column}={format_duration_seconds(seconds)}"
            for column, seconds in row.mode_durations_seconds.items()
        )
        lines.append(
            f"row#{row.row_index}: {row.section_name} | sum={format_duration_seconds(row.modes_sum_seconds)} | "
            f"{durations_text}"
        )
    return "\n".join(lines)


__all__ = [
    "LdsStatusReportParsed",
    "LdsStatusReportSectionRow",
    "build_lds_status_report_file_name",
    "format_duration_seconds",
    "format_section_rows_for_allure",
    "is_duration_cell_filled",
    "is_xlsx_extension",
    "is_xlsx_file_bytes",
    "load_lds_status_report_worksheet",
    "parse_duration_seconds",
    "parse_lds_status_report_worksheet",
    "parse_period_from_lds_status_report_file_name",
    "report_period_comparison_bounds",
    "save_lds_status_report_bytes_to_temp_file",
    "attach_report_file_to_allure",














конфтест
    'test_export_leaks_report': 'export_leaks_report_test',
    'test_export_lds_status_report': 'export_lds_status_report_test',





























тест конст



class ExportLdsStatusReportConstants:
    """Константы для теста формирования xlsx-отчёта о режиме работы СОУ"""

    LDS_STATUS_REPORT_NAME_PART: str = "Отчет о режиме работы СОУ"
    SECTION_NAMES: list[str] = [
        "НПС-5 Тихорецкая - НПС-3 Нововеличковская",
        "НПС-3 Нововеличковская - НПС-2 Крымская",
        "НПС-2 Крымская - НПС Грушовая",
    ]
    TOTAL_WORK_DURATION_LABEL: str = "Суммарное время работы:"
    ZERO_DURATION_TEXT: str = "0:00:00"
    TOTAL_DURATION_TOLERANCE_SECONDS: int = 5
    # Число частей времени при split(':') - часы:минуты:секунды (1:02:51) и минуты:секунды (02:51)
    DURATION_PARTS_COUNT_H_MM_SS: int = 3
    DURATION_PARTS_COUNT_MM_SS: int = 2

    REPORT_TITLE_ROW: int = 1
    REPORT_COLUMN_HEADERS_ROW: int = 2
    REPORT_DATA_FIRST_ROW: int = 3

    COL_SECTION: str = "Наименование участка"
    COL_FAULTY: str = "Неисправность"
    COL_DEGRADATION: str = "В ухудшенных характеристиках"
    COL_INITIALIZATION: str = "Инициализация"
    COL_SERVICEABLE: str = "Исправность"

    MODE_DURATION_COLUMNS: list = [
        COL_FAULTY,
        COL_DEGRADATION,
        COL_INITIALIZATION,
        COL_SERVICEABLE,
    ]

    EXPECTED_COLUMN_HEADERS: list = [COL_SECTION, *MODE_DURATION_COLUMNS]

    REPORT_HEADER_PERIOD_PATTERN: str = (
        r'Отчет о режиме работы СОУ с (?P<period_start>\d{2}\.\d{2}\.\d{4} \d{2}:\d{2}:\d{2})'
        r' по (?P<period_end>\d{2}\.\d{2}\.\d{4} \d{2}:\d{2}:\d{2})'
    )
    REPORT_FILE_NAME_PERIOD_PATTERN: str = (
        r'^Отчет о режиме работы СОУ\. (?P<tu>.+?) '
        r'(?P<period_start>\d{2}\.\d{2}\.\d{4} \d{2}_\d{2}_\d{2})'
        r' - '
        r'(?P<period_end>\d{2}\.\d{2}\.\d{4} \d{2}_\d{2}_\d{2})'
        r'\.xlsx$'
    )


























imi 17
        output_signals_test=CaseMarkers(test_case_id="158", offset=62),
        # ----- Тест ExportReports -----
        export_leaks_report_test=CaseMarkers(test_case_id="234", offset=62),
        export_lds_status_report_test=CaseMarkers(test_case_id="235", offset=63),
    ),
























мод фо тест
    export_leaks_report_test: Optional[CaseMarkers] = None
    export_lds_status_report_test: Optional[CaseMarkers] = None




    # Для наборов с несколькими утечками (select_19_20)
    leaks: list[LeakTestConfig] = field(default_factory=list)

    # Участки в xlsx-отчёте о режиме работы СОУ (export_lds_status_report)
    lds_status_report_section_names: list[str] = field(default_factory=list)








@dataclass
class ExportLdsStatusReportState:
    """Состояние сценария формирования xlsx-отчёта о режиме работы СОУ."""

    report_test: Optional[CaseMarkers] = None
    period_start: Optional[datetime] = None
    period_end: Optional[datetime] = None
    period_start_naive: Optional[datetime] = None
    period_end_naive: Optional[datetime] = None
    time_offset_hours: Optional[int] = None
    tu_description_lower: str = ""
    notification: Optional[ReportDataExportedNotification] = None
    report_item: Optional[ExportedDataItem] = None
    report_file_name: str = ""
    download_invocation_id: Optional[str] = None
    download_reply: Optional[DownloadExportedDataReply] = None
    file_bytes: Optional[bytes] = None
    temp_file_path: Optional[Path] = None
    worksheet: Any = None
    parsed_report: Any = None





























инит сцен
    scenarios.export_leaks_report,
    scenarios.export_lds_status_report,




















сцен
   except Exception:
        with allure.step("Прикрепление xlsx отчёта к Allure при падении теста"):
            if actual_report_state.temp_file_path and actual_report_state.report_file_name:
                report_utils.attach_report_file_to_allure(
                    actual_report_state.temp_file_path, actual_report_state.report_file_name
                )
        raise

    with allure.step("Проверка имени файла отчёта"):
        with SoftAssertions() as soft_failures:
            StepCheck(f"Имя файла оканчивается на {ReportConst.XLSX_EXTENSION}", "file_name", soft_failures).actual(
                has_xlsx_extension
            ).expected(True).equal_to()
            StepCheck(
                f"Имя файла содержит '{ReportConst.LEAKS_REPORT_NAME_PART}'", "file_name", soft_failures
            ).contains(report_file_name_lower, leaks_report_name_part_lower)
            StepCheck(
                f"Имя файла содержит описание ТУ '{cfg.technological_unit.description}'", "file_name", soft_failures
            ).contains(report_file_name_lower, actual_report_state.tu_description_lower)
            StepCheck(
                "Дата начала периода в имени файла совпадает с фильтром запроса (+-1 мин)",
                "period_start_in_file_name",
                soft_failures,
            ).actual(file_name_period_start).is_between(period_start_lo, period_start_hi)
            StepCheck(
                "Дата конца периода в имени файла совпадает с фильтром запроса (+-1 мин)",
                "period_end_in_file_name",
                soft_failures,
            ).actual(file_name_period_end).is_between(period_end_lo, period_end_hi)

    with allure.step("Проверка двойной шапки отчёта"):
        StepCheck("Лист xlsx открыт", "worksheet").actual(actual_report_state.worksheet).is_not_none()
        with SoftAssertions() as soft_failures:
            StepCheck(
                f"В шапке отчёта присутствует '{ReportConst.LEAKS_REPORT_NAME_PART}'",
                "report_title",
                soft_failures,
            ).contains(report_title_lower, leaks_report_name_part_lower)

            StepCheck(
                "Время начала периода в шапке совпадает с фильтром запроса (+-1 мин)",
                "period_start",
                soft_failures,
            ).actual(header_period_start).is_between(period_start_lo, period_start_hi)
            StepCheck(
                "Время конца периода в шапке совпадает с фильтром запроса (+-1 мин)",
                "period_end",
                soft_failures,
            ).actual(header_period_end).is_between(period_end_lo, period_end_hi)
            StepCheck(
                "Названия колонок в шапке отчёта",
                "column_headers",
                soft_failures,
            ).actual(
                column_headers
            ).expected(ReportConst.EXPECTED_COLUMN_HEADERS).equal_to()

    with allure.step("Проверка пуш-нотификации о готовности отчёта"):
        with SoftAssertions() as soft_failures:
            StepCheck("Получена пуш-нотификация о готовности отчёта", "notification", soft_failures).actual(
                actual_report_state.notification
            ).is_not_none()
            StepCheck("Проверка статуса пуш-нотификации", "replyStatus", soft_failures).actual(
                notification_reply_status
            ).expected(ReplyStatus.OK.value).equal_to()
            StepCheck("Проверка наличия контента нотификации", "replyContent", soft_failures).actual(
                notification_reply_content
            ).is_not_none()
            StepCheck("Проверка exportStatus в нотификации", "exportStatus", soft_failures).actual(
                notification_export_status
            ).expected(ExportStatus.DONE).equal_to()
            StepCheck("В нотификации нет текста ошибки", "errorMessage", soft_failures).actual(
                notification_error_message
            ).is_empty()


async def export_lds_status_report(
    ws_client, cfg: SmokeSuiteConfig, leak: LeakTestConfig, imitator_start_time: datetime
):
    """
    Сценарий формирования xlsx-отчёта о режиме работы СОУ.
    """
    report_state = ExportLdsStatusReportState()

    with allure.step("Подготовка параметров сценария формирования отчёта о режиме работы СОУ"):
        report_state.report_test = leak.export_lds_status_report_test
        StepCheck("В конфигурации задан export_lds_status_report_test", "export_lds_status_report_test").actual(
            report_state.report_test
        ).is_not_none()
        report_state.period_start = t_utils.localize_as_moscow(imitator_start_time)
        report_state.period_end = t_utils.localize_as_moscow(
            imitator_start_time + timedelta(minutes=report_state.report_test.offset)
        )
        report_state.period_start_naive = report_utils.normalize_report_period_naive(report_state.period_start)
        report_state.period_end_naive = report_utils.normalize_report_period_naive(report_state.period_end)
        report_state.tu_description_lower = cfg.technological_unit.description.lower()
        time_offset_hours = t_utils.report_time_offset_hours()
        StepCheck(
            f"Смещение timeOffset для запросов отчёта (часовой пояс {TestConst.ZONE_INFO})",
            "time_offset_hours",
        ).actual(time_offset_hours).is_not_none()
        report_state.time_offset_hours = time_offset_hours

        allure.attach(
            f"period.start={report_state.period_start}\n"
            f"period.end={report_state.period_end}\n"
            f"offset_minutes={report_state.report_test.offset}\n"
            f"sections={LdsReportConst.SECTION_NAMES}",
            name="Фильтр периода отчёта о режиме СОУ",
            attachment_type=allure.attachment_type.TEXT,
        )

    with allure.step(f"Этап 1. Подписка на пуш-нотификации {ReportConst.SUBSCRIBE_REPORTS_DATA_EXPORTED_REQUEST}"):
        await t_utils.connect(ws_client, ReportConst.SUBSCRIBE_REPORTS_DATA_EXPORTED_REQUEST, [])

    with allure.step(f"Этап 2. Запрос формирования отчёта {ReportConst.EXPORT_REPORTS_COMMAND_REQUEST}"):
        request_payload = {
            "tuId": cfg.tu_id,
            "exportedDataTypes": [ExportedDataType.LDS_STATUS_REPORT.value],
            "timeOffset": report_state.time_offset_hours,
            "period": {
                "start": t_utils.datetime_to_msgpack_timestamp(report_state.period_start),
                "end": t_utils.datetime_to_msgpack_timestamp(report_state.period_end),
                "additionalProperties": {},
            },
        }
        await t_utils.connect(ws_client, ReportConst.EXPORT_REPORTS_COMMAND_REQUEST, request_payload)

    with allure.step(
        f"Этап 3. Ожидание пуш-нотификации {ReportConst.REPORT_DATA_EXPORTED_NOTIFICATION} о готовности отчёта"
    ):
        report_state.notification = await t_utils.poll_for_report_export_notification(
            ws_client=ws_client,
            parser=parser,
            total_wait_seconds=ReportConst.NOTIFICATION_TIMEOUT_SECONDS,
            poll_interval_seconds=ReportConst.LIST_POLL_INTERVAL_SECONDS,
        )

    with allure.step(f"Этап 4. Лонг-поллинг {ReportConst.GET_EXPORTED_DATA_LIST_REQUEST} до появления отчёта в списке"):
        report_state.report_item = await t_utils.poll_for_exported_file(
            ws_client=ws_client,
            parser=parser,
            list_limit=ReportConst.EXPORTED_DATA_LIST_LIMIT,
            expected_data_type=ExportedDataType.LDS_STATUS_REPORT,
            name_substring=LdsReportConst.LDS_STATUS_REPORT_NAME_PART,
            tu_name_substring=cfg.technological_unit.description,
            period_start=report_state.period_start,
            period_end=report_state.period_end,
            total_wait_seconds=ReportConst.LIST_POLL_TOTAL_WAIT_SECONDS,
            poll_interval_seconds=ReportConst.LIST_POLL_INTERVAL_SECONDS,
        )

    with allure.step("Подготовка данных найденного отчёта в списке"):
        report_item = report_state.report_item
        if report_item is not None:
            allure.attach(
                f"id={report_item.id}, name={report_item.name}, "
                f"exportedDataType={report_item.exportedDataType}, "
                f"start={t_utils.format_datetime_moscow(report_item.start)}, "
                f"end={t_utils.format_datetime_moscow(report_item.end)}",
                name="Найденный отчёт в списке",
                attachment_type=allure.attachment_type.TEXT,
            )
        report_state.report_file_name = lds_report_utils.build_lds_status_report_file_name(
            cfg.technological_unit.description,
            report_state.period_start,
            report_state.period_end,
        )

    with allure.step("Проверка: отчёт найден в списке сформированных файлов"):
        StepCheck("Отчёт найден в списке сформированных файлов", "report_item").actual(
            report_state.report_item
        ).is_not_none()

    with allure.step(
        f"Этап 5. Streaming-вызов {ReportConst.DOWNLOAD_EXPORTED_DATA_REQUEST} по id={report_state.report_item.id}"
    ):
        download_request = {
            "exportedDataId": report_state.report_item.id,
            "exportedDataType": ExportedDataType.LDS_STATUS_REPORT.to_download_name(),
            "additionalProperties": None,
            "timeOffset": report_state.time_offset_hours,
        }
        download_purpose = (
            f"скачивание xlsx-отчёта о режиме СОУ (exportedDataId={report_state.report_item.id}) "
            f"после формирования отчёта и выбора файла в списке GetExportedDataListRequest"
        )
        await t_utils.connect_stream(
            ws_client,
            ReportConst.DOWNLOAD_EXPORTED_DATA_REQUEST,
            download_request,
            purpose=download_purpose,
        )
        report_state.download_invocation_id = ws_client.invocation_id

    with allure.step("Этап 6. Получение fileChunk - скачивание отчёта о режиме СОУ"):
        report_state.download_reply = await t_utils.receive_download_exported_data_reply(
            ws_client=ws_client,
            parser=parser,
            invocation_id=report_state.download_invocation_id,
            request_name=ReportConst.DOWNLOAD_EXPORTED_DATA_REQUEST,
            total_wait_seconds=ReportConst.DOWNLOAD_TIMEOUT_SECONDS,
            purpose=download_purpose,
        )

    with allure.step("Извлечение данных ответа на скачивание"):
        download_reply = report_state.download_reply
        download_reply_status = download_reply.replyStatus
        has_download_reply_content = download_reply.replyContent is not None
        report_state.file_bytes = download_reply.replyContent.fileChunk if has_download_reply_content else None
        is_xlsx_signature = (
            lds_report_utils.is_xlsx_file_bytes(report_state.file_bytes) if report_state.file_bytes else False
        )

    with allure.step("Проверка ответа на скачивание и формата xlsx"):
        StepCheck("Проверка статуса ответа на скачивание", "replyStatus").actual(download_reply_status).expected(
            ReplyStatus.OK.value
        ).equal_to()
        StepCheck("Проверка наличия контента ответа на скачивание", "replyContent").actual(
            has_download_reply_content
        ).expected(True).equal_to()
        StepCheck("Проверка наличия байт файла", "fileChunk").actual(report_state.file_bytes).is_not_empty()
        StepCheck("Проверка xlsx (zip) сигнатуры файла", "file_signature").actual(is_xlsx_signature).expected(
            True
        ).equal_to()

    with allure.step("Подготовка данных для проверки имени файла отчёта"):
        report_file_name = report_state.report_file_name
        report_file_name_lower = report_file_name.lower()
        file_name_period_start, file_name_period_end = lds_report_utils.parse_period_from_lds_status_report_file_name(
            report_file_name
        )
        period_start_lo, period_start_hi, period_end_lo, period_end_hi = lds_report_utils.report_period_comparison_bounds(
            report_state.period_start_naive,
            report_state.period_end_naive,
        )
        has_xlsx_extension = lds_report_utils.is_xlsx_extension(report_file_name)
        lds_report_name_part_lower = LdsReportConst.LDS_STATUS_REPORT_NAME_PART.lower()

    try:
        with allure.step("Этап 7. Сохранение и разбор xlsx-отчёта о режиме СОУ"):
            report_state.temp_file_path = lds_report_utils.save_lds_status_report_bytes_to_temp_file(
                report_state.file_bytes
            )
            StepCheck("Временный xlsx файл создан", "temp_file_path").actual(report_state.temp_file_path).is_not_none()
            report_state.worksheet = lds_report_utils.load_lds_status_report_worksheet(report_state.temp_file_path)
            report_state.parsed_report = lds_report_utils.parse_lds_status_report_worksheet(
                report_state.worksheet,
                LdsReportConst.SECTION_NAMES,
            )
            allure.attach(
                f"Шапка (raw): {report_state.parsed_report.title_info.raw_title}\n"
                f"period_start: {report_state.parsed_report.title_info.period_start}\n"
                f"period_end: {report_state.parsed_report.title_info.period_end}\n"
                f"total_duration: {report_state.parsed_report.total_duration_raw}",
                name="Шапка отчёта о режиме СОУ",
                attachment_type=allure.attachment_type.TEXT,
            )
            allure.attach(
                lds_report_utils.format_section_rows_for_allure(report_state.parsed_report.section_rows),
                name="Строки участков отчёта",
                attachment_type=allure.attachment_type.TEXT,
            )

        with allure.step("Подготовка данных таблицы отчёта для проверки"):
            parsed_report = report_state.parsed_report
            expected_section_names = LdsReportConst.SECTION_NAMES
            section_rows = parsed_report.section_rows
            total_duration_seconds = parsed_report.total_duration_seconds
            duration_tolerance = LdsReportConst.TOTAL_DURATION_TOLERANCE_SECONDS

        with allure.step("Проверка содержимого таблицы отчёта о режиме СОУ"):
            StepCheck("Лист xlsx открыт", "worksheet").actual(report_state.worksheet).is_not_none()
            with SoftAssertions() as soft_failures:
                StepCheck(
                    "Количество строк участков в отчёте",
                    "section_rows_count",
                    soft_failures,
                ).actual(len(section_rows)).expected(len(expected_section_names)).equal_to()

                for section_index, expected_section_name in enumerate(expected_section_names):
                    actual_section_name = (
                        section_rows[section_index].section_name if section_index < len(section_rows) else None
                    )
                    StepCheck(
                        f"Наименование участка #{section_index + 1}",
                        LdsReportConst.COL_SECTION,
                        soft_failures,
                    ).actual(actual_section_name).expected(expected_section_name).equal_to()

                for section_row in section_rows:
                    for column_name in LdsReportConst.MODE_DURATION_COLUMNS:
                        cell_value = section_row.cells.get(column_name)
                        StepCheck(
                            f"Длительность '{column_name}' для участка '{section_row.section_name}' заполнена",
                            column_name,
                            soft_failures,
                        ).actual(lds_report_utils.is_duration_cell_filled(cell_value)).expected(True).equal_to()

                StepCheck(
                    "В отчёте найдена строка 'Суммарное время работы:'",
                    "total_work_duration_label",
                    soft_failures,
                ).actual(parsed_report.total_label_row_index).is_not_none()
                StepCheck(
                    "Суммарное время работы в отчёте не нулевое",
                    "total_work_duration",
                    soft_failures,
                ).actual(total_duration_seconds).is_greater_than(0, LdsReportConst.ZERO_DURATION_TEXT)

                for section_row in section_rows:
                    duration_diff = abs(section_row.modes_sum_seconds - (total_duration_seconds or 0))
                    StepCheck(
                        f"Сумма режимов СОУ для '{section_row.section_name}' "
                        f"совпадает с суммарным временем (+-{duration_tolerance} с)",
                        "modes_sum_seconds",
                        soft_failures,
                    ).actual(duration_diff).is_less_than(
                        duration_tolerance + 1,
                        f"погрешность {duration_tolerance} с",
                    )

        with allure.step("Подготовка данных шапки отчёта для проверки"):
            title_info = parsed_report.title_info
            report_title_lower = title_info.raw_title.lower()
            lds_report_name_part_lower = LdsReportConst.LDS_STATUS_REPORT_NAME_PART.lower()
            column_headers = parsed_report.column_headers
            period_start_lo, period_start_hi, period_end_lo, period_end_hi = lds_report_utils.report_period_comparison_bounds(
                report_state.period_start_naive,
                report_state.period_end_naive,
            )
            header_period_start = title_info.period_start
            header_period_end = title_info.period_end

        with allure.step("Проверка двойной шапки отчёта о режиме СОУ"):
            with SoftAssertions() as soft_failures:
                StepCheck(
                    f"В шапке отчёта присутствует '{LdsReportConst.LDS_STATUS_REPORT_NAME_PART}'",
                    "report_title",
                    soft_failures,
                ).contains(report_title_lower, lds_report_name_part_lower)
                StepCheck(
                    "Время начала периода в шапке совпадает с фильтром запроса (+-1 мин)",
                    "period_start",
                    soft_failures,
                ).actual(header_period_start).is_between(period_start_lo, period_start_hi)
                StepCheck(
                    "Время конца периода в шапке совпадает с фильтром запроса (+-1 мин)",
                    "period_end",
                    soft_failures,
                ).actual(header_period_end).is_between(period_end_lo, period_end_hi)
                StepCheck(
                    "Названия колонок в шапке отчёта",
                    "column_headers",
                    soft_failures,
                ).actual(column_headers).expected(LdsReportConst.EXPECTED_COLUMN_HEADERS).equal_to()

    except Exception:
        with allure.step("Прикрепление xlsx отчёта к Allure при падении теста"):
            if report_state.temp_file_path and report_state.report_file_name:
                lds_report_utils.attach_report_file_to_allure(
                    report_state.temp_file_path, report_state.report_file_name
                )
        raise

    with allure.step("Проверка имени файла отчёта о режиме СОУ"):
        with SoftAssertions() as soft_failures:
            StepCheck(f"Имя файла оканчивается на {ReportConst.XLSX_EXTENSION}", "file_name", soft_failures).actual(
                has_xlsx_extension
            ).expected(True).equal_to()
            StepCheck(
                f"Имя файла содержит '{LdsReportConst.LDS_STATUS_REPORT_NAME_PART}'",
                "file_name",
                soft_failures,
            ).contains(report_file_name_lower, lds_report_name_part_lower)
            StepCheck(
                f"Имя файла содержит описание ТУ '{cfg.technological_unit.description}'",
                "file_name",
                soft_failures,
            ).contains(report_file_name_lower, report_state.tu_description_lower)
            StepCheck(
                "Дата начала периода в имени файла совпадает с фильтром запроса (+-1 мин)",
                "period_start_in_file_name",
                soft_failures,
            ).actual(file_name_period_start).is_between(period_start_lo, period_start_hi)
            StepCheck(
                "Дата конца периода в имени файла совпадает с фильтром запроса (+-1 мин)",
                "period_end_in_file_name",
                soft_failures,
            ).actual(file_name_period_end).is_between(period_end_lo, period_end_hi)

    with allure.step("Проверка пуш-нотификации о готовности отчёта"):
        notification = report_state.notification
        notification_reply_status = notification.replyStatus if notification else None
        notification_reply_content = notification.replyContent if notification else None
        notification_export_status = (
            notification_reply_content.exportStatus if notification_reply_content else None
        )
        notification_error_message = (
            (notification_reply_content.errorMessage or "") if notification_reply_content else ""
        )
        with SoftAssertions() as soft_failures:
            StepCheck("Получена пуш-нотификация о готовности отчёта", "notification", soft_failures).actual(
                notification
            ).is_not_none()
            StepCheck("Проверка статуса пуш-нотификации", "replyStatus", soft_failures).actual(
                notification_reply_status
            ).expected(ReplyStatus.OK.value).equal_to()
            StepCheck("Проверка наличия контента нотификации", "replyContent", soft_failures).actual(
                notification_reply_content
            ).is_not_none()
            StepCheck("Проверка exportStatus в нотификации", "exportStatus", soft_failures).actual(
                notification_export_status
            ).expected(ExportStatus.DONE).equal_to()
            StepCheck("В нотификации нет текста ошибки", "errorMessage", soft_failures).actual(
                notification_error_message
            ).is_empty()





















    @pytest.mark.asyncio
    async def test_export_lds_status_report(
        self,
        ws_client: WebSocketClient,
        config: SmokeSuiteConfig,
        leak: LeakTestConfig,
        leak_number: int,
        imitator_start_time: datetime,
    ) -> None:
        tag = "ExportReports"
        title = f"[{tag}] Проверка формирования отчёта о режиме работы СОУ. ЭФ: Выпадашка отчётов"
        _apply_allure_markers(
            leak.export_lds_status_report_test,
            tag,
            title,
            (
                f"Проверка формирования и содержимого xlsx-отчёта о режиме работы СОУ на наборе данных "
                f"{config.suite_name},\n"
                f"на технологическом участке {config.technological_unit.description}\n"
                f"Период отчёта: от старта имитатора до старта + сдвиг теста "
                f"{leak.export_lds_status_report_test.offset} мин.\n"
                "Этапы сценария:\n"
                "1) SubscribeReportsDataExportedRequest - подписка на пуш-нотификации\n"
                "2) ExportReportsCommandRequest - запрос формирования отчёта (тип LdsStatusReport, фильтр по периоду)\n"
                "3) Ожидание ReportDataExportedNotification\n"
                "4) Лонг-поллинг GetExportedDataListRequest - поиск отчёта в списке\n"
                "5) DownloadExportedDataRequest (StreamInvocation) - скачивание по exportedDataId\n"
                "6) Проверка xlsx: участки, длительности режимов СОУ, суммарное время работы\n"
                "7) Проверка двойной шапки и названий колонок\n"
                "8) Проверка имени файла (.xlsx, название отчёта, ТУ, период +-1 мин)\n"
                "Во вложениях Allure xlsx прикладывается только при падении теста"
            ),
        )
        if config.has_multiple_leaks:
            allure.dynamic.title(f"{title} (утечка #{leak_number})")
        await scenarios.export_lds_status_report(ws_client, config, leak, imitator_start_time)