получение сводной осв по id версии консолидации

api/alembic/versions/c94eb1503a26_add_osv_counterparty_alias_flag.py

@@ -1,13 +1,13 @@
-"""Expose whether an OSV counterparty was resolved by alias.
+"""Add OSV entries for group with counterparty alias resolution.
 
 Revision ID: c94eb1503a26
-Revises: d614b414a32f
+Revises: c89d082850bd
 """
 from alembic import op
 from src.core.config import settings
 
 revision = "c94eb1503a26"
-down_revision = "d614b414a32f"
+down_revision = "c89d082850bd"
 branch_labels = None
 depends_on = None
 
@@ -20,9 +20,6 @@ def _quoted_db_schema() -> str:
 
 def upgrade() -> None:
     schema = _quoted_db_schema()
-    # PostgreSQL requires recreation when RETURNS TABLE changes.
-    # No CASCADE: dependent database objects must not be silently removed.
-    op.execute(f"DROP FUNCTION {schema}.get_osv_entries_for_group(uuid, integer, text)")
     op.execute(f"""
        CREATE OR REPLACE FUNCTION {schema}.get_osv_entries_for_group(
             p_company_group_id uuid,
@@ -148,120 +145,6 @@ def upgrade() -> None:
 
 def downgrade() -> None:
     schema = _quoted_db_schema()
-    op.execute(f"DROP FUNCTION {schema}.get_osv_entries_for_group(uuid, integer, text)")
+    op.execute(
-    op.execute(f"""
+        f"DROP FUNCTION IF EXISTS {schema}.get_osv_entries_for_group(uuid, integer, text)"
-       CREATE OR REPLACE FUNCTION {schema}.get_osv_entries_for_group(
+    )
-            p_company_group_id uuid,
-            p_period_year integer,
-            p_period_quarter text
-        )
-        RETURNS TABLE (
-            osv_entry_id uuid,
-            recognition_job_id uuid,
-            legal_entity_id uuid,
-            stored_file_id uuid,
-            account_code varchar,
-            account_id uuid,
-            account_root varchar,
-            counterparty_name varchar,
-            counterparty_inn varchar,
-            counterparty_id uuid,
-            opening_debit numeric,
-            opening_credit numeric,
-            turnover_debit numeric,
-            turnover_credit numeric,
-            closing_debit numeric,
-            closing_credit numeric,
-            legal_entity_name varchar,
-            company_group_id uuid,
-            company_group_name varchar,
-            original_name varchar,
-            file_status varchar,
-            period_year smallint,
-            period_quarter varchar,
-            uploaded_at timestamptz,
-            uploaded_by varchar,
-            finished_at timestamptz,
-            is_gsk boolean,
-            is_gk boolean,
-            entity_group_flag text
-        )
-        LANGUAGE sql
-        STABLE
-        STRICT
-        AS $function$
-            WITH classified_entries AS (
-                SELECT
-                    oe.id AS osv_entry_id,
-                    oe.recognition_job_id,
-                    oe.legal_entity_id,
-                    oe.stored_file_id,
-                    CASE
-                        WHEN oe.account_id IS NOT NULL THEN coa.code
-                        ELSE oe.account_code
-                    END AS account_code,
-                    oe.account_id,
-                    oe.account_root,
-                    oe.counterparty_name,
-                    oe.counterparty_inn,
-                    COALESCE(oe.counterparty_id, counterparty_by_inn.id, counterparty_by_alias.legal_entity_id) AS counterparty_id,
-                    oe.opening_debit,
-                    oe.opening_credit,
-                    oe.turnover_debit,
-                    oe.turnover_credit,
-                    oe.closing_debit,
-                    oe.closing_credit,
-                    le.name AS legal_entity_name,
-                    le.company_group_id,
-                    cg.name AS company_group_name,
-                    sf.original_name,
-                    sf.status AS file_status,
-                    sf.period_year,
-                    sf.period_quarter,
-                    sf.uploaded_at,
-                    sf.uploaded_by,
-                    rj.finished_at,
-                    EXISTS (
-                        SELECT 1
-                        FROM {schema}.related_group_member AS member
-                        JOIN {schema}.related_group AS rg
-                        ON rg.id = member.related_group_id
-                        WHERE member.legal_entity_id = oe.legal_entity_id
-                        AND member.is_current IS TRUE
-                        AND rg.company_group_id = p_company_group_id
-                        AND rg.kind = 'GSK'
-                    ) AS is_gsk,
-                    le.company_group_id = p_company_group_id AS is_gk
-                FROM {schema}.osv_entry AS oe
-                LEFT JOIN {schema}.legal_entity AS counterparty_by_inn
-                  ON oe.counterparty_id IS NULL
-                 AND counterparty_by_inn.inn = NULLIF(BTRIM(oe.counterparty_inn), '')
-                LEFT JOIN {schema}.legal_entity_alias AS counterparty_by_alias
-                  ON oe.counterparty_id IS NULL
-                 AND counterparty_by_inn.id IS NULL
-                 AND counterparty_by_alias.normalized_alias = NULLIF(BTRIM(oe.counterparty_name), '')
-                LEFT JOIN {schema}.chart_of_account AS coa ON coa.id = oe.account_id
-                JOIN {schema}.stored_file AS sf ON sf.id = oe.stored_file_id
-                JOIN {schema}.legal_entity AS le ON le.id = oe.legal_entity_id
-                JOIN {schema}.company_group AS cg ON cg.id = le.company_group_id
-                LEFT JOIN {schema}.recognition_job AS rj
-                ON rj.id = oe.recognition_job_id
-                AND rj.file_id = sf.id
-                WHERE sf.status = 'UPLOADED'
-                AND sf.period_year = p_period_year
-                AND sf.period_quarter = p_period_quarter
-                AND EXISTS (
-                    SELECT 1 FROM {schema}.company_group AS selected_group
-                    WHERE selected_group.id = p_company_group_id
-                )
-            )
-            SELECT
-                ce.*,
-                CASE
-                    WHEN ce.is_gsk THEN 'ГСК'
-                    WHEN ce.is_gk THEN 'ГК'
-                    ELSE 'Внешний'
-                END AS entity_group_flag
-            FROM classified_entries AS ce;
-        $function$;
-    """)

api/alembic/versions/e28f709a6b31_add_osv_entries_for_consolidation_version.py

@@ -1,22 +1,15 @@
-"""add osv entries for group function
+"""Add OSV entries scoped to a consolidation version's members and period.
-
-Revision ID: d614b414a32f
-Revises: c89d082850bd
-Create Date: 2026-09-17 17:15:16.421048
 
+Revision ID: e28f709a6b31
+Revises: c94eb1503a26
 """
-from typing import Sequence, Union
-
 from alembic import op
-import sqlalchemy as sa
 from src.core.config import settings
 
-
+revision = "e28f709a6b31"
-# revision identifiers, used by Alembic.
+down_revision = "c94eb1503a26"
-revision: str = 'd614b414a32f'
+branch_labels = None
-down_revision: Union[str, Sequence[str], None] = 'c89d082850bd'
+depends_on = None
-branch_labels: Union[str, Sequence[str], None] = None
-depends_on: Union[str, Sequence[str], None] = None
 
 
 def _quoted_db_schema() -> str:
@@ -28,10 +21,8 @@ def _quoted_db_schema() -> str:
 def upgrade() -> None:
     schema = _quoted_db_schema()
     op.execute(f"""
-       CREATE OR REPLACE FUNCTION {schema}.get_osv_entries_for_group(
+       CREATE OR REPLACE FUNCTION {schema}.get_osv_entries_for_consolidation_version(
-            p_company_group_id uuid,
+            p_consolidation_version_id uuid
-            p_period_year integer,
-            p_period_quarter text
         )
         RETURNS TABLE (
             osv_entry_id uuid,
@@ -44,6 +35,7 @@ def upgrade() -> None:
             counterparty_name varchar,
             counterparty_inn varchar,
             counterparty_id uuid,
+            counterparty_resolved_by_alias boolean,
             opening_debit numeric,
             opening_credit numeric,
             turnover_debit numeric,
@@ -68,7 +60,16 @@ def upgrade() -> None:
         STABLE
         STRICT
         AS $function$
-            WITH classified_entries AS (
+            WITH consolidation_info AS (
+                SELECT cv.id AS consolidation_version_id,
+                       c.year AS period_year,
+                       c.quarter AS period_quarter,
+                       rg.company_group_id
+                FROM {schema}.consolidation_version AS cv
+                JOIN {schema}.consolidation AS c ON c.id = cv.consolidation_id
+                JOIN {schema}.related_group AS rg ON rg.id = c.related_group_id
+                WHERE cv.id = p_consolidation_version_id
+            ), classified_entries AS (
                 SELECT
                     oe.id AS osv_entry_id,
                     oe.recognition_job_id,
@@ -82,7 +83,11 @@ def upgrade() -> None:
                     oe.account_root,
                     oe.counterparty_name,
                     oe.counterparty_inn,
-                    oe.counterparty_id,
+                    COALESCE(oe.counterparty_id, counterparty_by_inn.id, counterparty_by_alias.legal_entity_id) AS counterparty_id,
+                    (oe.counterparty_id IS NULL
+                     AND counterparty_by_inn.id IS NULL
+                     AND counterparty_by_alias.legal_entity_id IS NOT NULL)
+                        AS counterparty_resolved_by_alias,
                     oe.opening_debit,
                     oe.opening_credit,
                     oe.turnover_debit,
@@ -106,11 +111,21 @@ def upgrade() -> None:
                         ON rg.id = member.related_group_id
                         WHERE member.legal_entity_id = oe.legal_entity_id
                         AND member.is_current IS TRUE
-                        AND rg.company_group_id = p_company_group_id
+                        AND rg.company_group_id = ci.company_group_id
                         AND rg.kind = 'GSK'
                     ) AS is_gsk,
-                    le.company_group_id = p_company_group_id AS is_gk
+                    le.company_group_id = ci.company_group_id AS is_gk
-                FROM {schema}.osv_entry AS oe
+                FROM consolidation_info AS ci
+                JOIN {schema}.consolidation_member AS cm
+                  ON cm.consolidation_version_id = ci.consolidation_version_id
+                JOIN {schema}.osv_entry AS oe ON oe.legal_entity_id = cm.legal_entity_id
+                LEFT JOIN {schema}.legal_entity AS counterparty_by_inn
+                  ON oe.counterparty_id IS NULL
+                 AND counterparty_by_inn.inn = NULLIF(BTRIM(oe.counterparty_inn), '')
+                LEFT JOIN {schema}.legal_entity_alias AS counterparty_by_alias
+                  ON oe.counterparty_id IS NULL
+                 AND counterparty_by_inn.id IS NULL
+                 AND counterparty_by_alias.normalized_alias = NULLIF(BTRIM(oe.counterparty_name), '')
                 LEFT JOIN {schema}.chart_of_account AS coa ON coa.id = oe.account_id
                 JOIN {schema}.stored_file AS sf ON sf.id = oe.stored_file_id
                 JOIN {schema}.legal_entity AS le ON le.id = oe.legal_entity_id
@@ -119,12 +134,8 @@ def upgrade() -> None:
                 ON rj.id = oe.recognition_job_id
                 AND rj.file_id = sf.id
                 WHERE sf.status = 'UPLOADED'
-                AND sf.period_year = p_period_year
+                AND sf.period_year = ci.period_year
-                AND sf.period_quarter = p_period_quarter
+                AND sf.period_quarter = ci.period_quarter
-                AND EXISTS (
-                    SELECT 1 FROM {schema}.company_group AS selected_group
-                    WHERE selected_group.id = p_company_group_id
-                )
             )
             SELECT
                 ce.*,
@@ -140,7 +151,6 @@ def upgrade() -> None:
 
 def downgrade() -> None:
     schema = _quoted_db_schema()
-    op.execute(f"""
+    op.execute(
-        DROP FUNCTION IF EXISTS
+        f"DROP FUNCTION {schema}.get_osv_entries_for_consolidation_version(uuid)"
-        {schema}.get_osv_entries_for_group(uuid, integer, text)
+    )
-    """)

api/src/api/endpoints/osv.py

@@ -23,6 +23,21 @@ async def get_osv_by_company_group(
         company_group_id=company_group_id, year=year, quarter=quarter
     )
 
+
+@router.get('/by-consolidation-version', response_model=list[OsvGroupEntryResponse])
+async def get_osv_by_consolidation_version(
+    consolidation_version_id: uuid.UUID,
+    service: OsvService = Depends(get_osv_service),
+):
+    """ОСВ за версию консолидации.
+
+    Пустая выборка (в том числе неизвестная версия) возвращается как [].
+    """
+    return await service.get_by_consolidation_version(
+        consolidation_version_id=consolidation_version_id
+    )
+
+
 @router.get('/by-file-id')
 async def get_osv_by_file(file_id: uuid.UUID, service: OsvService = Depends(get_osv_service)):
     result = await service.get_by_file_id(file_id)

api/src/repositories/osv_repository.py

@@ -20,6 +20,17 @@ class OsvRepository(BaseRepository):
         })
         return [dict(row) for row in result.mappings().all()]
 
+    async def get_by_consolidation_version(self, consolidation_version_id):
+        stmt = text("""
+            SELECT * FROM dkb.get_osv_entries_for_consolidation_version(
+                CAST(:consolidation_version_id AS uuid)
+            )
+        """)
+        result = await self.session.execute(stmt, {
+            "consolidation_version_id": consolidation_version_id,
+        })
+        return [dict(row) for row in result.mappings().all()]
+
     async def get_by_file_id(self, file_id):
         stmt = select(self.model).where(self.model.stored_file_id == file_id)
         result = await self.session.execute(stmt)

api/src/services/osv_service.py

@@ -42,6 +42,32 @@ class OsvService:
 
         return result
 
+    async def get_by_consolidation_version(
+        self, consolidation_version_id: uuid.UUID
+    ) -> list[OsvGroupEntryResponse]:
+        entries = await self.osv_repo.get_by_consolidation_version(
+            consolidation_version_id=consolidation_version_id
+        )
+
+        result: list[OsvGroupEntryResponse] = []
+        for row_index, entry in enumerate(entries):
+            try:
+                result.append(OsvGroupEntryResponse.model_validate(entry))
+            except ValidationError as exc:
+                logger.error(
+                    "Некорректный результат get_osv_entries_for_consolidation_version: "
+                    "consolidation_version_id={consolidation_version_id}, "
+                    "row_index={row_index}, osv_entry_id={osv_entry_id}, "
+                    "validation_errors={validation_errors}",
+                    consolidation_version_id=consolidation_version_id,
+                    row_index=row_index,
+                    osv_entry_id=entry.get("osv_entry_id"),
+                    validation_errors=exc.errors(include_input=False),
+                )
+                raise
+
+        return result
+
 
     async def get_by_file_id(self, file_id: uuid.UUID):
         try: