optimization

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] = []
+        values: list[object] = []
-        for column_name in self.columns:
+        for index, column_name in enumerate(self.columns):
-            cell = WriteOnlyCell(self.sheet, value=data.get(column_name, ""))
+            raw_value = data.get(column_name, "")
-            if "Дата" in column_name:
+            number_format = self._column_formats[index]
-                cell.number_format = "DD.MM.YYYY"
+            if number_format is None:
-            if "Сумм" in column_name or "Сумма" in column_name:
+                values.append(raw_value)
-                cell.number_format = "#,##0.00"
+                continue
-            cells.append(cell)
+            cell = WriteOnlyCell(self.sheet, value=raw_value)
-        self.sheet.append(cells)
+            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()

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,14 +143,28 @@ 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()
 
+        profiler.enable()
+        try:
             _process_files(
                 scan_result=scan_result,
                 run_context=run_context,
@@ -154,7 +173,10 @@ def run_batch(
                 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")