Загрузка данных
def format_duration_seconds(total_seconds: int) -> str:
"""Форматирует длительность в секундах в строку H:MM:SS (минуты и секунды с ведущим нулём)."""
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:
"""
Парсит первую строку шапки отчёта о режиме СОУ.
Извлекает period_start/period_end по REPORT_HEADER_PERIOD_PATTERN.
"""
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:
"""
Ожидаемое имя xlsx при скачивании:
«Отчет о режиме работы СОУ. Тихорецк-Новороссийск-3 DD.MM.YYYY HH_MM_SS - DD.MM.YYYY HH_MM_SS.xlsx».
"""
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]]:
"""Извлекает границы периода из имени xlsx-отчёта о режиме СОУ."""
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]:
"""Возвращает непустые заголовки колонок из строки REPORT_COLUMN_HEADERS_ROW."""
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]]:
"""
Ищет строку «Суммарное время работы:» и парсит длительность рядом (в той же или следующей строке).
Returns:
(секунды, сырое значение ячейки, номер строки с меткой) или (None, "", None).
"""
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:
"""
Разбирает лист xlsx-отчёта о режиме СОУ: шапка, колонки, строки участков и суммарное время.
В section_rows попадают только участки из expected_section_names (без учёта регистра).
"""
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]:
"""Сохраняет байты отчёта во временный xlsx-файл. При ошибке возвращает None."""
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]:
"""Открывает первый лист xlsx-отчёта о режиме СОУ. При ошибке возвращает None."""
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",
]