Export

api/requirements.txt

@@ -15,3 +15,4 @@ asyncpg==0.30.0
 pytest==8.3.2
 pytest-asyncio==0.24.0
 raisa-fastapi-protected-api==1.0.0
+openpyxl

api/src/api/v1/export.py

@@ -0,0 +1,117 @@
+from fastapi import APIRouter, Depends, HTTPException, status
+from fastapi.responses import StreamingResponse
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from src.api.v1.deps import get_current_active_user_with_set_db
+from src.db.models.app_user import AppUser
+from src.db.session import get_db
+from src.domain.schemas import DirectionSchemaEnum, ExportBulkRequest
+from src.services.export_service import ExportService
+
+router = APIRouter(prefix="/export", tags=["export"])
+
+
+def _parse_sections_csv(sections: str | None) -> list[str] | None:
+    if sections is None:
+        return None
+    raw = [section.strip() for section in sections.split(",") if section.strip()]
+    normalized: list[str] = []
+    seen: set[str] = set()
+    quarter_tokens = {"q1", "q2", "q3", "q4", "totals"}
+    for section in raw:
+        if "." in section:
+            base_section, nested_section = section.split(".", 1)
+            nested_section = nested_section.strip()
+            tokens = [base_section.strip()]
+            if nested_section in quarter_tokens:
+                tokens.append(nested_section)
+            for token in tokens:
+                if token and token not in seen:
+                    seen.add(token)
+                    normalized.append(token)
+            continue
+        if section not in seen:
+            seen.add(section)
+            normalized.append(section)
+    return normalized or None
+
+
+@router.get("/form/{form_id}")
+async def export_form(
+    form_id: int,
+    direction: DirectionSchemaEnum | None = None,
+    sections: str | None = None,
+    db: AsyncSession = Depends(get_db),
+    current_user: AppUser = Depends(get_current_active_user_with_set_db),
+):
+    export_service = ExportService(db)
+    try:
+        stream, filename, media_type = await export_service.export_form_payload(
+            form_id=form_id,
+            current_user=current_user,
+            direction=direction.value if direction else None,
+            sections=_parse_sections_csv(sections),
+            ignore_non_applicable_query_params=True,
+        )
+    except ValueError as exc:
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc))
+    return StreamingResponse(
+        stream,
+        media_type=media_type,
+        headers={"Content-Disposition": export_service.build_content_disposition(filename)},
+    )
+
+
+@router.get("/form/{form_id}/sheet/{sheet}")
+async def export_form_sheet(
+    form_id: int,
+    sheet: str,
+    direction: DirectionSchemaEnum | None = None,
+    sections: str | None = None,
+    db: AsyncSession = Depends(get_db),
+    current_user: AppUser = Depends(get_current_active_user_with_set_db),
+):
+    export_service = ExportService(db)
+    try:
+        stream, filename, media_type = await export_service.export_form_payload(
+            form_id=form_id,
+            current_user=current_user,
+            sheet=sheet,
+            direction=direction.value if direction else None,
+            sections=_parse_sections_csv(sections),
+        )
+    except ValueError as exc:
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc))
+    return StreamingResponse(
+        stream,
+        media_type=media_type,
+        headers={"Content-Disposition": export_service.build_content_disposition(filename)},
+    )
+
+
+@router.post("/bulk")
+async def export_forms_bulk(
+    payload: ExportBulkRequest,
+    direction: DirectionSchemaEnum | None = None,
+    sections: str | None = None,
+    db: AsyncSession = Depends(get_db),
+    current_user: AppUser = Depends(get_current_active_user_with_set_db),
+):
+    export_service = ExportService(db)
+    try:
+        stream, filename, media_type = await export_service.export_bulk_payload(
+            form_ids=payload.form_ids,
+            current_user=current_user,
+            skip_failed=payload.skip_failed,
+            direction=direction.value if direction else None,
+            sections=_parse_sections_csv(sections),
+        )
+    except ValueError as exc:
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc))
+    return StreamingResponse(
+        stream,
+        media_type=media_type,
+        headers={"Content-Disposition": export_service.build_content_disposition(filename)},
+    )
+
+

api/src/api/v1/router.py

@@ -1,6 +1,6 @@
 from fastapi import APIRouter
 
-from src.api.v1 import auth, users, admin, audit, forms, form_phases, projects
+from src.api.v1 import auth, users, admin, audit, forms, form_phases, projects, export
 from src.api.v1 import websocket
 
 
@@ -12,4 +12,5 @@ api_router.include_router(audit.router)
 api_router.include_router(forms.router)
 api_router.include_router(projects.router)
 api_router.include_router(form_phases.router)
+api_router.include_router(export.router)
 api_router.include_router(websocket.router)

api/src/domain/schemas.py

@@ -222,6 +222,11 @@ class CellsPatch(BaseModel):
     changes: list[CellPatch]
 
 
+class ExportBulkRequest(BaseModel):
+    form_ids: list[int] = Field(min_length=1)
+    skip_failed: bool = True
+
+
 class AddForm3LineBody(BaseModel):
     expense_item_id: int
 

api/src/services/export_normalizers.py

@@ -0,0 +1,211 @@
+from __future__ import annotations
+
+import json
+from typing import Any
+
+SMETA_VALUE_PATHS: list[tuple[str, tuple[str, ...]]] = [
+    ("section_code", ("section_code",)),
+    ("name", ("name",)),
+    ("plan_support_q1", ("plan", "support", "q1")),
+    ("plan_support_q2", ("plan", "support", "q2")),
+    ("plan_support_q3", ("plan", "support", "q3")),
+    ("plan_support_q4", ("plan", "support", "q4")),
+    ("plan_support_year", ("plan", "support", "year")),
+    ("plan_development_q1", ("plan", "development", "q1")),
+    ("plan_development_q2", ("plan", "development", "q2")),
+    ("plan_development_q3", ("plan", "development", "q3")),
+    ("plan_development_q4", ("plan", "development", "q4")),
+    ("plan_development_year", ("plan", "development", "year")),
+    ("plan_total_year", ("plan", "total_year")),
+    ("approved_support_q1", ("approved", "support", "q1")),
+    ("approved_support_q2", ("approved", "support", "q2")),
+    ("approved_support_q3", ("approved", "support", "q3")),
+    ("approved_support_q4", ("approved", "support", "q4")),
+    ("approved_support_year", ("approved", "support", "year")),
+    ("approved_development_q1", ("approved", "development", "q1")),
+    ("approved_development_q2", ("approved", "development", "q2")),
+    ("approved_development_q3", ("approved", "development", "q3")),
+    ("approved_development_q4", ("approved", "development", "q4")),
+    ("approved_development_year", ("approved", "development", "year")),
+    ("approved_total_year", ("approved", "total_year")),
+    ("fact_support_q1", ("fact", "support", "q1")),
+    ("fact_support_q2", ("fact", "support", "q2")),
+    ("fact_support_q3", ("fact", "support", "q3")),
+    ("fact_support_q4", ("fact", "support", "q4")),
+    ("fact_support_year", ("fact", "support", "year")),
+    ("fact_development_q1", ("fact", "development", "q1")),
+    ("fact_development_q2", ("fact", "development", "q2")),
+    ("fact_development_q3", ("fact", "development", "q3")),
+    ("fact_development_q4", ("fact", "development", "q4")),
+    ("fact_development_year", ("fact", "development", "year")),
+    ("fact_total_year", ("fact", "total_year")),
+    ("corrected_support_q2", ("corrected", "support", "q2")),
+    ("corrected_support_q3", ("corrected", "support", "q3")),
+    ("corrected_support_q4", ("corrected", "support", "q4")),
+    ("corrected_development_q2", ("corrected", "development", "q2")),
+    ("corrected_development_q3", ("corrected", "development", "q3")),
+    ("corrected_development_q4", ("corrected", "development", "q4")),
+]
+QUARTER_KEYS = ("q1", "q2", "q3", "q4", "totals")
+FORM1_SECTION_BLOCK_KEYS = (
+    "sequestration",
+    "allocation",
+    "reserve",
+    "approved",
+    "collegial",
+    "ckk",
+    "contract",
+    "plan",
+)
+FORM1_METRIC_BLOCK_KEYS = (
+    "plan",
+    "reserve",
+    "approved",
+    "collegial",
+    "allocation",
+    "sequestration",
+    "contract",
+    "ckk",
+    "contract_summary",
+)
+
+
+class ExcelValueSerializer:
+    def to_excel_value(self, value: Any) -> Any:
+        if value is None or isinstance(value, (str, int, float, bool)):
+            return value
+        if isinstance(value, dict):
+            for preferred_key in ("name", "title", "label", "value"):
+                preferred_value = value.get(preferred_key)
+                if isinstance(preferred_value, (str, int, float, bool)) and preferred_value is not None:
+                    return preferred_value
+            if len(value) == 1:
+                only_value = next(iter(value.values()))
+                if isinstance(only_value, (str, int, float, bool)) or only_value is None:
+                    return only_value
+            return self._dict_to_readable_text(value)
+        if isinstance(value, (list, tuple)):
+            if all(isinstance(item, (str, int, float, bool)) or item is None for item in value):
+                return ", ".join("" if item is None else str(item) for item in value)
+            return json.dumps(value, ensure_ascii=False)
+        return str(value)
+
+    def _dict_to_readable_text(self, value: dict[str, Any]) -> Any:
+        parts: list[str] = []
+        for key, raw in value.items():
+            normalized = self.to_excel_value(raw)
+            if self._is_empty_excel_value(normalized):
+                continue
+            parts.append(f"{key}={normalized}")
+        if not parts:
+            return None
+        return "; ".join(parts)
+
+    def dict_to_readable_text(self, value: dict[str, Any]) -> Any:
+        return self._dict_to_readable_text(value)
+
+    @staticmethod
+    def _is_empty_excel_value(value: Any) -> bool:
+        if value is None:
+            return True
+        if isinstance(value, str) and value.strip() == "":
+            return True
+        return False
+
+
+class Form1RowNormalizer:
+    def __init__(self, serializer: ExcelValueSerializer):
+        self.serializer = serializer
+
+    def normalize(self, data: dict[str, Any]) -> dict[str, Any]:
+        normalized = dict(data)
+        self._merge_header_fields_into_row(normalized=normalized, original=data)
+        self._promote_quarter_values_from_section_blocks(normalized=normalized)
+        self._collapse_metric_blocks_to_scalar(normalized=normalized)
+        self._fill_contract_fields(normalized=normalized)
+        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")
+        normalized.pop("header", None)
+        normalized.pop("line_id", None)
+
+    def _promote_quarter_values_from_section_blocks(self, normalized: dict[str, Any]) -> None:
+        for section_key in FORM1_SECTION_BLOCK_KEYS:
+            section_payload = normalized.get(section_key)
+            if not isinstance(section_payload, dict):
+                continue
+            for quarter_key in QUARTER_KEYS:
+                if normalized.get(quarter_key) is None and section_payload.get(quarter_key) is not None:
+                    normalized[quarter_key] = section_payload.get(quarter_key)
+
+    def _collapse_metric_blocks_to_scalar(self, normalized: dict[str, Any]) -> None:
+        for block_key in FORM1_METRIC_BLOCK_KEYS:
+            block_value = normalized.get(block_key)
+            if isinstance(block_value, dict):
+                normalized[block_key] = self._extract_preferred_metric(block_value)
+
+    def _fill_contract_fields(self, normalized: dict[str, Any]) -> None:
+        contract_summary = normalized.get("contract_summary")
+        contract_detail = normalized.get("contract_detail")
+        if normalized.get("contract") is None:
+            normalized["contract"] = self._extract_contract_label(contract_detail, contract_summary)
+        if isinstance(contract_detail, dict):
+            normalized["contract_sum"] = self._extract_preferred_metric(contract_detail)
+
+    def _extract_preferred_metric(self, payload: dict[str, Any]) -> Any:
+        priority_keys = (
+            "total_year",
+            "year",
+            "totals",
+            "total",
+            "sum",
+            "amount",
+            "value",
+            "approved_amount",
+            "reserved_amount",
+            "allocated_amount",
+            "booked_amount",
+        )
+        for key in priority_keys:
+            if payload.get(key) is not None:
+                return payload.get(key)
+        for quarter_key in ("q1", "q2", "q3", "q4"):
+            if payload.get(quarter_key) is not None:
+                return payload.get(quarter_key)
+        return self.serializer.dict_to_readable_text(payload)
+
+    @staticmethod
+    def _extract_contract_label(contract_detail: Any, contract_summary: Any) -> Any:
+        for payload in (contract_detail, contract_summary):
+            if not isinstance(payload, dict):
+                continue
+            for key in ("reference", "contract_ref", "counterparty", "subject"):
+                value = payload.get(key)
+                if value not in (None, ""):
+                    return value
+        return None
+
+
+class SmetaRowNormalizer:
+    def normalize(self, data: dict[str, Any]) -> dict[str, Any]:
+        return {
+            field: self._pick_nested_value(data, *path)
+            for field, path in SMETA_VALUE_PATHS
+        }
+
+    @staticmethod
+    def _pick_nested_value(container: Any, *path: str) -> Any:
+        cur = container
+        for key in path:
+            if not isinstance(cur, dict):
+                return None
+            cur = cur.get(key)
+        return cur

api/src/services/export_service.py

@@ -0,0 +1,333 @@
+from __future__ import annotations
+
+import io
+import json
+import zipfile
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from typing import Any
+from urllib.parse import quote
+
+from openpyxl import Workbook
+from openpyxl.worksheet.worksheet import Worksheet
+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 (
+    ExcelValueSerializer,
+    Form1RowNormalizer,
+    SmetaRowNormalizer,
+)
+from src.services.export_writers import (
+    FORM1_FLAT_COLUMNS,
+    SMETA_COLUMNS,
+    TabularSheetWriter,
+)
+from src.services.sheet_service import SheetService
+
+XLSX_MEDIA_TYPE = "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"
+ZIP_MEDIA_TYPE = "application/zip"
+FORM1_DIRECTION_REQUIRED_SHEETS = {"AHR", "CAP"}
+SHEETS_WITH_SECTIONS = {"AHR", "CAP", "OPER"}
+
+
+@dataclass(frozen=True)
+class ExportSheetPlan:
+    sheet_name: str
+    effective_direction: str | None
+    effective_sections: list[str] | None
+    validation_direction: str | None
+    validation_sections: list[str] | None
+
+
+class ExportService:
+    def __init__(self, db: AsyncSession):
+        self.budget_form_service = BudgetFormService(db)
+        self.sheet_service = SheetService(db)
+        self.value_serializer = ExcelValueSerializer()
+        self.form1_row_normalizer = Form1RowNormalizer(self.value_serializer)
+        self.smeta_row_normalizer = SmetaRowNormalizer()
+        self.sheet_writers: dict[str, TabularSheetWriter] = {
+            "__default__": TabularSheetWriter(
+                columns=FORM1_FLAT_COLUMNS,
+                row_normalizer=self.form1_row_normalizer.normalize,
+                value_serializer=self.value_serializer.to_excel_value,
+            ),
+            "SMETA": TabularSheetWriter(
+                columns=SMETA_COLUMNS,
+                row_normalizer=self.smeta_row_normalizer.normalize,
+                value_serializer=self.value_serializer.to_excel_value,
+            ),
+        }
+
+    async def export_form_payload(
+        self,
+        form_id: int,
+        current_user: AppUser,
+        sheet: str | None = None,
+        direction: str | None = None,
+        sections: list[str] | None = None,
+        ignore_non_applicable_query_params: bool = False,
+    ) -> tuple[io.BytesIO, str, str]:
+        stream, filename = await self.export_form_to_xlsx(
+            form_id=form_id,
+            current_user=current_user,
+            sheet=sheet,
+            direction=direction,
+            sections=sections,
+            ignore_non_applicable_query_params=ignore_non_applicable_query_params,
+        )
+        return stream, filename, XLSX_MEDIA_TYPE
+
+    async def export_bulk_payload(
+        self,
+        form_ids: list[int],
+        current_user: AppUser,
+        skip_failed: bool = True,
+        direction: str | None = None,
+        sections: list[str] | None = None,
+        ignore_non_applicable_query_params: bool = False,
+    ) -> tuple[io.BytesIO, str, str]:
+        archive_stream = io.BytesIO()
+        exported_count = 0
+        errors: list[dict[str, Any]] = []
+        used_names: set[str] = set()
+
+        with zipfile.ZipFile(
+            archive_stream,
+            mode="w",
+            compression=zipfile.ZIP_DEFLATED,
+        ) as archive:
+            for form_id in self._deduplicate_ids(form_ids):
+                try:
+                    form_stream, filename = await self.export_form_to_xlsx(
+                        form_id=form_id,
+                        current_user=current_user,
+                        direction=direction,
+                        sections=sections,
+                        ignore_non_applicable_query_params=ignore_non_applicable_query_params,
+                    )
+                    archive_name = self._make_unique_filename(filename, used_names)
+                    archive.writestr(archive_name, form_stream.getvalue())
+                    exported_count += 1
+                except ValueError as exc:
+                    if not skip_failed:
+                        raise
+                    errors.append({"form_id": form_id, "error": str(exc)})
+
+            if errors:
+                archive.writestr(
+                    "errors.json",
+                    json.dumps({"errors": errors}, ensure_ascii=False, indent=2).encode("utf-8"),
+                )
+
+        if exported_count == 0:
+            raise ValueError("Не удалось экспортировать ни одной формы.")
+
+        archive_stream.seek(0)
+        filename = f"export_forms_{datetime.now(timezone.utc).strftime('%Y-%m-%d')}.zip"
+        return archive_stream, filename, ZIP_MEDIA_TYPE
+
+    async def export_form_to_xlsx(
+        self,
+        form_id: int,
+        current_user: AppUser,
+        sheet: str | None = None,
+        direction: str | None = None,
+        sections: list[str] | None = None,
+        ignore_non_applicable_query_params: bool = False,
+    ) -> tuple[io.BytesIO, str]:
+        form = await self._get_form_for_export(form_id=form_id, current_user=current_user)
+        sheets = self._resolve_requested_sheets(form=form, requested_sheet=sheet)
+        self._validate_sections_for_form(form=form, sections=sections)
+        plans = self._build_sheet_plans(
+            form=form,
+            sheets=sheets,
+            direction=direction,
+            sections=sections,
+            ignore_non_applicable_query_params=ignore_non_applicable_query_params,
+        )
+
+        workbook = self._create_workbook_with_metadata(form=form)
+        exported_sheets = 0
+        for plan in plans:
+            self._validate_sheet_query_params(
+                form_type_code=form.form_type.code,
+                sheet=plan.sheet_name,
+                direction=plan.validation_direction,
+                sections=plan.validation_sections,
+            )
+            rows = await self.sheet_service.get(
+                form_id=form_id,
+                sheet=plan.sheet_name,
+                direction=plan.effective_direction,
+                sections=plan.effective_sections,
+            )
+            worksheet = workbook.create_sheet(title=self._safe_sheet_name(plan.sheet_name))
+            self._write_sheet_rows(
+                worksheet=worksheet,
+                rows=rows,
+                sheet_name=plan.sheet_name,
+            )
+            exported_sheets += 1
+
+        if exported_sheets == 0:
+            raise ValueError("Не удалось экспортировать листы формы.")
+
+        stream = io.BytesIO()
+        workbook.save(stream)
+        workbook.close()
+        stream.seek(0)
+        filename = self._build_filename(form_id=form.id, form_type_code=form.form_type.code, year=form.year, sheet=sheet)
+        return stream, filename
+
+    @staticmethod
+    def build_content_disposition(filename: str) -> str:
+        encoded = quote(filename)
+        return f"attachment; filename*=UTF-8''{encoded}"
+
+    def _fill_metadata(self, sheet: Worksheet, form: Any) -> None:
+        metadata_rows = [
+            ("Форма ID", form.id),
+            ("Тип формы", form.form_type_code),
+            ("Год", form.year),
+            ("Org Unit ID", form.org_unit_id),
+            ("Org Unit", form.org_unit.title if form.org_unit else None),
+            ("Экспортировано", datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S")),
+        ]
+        for row_idx, (key, value) in enumerate(metadata_rows, start=1):
+            sheet.cell(row=row_idx, column=1, value=key)
+            sheet.cell(row=row_idx, column=2, value=value)
+
+    def _write_sheet_rows(
+        self,
+        worksheet: Worksheet,
+        rows: list[tuple],
+        sheet_name: str,
+    ) -> None:
+        writer = self.sheet_writers.get(sheet_name, self.sheet_writers["__default__"])
+        writer.write(worksheet, rows)
+
+    async def _get_form_for_export(self, form_id: int, current_user: AppUser) -> Any:
+        form = await self.budget_form_service.get(
+            budget_form_id=form_id,
+            user=current_user,
+            load_form_type=True,
+            load_org=True,
+        )
+        if not form:
+            raise ValueError(f"Форма {form_id} не найдена")
+        return form
+
+    def _resolve_requested_sheets(self, form: Any, requested_sheet: str | None) -> list[str]:
+        if requested_sheet and requested_sheet not in form.form_type.sheet_list:
+            raise ValueError(f"Лист {requested_sheet} формы {form.id} не найден")
+        if requested_sheet:
+            return [requested_sheet]
+        return list(form.form_type.sheet_list)
+
+    @staticmethod
+    def _validate_sections_for_form(form: Any, sections: list[str] | None) -> None:
+        if not sections:
+            return
+        invalid = sorted(set(sections) - set(form.form_type.section_list))
+        if invalid:
+            raise ValueError(f"Недопустимые sections: {', '.join(invalid)}")
+
+    def _create_workbook_with_metadata(self, form: Any) -> Workbook:
+        workbook = Workbook()
+        meta_sheet = workbook.active
+        meta_sheet.title = "metadata"
+        self._fill_metadata(meta_sheet, form)
+        return workbook
+
+    def _build_sheet_plans(
+        self,
+        form: Any,
+        sheets: list[str],
+        direction: str | None,
+        sections: list[str] | None,
+        ignore_non_applicable_query_params: bool,
+    ) -> list[ExportSheetPlan]:
+        plans: list[ExportSheetPlan] = []
+        for sheet_name in sheets:
+            effective_direction = direction if self._requires_direction(form.form_type.code, sheet_name) else None
+            effective_sections = sections if sheet_name in SHEETS_WITH_SECTIONS else None
+            validation_direction = effective_direction if ignore_non_applicable_query_params else direction
+            validation_sections = effective_sections if ignore_non_applicable_query_params else sections
+            plans.append(
+                ExportSheetPlan(
+                    sheet_name=sheet_name,
+                    effective_direction=effective_direction,
+                    effective_sections=effective_sections,
+                    validation_direction=validation_direction,
+                    validation_sections=validation_sections,
+                )
+            )
+        return plans
+
+
+    def _validate_sheet_query_params(
+        self,
+        form_type_code: FormTypeEnum,
+        sheet: str,
+        direction: str | None,
+        sections: list[str] | None,
+    ) -> None:
+        if form_type_code == FormTypeEnum.FORM_1 and sheet in FORM1_DIRECTION_REQUIRED_SHEETS and not direction:
+            raise ValueError("direction обязателен для FORM_1 листов AHR/CAP")
+        if direction and not self._requires_direction(form_type_code, sheet):
+            raise ValueError("direction допустим только для FORM_1 листов AHR/CAP")
+        if sections and sheet not in SHEETS_WITH_SECTIONS:
+            raise ValueError(f"sections не поддерживается для листа {sheet}")
+
+    @staticmethod
+    def _requires_direction(form_type_code: FormTypeEnum, sheet: str) -> bool:
+        return form_type_code == FormTypeEnum.FORM_1 and sheet in FORM1_DIRECTION_REQUIRED_SHEETS
+
+    @staticmethod
+    def _safe_sheet_name(value: str) -> str:
+        safe = value
+        for char in ("[", "]", ":", "*", "?", "/", "\\"):
+            safe = safe.replace(char, "_")
+        safe = safe.strip()
+        return (safe or "Sheet")[:31]
+
+    @staticmethod
+    def _build_filename(form_id: int, form_type_code: str, year: int, sheet: str | None = None) -> str:
+        date_str = datetime.now(timezone.utc).strftime("%Y-%m-%d")
+        if sheet:
+            return f"{year}_{form_type_code}_{form_id}_{sheet}_{date_str}.xlsx"
+        return f"{year}_{form_type_code}_{form_id}_{date_str}.xlsx"
+
+    @staticmethod
+    def _deduplicate_ids(form_ids: list[int]) -> list[int]:
+        seen: set[int] = set()
+        result: list[int] = []
+        for form_id in form_ids:
+            if form_id not in seen:
+                seen.add(form_id)
+                result.append(form_id)
+        return result
+
+    @staticmethod
+    def _make_unique_filename(filename: str, used_names: set[str]) -> str:
+        if filename not in used_names:
+            used_names.add(filename)
+            return filename
+
+        if "." in filename:
+            base, ext = filename.rsplit(".", 1)
+            ext = f".{ext}"
+        else:
+            base, ext = filename, ""
+
+        suffix = 2
+        while True:
+            candidate = f"{base}_{suffix}{ext}"
+            if candidate not in used_names:
+                used_names.add(candidate)
+                return candidate
+            suffix += 1

api/src/services/export_writers.py

@@ -0,0 +1,93 @@
+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 квартал"),
+]
+
+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

@@ -0,0 +1,51 @@
+from src.services.export_normalizers import (
+    ExcelValueSerializer,
+    Form1RowNormalizer,
+    SmetaRowNormalizer,
+)
+
+
+def test_form1_normalizer_merges_header_and_contract_fields():
+    serializer = ExcelValueSerializer()
+    normalizer = Form1RowNormalizer(serializer)
+    row = {
+        "header": {"section": "1.00.0.", "item_id": 1, "name": "A"},
+        "contract_detail": {"reference": "DOG-1", "amount": 10},
+        "plan": {"q1": 7},
+    }
+
+    normalized = normalizer.normalize(row)
+
+    assert normalized["section"] == "1.00.0."
+    assert normalized["item_id"] == 1
+    assert normalized["name"] == "A"
+    assert normalized["q1"] == 7
+    assert normalized["contract"] == "DOG-1"
+    assert normalized["contract_sum"] == 10
+    assert "header" not in normalized
+
+
+def test_smeta_normalizer_extracts_nested_values():
+    normalizer = SmetaRowNormalizer()
+    row = {
+        "section_code": "1",
+        "name": "SMETA row",
+        "plan": {"support": {"q1": 5}, "total_year": 20},
+        "approved": {"development": {"year": 8}},
+    }
+
+    normalized = normalizer.normalize(row)
+
+    assert normalized["section_code"] == "1"
+    assert normalized["name"] == "SMETA row"
+    assert normalized["plan_support_q1"] == 5
+    assert normalized["plan_total_year"] == 20
+    assert normalized["approved_development_year"] == 8
+
+
+def test_excel_value_serializer_formats_collections():
+    serializer = ExcelValueSerializer()
+
+    assert serializer.to_excel_value(["a", 1, None]) == "a, 1, "
+    assert serializer.to_excel_value({"name": "X"}) == "X"
+    assert serializer.to_excel_value({"k": {"nested": 1}}) == "k=1"

api/tests/unit/test_export_service_query_params.py

@@ -0,0 +1,115 @@
+import io
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, call
+
+import pytest
+
+from src.db.models.form_type import FormTypeEnum
+from src.domain.schemas import DirectionSchemaEnum
+from src.services.export_service import ExportService
+
+
+def _build_form(
+    *,
+    form_type_code: FormTypeEnum = FormTypeEnum.FORM_1,
+    sheet_list: list[str] | None = None,
+    section_list: list[str] | None = None,
+) -> SimpleNamespace:
+    return SimpleNamespace(
+        id=1,
+        year=2026,
+        form_type_code=form_type_code.value,
+        org_unit_id=10,
+        org_unit=SimpleNamespace(title="Org"),
+        form_type=SimpleNamespace(
+            code=form_type_code,
+            sheet_list=sheet_list or ["AHR", "SMETA"],
+            section_list=section_list or ["plan", "q1"],
+        ),
+    )
+
+
+@pytest.mark.asyncio
+async def test_export_sheet_rejects_direction_for_unsupported_sheet():
+    service = ExportService(AsyncMock())
+    service.budget_form_service.get = AsyncMock(
+        return_value=_build_form(sheet_list=["SMETA"])
+    )
+    service.sheet_service.get = AsyncMock(return_value=[])
+
+    with pytest.raises(ValueError, match="direction допустим только"):
+        await service.export_form_to_xlsx(
+            form_id=1,
+            current_user=SimpleNamespace(),
+            sheet="SMETA",
+            direction=DirectionSchemaEnum.SUPPORT.value,
+        )
+
+
+@pytest.mark.asyncio
+async def test_export_sheet_rejects_sections_for_unsupported_sheet():
+    service = ExportService(AsyncMock())
+    service.budget_form_service.get = AsyncMock(
+        return_value=_build_form(sheet_list=["SMETA"])
+    )
+    service.sheet_service.get = AsyncMock(return_value=[])
+
+    with pytest.raises(ValueError, match="sections не поддерживается"):
+        await service.export_form_to_xlsx(
+            form_id=1,
+            current_user=SimpleNamespace(),
+            sheet="SMETA",
+            sections=["q1"],
+        )
+
+
+@pytest.mark.asyncio
+async def test_export_full_form_ignores_non_applicable_params_for_other_sheets():
+    service = ExportService(AsyncMock())
+    service.budget_form_service.get = AsyncMock(return_value=_build_form())
+    service.sheet_service.get = AsyncMock(return_value=[])
+
+    stream, filename = await service.export_form_to_xlsx(
+        form_id=1,
+        current_user=SimpleNamespace(),
+        direction=DirectionSchemaEnum.SUPPORT.value,
+        sections=["q1"],
+        ignore_non_applicable_query_params=True,
+    )
+
+    assert stream.getbuffer().nbytes > 0
+    assert filename.endswith(".xlsx")
+    assert service.sheet_service.get.await_args_list == [
+        call(
+            form_id=1,
+            sheet="AHR",
+            direction=DirectionSchemaEnum.SUPPORT.value,
+            sections=["q1"],
+        ),
+        call(
+            form_id=1,
+            sheet="SMETA",
+            direction=None,
+            sections=None,
+        ),
+    ]
+
+
+@pytest.mark.asyncio
+async def test_export_form_payload_forwards_ignore_flag():
+    service = ExportService(AsyncMock())
+    service.export_form_to_xlsx = AsyncMock(return_value=(io.BytesIO(), "demo.xlsx"))
+
+    await service.export_form_payload(
+        form_id=1,
+        current_user=SimpleNamespace(),
+        ignore_non_applicable_query_params=True,
+    )
+
+    assert service.export_form_to_xlsx.await_count == 1
+    assert (
+        service.export_form_to_xlsx.await_args.kwargs[
+            "ignore_non_applicable_query_params"
+        ]
+        is True
+    )

api/tests/unit/test_export_writers.py

@@ -0,0 +1,60 @@
+from unittest.mock import AsyncMock
+
+from openpyxl import Workbook
+
+from src.services.export_service import ExportService
+from src.services.export_writers import (
+    FORM1_FLAT_COLUMNS,
+    SMETA_COLUMNS,
+    TabularSheetWriter,
+)
+
+
+def test_tabular_sheet_writer_writes_headers_and_rows():
+    writer = TabularSheetWriter(
+        columns=[("a", "A"), ("b", "B")],
+        row_normalizer=lambda row: row,
+        value_serializer=lambda value: value,
+    )
+    wb = Workbook()
+    ws = wb.active
+
+    rows = [(None, None, None, {"a": 1, "b": 2})]
+    writer.write(ws, rows)
+
+    assert ws.cell(row=1, column=1).value == "A"
+    assert ws.cell(row=1, column=2).value == "B"
+    assert ws.cell(row=2, column=1).value == 1
+    assert ws.cell(row=2, column=2).value == 2
+
+
+def test_sheet_column_contracts_are_stable():
+    assert len(FORM1_FLAT_COLUMNS) == 17
+    assert len(SMETA_COLUMNS) == 41
+    assert FORM1_FLAT_COLUMNS[0] == ("section", "Код раздела")
+    assert SMETA_COLUMNS[0] == ("section_code", "Код раздела")
+
+
+def test_export_service_uses_sheet_writer_strategy():
+    service = ExportService(AsyncMock())
+
+    class Recorder:
+        def __init__(self):
+            self.calls = 0
+
+        def write(self, worksheet, rows):
+            self.calls += 1
+            worksheet.append(["ok"])
+
+    default_writer = Recorder()
+    smeta_writer = Recorder()
+    service.sheet_writers = {"__default__": default_writer, "SMETA": smeta_writer}
+
+    wb = Workbook()
+    ws_default = wb.active
+    ws_smeta = wb.create_sheet("smeta")
+    service._write_sheet_rows(ws_default, [], "AHR")
+    service._write_sheet_rows(ws_smeta, [], "SMETA")
+
+    assert default_writer.calls == 1
+    assert smeta_writer.calls == 1