From ad6eae9c0b8fa436864a3bcefdc3834f746eea9e Mon Sep 17 00:00:00 2001 From: Raykov-MS Date: Mon, 25 May 2026 17:31:29 +0300 Subject: [PATCH 1/2] Export --- api/requirements.txt | 1 + api/src/api/v1/export.py | 117 ++++++ api/src/api/v1/router.py | 3 +- api/src/domain/schemas.py | 5 + api/src/services/export_normalizers.py | 211 +++++++++++ api/src/services/export_service.py | 333 ++++++++++++++++++ api/src/services/export_writers.py | 93 +++++ api/tests/unit/test_export_normalizers.py | 51 +++ .../unit/test_export_service_query_params.py | 115 ++++++ api/tests/unit/test_export_writers.py | 60 ++++ 10 files changed, 988 insertions(+), 1 deletion(-) create mode 100644 api/src/api/v1/export.py create mode 100644 api/src/services/export_normalizers.py create mode 100644 api/src/services/export_service.py create mode 100644 api/src/services/export_writers.py create mode 100644 api/tests/unit/test_export_normalizers.py create mode 100644 api/tests/unit/test_export_service_query_params.py create mode 100644 api/tests/unit/test_export_writers.py diff --git a/api/requirements.txt b/api/requirements.txt index a7407f8..db3ab58 100644 --- a/api/requirements.txt +++ b/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 diff --git a/api/src/api/v1/export.py b/api/src/api/v1/export.py new file mode 100644 index 0000000..5b9c515 --- /dev/null +++ b/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)}, + ) + + diff --git a/api/src/api/v1/router.py b/api/src/api/v1/router.py index 915dead..e89635c 100644 --- a/api/src/api/v1/router.py +++ b/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) diff --git a/api/src/domain/schemas.py b/api/src/domain/schemas.py index ac85ccc..367aa41 100644 --- a/api/src/domain/schemas.py +++ b/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 diff --git a/api/src/services/export_normalizers.py b/api/src/services/export_normalizers.py new file mode 100644 index 0000000..6962598 --- /dev/null +++ b/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 diff --git a/api/src/services/export_service.py b/api/src/services/export_service.py new file mode 100644 index 0000000..fc5cb8d --- /dev/null +++ b/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 diff --git a/api/src/services/export_writers.py b/api/src/services/export_writers.py new file mode 100644 index 0000000..1c8c43e --- /dev/null +++ b/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] + ) diff --git a/api/tests/unit/test_export_normalizers.py b/api/tests/unit/test_export_normalizers.py new file mode 100644 index 0000000..72b9c17 --- /dev/null +++ b/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" diff --git a/api/tests/unit/test_export_service_query_params.py b/api/tests/unit/test_export_service_query_params.py new file mode 100644 index 0000000..bf43156 --- /dev/null +++ b/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 + ) diff --git a/api/tests/unit/test_export_writers.py b/api/tests/unit/test_export_writers.py new file mode 100644 index 0000000..f16bd81 --- /dev/null +++ b/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 From fc55a7b0429004c6171b65804dd150c5fb1e888a Mon Sep 17 00:00:00 2001 From: Raykov-MS Date: Mon, 25 May 2026 18:46:11 +0300 Subject: [PATCH 2/2] =?UTF-8?q?=D0=9F=D1=80=D0=B0=D0=B2=D0=BA=D0=B8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- api/src/api/v1/export.py | 38 +++++++++++ api/src/db/models/form_type.py | 2 + api/src/services/export_normalizers.py | 37 ++++++++++ api/src/services/export_service.py | 67 +++++++++++++++++++ api/src/services/export_writers.py | 10 +++ .../test_export_projects_integration.py | 38 +++++++++++ api/tests/unit/test_export_normalizers.py | 24 +++++++ 7 files changed, 216 insertions(+) create mode 100644 api/tests/integration/test_export_projects_integration.py diff --git a/api/src/api/v1/export.py b/api/src/api/v1/export.py index 5b9c515..b6d9c1b 100644 --- a/api/src/api/v1/export.py +++ b/api/src/api/v1/export.py @@ -9,6 +9,7 @@ from src.domain.schemas import DirectionSchemaEnum, ExportBulkRequest from src.services.export_service import ExportService router = APIRouter(prefix="/export", tags=["export"]) +FORM3_ALLOWED_SECTIONS = {"q1", "q2", "q3", "q4", "year"} def _parse_sections_csv(sections: str | None) -> list[str] | None: @@ -36,6 +37,16 @@ def _parse_sections_csv(sections: str | None) -> list[str] | None: return normalized or None +def _parse_form3_sections_csv(sections: str | None) -> list[str] | None: + if not sections: + return None + parsed = [section.strip() for section in sections.split(",") if section.strip()] + invalid = sorted(set(parsed) - FORM3_ALLOWED_SECTIONS) + if invalid: + raise ValueError(f"Недопустимые sections для FORM_3: {', '.join(invalid)}") + return parsed + + @router.get("/form/{form_id}") async def export_form( form_id: int, @@ -115,3 +126,30 @@ async def export_forms_bulk( ) +@router.get("/project/{project_id}/report/{year}/{report_type}") +async def export_project_report( + project_id: int, + year: int, + report_type: str, + 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_project_report_payload( + project_id=project_id, + year=year, + report_type=report_type, + current_user=current_user, + sections=_parse_form3_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)}, + ) + + diff --git a/api/src/db/models/form_type.py b/api/src/db/models/form_type.py index 9194e9a..c5ede4b 100644 --- a/api/src/db/models/form_type.py +++ b/api/src/db/models/form_type.py @@ -34,11 +34,13 @@ class FormType(Base): FormTypeEnum.FORM_1: ["AHR", "CAP", "OPER", "SMETA"], FormTypeEnum.FORM_2: ["AHR", "CAP", "OPER", "AHR_LIMIT", "AHR_RENT", "AHR_UTILITY", "AHR_SECURITY", "OTCH9F", "SMETA"], + FormTypeEnum.FORM_3: ["LIMIT", "CURRENT_EXPENSES"], FormTypeEnum.FORM_4: ["AHR", "CAP", "OPER", "STRUCTURE", "SMETA"], } SECTIONS_BY_FORM_TYPE = { FormTypeEnum.FORM_1: ["plan", "contract_summary", "allocation", "sequestration", "reserve", "approved", "collegial", "ckk", "contract_detail", "q1", "q2", "q3", "q4", "totals"], FormTypeEnum.FORM_2: ["plan", "seq_dfip", "seq_ssp", "approved", "contract", "booking", "q1", "q2", "q3", "q4", "totals"], + FormTypeEnum.FORM_3: ["q1", "q2", "q3", "q4", "year"], FormTypeEnum.FORM_4: ["plan", "seq_dfip", "approved", "contract", "booking", "q1", "q2", "q3", "q4", "totals", "contract_summary", "allocation", "reserve", "collegial", "ckk"], } diff --git a/api/src/services/export_normalizers.py b/api/src/services/export_normalizers.py index 6962598..c8fe28d 100644 --- a/api/src/services/export_normalizers.py +++ b/api/src/services/export_normalizers.py @@ -209,3 +209,40 @@ class SmetaRowNormalizer: return None cur = cur.get(key) return cur + + +class Form3ReportRowNormalizer: + def normalize(self, data: dict[str, Any]) -> dict[str, Any]: + normalized = dict(data) + header = data.get("header") + if isinstance(header, dict): + if "section_code" in header and "section" not in normalized: + normalized["section"] = header.get("section_code") + 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") + for quarter_key in ("q1", "q2", "q3", "q4"): + normalized[quarter_key] = self._extract_quarter_metric(normalized.get(quarter_key)) + normalized["totals"] = self._extract_totals_metric(normalized.get("totals")) + normalized.pop("header", None) + normalized.pop("line_id", None) + return normalized + + @staticmethod + def _extract_quarter_metric(value: Any) -> Any: + if not isinstance(value, dict): + return value + for key in ("quarter_actual", "total_corr", "economy"): + if value.get(key) is not None: + return value.get(key) + return None + + @staticmethod + def _extract_totals_metric(value: Any) -> Any: + if not isinstance(value, dict): + return value + for key in ("total_actual", "total_corr", "economy"): + if value.get(key) is not None: + return value.get(key) + return None diff --git a/api/src/services/export_service.py b/api/src/services/export_service.py index fc5cb8d..e6dd1e9 100644 --- a/api/src/services/export_service.py +++ b/api/src/services/export_service.py @@ -18,13 +18,16 @@ from src.services.budget_form_service import BudgetFormService from src.services.export_normalizers import ( ExcelValueSerializer, Form1RowNormalizer, + Form3ReportRowNormalizer, SmetaRowNormalizer, ) from src.services.export_writers import ( FORM1_FLAT_COLUMNS, + FORM3_REPORT_COLUMNS, SMETA_COLUMNS, TabularSheetWriter, ) +from src.services.project_service import ProjectService from src.services.sheet_service import SheetService XLSX_MEDIA_TYPE = "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet" @@ -46,8 +49,10 @@ class ExportService: def __init__(self, db: AsyncSession): self.budget_form_service = BudgetFormService(db) self.sheet_service = SheetService(db) + self.project_service = ProjectService(db) self.value_serializer = ExcelValueSerializer() self.form1_row_normalizer = Form1RowNormalizer(self.value_serializer) + self.form3_report_row_normalizer = Form3ReportRowNormalizer() self.smeta_row_normalizer = SmetaRowNormalizer() self.sheet_writers: dict[str, TabularSheetWriter] = { "__default__": TabularSheetWriter( @@ -61,6 +66,11 @@ class ExportService: value_serializer=self.value_serializer.to_excel_value, ), } + self.form3_report_writer = TabularSheetWriter( + columns=FORM3_REPORT_COLUMNS, + row_normalizer=self.form3_report_row_normalizer.normalize, + value_serializer=self.value_serializer.to_excel_value, + ) async def export_form_payload( self, @@ -81,6 +91,23 @@ class ExportService: ) return stream, filename, XLSX_MEDIA_TYPE + async def export_project_report_payload( + self, + project_id: int, + year: int, + report_type: str, + current_user: AppUser, + sections: list[str] | None = None, + ) -> tuple[io.BytesIO, str, str]: + stream, filename = await self.export_project_report_to_xlsx( + project_id=project_id, + year=year, + report_type=report_type, + current_user=current_user, + sections=sections, + ) + return stream, filename, XLSX_MEDIA_TYPE + async def export_bulk_payload( self, form_ids: list[int], @@ -183,6 +210,46 @@ class ExportService: filename = self._build_filename(form_id=form.id, form_type_code=form.form_type.code, year=form.year, sheet=sheet) return stream, filename + async def export_project_report_to_xlsx( + self, + project_id: int, + year: int, + report_type: str, + current_user: AppUser, + sections: list[str] | None = None, + ) -> tuple[io.BytesIO, str]: + rows = await self.project_service.get_report_rows( + project_id=project_id, + year=year, + report_type=report_type, + sections=sections, + user=current_user, + ) + report_type_upper = report_type.upper() + workbook = Workbook() + meta_sheet = workbook.active + meta_sheet.title = "metadata" + metadata_rows = [ + ("Тип формы", "FORM_3"), + ("Project ID", project_id), + ("Year", year), + ("Report Type", report_type_upper), + ("Sections", ", ".join(sections) if sections else None), + ("Экспортировано", datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S")), + ] + for row_idx, (key, value) in enumerate(metadata_rows, start=1): + meta_sheet.cell(row=row_idx, column=1, value=key) + meta_sheet.cell(row=row_idx, column=2, value=value) + worksheet = workbook.create_sheet(title=self._safe_sheet_name(report_type_upper)) + self.form3_report_writer.write(worksheet, rows) + stream = io.BytesIO() + workbook.save(stream) + workbook.close() + stream.seek(0) + date_str = datetime.now(timezone.utc).strftime("%Y-%m-%d") + filename = f"{year}_FORM_3_{project_id}_{report_type_upper}_{date_str}.xlsx" + return stream, filename + @staticmethod def build_content_disposition(filename: str) -> str: encoded = quote(filename) diff --git a/api/src/services/export_writers.py b/api/src/services/export_writers.py index 1c8c43e..fb70fdf 100644 --- a/api/src/services/export_writers.py +++ b/api/src/services/export_writers.py @@ -67,6 +67,16 @@ SMETA_COLUMNS: list[tuple[str, str]] = [ ("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] diff --git a/api/tests/integration/test_export_projects_integration.py b/api/tests/integration/test_export_projects_integration.py new file mode 100644 index 0000000..77fa3c6 --- /dev/null +++ b/api/tests/integration/test_export_projects_integration.py @@ -0,0 +1,38 @@ +import pytest + + +def _get_any_project_report(client, admin_tokens, auth_headers) -> tuple[int, int, str]: + list_response = client.get( + "/api/v1/projects?limit=50", + headers=auth_headers(admin_tokens), + ) + assert list_response.status_code == 200 + projects = list_response.json().get("result") or [] + for project in projects: + project_id = int(project["id"]) + details_response = client.get( + f"/api/v1/projects/{project_id}", + headers=auth_headers(admin_tokens), + ) + if details_response.status_code != 200: + continue + reports = (details_response.json().get("result") or {}).get("reports") or [] + if reports: + report = reports[0] + return project_id, int(report["year"]), str(report["report_type"]) + pytest.skip("Не найден проект FORM_3 с отчетами для smoke-экспорта") + + +def test_export_project_report_smoke(client, admin_tokens, auth_headers): + project_id, report_year, report_type = _get_any_project_report(client, admin_tokens, auth_headers) + + response = client.get( + f"/api/v1/export/project/{project_id}/report/{report_year}/{report_type}", + headers=auth_headers(admin_tokens), + ) + assert response.status_code == 200 + assert response.headers["content-type"] == ( + "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet" + ) + assert "attachment; filename*=UTF-8''" in response.headers["content-disposition"] + assert len(response.content) > 0 diff --git a/api/tests/unit/test_export_normalizers.py b/api/tests/unit/test_export_normalizers.py index 72b9c17..933aaf6 100644 --- a/api/tests/unit/test_export_normalizers.py +++ b/api/tests/unit/test_export_normalizers.py @@ -1,6 +1,7 @@ from src.services.export_normalizers import ( ExcelValueSerializer, Form1RowNormalizer, + Form3ReportRowNormalizer, SmetaRowNormalizer, ) @@ -49,3 +50,26 @@ def test_excel_value_serializer_formats_collections(): 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" + + +def test_form3_normalizer_flattens_quarters_and_totals_to_scalars(): + normalizer = Form3ReportRowNormalizer() + row = { + "header": {"section_code": "1.00.0.", "item_id": "R001", "name": "Строка"}, + "q1": {"quarter_actual": 10, "total_corr": 12}, + "q2": {"total_corr": 22}, + "q3": {"economy": 1.5}, + "q4": {"m1": 100}, + "totals": {"total_actual": 40, "economy": 2.1}, + } + + normalized = normalizer.normalize(row) + + assert normalized["section"] == "1.00.0." + assert normalized["item_id"] == "R001" + assert normalized["name"] == "Строка" + assert normalized["q1"] == 10 + assert normalized["q2"] == 22 + assert normalized["q3"] == 1.5 + assert normalized["q4"] is None + assert normalized["totals"] == 40