diff --git a/app/pipeline/report.py b/app/pipeline/report.py index 82edff6..45cc655 100644 --- a/app/pipeline/report.py +++ b/app/pipeline/report.py @@ -23,6 +23,7 @@ class ReportRow: class StreamingReportWriter: def __init__(self) -> None: self.columns = list(FIXED_REPORT_COLUMNS) + 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" @@ -36,15 +37,17 @@ class StreamingReportWriter: operation_fields=row.operation_fields, participant_fields=row.participant_fields, ) - cells: list[WriteOnlyCell] = [] - for column_name in self.columns: - cell = WriteOnlyCell(self.sheet, value=data.get(column_name, "")) - if "Дата" in column_name: - cell.number_format = "DD.MM.YYYY" - if "Сумм" in column_name or "Сумма" in column_name: - cell.number_format = "#,##0.00" - cells.append(cell) - self.sheet.append(cells) + 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) def save(self, destination: Path) -> None: destination.parent.mkdir(parents=True, exist_ok=True) @@ -65,6 +68,13 @@ class StreamingReportWriter: result.append(cell) return result + def _resolve_format(self, column_name: str) -> str | None: + if "Дата" in column_name: + return "DD.MM.YYYY" + if "Сумм" in column_name or "Сумма" in column_name: + return "#,##0.00" + return None + def write_report(rows: Iterable[ReportRow], destination: Path) -> None: writer = StreamingReportWriter() diff --git a/app/pipeline/service.py b/app/pipeline/service.py index 9620c25..3a08ef4 100644 --- a/app/pipeline/service.py +++ b/app/pipeline/service.py @@ -1,5 +1,8 @@ from __future__ import annotations +import cProfile +import io +import pstats import tempfile import time from dataclasses import dataclass, field @@ -28,6 +31,7 @@ from .protocol import ProtocolRow, write_protocol from .report import ReportRow, StreamingReportWriter EMPTY_BATCH_STATUS = "info" +CPROFILE_TOP_LIMIT = 30 @dataclass @@ -41,6 +45,7 @@ class BatchResult: log_path: str messages: list[str] = field(default_factory=list) duration_seconds: float = 0.0 + profile_path: str = "" def run_batch( @@ -138,23 +143,40 @@ def run_batch( report_path = join_path(run_context.run_dir, report_file_name) protocol_path = join_path(run_context.run_dir, protocol_file_name) log_path = join_path(run_context.run_dir, log_file_name) + profile_path = join_path( + run_context.run_dir, + f"profile_{run_context.date_key}_{run_context.run_number}.txt", + ) with tempfile.TemporaryDirectory() as temp_dir: temp_dir_path = Path(temp_dir) local_report = temp_dir_path / "report.xlsx" local_protocol = temp_dir_path / "protocol.xlsx" local_log = temp_dir_path / "process.log" + local_profile = temp_dir_path / "profile.txt" report_writer = StreamingReportWriter() + stage_timings: dict[str, float] = { + "smb_read_seconds": 0.0, + "parse_seconds": 0.0, + "report_append_seconds": 0.0, + "move_seconds": 0.0, + } + profiler = cProfile.Profile() - _process_files( - scan_result=scan_result, - run_context=run_context, - samba_conn=samba_conn, - launcher=launcher, - report_writer=report_writer, - protocol_rows=protocol_rows, - log_lines=log_lines, - ) + profiler.enable() + try: + _process_files( + scan_result=scan_result, + run_context=run_context, + samba_conn=samba_conn, + launcher=launcher, + report_writer=report_writer, + protocol_rows=protocol_rows, + log_lines=log_lines, + stage_timings=stage_timings, + ) + finally: + profiler.disable() if not protocol_rows: protocol_rows.append( @@ -170,8 +192,22 @@ def run_batch( ) log_lines.append(f"{_now()} | empty batch") + log_lines.append( + f"{_now()} | stage timings (sec): " + f"smb_read={stage_timings['smb_read_seconds']:.3f}, " + f"parse={stage_timings['parse_seconds']:.3f}, " + f"report_append={stage_timings['report_append_seconds']:.3f}, " + f"move={stage_timings['move_seconds']:.3f}" + ) + report_writer.save(local_report) write_protocol(protocol_rows, local_protocol) + profile_warning = "" + try: + _write_profile_report(profiler, local_profile) + except Exception as exc: # noqa: BLE001 + profile_warning = f"Не удалось сформировать cProfile-отчет: {exc}" + log_lines.append(f"{_now()} | profile report failed: {exc}") # Use CRLF to keep one-record-per-line in Windows viewers. local_log.write_text("\r\n".join(log_lines) + "\r\n", encoding="utf-8") @@ -184,6 +220,15 @@ def run_batch( log_path=log_path, samba_conn=samba_conn, ) + if not profile_warning: + try: + upload_local_file(local_profile, profile_path, samba_conn=samba_conn) + except Exception as exc: # noqa: BLE001 + profile_warning = f"Не удалось выгрузить cProfile-отчет: {exc}" + log_lines.append(f"{_now()} | profile upload failed: {exc}") + + if profile_warning: + messages.append(profile_warning) processed = sum(1 for row in protocol_rows if row.status == "success") partial = sum(1 for row in protocol_rows if row.status == "partial") @@ -211,17 +256,25 @@ def run_batch( log_path=log_path, messages=messages, duration_seconds=duration_seconds, + profile_path=profile_path, ) -def _parse_remote_xml(file_name: str, remote_path: str, samba_conn) -> FileParseResult: +def _parse_remote_xml( + file_name: str, remote_path: str, samba_conn, stage_timings: dict[str, float] +) -> FileParseResult: try: + read_started = time.perf_counter() content = read_remote_file_bytes(remote_path=remote_path, samba_conn=samba_conn) + stage_timings["smb_read_seconds"] += time.perf_counter() - read_started except Exception as exc: # noqa: BLE001 result = FileParseResult(file_name=file_name) result.fatal_error = f"Ошибка чтения файла из SAMBA: {exc}" return result - return parse_xml_content(file_name=file_name, xml_content=content) + parse_started = time.perf_counter() + result = parse_xml_content(file_name=file_name, xml_content=content) + stage_timings["parse_seconds"] += time.perf_counter() - parse_started + return result def _process_files( @@ -232,18 +285,22 @@ def _process_files( report_writer: StreamingReportWriter, protocol_rows: list[ProtocolRow], log_lines: list[str], + stage_timings: dict[str, float], ) -> None: for remote_file in scan_result.invalid_files: reason = "Некорректное имя XML-файла" moved_to = "" try: + move_started = time.perf_counter() moved_to = move_file_to_error( file_path=remote_file.path, file_name=remote_file.name, error_dir=run_context.error_dir, samba_conn=samba_conn, ) + stage_timings["move_seconds"] += time.perf_counter() - move_started except Exception as exc: # noqa: BLE001 + stage_timings["move_seconds"] += time.perf_counter() - move_started reason = f"{reason}. Не удалось переместить в Err: {exc}" log_lines.append( f"{_now()} | move to Err failed: {remote_file.path}. {exc}" @@ -265,8 +322,11 @@ def _process_files( ) for remote_file in scan_result.valid_files: - parse_result = _parse_remote_xml(remote_file.name, remote_file.path, samba_conn) + parse_result = _parse_remote_xml( + remote_file.name, remote_file.path, samba_conn, stage_timings + ) for row in parse_result.rows: + report_started = time.perf_counter() report_writer.append_row( ReportRow( file_name=row.file_name, @@ -276,8 +336,12 @@ def _process_files( participant_fields=row.participant_fields, ) ) + stage_timings["report_append_seconds"] += ( + time.perf_counter() - report_started + ) status, message = _resolve_parse_status(parse_result) if status == "success": + move_started = time.perf_counter() try: moved_to = move_file_to_processed( file_path=remote_file.path, @@ -294,7 +358,10 @@ def _process_files( log_lines.append( f"{_now()} | move to Processed failed: {remote_file.path}. {exc}" ) + finally: + stage_timings["move_seconds"] += time.perf_counter() - move_started else: + move_started = time.perf_counter() try: moved_to = move_file_to_error( file_path=remote_file.path, @@ -312,6 +379,8 @@ def _process_files( log_lines.append( f"{_now()} | move to Err failed: {remote_file.path}. {exc}" ) + finally: + stage_timings["move_seconds"] += time.perf_counter() - move_started protocol_rows.append( ProtocolRow( file_name=remote_file.name, @@ -376,3 +445,21 @@ def _format_duration(duration_seconds: float) -> str: if minutes: return f"{minutes} мин {seconds} сек" return f"{seconds} сек" + + +def _write_profile_report(profiler: cProfile.Profile, destination: Path) -> None: + stream = io.StringIO() + cumulative_stats = pstats.Stats(profiler, stream=stream) + cumulative_stats.sort_stats("cumulative") + stream.write( + f"=== TOP {CPROFILE_TOP_LIMIT} cumulative functions " + "(sorted by cumulative) ===\n" + ) + cumulative_stats.print_stats(CPROFILE_TOP_LIMIT) + tottime_stats = pstats.Stats(profiler, stream=stream) + stream.write( + f"\n=== TOP {CPROFILE_TOP_LIMIT} tottime functions (sorted by tottime) ===\n" + ) + tottime_stats.sort_stats("tottime") + tottime_stats.print_stats(CPROFILE_TOP_LIMIT) + destination.write_text(stream.getvalue(), encoding="utf-8")