excel-fix: пофиксил экспорт первой формы

api/src/services/export_service/__init__.py

@@ -0,0 +1 @@
+from .export_service import ExportService

api/src/services/export_service/export_normalizers.py

@@ -194,6 +194,41 @@ class Form1RowNormalizer:
         return None
 
 
+class Form1DepthRowNormalizer:
+    def __init__(self, serializer: ExcelValueSerializer):
+        self.serializer = serializer
+
+    def normalize(self, data: dict[str, Any]) -> dict[str, Any]:
+        normalized = dict()
+        self._merge_header_fields_into_row(normalized=normalized, original=data)
+        self._normilize_depth(normalized=normalized, original=data)
+        return normalized
+
+    def _merge_header_fields_into_row(self, normalized: dict[str, Any], original: dict[str, Any]) -> None:
+        header = original.get("header")
+        if isinstance(header, dict):
+            if "section" in header and "section" not in normalized:
+                normalized["section"] = header.get("section")
+            if "item_id" in header and "item_id" not in normalized:
+                normalized["item_id"] = header.get("item_id")
+            if "name" in header and "name" not in normalized:
+                normalized["name"] = header.get("name")
+            if "num_group" in header and "num_group" not in normalized:
+                normalized["num_group"] = header.get("num_group")
+            if "internal_order" in header and "internal_order" not in normalized:
+                normalized["internal_order"] = header.get("internal_order")
+
+        normalized.pop("header", None)
+        normalized.pop("line_id", None)
+    
+    def _normilize_depth(self, normalized: dict, original: dict, prefix: str = "") -> None:
+        for key, value in original.items():
+            if isinstance(value, dict):
+                self._normilize_depth(normalized=normalized, original=value, prefix=f"{prefix}{key}.")
+            else:
+                normalized[f"{prefix}{key}"] = value
+    
+
 class SmetaRowNormalizer:
     def normalize(self, data: dict[str, Any]) -> dict[str, Any]:
         return {

api/src/services/export_service/export_service.py

@@ -15,16 +15,20 @@ from sqlalchemy.ext.asyncio import AsyncSession
 from src.db.models.app_user import AppUser
 from src.db.models.form_type import FormTypeEnum
 from src.services.budget_form_service import BudgetFormService
-from src.services.export_normalizers import (
+from src.services.export_service.export_normalizers import (
     ExcelValueSerializer,
-    Form1RowNormalizer,
+    Form1DepthRowNormalizer,
     Form3ReportRowNormalizer,
     SmetaRowNormalizer,
 )
-from src.services.export_writers import (
-    FORM1_FLAT_COLUMNS,
+from src.services.export_service.export_mappings import (
+    FORM1_COLORS,
+    FORM1_DEPTH_COLUMNS,
     FORM3_REPORT_COLUMNS,
     SMETA_COLUMNS,
+)
+from src.services.export_service.export_writers import (
+    DepthSheetWriter,
     TabularSheetWriter,
 )
 from src.services.project_service import ProjectService
@@ -51,12 +55,12 @@ class ExportService:
         self.sheet_service = SheetService(db)
         self.project_service = ProjectService(db)
         self.value_serializer = ExcelValueSerializer()
-        self.form1_row_normalizer = Form1RowNormalizer(self.value_serializer)
+        self.form1_row_normalizer = Form1DepthRowNormalizer(self.value_serializer)
         self.form3_report_row_normalizer = Form3ReportRowNormalizer()
         self.smeta_row_normalizer = SmetaRowNormalizer()
         self.sheet_writers: dict[str, TabularSheetWriter] = {
-            "__default__": TabularSheetWriter(
-                columns=FORM1_FLAT_COLUMNS,
+            "__default__": DepthSheetWriter(
+                columns_dict=FORM1_DEPTH_COLUMNS,
                 row_normalizer=self.form1_row_normalizer.normalize,
                 value_serializer=self.value_serializer.to_excel_value,
             ),
@@ -65,6 +69,24 @@ class ExportService:
                 row_normalizer=self.smeta_row_normalizer.normalize,
                 value_serializer=self.value_serializer.to_excel_value,
             ),
+            "OPER": DepthSheetWriter(
+                columns_dict=FORM1_DEPTH_COLUMNS,
+                row_normalizer=self.form1_row_normalizer.normalize,
+                value_serializer=self.value_serializer.to_excel_value,
+                colors_config=FORM1_COLORS["OPER"],
+            ),
+            "AHR": DepthSheetWriter(
+                columns_dict=FORM1_DEPTH_COLUMNS,
+                row_normalizer=self.form1_row_normalizer.normalize,
+                value_serializer=self.value_serializer.to_excel_value,
+                colors_config=FORM1_COLORS["AHR"],
+            ),
+            "CAP": DepthSheetWriter(
+                columns_dict=FORM1_DEPTH_COLUMNS,
+                row_normalizer=self.form1_row_normalizer.normalize,
+                value_serializer=self.value_serializer.to_excel_value,
+                colors_config=FORM1_COLORS["CAP"],
+            ),
         }
         self.form3_report_writer = TabularSheetWriter(
             columns=FORM3_REPORT_COLUMNS,

api/src/services/export_service/export_writers.py

@@ -0,0 +1,209 @@
+from __future__ import annotations
+
+from collections import defaultdict
+from typing import Any, Callable
+
+from openpyxl.styles import Alignment, Border, Color, Font, PatternFill, Side
+from openpyxl.worksheet.worksheet import Worksheet
+from openpyxl.utils import get_column_letter
+
+from src.services.export_service.export_mappings import PALETTE
+
+
+
+RowNormalizer = Callable[[dict[str, Any]], dict[str, Any]]
+ValueSerializer = Callable[[Any], Any]
+
+
+class TabularSheetWriter:
+    def __init__(
+        self,
+        columns: list[tuple[str, str]],
+        row_normalizer: RowNormalizer,
+        value_serializer: ValueSerializer,
+    ):
+        self.columns = columns
+        self.row_normalizer = row_normalizer
+        self.value_serializer = value_serializer
+
+    def write(self, worksheet: Worksheet, rows: list[tuple]) -> None:
+        worksheet.append([title for _, title in self.columns])
+        for row in rows:
+            raw_data = row[3] if len(row) > 3 and isinstance(row[3], dict) else {}
+            normalized = self.row_normalizer(raw_data)
+            worksheet.append(
+                [self.value_serializer(normalized.get(key)) for key, _ in self.columns]
+            )
+
+
+class DepthSheetWriter:
+    alignment = Alignment(horizontal='center', vertical='center')
+    thin_border = Border(
+        left=Side(style='thin'),
+        right=Side(style='thin'),
+        top=Side(style='thin'),
+        bottom=Side(style='thin')
+    )
+
+    def __init__(
+            self,
+            columns_dict: dict,
+            row_normalizer: RowNormalizer,
+            value_serializer: ValueSerializer,
+            colors_config: dict | None = None,
+    ):
+        self.columns_dict = columns_dict
+        self._flat_columns_to_depth = {}
+        self.columns = self._columns_dict_to_flat(columns_dict)
+        self.row_normalizer = row_normalizer
+        self.value_serializer = value_serializer
+        self.colors_config = colors_config
+        self.palette = None
+        if colors_config:
+            self._set_palette()
+    
+    def _columns_dict_to_flat(
+            self, 
+            columns_dict: dict | None = None, 
+            prefix: str = "",
+            depth_prefix: str = "",
+    ) -> None:
+        if columns_dict is None:
+            columns_dict = self.columns_dict
+        result = []
+        for key, sub_dict in columns_dict.items():
+            if "key" in sub_dict:
+                new_key = sub_dict['key']
+            else:
+                new_key = key
+            if "children" not in sub_dict:
+                if new_key:
+                    add_key = f"{prefix}{new_key}"
+                    result.append(add_key)
+                    self._flat_columns_to_depth[add_key] = f"{depth_prefix}{key}"
+            else:
+                result += self._columns_dict_to_flat(
+                    sub_dict["children"], 
+                    prefix=f"{prefix}{new_key}." if new_key else prefix,
+                    depth_prefix=f"{depth_prefix}{key}.",
+                )
+        return result
+    
+    def _set_palette(self) -> None:
+        self.palette = {}
+        for key in self.columns:
+            full_key = None
+            for sub_key in self._flat_columns_to_depth[key].split("."):
+                if full_key is None:
+                    full_key = sub_key
+                else:
+                    full_key = f"{full_key}.{sub_key}"
+                if full_key in self.colors_config and "palette" in self.colors_config[full_key]:
+                    self.palette[key] = self.colors_config[full_key]["palette"]
+
+    def _get_header_max_row(self, column_dict: dict) -> int:
+        max_depth = 0
+        for sub_dict in column_dict.values():
+            if "children" in sub_dict:
+                max_depth = max(max_depth, self._get_header_max_row(sub_dict["children"]))
+        return max_depth + 1
+
+    def _write_header(
+            self, 
+            worksheet: Worksheet, 
+            columns_dict: dict | None = None,
+            current_column: int = 1,
+            current_row: int = 1,
+            max_row: int | None = None,
+            full_key: str | None = None
+    ) -> int:
+        if columns_dict is None:
+            columns_dict = self.columns_dict
+            max_row = self._get_header_max_row(self.columns_dict)
+        for key, sub_dict in columns_dict.items():
+            if full_key:
+                key = f"{full_key}.{key}"
+
+            if "children" not in sub_dict:
+                cell = worksheet.cell(row=current_row, column=current_column, value=sub_dict["name"])
+                cell.border = self.thin_border
+                if max_row != current_row:
+                    worksheet.merge_cells(
+                        start_row=current_row, 
+                        start_column=current_column, 
+                        end_row=max_row, 
+                        end_column=current_column,
+                    )
+                current_column += 1
+            else:
+                new_column = self._write_header(
+                    worksheet=worksheet,
+                    columns_dict=sub_dict["children"],
+                    current_column=current_column,
+                    current_row=current_row+1,
+                    max_row=max_row,
+                    full_key=key,
+                )
+                cell = worksheet.cell(row=current_row, column=current_column, value=sub_dict["name"])
+                if new_column - current_column > 1:
+                    worksheet.merge_cells(
+                        start_row=current_row, 
+                        start_column=current_column, 
+                        end_row=current_row,
+                        end_column=new_column - 1,
+                    )
+                cell.border = self.thin_border
+                current_column = new_column
+            cell.alignment = self.alignment
+            
+            if self.colors_config and (color := self.colors_config.get(key)):
+                if "background" in color:
+                    cell.fill = PatternFill(
+                        start_color=Color(rgb=color["background"]),
+                        fill_type="solid",
+                    )
+                if "color" in color:
+                    cell.font = Font(color=color["color"])
+        return current_column
+
+    def _set_widths(self, worksheet: Worksheet):
+        for col in worksheet.columns:
+            max_len = 0
+            col_letter = get_column_letter(col[0].column)  # Get alphabetical column name (e.g., 'A')
+            
+            for cell in col:
+                if cell.value is not None:
+                    cell_len = len(str(cell.value))
+                    if cell_len > max_len:
+                        max_len = cell_len
+                        
+            adjusted_width = max(max_len + 3, 10)
+            worksheet.column_dimensions[col_letter].width = adjusted_width
+
+    def _set_color_to_row(self, ws: Worksheet, item_type: str) -> None:
+        if not self.palette:
+            return
+
+        row = ws.max_row
+
+        for i, key in enumerate(self.columns):
+            if key not in self.palette:
+                continue
+            cell = ws.cell(column=i+1, row=row)
+            if (palette := self.palette.get(key)):
+                cell.fill = PatternFill(
+                    start_color=Color(rgb=PALETTE[palette][item_type]),
+                    fill_type="solid",
+                )
+
+    def write(self, worksheet: Worksheet, rows: list[tuple]) -> None:
+        self._write_header(worksheet)
+        for row in rows:
+            raw_data = row[3] if len(row) > 3 and isinstance(row[3], dict) else {}
+            normalized = self.row_normalizer(raw_data)
+            worksheet.append(
+                [self.value_serializer(normalized.get(key)) for key in self.columns]
+            )
+            if len(row) > 3:
+                self._set_color_to_row(ws=worksheet, item_type=row[0])
+        self._set_widths(worksheet)

api/src/services/export_writers.py

@@ -1,103 +0,0 @@
-from __future__ import annotations
-
-from typing import Any, Callable
-
-from openpyxl.worksheet.worksheet import Worksheet
-
-FORM1_FLAT_COLUMNS: list[tuple[str, str]] = [
-    ("section", "Код раздела"),
-    ("item_id", "ID статьи"),
-    ("name", "Наименование"),
-    ("q1", "I квартал"),
-    ("q2", "II квартал"),
-    ("q3", "III квартал"),
-    ("q4", "IV квартал"),
-    ("totals", "ИТОГО"),
-    ("plan", "План"),
-    ("reserve", "Резерв"),
-    ("approved", "Согласовано"),
-    ("collegial", "Коллегиальный"),
-    ("allocation", "Аллокация"),
-    ("sequestration", "Секвестр"),
-    ("contract", "Договор"),
-    ("contract_sum", "Сумма договора"),
-    ("ckk", "ЦКК"),
-]
-
-SMETA_COLUMNS: list[tuple[str, str]] = [
-    ("section_code", "Код раздела"),
-    ("name", "Наименование"),
-    ("plan_support_q1", "План поддержка I квартал"),
-    ("plan_support_q2", "План поддержка II квартал"),
-    ("plan_support_q3", "План поддержка III квартал"),
-    ("plan_support_q4", "План поддержка IV квартал"),
-    ("plan_support_year", "План поддержка Год"),
-    ("plan_development_q1", "План развитие I квартал"),
-    ("plan_development_q2", "План развитие II квартал"),
-    ("plan_development_q3", "План развитие III квартал"),
-    ("plan_development_q4", "План развитие IV квартал"),
-    ("plan_development_year", "План развитие Год"),
-    ("plan_total_year", "План ИТОГО Год"),
-    ("approved_support_q1", "Утверждено поддержка I квартал"),
-    ("approved_support_q2", "Утверждено поддержка II квартал"),
-    ("approved_support_q3", "Утверждено поддержка III квартал"),
-    ("approved_support_q4", "Утверждено поддержка IV квартал"),
-    ("approved_support_year", "Утверждено поддержка Год"),
-    ("approved_development_q1", "Утверждено развитие I квартал"),
-    ("approved_development_q2", "Утверждено развитие II квартал"),
-    ("approved_development_q3", "Утверждено развитие III квартал"),
-    ("approved_development_q4", "Утверждено развитие IV квартал"),
-    ("approved_development_year", "Утверждено развитие Год"),
-    ("approved_total_year", "Утверждено ИТОГО Год"),
-    ("fact_support_q1", "Факт поддержка I квартал"),
-    ("fact_support_q2", "Факт поддержка II квартал"),
-    ("fact_support_q3", "Факт поддержка III квартал"),
-    ("fact_support_q4", "Факт поддержка IV квартал"),
-    ("fact_support_year", "Факт поддержка Год"),
-    ("fact_development_q1", "Факт развитие I квартал"),
-    ("fact_development_q2", "Факт развитие II квартал"),
-    ("fact_development_q3", "Факт развитие III квартал"),
-    ("fact_development_q4", "Факт развитие IV квартал"),
-    ("fact_development_year", "Факт развитие Год"),
-    ("fact_total_year", "Факт ИТОГО Год"),
-    ("corrected_support_q2", "Корректировка поддержка II квартал"),
-    ("corrected_support_q3", "Корректировка поддержка III квартал"),
-    ("corrected_support_q4", "Корректировка поддержка IV квартал"),
-    ("corrected_development_q2", "Корректировка развитие II квартал"),
-    ("corrected_development_q3", "Корректировка развитие III квартал"),
-    ("corrected_development_q4", "Корректировка развитие IV квартал"),
-]
-FORM3_REPORT_COLUMNS: list[tuple[str, str]] = [
-    ("section", "Код раздела"),
-    ("item_id", "ID статьи"),
-    ("name", "Наименование"),
-    ("q1", "I квартал"),
-    ("q2", "II квартал"),
-    ("q3", "III квартал"),
-    ("q4", "IV квартал"),
-    ("totals", "ИТОГО"),
-]
-
-RowNormalizer = Callable[[dict[str, Any]], dict[str, Any]]
-ValueSerializer = Callable[[Any], Any]
-
-
-class TabularSheetWriter:
-    def __init__(
-        self,
-        columns: list[tuple[str, str]],
-        row_normalizer: RowNormalizer,
-        value_serializer: ValueSerializer,
-    ):
-        self.columns = columns
-        self.row_normalizer = row_normalizer
-        self.value_serializer = value_serializer
-
-    def write(self, worksheet: Worksheet, rows: list[tuple]) -> None:
-        worksheet.append([title for _, title in self.columns])
-        for row in rows:
-            raw_data = row[3] if len(row) > 3 and isinstance(row[3], dict) else {}
-            normalized = self.row_normalizer(raw_data)
-            worksheet.append(
-                [self.value_serializer(normalized.get(key)) for key, _ in self.columns]
-            )

api/tests/unit/test_export_normalizers.py

@@ -1,4 +1,4 @@
-from src.services.export_normalizers import (
+from src.services.export_service.export_normalizers import (
     ExcelValueSerializer,
     Form1RowNormalizer,
     Form3ReportRowNormalizer,

api/tests/unit/test_export_writers.py

@@ -3,11 +3,11 @@ from unittest.mock import AsyncMock
 from openpyxl import Workbook
 
 from src.services.export_service import ExportService
-from src.services.export_writers import (
+from src.services.export_service.export_mappings import (
     FORM1_FLAT_COLUMNS,
     SMETA_COLUMNS,
-    TabularSheetWriter,
 )
+from src.services.export_service.export_writers import TabularSheetWriter
 
 
 def test_tabular_sheet_writer_writes_headers_and_rows():