RSHB/DFiP_Budget_planing
5
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

export-optimize: оптимизировал время экспорта

Browse Source
This commit is contained in:
tsygankoviva 2026-09-03 17:02:15 +03:00
parent 5dcfd73a10
commit bcf9cd7d6c
14 changed files with 1531 additions and 82 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
api/alembic/versions/0026_summary2_go_export.py Normal file
View File

@ -0,0 +1,19 @@
from alembic import op
from sql_migration import run_sql_file_migration
revision = "0026"
down_revision = "0025"
branch_labels = None
depends_on = None
def upgrade() -> None:
run_sql_file_migration("0026_summary2_go_export.sql")
def downgrade() -> None:
op.execute(
"DROP FUNCTION IF EXISTS "
"v3.v_svod2_go_export(integer, integer[], varchar, varchar)"
)

210
api/alembic/versions/sql/0026_summary2_go_export.sql Normal file
View File

@ -0,0 +1,210 @@
DROP FUNCTION IF EXISTS v3.v_svod2_go_export(integer, integer[], varchar, varchar);
CREATE OR REPLACE FUNCTION v3.v_svod2_go_export(
p_year integer,
p_org_unit_ids integer[] DEFAULT NULL,
p_sort_by varchar DEFAULT 'org_name',
p_sort_direction varchar DEFAULT 'asc'
)
RETURNS TABLE(sort_order bigint, row_values text[])
LANGUAGE sql
STABLE
SET search_path TO 'v3', 'pg_catalog'
AS $function$
WITH eligible_lines AS (
SELECT
bl.id AS line_id,
bf.form_type_code,
bf.id AS form_id,
bf.org_unit_id,
ou.title AS org_name,
ei.section_code,
ei.item_id,
ei.num_group_id,
COALESCE(bl.name, ei.name) AS name,
bl.justification,
CASE bl.direction
WHEN 'Support' THEN 'Поддержка'
WHEN 'Development' THEN 'Развитие'
END::varchar AS smeta_direction,
CASE ei.sheet
WHEN 'AHR' THEN 'АХР'
WHEN 'CAP' THEN 'КВП'
WHEN 'OPER' THEN 'Операц'
END::varchar AS smeta_type,
cd.counterparty,
cd.reference,
cd.subject,
cd.currency,
cd.ceiling_amount,
cd.expenses_q1,
cd.expenses_q2,
cd.expenses_q3,
cd.expenses_q4,
cd.vat_rate,
cd.deadline
FROM v3.budget_form bf
JOIN v3.budget_line bl ON bl.budget_form_id = bf.id
JOIN v3.expense_item ei ON ei.id = bl.expense_item_id
JOIN v3.contract_detail cd ON cd.line_id = bl.id
LEFT JOIN v3.org_unit ou ON ou.id = bf.org_unit_id
WHERE bf.year = p_year
AND bf.form_type_code IN ('FORM_1', 'FORM_4')
AND ei.sheet IN ('AHR', 'CAP', 'OPER')
AND (
p_org_unit_ids IS NULL
OR cardinality(p_org_unit_ids) = 0
OR bf.org_unit_id = ANY(p_org_unit_ids)
)
), page AS MATERIALIZED (
SELECT
el.*,
row_number() OVER (
ORDER BY
CASE WHEN p_sort_by = 'org_name' AND lower(p_sort_direction) = 'asc' THEN el.org_name END ASC NULLS LAST,
CASE WHEN p_sort_by = 'org_name' AND lower(p_sort_direction) = 'desc' THEN el.org_name END DESC NULLS LAST,
CASE WHEN p_sort_by = 'org_name' AND lower(p_sort_direction) = 'asc' THEN el.item_id END ASC NULLS LAST,
CASE WHEN p_sort_by = 'org_name' AND lower(p_sort_direction) = 'desc' THEN el.item_id END DESC NULLS LAST,
CASE WHEN p_sort_by = 'section_code' AND lower(p_sort_direction) = 'asc' THEN el.section_code END ASC NULLS LAST,
CASE WHEN p_sort_by = 'section_code' AND lower(p_sort_direction) = 'desc' THEN el.section_code END DESC NULLS LAST,
el.form_type_code,
el.form_id,
el.line_id
) AS sort_order
FROM eligible_lines el
)
SELECT
pg.sort_order,
ARRAY[
pg.smeta_type::text,
pg.smeta_direction::text,
pg.org_name::text,
pg.name::text,
pg.item_id::text,
pg.num_group_id::text,
pg.justification::text,
pl.plan_q1::text,
pl.plan_q2::text,
pl.plan_q3::text,
pl.plan_q4::text,
al.internal_order::text,
al.property_object::text,
sd.adj_q1::text,
sd.adj_q2::text,
sd.adj_q3::text,
sd.adj_q4::text,
rv.amount_q1::text,
rv.amount_q2::text,
rv.amount_q3::text,
rv.amount_q4::text,
(CASE WHEN pl.line_id IS NULL AND sd.id IS NULL AND rv.line_id IS NULL THEN NULL ELSE COALESCE(pl.plan_q1, 0) + COALESCE(sd.adj_q1, 0) + COALESCE(rv.amount_q1, 0) END)::text,
(CASE WHEN pl.line_id IS NULL AND sd.id IS NULL AND rv.line_id IS NULL THEN NULL ELSE COALESCE(pl.plan_q2, 0) + COALESCE(sd.adj_q2, 0) + COALESCE(rv.amount_q2, 0) END)::text,
(CASE WHEN pl.line_id IS NULL AND sd.id IS NULL AND rv.line_id IS NULL THEN NULL ELSE COALESCE(pl.plan_q3, 0) + COALESCE(sd.adj_q3, 0) + COALESCE(rv.amount_q3, 0) END)::text,
(CASE WHEN pl.line_id IS NULL AND sd.id IS NULL AND rv.line_id IS NULL THEN NULL ELSE COALESCE(pl.plan_q4, 0) + COALESCE(sd.adj_q4, 0) + COALESCE(rv.amount_q4, 0) END)::text,
pg.counterparty::text,
pg.reference::text,
pg.subject::text,
pg.currency::text,
pg.ceiling_amount::text,
pg.expenses_q1::text,
pg.expenses_q2::text,
pg.expenses_q3::text,
pg.expenses_q4::text,
pg.vat_rate::text,
pg.deadline::text,
q1.adj_current::text,
q1.adj_ssp::text,
q1.adj_rf::text,
q1.adj_reserve::text,
q1.payment_date::text,
q1.payment_amount::text,
q1.payment_amount_ho::text,
q1.payment_amount_rf::text,
q1.payment_act::text,
q1.booking_amount::text,
q1.actual_m1::text,
q1.actual_m2::text,
q1.actual_m3::text,
q1.transfer_to_q2::text,
q1.transfer_to_q3::text,
q1.transfer_to_q4::text,
q1.transfer_to_economy::text,
q2.adj_current::text,
q2.adj_ssp::text,
q2.adj_rf::text,
q2.adj_reserve::text,
q2.payment_date::text,
q2.payment_amount::text,
q2.payment_amount_ho::text,
q2.payment_amount_rf::text,
q2.payment_act::text,
q2.booking_amount::text,
q2.actual_m1::text,
q2.actual_m2::text,
q2.actual_m3::text,
q2.plan_revision_eco_change::text,
q2.plan_revision_item_adj::text,
q2.plan_revision_increase::text,
q2.plan_revision_sequester::text,
q2.transfer_to_q3::text,
q2.transfer_to_q4::text,
q2.transfer_to_economy::text,
q3.adj_current::text,
q3.adj_ssp::text,
q3.adj_rf::text,
q3.adj_reserve::text,
q3.payment_date::text,
q3.payment_amount::text,
q3.payment_amount_ho::text,
q3.payment_amount_rf::text,
q3.payment_act::text,
q3.booking_amount::text,
q3.actual_m1::text,
q3.actual_m2::text,
q3.actual_m3::text,
q3.plan_revision_eco_change::text,
q3.plan_revision_item_adj::text,
q3.plan_revision_increase::text,
q3.plan_revision_sequester::text,
q3.transfer_to_q4::text,
q3.transfer_to_economy::text,
q4.adj_current::text,
q4.adj_ssp::text,
q4.adj_rf::text,
q4.adj_reserve::text,
q4.payment_date::text,
q4.payment_amount::text,
q4.payment_amount_ho::text,
q4.payment_amount_rf::text,
q4.payment_act::text,
q4.booking_amount::text,
q4.actual_m1::text,
q4.actual_m2::text,
q4.actual_m3::text,
q4.actual_spod::text,
q4.plan_revision_eco_change::text,
q4.plan_revision_item_adj::text,
q4.plan_revision_increase::text,
q4.plan_revision_sequester::text
]::text[] AS row_values
FROM page pg
LEFT JOIN v3.plan pl ON pl.line_id = pg.line_id
LEFT JOIN v3.sequestration sd
ON sd.line_id = pg.line_id
AND sd.actor = 'DFIP'
LEFT JOIN v3.reserve rv ON rv.line_id = pg.line_id
LEFT JOIN v3.allocation al ON al.line_id = pg.line_id
LEFT JOIN v3.budget_line_quarter q1
ON q1.line_id = pg.line_id AND q1.quarter = 1
LEFT JOIN v3.budget_line_quarter q2
ON q2.line_id = pg.line_id AND q2.quarter = 2
LEFT JOIN v3.budget_line_quarter q3
ON q3.line_id = pg.line_id AND q3.quarter = 3
LEFT JOIN v3.budget_line_quarter q4
ON q4.line_id = pg.line_id AND q4.quarter = 4
ORDER BY pg.sort_order;
$function$;
COMMENT ON FUNCTION v3.v_svod2_go_export(integer, integer[], varchar, varchar) IS
'Плоская выборка свода 2 / ГО для XLSX-экспорта без построения JSONB';

1
api/requirements-dev.txt
View File

@ -15,3 +15,4 @@ asyncpg==0.30.0
pytest==8.3.2
pytest-asyncio==0.24.0
openpyxl==3.1.5
XlsxWriter>=3.2,<4

1
api/requirements.txt
View File

@ -13,3 +13,4 @@ alembic==1.16.1
asyncpg==0.30.0
raisa-fastapi-protected-api==1.0.0
openpyxl
XlsxWriter>=3.2,<4

88
api/src/api/v1/export.py
View File

@ -1,15 +1,34 @@
from fastapi import APIRouter, Depends, HTTPException, status
from typing import Annotated, AsyncGenerator, Literal
from fastapi import APIRouter, Depends, HTTPException, Path, status
from fastapi.responses import StreamingResponse
from sqlalchemy.ext.asyncio import AsyncSession
from starlette.background import BackgroundTask
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
from src.services.export_lock import ExportAlreadyRunningError, user_export_slot
router = APIRouter(prefix="/export", tags=["export"])
FORM3_ALLOWED_SECTIONS = {"q1", "q2", "q3", "q4", "year"}
SUMMARY1_ALLOWED_SHEETS = {"FORM_1", "FORM_2", "FORM_3", "FORM_4", "MAIN"}
SUMMARY2_ALLOWED_SHEETS = {"GO", "RF"}
async def hold_user_export_slot(
current_user: AppUser = Depends(get_current_active_user_with_set_db),
) -> AsyncGenerator[None, None]:
try:
async with user_export_slot(current_user.id):
yield
except ExportAlreadyRunningError:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail="Экспорт уже выполняется. Дождитесь завершения предыдущего экспорта.",
)
def _parse_sections_csv(sections: str | None) -> list[str] | None:
@ -47,6 +66,19 @@ def _parse_form3_sections_csv(sections: str | None) -> list[str] | None:
return parsed
def _parse_org_ids_csv(org_ids: str | None) -> list[int] | None:
if not org_ids:
return None
try:
parsed = [int(value.strip()) for value in org_ids.split(",") if value.strip()]
except ValueError:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="org_ids должен быть списком целых чисел",
)
return parsed or None
@router.get("/form/{form_id}")
async def export_form(
form_id: int,
@ -54,6 +86,7 @@ async def export_form(
sections: str | None = None,
db: AsyncSession = Depends(get_db),
current_user: AppUser = Depends(get_current_active_user_with_set_db),
_export_slot: None = Depends(hold_user_export_slot),
):
export_service = ExportService(db)
try:
@ -62,7 +95,7 @@ async def export_form(
current_user=current_user,
direction=direction.value if direction else None,
sections=_parse_sections_csv(sections),
ignore_non_applicable_query_params=True,
ignore_unsupported_sheet_params=True,
)
except ValueError as exc:
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc))
@ -70,6 +103,7 @@ async def export_form(
stream,
media_type=media_type,
headers={"Content-Disposition": export_service.build_content_disposition(filename)},
background=BackgroundTask(stream.close),
)
@ -81,6 +115,7 @@ async def export_form_sheet(
sections: str | None = None,
db: AsyncSession = Depends(get_db),
current_user: AppUser = Depends(get_current_active_user_with_set_db),
_export_slot: None = Depends(hold_user_export_slot),
):
export_service = ExportService(db)
try:
@ -97,6 +132,7 @@ async def export_form_sheet(
stream,
media_type=media_type,
headers={"Content-Disposition": export_service.build_content_disposition(filename)},
background=BackgroundTask(stream.close),
)
@ -107,6 +143,7 @@ async def export_forms_bulk(
sections: str | None = None,
db: AsyncSession = Depends(get_db),
current_user: AppUser = Depends(get_current_active_user_with_set_db),
_export_slot: None = Depends(hold_user_export_slot),
):
export_service = ExportService(db)
try:
@ -123,6 +160,47 @@ async def export_forms_bulk(
stream,
media_type=media_type,
headers={"Content-Disposition": export_service.build_content_disposition(filename)},
background=BackgroundTask(stream.close),
)
@router.get("/summary/{summary_number}/{sheet}/{year}")
async def export_summary(
summary_number: Annotated[int, Path(ge=1, le=2)],
sheet: str,
year: int,
org_ids: str | None = None,
sort_by: Literal["org_name", "section_code"] = "org_name",
sort_direction: Literal["asc", "desc"] = "asc",
db: AsyncSession = Depends(get_db),
current_user: AppUser = Depends(get_current_active_user_with_set_db),
_export_slot: None = Depends(hold_user_export_slot),
):
sheet = sheet.upper()
allowed_sheets = (
SUMMARY1_ALLOWED_SHEETS
if summary_number == 1
else SUMMARY2_ALLOWED_SHEETS
)
if sheet not in allowed_sheets:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Неизвестный лист свода {summary_number}: {sheet}",
)
stream, filename, media_type = await ExportService(db).export_summary_payload(
summary_number=summary_number,
year=year,
sheet=sheet,
current_user=current_user,
org_ids=_parse_org_ids_csv(org_ids),
sort_by=sort_by,
sort_direction=sort_direction,
)
return StreamingResponse(
stream,
media_type=media_type,
headers={"Content-Disposition": ExportService.build_content_disposition(filename)},
background=BackgroundTask(stream.close),
)
@ -131,6 +209,7 @@ async def export_project(
project_id: int,
db: AsyncSession = Depends(get_db),
current_user: AppUser = Depends(get_current_active_user_with_set_db),
_export_slot: None = Depends(hold_user_export_slot),
):
export_service = ExportService(db)
try:
@ -144,6 +223,7 @@ async def export_project(
stream,
media_type=media_type,
headers={"Content-Disposition": export_service.build_content_disposition(filename)},
background=BackgroundTask(stream.close),
)
@ -155,6 +235,7 @@ async def export_project_report(
sections: str | None = None,
db: AsyncSession = Depends(get_db),
current_user: AppUser = Depends(get_current_active_user_with_set_db),
_export_slot: None = Depends(hold_user_export_slot),
):
export_service = ExportService(db)
try:
@ -171,6 +252,5 @@ async def export_project_report(
stream,
media_type=media_type,
headers={"Content-Disposition": export_service.build_content_disposition(filename)},
background=BackgroundTask(stream.close),
)

14
api/src/db/session.py
View File

@ -1,16 +1,17 @@
import asyncio
from typing import AsyncGenerator
from contextlib import asynccontextmanager
from typing import AsyncGenerator, AsyncIterator
from alembic import command
from alembic.config import Config
from sqlalchemy.ext.asyncio import AsyncSession
from src.core.config import settings
from typing import AsyncGenerator
from src.db.base import SessionLocal
async def get_db() -> AsyncGenerator:
@asynccontextmanager
async def db_session() -> AsyncGenerator[AsyncSession, None]:
db = SessionLocal()
try:
yield db
@ -22,6 +23,11 @@ async def get_db() -> AsyncGenerator:
await db.close()
async def get_db() -> AsyncGenerator[AsyncSession, None]:
async with db_session() as db:
yield db
async def run_migrations() -> None:
alembic_cfg = Config("alembic.ini")
await asyncio.to_thread(command.upgrade, alembic_cfg, "head")

120
api/src/repository/summary_repository.py
View File

@ -1,3 +1,5 @@
from typing import Any
from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncSession
@ -58,7 +60,6 @@ class SummaryRepository:
CAST(:sort_by AS VARCHAR),
CAST(:sort_direction AS VARCHAR)
)
ORDER BY sort_order
"""
)
return (
@ -74,3 +75,120 @@ class SummaryRepository:
},
)
).all()
async def get_svod2_go_export_rows(
self,
year: int,
org_ids: list[int] | None,
sort_by: str,
sort_direction: str,
) -> list[tuple]:
"""Load the compact flat representation produced for XLSX export."""
common_quarter_fields = (
"adj_current", "adj_ssp", "adj_rf", "adj_reserve",
"payment_date", "payment_amount", "payment_amount_ho",
"payment_amount_rf", "payment_act", "booking_amount",
"actual_m1", "actual_m2", "actual_m3",
)
export_keys = [
"header.smeta_type",
"header.smeta_direction",
"header.org_name",
"header.name",
"header.item_id",
"header.num_group",
"header.justification",
*(f"plan.plan_q{quarter}" for quarter in range(1, 5)),
"allocation.internal_order",
"allocation.property_object",
*(f"sequestration.DFIP.adj_q{quarter}" for quarter in range(1, 5)),
*(f"reserve.amount_q{quarter}" for quarter in range(1, 5)),
*(f"approved.approved_q{quarter}" for quarter in range(1, 5)),
*(
f"contract_detail.{field}"
for field in (
"counterparty", "reference", "subject", "currency",
"ceiling_amount", "expenses_q1", "expenses_q2",
"expenses_q3", "expenses_q4", "vat_rate", "deadline",
)
),
]
for quarter in range(1, 5):
quarter_fields = list(common_quarter_fields)
if quarter == 4:
quarter_fields.append("actual_spod")
if quarter > 1:
quarter_fields.extend(
(
"plan_revision_eco_change",
"plan_revision_item_adj",
"plan_revision_increase",
"plan_revision_sequester",
)
)
quarter_fields.extend(
f"transfer_to_q{target_quarter}"
for target_quarter in range(quarter + 1, 5)
)
if quarter < 4:
quarter_fields.append("transfer_to_economy")
export_keys.extend(f"q{quarter}.{field}" for field in quarter_fields)
query = text(
"""
SELECT sort_order, row_values
FROM v3.v_svod2_go_export(
CAST(:year AS INT),
CAST(:org_ids AS INT[]),
CAST(:sort_by AS VARCHAR),
CAST(:sort_direction AS VARCHAR)
)
"""
)
result = await self.db.execute(
query,
{
"year": year,
"org_ids": org_ids,
"sort_by": sort_by,
"sort_direction": sort_direction.lower(),
},
)
rows = []
for row in result.mappings():
data = {
key: self._parse_svod2_export_value(key, value)
for key, value in zip(export_keys, row["row_values"])
}
rows.append(("INPUT", 3, row["sort_order"], data))
return rows
@staticmethod
def _parse_svod2_export_value(key: str, value: str | None) -> Any:
if value is None:
return None
numeric_prefixes = (
"plan.",
"sequestration.",
"reserve.",
"approved.",
)
numeric_contract_fields = {
"contract_detail.ceiling_amount",
"contract_detail.expenses_q1",
"contract_detail.expenses_q2",
"contract_detail.expenses_q3",
"contract_detail.expenses_q4",
}
quarter_text_suffixes = (".payment_date", ".payment_act")
is_numeric_quarter = (
key.startswith(("q1.", "q2.", "q3.", "q4."))
and not key.endswith(quarter_text_suffixes)
)
if (
key.startswith(numeric_prefixes)
or key in numeric_contract_fields
or is_numeric_quarter
):
return float(value)
return value

24
api/src/services/export_lock.py Normal file
View File

@ -0,0 +1,24 @@
import asyncio
from contextlib import asynccontextmanager
from typing import AsyncGenerator, AsyncIterator
class ExportAlreadyRunningError(Exception):
pass
_local_active_users: set[int] = set()
_local_guard = asyncio.Lock()
@asynccontextmanager
async def user_export_slot(user_id: int) -> AsyncGenerator[None, None]:
async with _local_guard:
if user_id in _local_active_users:
raise ExportAlreadyRunningError
_local_active_users.add(user_id)
try:
yield
finally:
async with _local_guard:
_local_active_users.discard(user_id)

170
api/src/services/export_service/export_mappings.py
View File

@ -3527,3 +3527,173 @@ FORM3_COLORS: dict = {
"q4.ekonomiya": {"background": "C2D69B", "palette": "green"},
"q4.ekonomiya.economy": {"background": "C2D69B", "palette": "green"},
}
# Маппинги свода
def build_summary1_columns(sheet: str) -> dict:
metric_labels = {
"plan": "План",
"approved": "Утверждено",
"fact": "Факт",
"corrected": "Скорректированный план",
}
direction_labels = {"support": "Поддержка", "development": "Развитие"}
period_labels = {
"q1": "I кв.", "q2": "II кв.", "q3": "III кв.",
"q4": "IV кв.", "year": "Год",
}
directions = ["support", "development"]
if sheet == "FORM_2":
directions = ["support"]
elif sheet in {"FORM_3", "FORM_4"}:
directions = ["development"]
columns: dict = {
"section_code": {"name": "Код"},
"name": {"name": "Статья расходов"},
}
for metric, metric_label in metric_labels.items():
periods = ["q2", "q3", "q4"] if metric == "corrected" else list(period_labels)
children = {
direction: {
"name": direction_labels[direction],
"children": {
period: {"name": period_labels[period]} for period in periods
},
}
for direction in directions
}
if metric != "corrected" and sheet in {"FORM_1", "MAIN"}:
children["total_year"] = {"name": "Итого за год"}
columns[metric] = {"name": metric_label, "children": children}
return columns
def _quarter_columns() -> dict:
months = {
1: ("Январь", "Февраль", "Март"),
2: ("Апрель", "Май", "Июнь"),
3: ("Июль", "Август", "Сентябрь"),
4: ("Октябрь", "Ноябрь", "Декабрь"),
}
result: dict = {}
for quarter in range(1, 5):
children = {
"adj_current": {"name": "Текущие корректировки (= 0)"},
"adj_ssp": {"name": "Корректировки с ССП / сметой развития"},
"adj_rf": {"name": "Корректировки с РФ"},
"adj_reserve": {"name": "Корректировки из резерва"},
"payment_date": {"name": "Дата платежа"},
"payment_amount": {"name": "Сумма платежа, тыс. руб. (без НДС)"},
"payment_amount_ho": {"name": "в т.ч. сумма ГО (без НДС)"},
"payment_amount_rf": {"name": "в т.ч. сумма РФ (без НДС)"},
"payment_act": {"name": "Предоставление акта"},
"booking_amount": {"name": f"Бронь {quarter} кв."},
}
for month_number, month in enumerate(months[quarter], start=1):
children[f"actual_m{month_number}"] = {"name": f"Факт {month}"}
if quarter == 4:
children["actual_spod"] = {"name": "Факт СПОД"}
if quarter > 1:
children.update(
{
"plan_revision_eco_change": {
"name": "Изменение целевого назначения перенесенной экономии (= 0)"
},
"plan_revision_item_adj": {
"name": "Корректировка статей базового плана (= 0)"
},
"plan_revision_increase": {
"name": "Увеличение базового плана (> 0)"
},
"plan_revision_sequester": {
"name": "Секвестр базового плана (< 0)"
},
}
)
for target_quarter in range(quarter + 1, 5):
children[f"transfer_to_q{target_quarter}"] = {
"name": f"Перенос в {target_quarter} кв."
}
if quarter < 4:
children["transfer_to_economy"] = {"name": "Перенос в фонд экономии"}
result[f"q{quarter}"] = {
"name": f"{quarter} квартал",
"children": children,
}
return result
def build_summary2_columns(sheet: str) -> dict:
estimate_title = (
"Смета расходов ГО в разрезе договоров"
if sheet == "GO"
else "Смета расходов РФ в разрезе договоров"
)
columns: dict = {
"smeta_type": {"key": "header.smeta_type", "name": "Вид сметы"},
"smeta_direction": {"key": "header.smeta_direction", "name": "Направление"},
"org_name": {"key": "header.org_name", "name": "ССП"},
"name": {"key": "header.name", "name": "Наименование статьи"},
"estimate": {
"key": None,
"name": estimate_title,
"children": {
"item_id": {"key": "header.item_id", "name": "ID статьи"},
"num_group": {"key": "header.num_group", "name": "ID группы номенклатуры"},
"justification": {"key": "header.justification", "name": "Конкретный вид расхода"},
**{
f"plan_q{q}": {"key": f"plan.plan_q{q}", "name": f"{q} кв."}
for q in range(1, 5)
},
},
},
"allocation": {
"name": "Аллокация расходов",
"children": {
"internal_order": {"name": "Внутренний заказ"},
"property_object": {"name": "Объект недвижимости"},
},
},
"sequestration": {
"name": "Секвестирование ДФиП",
"children": {
"DFIP": {
"name": "ДФиП",
"children": {
f"adj_q{q}": {"name": f"{q} кв."} for q in range(1, 5)
},
}
},
},
"reserve": {
"name": "Отнесение в резерв",
"children": {
f"amount_q{q}": {"name": f"{q} кв."} for q in range(1, 5)
},
},
"approved": {
"name": "Итоговая смета расходов",
"children": {
f"approved_q{q}": {"name": f"{q} кв."} for q in range(1, 5)
},
},
"contract_detail": {
"name": "Договор",
"children": {
"counterparty": {"name": "Контрагент"},
"reference": {"name": "Договор (реквизиты)"},
"subject": {"name": "Предмет"},
"currency": {"name": "Валюта договора"},
"ceiling_amount": {"name": "Предельная стоимость, тыс. руб. (без НДС)"},
**{
f"expenses_q{q}": {"name": f"Расходы {q} кв., тыс. руб. (без НДС)"}
for q in range(1, 5)
},
"vat_rate": {"name": "Ставка НДС (%)"},
"deadline": {"name": "Срок"},
},
},
}
columns.update(_quarter_columns())
return columns

3
api/src/services/export_service/export_normalizers.py
View File

@ -1,6 +1,7 @@
from __future__ import annotations
import json
from decimal import Decimal
from typing import Any
SMETA_VALUE_PATHS: list[tuple[str, tuple[str, ...]]] = [
@ -72,6 +73,8 @@ FORM1_METRIC_BLOCK_KEYS = (
class ExcelValueSerializer:
def to_excel_value(self, value: Any) -> Any:
if isinstance(value, Decimal):
return float(value)
if value is None or isinstance(value, (str, int, float, bool)):
return value
if isinstance(value, dict):

537
api/src/services/export_service/export_service.py
View File

@ -1,19 +1,28 @@
from __future__ import annotations
import asyncio
import io
import json
import logging
import os
import shutil
import tempfile
import zipfile
from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Any
from functools import partial
from time import perf_counter
from typing import Any, BinaryIO
from urllib.parse import quote
from openpyxl import Workbook
from openpyxl.worksheet.worksheet import Worksheet
from sqlalchemy.ext.asyncio import AsyncSession
from xlsxwriter import Workbook as XlsxWorkbook
from src.db.models.app_user import AppUser
from src.db.models.form_type import FormTypeEnum
from src.db.session import db_session
from src.services.budget_form_service import BudgetFormService
from src.services.export_service.export_normalizers import (
ExcelValueSerializer,
@ -28,17 +37,24 @@ from src.services.export_service.export_mappings import (
FORM3_COLORS,
FORM3_CURRENT_EXPENSES_DEPTH_COLUMNS,
FORM3_LIMIT_DEPTH_COLUMNS,
PALETTE,
SMETA_COLUMNS,
build_summary1_columns,
build_summary2_columns,
)
from src.services.export_service.export_writers import (
DepthSheetWriter,
TabularSheetWriter,
XlsxDepthSheetWriter,
XlsxTabularSheetWriter,
)
from src.services.project_service import ProjectService
from src.services.sheet_service import SheetService
from src.services.summary_service import SummaryService
XLSX_MEDIA_TYPE = "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"
ZIP_MEDIA_TYPE = "application/zip"
SUMMARY_EXPORT_COLUMN_WIDTH = 15
FORM1_DIRECTION_REQUIRED_SHEETS = {"AHR", "CAP"}
SHEETS_WITH_SECTIONS = {"AHR", "CAP", "OPER"}
@ -56,6 +72,11 @@ SHEET_NAMES = {
'CURRENT_EXPENSES': 'Текущие расходы',
}
SKIP_SHEETS = ('OTCH9F',)
BULK_EXPORT_CONCURRENCY = 2
EXPORT_TMPDIR = os.getenv("EXPORT_TMPDIR") or tempfile.gettempdir()
logger = logging.getLogger()
@dataclass(frozen=True)
class ExportSheetPlan:
@ -71,6 +92,7 @@ class ExportService:
self.budget_form_service = BudgetFormService(db)
self.sheet_service = SheetService(db)
self.project_service = ProjectService(db)
self.summary_service = SummaryService(db)
self.value_serializer = ExcelValueSerializer()
self.form1_row_normalizer = Form1DepthRowNormalizer(self.value_serializer)
self.form3_row_normalizer = Form3DepthRowNormalizer(self.value_serializer)
@ -134,15 +156,15 @@ class ExportService:
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]:
ignore_unsupported_sheet_params: bool = False,
) -> tuple[BinaryIO, 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,
ignore_unsupported_sheet_params=ignore_unsupported_sheet_params,
)
return stream, filename, XLSX_MEDIA_TYPE
@ -153,7 +175,7 @@ class ExportService:
report_type: str,
current_user: AppUser,
sections: list[str] | None = None,
) -> tuple[io.BytesIO, str, str]:
) -> tuple[BinaryIO, str, str]:
stream, filename = await self.export_project_report_to_xlsx(
project_id=project_id,
year=year,
@ -167,13 +189,74 @@ class ExportService:
self,
project_id: int,
current_user: AppUser,
) -> tuple[io.BytesIO, str, str]:
) -> tuple[BinaryIO, str, str]:
stream, filename = await self.export_project_to_xlsx(
project_id=project_id,
current_user=current_user,
)
return stream, filename, XLSX_MEDIA_TYPE
async def export_summary_payload(
self,
summary_number: int,
year: int,
sheet: str,
current_user: AppUser,
org_ids: list[int] | None = None,
sort_by: str = "org_name",
sort_direction: str = "asc",
) -> tuple[BinaryIO, str, str]:
export_started_at = perf_counter()
load_started_at = perf_counter()
if summary_number == 1:
rows = await self.summary_service.get_svod1_rows(
year=year,
sheet=sheet,
user=current_user,
org_ids=org_ids,
)
columns = build_summary1_columns(sheet)
elif sheet == "GO":
rows = await self.summary_service.get_svod2_go_export_rows(
year=year,
user=current_user,
org_ids=org_ids,
sort_by=sort_by,
sort_direction=sort_direction,
)
columns = build_summary2_columns(sheet)
else:
rows = await self.summary_service.get_svod2_rows(
year=year,
sheet=sheet,
user=current_user,
org_ids=org_ids,
offset=None,
limit=None,
sort_by=sort_by,
sort_direction=sort_direction,
)
columns = build_summary2_columns(sheet)
logger.info(
f"[export timing] summary={summary_number} sheet={sheet} year={year}: "
f"load_rows={perf_counter() - load_started_at:.3f}s, rows={len(rows)}",
)
stream = await self._export_summary_rows(
rows=rows,
columns=columns,
metadata=(("Свод", summary_number), ("Лист", sheet), ("Год", year)),
worksheet_name=f"Свод {summary_number} {sheet}",
summary_number=summary_number,
year=year,
sheet=sheet,
)
logger.info(
f"[export timing] summary={summary_number} sheet={sheet} year={year}: "
f"total={perf_counter() - export_started_at:.3f}s",
)
filename = self._build_summary_filename(summary_number, sheet, year)
return stream, filename, XLSX_MEDIA_TYPE
async def export_bulk_payload(
self,
form_ids: list[int],
@ -181,9 +264,12 @@ class ExportService:
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()
ignore_unsupported_sheet_params: bool = False,
) -> tuple[BinaryIO, str, str]:
# Общий таймер bulk-экспорта: включает создание XLSX и упаковку в ZIP.
bulk_started_at = perf_counter()
# ZIP пишется во временный файл, чтобы размер bulk-ответа не увеличивал RAM процесса.
archive_stream = tempfile.TemporaryFile(mode="w+b", dir=EXPORT_TMPDIR)
exported_count = 0
errors: list[dict[str, Any]] = []
used_names: set[str] = set()
@ -193,22 +279,49 @@ class ExportService:
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,
)
unique_form_ids = self._deduplicate_ids(form_ids)
for batch_start in range(0, len(unique_form_ids), BULK_EXPORT_CONCURRENCY):
batch_ids = unique_form_ids[
batch_start:batch_start + BULK_EXPORT_CONCURRENCY
]
batch_results = await asyncio.gather(
*[
self._export_bulk_form(
form_id=form_id,
current_user=current_user,
direction=direction,
sections=sections,
ignore_unsupported_sheet_params=ignore_unsupported_sheet_params,
)
for form_id in batch_ids
],
return_exceptions=True,
)
for form_id, result in zip(batch_ids, batch_results):
if isinstance(result, BaseException):
if not isinstance(result, ValueError) or not skip_failed:
archive_stream.close()
raise result
errors.append({"form_id": form_id, "error": str(result)})
continue
form_stream, filename, xlsx_elapsed = result
xlsx_finished_at = perf_counter()
archive_name = self._make_unique_filename(filename, used_names)
archive.writestr(archive_name, form_stream.getvalue())
try:
form_stream.seek(0)
# Копируем XLSX порциями, не создавая его полную bytes-копию в памяти.
with archive.open(archive_name, mode="w") as archive_entry:
shutil.copyfileobj(form_stream, archive_entry, length=1024 * 1024)
finally:
form_stream.close()
logger.info(
f"[export timing] bulk form_id={form_id}: "
f"xlsx={xlsx_elapsed:.3f}s, "
f"zip_write={perf_counter() - xlsx_finished_at:.3f}s",
)
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(
@ -217,12 +330,38 @@ class ExportService:
)
if exported_count == 0:
archive_stream.close()
raise ValueError("Не удалось экспортировать ни одной формы.")
archive_stream.seek(0)
filename = f"export_forms_{datetime.now(timezone.utc).strftime('%Y-%m-%d')}.zip"
logger.info(
f"[export timing] bulk total: forms={exported_count}, "
f"elapsed={perf_counter() - bulk_started_at:.3f}s",
)
return archive_stream, filename, ZIP_MEDIA_TYPE
async def _export_bulk_form(
self,
form_id: int,
current_user: AppUser,
direction: str | None,
sections: list[str] | None,
ignore_unsupported_sheet_params: bool,
) -> tuple[BinaryIO, str, float]:
started_at = perf_counter()
# Конкурентные задачи не должны разделять один AsyncSession.
async with db_session() as db:
export_service = ExportService(db)
stream, filename = await export_service.export_form_to_xlsx(
form_id=form_id,
current_user=current_user,
direction=direction,
sections=sections,
ignore_unsupported_sheet_params=ignore_unsupported_sheet_params,
)
return stream, filename, perf_counter() - started_at
async def export_form_to_xlsx(
self,
form_id: int,
@ -230,9 +369,20 @@ class ExportService:
sheet: str | None = None,
direction: str | None = None,
sections: list[str] | None = None,
ignore_non_applicable_query_params: bool = False,
) -> tuple[io.BytesIO, str]:
ignore_unsupported_sheet_params: bool = False,
) -> tuple[BinaryIO, str]:
# perf_counter используется для измерения длительности и не зависит от изменения системного времени.
export_started_at = perf_counter()
# Этап 1: получение формы, её типа и подразделения из БД.
stage_started_at = perf_counter()
form = await self._get_form_for_export(form_id=form_id, current_user=current_user)
logger.info(
f"[export timing] form_id={form_id}: load_form={perf_counter() - stage_started_at:.3f}s",
)
# Этап 2: проверка параметров и построение списка экспортируемых листов.
stage_started_at = perf_counter()
sheets = self._resolve_requested_sheets(form=form, requested_sheet=sheet)
self._validate_sections_for_form(form=form, sections=sections)
@ -242,7 +392,7 @@ class ExportService:
sheets=sheets,
direction=direction,
sections=sections,
ignore_non_applicable_query_params=ignore_non_applicable_query_params,
ignore_unsupported_sheet_params=ignore_unsupported_sheet_params,
)
else:
sheets_by_directions = self._resolve_sheets_by_directions(form=form, requested_sheet=sheet)
@ -253,12 +403,24 @@ class ExportService:
sheets=shts,
direction=dr,
sections=sections,
ignore_non_applicable_query_params=ignore_non_applicable_query_params,
ignore_unsupported_sheet_params=ignore_unsupported_sheet_params,
)
logger.info(
f"[export timing] form_id={form_id}: build_plan={perf_counter() - stage_started_at:.3f}s, "
f"sheets={len(plans)}",
)
workbook = self._create_workbook_with_metadata(form=form)
# XlsxWriter хранит XML-части и итоговую книгу во временных файлах вместо RAM.
stream = tempfile.TemporaryFile(mode="w+b", dir=EXPORT_TMPDIR)
workbook = XlsxWorkbook(
stream,
{
"tmpdir": EXPORT_TMPDIR,
"strings_to_urls": False,
},
)
self._write_xlsxwriter_metadata(workbook.add_worksheet("metadata"), form)
sheet_writers = self._create_xlsxwriter_sheet_writers(workbook)
exported_sheets = 0
for plan in plans:
if not plan.validation_direction and plan.sheet_name in SKIP_SHEETS:
@ -270,27 +432,44 @@ class ExportService:
direction=plan.validation_direction,
sections=plan.validation_sections,
)
# Этап 3: отдельно измеряем запрос данных каждого листа к БД.
sheet_load_started_at = perf_counter()
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, plan.effective_direction))
self._write_sheet_rows(
worksheet=worksheet,
rows=rows,
sheet_name=plan.sheet_name,
sheet_loaded_at = perf_counter()
# Этап 4: рендер листа уводим с event loop, чтобы экспорт не блокировал API.
worksheet = workbook.add_worksheet(
self._safe_sheet_name(plan.sheet_name, plan.effective_direction)
)
writer = sheet_writers.get(plan.sheet_name, sheet_writers["__default__"])
await asyncio.to_thread(writer.write, worksheet, rows)
logger.info(
f"[export timing] form_id={form_id} sheet={plan.sheet_name} "
f"direction={plan.effective_direction or '-'} rows={len(rows)}: "
f"load_rows={sheet_loaded_at - sheet_load_started_at:.3f}s, "
f"write_sheet={perf_counter() - sheet_loaded_at:.3f}s",
)
exported_sheets += 1
if exported_sheets == 0:
raise ValueError("Не удалось экспортировать листы формы.")
stream = io.BytesIO()
workbook.save(stream)
workbook.close()
# Этап 5: сериализация всей книги в XLSX обычно заметна на больших формах.
save_started_at = perf_counter()
# Формирование XML и ZIP-контейнера XLSX также не должно блокировать event loop.
await asyncio.to_thread(workbook.close)
xlsx_size = stream.seek(0, io.SEEK_END)
stream.seek(0)
logger.info(
f"[export timing] form_id={form_id}: save_xlsx={perf_counter() - save_started_at:.3f}s, "
f"total={perf_counter() - export_started_at:.3f}s, size={xlsx_size} bytes",
)
filename = self._build_filename(form_id=form.id, form_type_code=form.form_type.code, year=form.year, sheet=sheet)
return stream, filename
@ -482,20 +661,291 @@ class ExportService:
self._fill_metadata(meta_sheet, form)
return workbook
def _write_xlsxwriter_metadata(self, sheet: Any, form: Any) -> None:
metadata_rows = [
("Форма ID", form.id),
("Тип формы", form.form_type_code),
("Год", form.year),
("ID ССП.РФ", form.org_unit_id),
("ССП/РФ", form.org_unit.title if form.org_unit else None),
("Экспортировано", datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S")),
]
for row, (key, value) in enumerate(metadata_rows):
sheet.write(row, 0, key)
sheet.write(row, 1, value)
async def _export_summary_rows(
self,
rows: list[tuple],
columns: dict,
metadata: tuple[tuple[str, Any], ...],
worksheet_name: str,
summary_number: int,
year: int,
sheet: str,
) -> BinaryIO:
workbook_started_at = perf_counter()
stream = tempfile.TemporaryFile(mode="w+b", dir=EXPORT_TMPDIR)
workbook = XlsxWorkbook(
stream,
{"tmpdir": EXPORT_TMPDIR, "strings_to_urls": False},
)
try:
metadata_sheet = workbook.add_worksheet("metadata")
metadata_rows = (
*metadata,
("Строк", len(rows)),
(
"Экспортировано",
datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S"),
),
)
for row_number, (key, value) in enumerate(metadata_rows):
metadata_sheet.write(row_number, 0, key)
metadata_sheet.write(row_number, 1, value)
worksheet = workbook.add_worksheet(self._safe_sheet_name(worksheet_name))
(
header_format_resolver,
row_format_resolver,
column_format_resolver,
) = (
self._summary_format_resolvers(columns, summary_number)
)
writer = XlsxDepthSheetWriter(
workbook=workbook,
columns_dict=columns,
row_normalizer=(
self._identity_row
if summary_number == 2 and sheet == "GO"
else self.form1_row_normalizer.normalize
),
value_serializer=self.value_serializer.to_excel_value,
header_format_resolver=header_format_resolver,
row_format_resolver=row_format_resolver,
column_format_resolver=column_format_resolver,
fixed_column_width=SUMMARY_EXPORT_COLUMN_WIDTH,
)
logger.info(
f"[export timing] summary={summary_number} sheet={sheet} year={year}: "
f"build_workbook={perf_counter() - workbook_started_at:.3f}s",
)
write_started_at = perf_counter()
await asyncio.to_thread(writer.write, worksheet, rows)
logger.info(
f"[export timing] summary={summary_number} sheet={sheet} year={year}: "
f"write_sheet={perf_counter() - write_started_at:.3f}s",
)
save_started_at = perf_counter()
await asyncio.to_thread(workbook.close)
logger.info(
f"[export timing] summary={summary_number} sheet={sheet} year={year}: "
f"save_xlsx={perf_counter() - save_started_at:.3f}s",
)
except BaseException:
stream.close()
raise
stream.seek(0)
return stream
@staticmethod
def _identity_row(row: dict[str, Any]) -> dict[str, Any]:
return row
@staticmethod
def _summary_format_resolvers(
columns: dict,
summary_number: int,
) -> tuple[Any, Any, Any]:
if summary_number == 1:
return (
ExportService._summary1_header_format,
ExportService._summary1_row_format,
None,
)
roots = list(columns)
band_by_root = {root: index % 2 for index, root in enumerate(roots)}
leaf_bands: list[int] = []
group_end_columns: set[int] = set()
group_end_header_keys: set[str] = set()
for root in roots:
ExportService._add_summary_leaf_bands(
config={root: columns[root]},
band=band_by_root[root],
leaf_bands=leaf_bands,
)
group_end_columns.add(len(leaf_bands) - 1)
group_end_header_keys.add(
ExportService._summary_last_leaf_key(root, columns[root])
)
return (
partial(
ExportService._summary2_header_format,
band_by_root=band_by_root,
roots=set(roots),
group_end_header_keys=group_end_header_keys,
),
None,
partial(
ExportService._summary2_column_format,
leaf_bands=leaf_bands,
group_end_columns=group_end_columns,
),
)
@staticmethod
def _summary1_header_format(_full_key: str) -> dict[str, Any]:
return {
"bg_color": "#F8F9FA",
"pattern": 1,
"font_color": "#4A5565",
"bold": True,
}
@staticmethod
def _summary1_row_format(
row: tuple,
_column_key: str,
_column: int,
) -> dict[str, Any]:
thin_border = {"right": 1, "right_color": "#EDF0F2"}
data = row[3] if len(row) > 3 and isinstance(row[3], dict) else {}
section_code = data.get("section_code")
if section_code in (None, "") and isinstance(data.get("header"), dict):
section_code = data["header"].get("section_code")
if section_code in (None, ""):
return thin_border
palette_name = {"1": "green", "3": "orange", "4": "blue"}.get(
str(section_code)[:1],
"blue",
)
item_type = row[0] if row else "INPUT"
background = PALETTE[palette_name].get(item_type, "FFFFFF")
red, green, blue = (
int(background[index:index + 2], 16) for index in (0, 2, 4)
)
is_dark = 0.299 * red + 0.587 * green + 0.114 * blue < 128
return {
"bg_color": f"#{background}",
"pattern": 1,
"font_color": "#FFFFFF" if is_dark else "#000000",
**thin_border,
}
@staticmethod
def _add_summary_leaf_bands(
config: dict,
band: int,
leaf_bands: list[int],
) -> None:
for child in config.values():
if "children" in child:
ExportService._add_summary_leaf_bands(
child["children"],
band,
leaf_bands,
)
else:
leaf_bands.append(band)
@staticmethod
def _summary_last_leaf_key(key: str, config: dict) -> str:
if "children" not in config:
return key
child_key, child_config = next(reversed(config["children"].items()))
return ExportService._summary_last_leaf_key(
f"{key}.{child_key}",
child_config,
)
@staticmethod
def _summary2_header_format(
full_key: str,
*,
band_by_root: dict[str, int],
roots: set[str],
group_end_header_keys: set[str],
) -> dict[str, Any]:
band = band_by_root.get(full_key.split(".", 1)[0], 0)
result = {
"bg_color": ("#F8F9FA", "#F1F7F3")[band],
"pattern": 1,
"font_color": "#4A5565",
"bold": True,
}
if full_key in roots or full_key in group_end_header_keys:
result.update({"right": 2, "right_color": "#9EAFA3"})
return result
@staticmethod
def _summary2_column_format(
_column_key: str,
column: int,
*,
leaf_bands: list[int],
group_end_columns: set[int],
) -> dict[str, Any]:
is_group_end = column in group_end_columns
return {
"bg_color": ("#FFFFFF", "#F8FBF9")[leaf_bands[column]],
"pattern": 1,
"right": 2 if is_group_end else 1,
"right_color": "#9EAFA3" if is_group_end else "#EDF0F2",
}
def _create_xlsxwriter_sheet_writers(self, workbook: XlsxWorkbook) -> dict[str, Any]:
default_writer = XlsxDepthSheetWriter(
workbook=workbook,
columns_dict=FORM1_DEPTH_COLUMNS,
row_normalizer=self.form1_row_normalizer.normalize,
value_serializer=self.value_serializer.to_excel_value,
)
writers: dict[str, Any] = {
"__default__": default_writer,
"SMETA": XlsxTabularSheetWriter(
workbook=workbook,
columns=SMETA_COLUMNS,
row_normalizer=self.smeta_row_normalizer.normalize,
value_serializer=self.value_serializer.to_excel_value,
),
}
writer_configs = {
"OPER": (FORM1_DEPTH_COLUMNS, FORM1_COLORS["OPER"]),
"AHR": (FORM1_DEPTH_COLUMNS, FORM1_COLORS["AHR"]),
"CAP": (FORM1_DEPTH_COLUMNS, FORM1_COLORS["CAP"]),
"AHR_RENT": (FORM2_DEPTH_AHR_SUB_COLUMNS, None),
"AHR_UTILITY": (FORM2_DEPTH_AHR_SUB_COLUMNS, None),
"AHR_SECURITY": (FORM2_DEPTH_AHR_SUB_COLUMNS, None),
"STRUCTURE": (FORM1_DEPTH_COLUMNS, FORM1_COLORS["AHR"]),
}
for sheet_name, (columns, colors) in writer_configs.items():
writers[sheet_name] = XlsxDepthSheetWriter(
workbook=workbook,
columns_dict=columns,
row_normalizer=self.form1_row_normalizer.normalize,
value_serializer=self.value_serializer.to_excel_value,
colors_config=colors,
)
return writers
def _build_sheet_plans(
self,
form: Any,
sheets: list[str],
direction: str | None,
sections: list[str] | None,
ignore_non_applicable_query_params: bool,
ignore_unsupported_sheet_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
validation_direction = effective_direction if ignore_unsupported_sheet_params else direction
validation_sections = effective_sections if ignore_unsupported_sheet_params else sections
plans.append(
ExportSheetPlan(
sheet_name=sheet_name,
@ -549,6 +999,11 @@ class ExportService:
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 _build_summary_filename(summary: int, sheet: str, year: int) -> str:
date_str = datetime.now(timezone.utc).strftime("%Y-%m-%d")
return f"{year}_SVOD_{summary}_{sheet}_{date_str}.xlsx"
@staticmethod
def _deduplicate_ids(form_ids: list[int]) -> list[int]:
seen: set[int] = set()

379
api/src/services/export_service/export_writers.py
View File

@ -7,6 +7,8 @@ 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 xlsxwriter import Workbook as XlsxWorkbook
from xlsxwriter.worksheet import Worksheet as XlsxWorksheet
from src.services.export_service.export_mappings import PALETTE
@ -14,23 +16,28 @@ from src.services.export_service.export_mappings import PALETTE
RowNormalizer = Callable[[dict[str, Any]], dict[str, Any]]
ValueSerializer = Callable[[Any], Any]
HeaderFormatResolver = Callable[[str], dict[str, Any] | None]
RowFormatResolver = Callable[[tuple, str, int], dict[str, Any] | None]
ColumnFormatResolver = Callable[[str, int], dict[str, Any] | None]
class AbstractWriter(abc.ABC):
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')
@staticmethod
def _track_width(column_widths: dict[int, int], column: int, value: Any) -> None:
if value is None:
return
column_widths[column] = max(column_widths.get(column, 0), len(str(value)))
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
def _track_row_widths(self, column_widths: dict[int, int], values: list[Any]) -> None:
for column, value in enumerate(values, start=1):
self._track_width(column_widths, column, value)
adjusted_width = max(max_len + 3, 10)
worksheet.column_dimensions[col_letter].width = adjusted_width
@staticmethod
def _apply_widths(worksheet: Worksheet, column_widths: dict[int, int]) -> None:
for column, max_len in column_widths.items():
column_letter = get_column_letter(column)
worksheet.column_dimensions[column_letter].width = max(max_len + 3, 10)
@abc.abstractmethod
def write(self, *args, **kwargs):
@ -49,14 +56,17 @@ class TabularSheetWriter(AbstractWriter):
self.value_serializer = value_serializer
def write(self, worksheet: Worksheet, rows: list[tuple]) -> None:
worksheet.append([title for _, title in self.columns])
column_widths: dict[int, int] = {}
headers = [title for _, title in self.columns]
worksheet.append(headers)
self._track_row_widths(column_widths, headers)
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]
)
self._set_widths(worksheet)
values = [self.value_serializer(normalized.get(key)) for key, _ in self.columns]
worksheet.append(values)
self._track_row_widths(column_widths, values)
self._apply_widths(worksheet, column_widths)
class DepthSheetWriter(AbstractWriter):
@ -82,6 +92,7 @@ class DepthSheetWriter(AbstractWriter):
self.value_serializer = value_serializer
self.colors_config = colors_config
self.palette = None
self._row_fill_cache: dict[tuple[str, str], PatternFill] = {}
if colors_config:
self._set_palette()
@ -138,17 +149,21 @@ class DepthSheetWriter(AbstractWriter):
current_column: int = 1,
current_row: int = 1,
max_row: int | None = None,
full_key: str | None = None
full_key: str | None = None,
column_widths: dict[int, int] | None = None,
) -> int:
if columns_dict is None:
columns_dict = self.columns_dict
max_row = self._get_header_max_row(self.columns_dict)
if column_widths is None:
column_widths = {}
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"])
self._track_width(column_widths, current_column, sub_dict["name"])
cell.border = self.thin_border
if max_row != current_row:
worksheet.merge_cells(
@ -166,8 +181,10 @@ class DepthSheetWriter(AbstractWriter):
current_row=current_row+1,
max_row=max_row,
full_key=key,
column_widths=column_widths,
)
cell = worksheet.cell(row=current_row, column=current_column, value=sub_dict["name"])
self._track_width(column_widths, current_column, sub_dict["name"])
if new_column - current_column > 1:
worksheet.merge_cells(
start_row=current_row,
@ -189,24 +206,27 @@ class DepthSheetWriter(AbstractWriter):
cell.font = Font(color=color["color"])
return current_column
def _set_color_to_row(self, ws: Worksheet, item_type: str) -> None:
def _set_color_to_row(self, ws: Worksheet, row: int, item_type: str) -> None:
if not self.palette or item_type in ("SUB_ITEM", "INPUT"):
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",
)
cache_key = (palette, item_type)
if cache_key not in self._row_fill_cache:
self._row_fill_cache[cache_key] = PatternFill(
start_color=Color(rgb=PALETTE[palette][item_type]),
fill_type="solid",
)
cell.fill = self._row_fill_cache[cache_key]
def write(self, worksheet: Worksheet, rows: list[tuple]) -> None:
self._write_header(worksheet)
column_widths: dict[int, int] = {}
self._write_header(worksheet, column_widths=column_widths)
next_row = worksheet.max_row + 1
for row in rows:
if len(row) > 3 and isinstance(row[3], dict):
if row[0] == 'META':
@ -215,9 +235,308 @@ class DepthSheetWriter(AbstractWriter):
else:
raw_data = {}
normalized = self.row_normalizer(raw_data)
worksheet.append(
[self.value_serializer(normalized.get(key)) for key in self.columns]
)
values = [self.value_serializer(normalized.get(key)) for key in self.columns]
worksheet.append(values)
self._track_row_widths(column_widths, values)
if len(row) > 3:
self._set_color_to_row(ws=worksheet, item_type=row[0])
self._set_widths(worksheet)
self._set_color_to_row(ws=worksheet, row=next_row, item_type=row[0])
next_row += 1
self._apply_widths(worksheet, column_widths)
class XlsxTabularSheetWriter(AbstractWriter):
def __init__(
self,
workbook: XlsxWorkbook,
columns: list[tuple[str, str]],
row_normalizer: RowNormalizer,
value_serializer: ValueSerializer,
):
self.workbook = workbook
self.columns = columns
self.row_normalizer = row_normalizer
self.value_serializer = value_serializer
@staticmethod
def _apply_widths(
worksheet: XlsxWorksheet,
column_widths: dict[int, int],
) -> None:
for column, max_len in column_widths.items():
zero_based_column = column - 1
worksheet.set_column(
zero_based_column,
zero_based_column,
max(max_len + 3, 10),
)
def write(self, worksheet: XlsxWorksheet, rows: list[tuple]) -> None:
column_widths: dict[int, int] = {}
headers = [title for _, title in self.columns]
worksheet.write_row(0, 0, headers)
self._track_row_widths(column_widths, headers)
output_row = 1
for row in rows:
raw_data = row[3] if len(row) > 3 and isinstance(row[3], dict) else {}
normalized = self.row_normalizer(raw_data)
values = [self.value_serializer(normalized.get(key)) for key, _ in self.columns]
worksheet.write_row(output_row, 0, values)
self._track_row_widths(column_widths, values)
output_row += 1
self._apply_widths(worksheet, column_widths)
class XlsxDepthSheetWriter(AbstractWriter):
def __init__(
self,
workbook: XlsxWorkbook,
columns_dict: dict,
row_normalizer: RowNormalizer,
value_serializer: ValueSerializer,
colors_config: dict | None = None,
header_format_resolver: HeaderFormatResolver | None = None,
row_format_resolver: RowFormatResolver | None = None,
column_format_resolver: ColumnFormatResolver | None = None,
fixed_column_width: int | None = None,
):
self.workbook = workbook
self.columns_dict = columns_dict
self._flat_columns_to_depth: dict[str, str] = {}
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.header_format_resolver = header_format_resolver
self.row_format_resolver = row_format_resolver
self.column_format_resolver = column_format_resolver
self.fixed_column_width = fixed_column_width
self.palette: dict[str, str] = {}
self._format_cache: dict[tuple[Any, ...], Any] = {}
if colors_config:
self._set_palette()
def _apply_widths(
self,
worksheet: XlsxWorksheet,
column_widths: dict[int, int],
column_formats: list[Any] | None = None,
) -> None:
if self.fixed_column_width is not None:
for column in range(len(self.columns)):
worksheet.set_column(
column,
column,
self.fixed_column_width,
column_formats[column] if column_formats else None,
)
return
for column, max_len in column_widths.items():
zero_based_column = column - 1
worksheet.set_column(
zero_based_column,
zero_based_column,
max(max_len + 3, 10),
column_formats[zero_based_column] if column_formats else None,
)
def _resolved_format(self, properties: dict[str, Any]) -> Any:
cache_key = ("resolved", tuple(sorted(properties.items())))
if cache_key not in self._format_cache:
self._format_cache[cache_key] = self.workbook.add_format(properties)
return self._format_cache[cache_key]
def _columns_dict_to_flat(
self,
columns_dict: dict | None = None,
prefix: str = "",
depth_prefix: str = "",
) -> list[str]:
if columns_dict is None:
columns_dict = self.columns_dict
result: list[str] = []
for key, sub_dict in columns_dict.items():
new_key = sub_dict.get("key", key)
if "children" not in sub_dict:
if new_key:
flat_key = f"{prefix}{new_key}"
result.append(flat_key)
self._flat_columns_to_depth[flat_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:
for key in self.columns:
full_key = None
for sub_key in self._flat_columns_to_depth[key].split("."):
full_key = sub_key if full_key is None else f"{full_key}.{sub_key}"
color = self.colors_config.get(full_key)
if color and "palette" in color:
self.palette[key] = color["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 _header_format(self, full_key: str):
properties: dict[str, Any] = {
"border": 1,
"align": "center",
"valign": "vcenter",
}
if self.colors_config and (color := self.colors_config.get(full_key)):
if "background" in color:
properties["bg_color"] = f"#{color['background']}"
properties["pattern"] = 1
if "color" in color:
properties["font_color"] = f"#{color['color']}"
if self.header_format_resolver:
properties.update(self.header_format_resolver(full_key) or {})
return self.workbook.add_format(properties)
def _write_header(
self,
worksheet: XlsxWorksheet,
column_widths: dict[int, int],
columns_dict: dict | None = None,
current_column: int = 0,
current_row: int = 0,
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(columns_dict) - 1
for key, sub_dict in columns_dict.items():
current_key = f"{full_key}.{key}" if full_key else key
title = sub_dict["name"]
start_column = current_column
if "children" in sub_dict:
current_column = self._write_header(
worksheet=worksheet,
column_widths=column_widths,
columns_dict=sub_dict["children"],
current_column=current_column,
current_row=current_row + 1,
max_row=max_row,
full_key=current_key,
)
end_column = current_column - 1
if end_column > start_column:
worksheet.merge_range(
current_row,
start_column,
current_row,
end_column,
title,
self._header_format(current_key),
)
else:
worksheet.write(
current_row,
start_column,
title,
self._header_format(current_key),
)
else:
if current_row != max_row:
worksheet.merge_range(
current_row,
current_column,
max_row,
current_column,
title,
self._header_format(current_key),
)
else:
worksheet.write(
current_row,
current_column,
title,
self._header_format(current_key),
)
current_column += 1
if self.fixed_column_width is None:
self._track_width(column_widths, start_column + 1, title)
return current_column
def _row_format(self, palette: str, item_type: str):
cache_key = (palette, item_type)
if cache_key not in self._format_cache:
self._format_cache[cache_key] = self.workbook.add_format(
{
"bg_color": f"#{PALETTE[palette][item_type]}",
"pattern": 1,
}
)
return self._format_cache[cache_key]
def write(self, worksheet: XlsxWorksheet, rows: list[tuple]) -> None:
column_widths: dict[int, int] = {}
header_rows = self._get_header_max_row(self.columns_dict)
self._write_header(worksheet, column_widths=column_widths)
output_row = header_rows
column_formats = None
if self.column_format_resolver:
column_formats = [
self._resolved_format(properties)
if (properties := self.column_format_resolver(column_key, column))
else None
for column, column_key in enumerate(self.columns)
]
for row in rows:
if len(row) > 3 and isinstance(row[3], dict):
if row[0] == "META":
continue
raw_data = row[3]
else:
raw_data = {}
normalized = self.row_normalizer(raw_data)
values = [self.value_serializer(normalized.get(key)) for key in self.columns]
if column_formats is not None:
worksheet.write_row(output_row, 0, values)
if self.fixed_column_width is None:
self._track_row_widths(column_widths, values)
output_row += 1
continue
item_type = row[0] if len(row) > 3 else None
for column, value in enumerate(values):
cell_format = None
column_key = self.columns[column]
resolved_format = (
self.row_format_resolver(row, column_key, column)
if self.row_format_resolver
else None
)
if resolved_format:
cell_format = self._resolved_format(resolved_format)
palette = self.palette.get(column_key)
if (
cell_format is None
and palette
and item_type not in (None, "SUB_ITEM", "INPUT")
):
cell_format = self._row_format(palette, item_type)
worksheet.write(output_row, column, value, cell_format)
if self.fixed_column_width is None:
self._track_row_widths(column_widths, values)
output_row += 1
self._apply_widths(worksheet, column_widths, column_formats)

18
api/src/services/summary_service.py
View File

@ -52,6 +52,24 @@ class SummaryService:
sort_direction=sort_direction,
)
async def get_svod2_go_export_rows(
self,
year: int,
user: AppUser,
org_ids: list[int] | None = None,
sort_by: str = "org_name",
sort_direction: str = "asc",
) -> list[tuple]:
org_ids = await self._allowed_org_ids(user, org_ids)
if org_ids == []:
return []
return await self.summary_repo.get_svod2_go_export_rows(
year=year,
org_ids=org_ids,
sort_by=sort_by,
sort_direction=sort_direction,
)
async def _allowed_org_ids(
self,
user: AppUser,

25
api/tests/integration/test_summary_api_smoke.py
View File

@ -1,6 +1,9 @@
import pytest
XLSX_CONTENT_TYPE = "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"
@pytest.mark.parametrize(
"sheet",
["FORM_1", "FORM_2", "FORM_3", "FORM_4", "MAIN"],
@ -139,3 +142,25 @@ def test_summary2_sheet_unknown(client, admin_tokens, auth_headers):
headers=auth_headers(admin_tokens),
)
assert response.status_code == 404
def test_export_summary1_smoke(client, admin_tokens, auth_headers):
response = client.get(
"/api/v1/export/summary/1/MAIN/2026",
headers=auth_headers(admin_tokens),
)
assert response.status_code == 200
assert response.headers["content-type"] == XLSX_CONTENT_TYPE
assert response.content.startswith(b"PK")
def test_export_summary2_smoke(client, admin_tokens, auth_headers):
response = client.get(
"/api/v1/export/summary/2/GO/2026",
headers=auth_headers(admin_tokens),
)
assert response.status_code == 200
assert response.headers["content-type"] == XLSX_CONTENT_TYPE
assert response.content.startswith(b"PK")