Merge remote-tracking branch 'origin/AURORA-1751' into feat/corrections-back

# Conflicts:
#	api/src/api/dependencies/consolidations.py
#	api/src/services/vgo.py

api/alembic/versions/0012_vgo_rule_not_excluded.py

@@ -0,0 +1,29 @@
+"""Запретить исключение из КФО строк с назначенным правилом ВГО."""
+
+import sqlalchemy as sa
+
+from alembic import op
+from src.models.base import SCHEMA
+
+revision = "0012_vgo_rule_not_excluded"
+down_revision = "0011_vgo_assignment"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.execute(sa.text(f'UPDATE "{SCHEMA}".vgo_assignment SET is_exclude = false WHERE rule_id IS NOT NULL'))
+    op.alter_column("vgo_assignment", "is_exclude", server_default=sa.text("true"), schema=SCHEMA)
+    op.create_check_constraint(
+        "ck_vgo_assignment_rule_not_excluded",
+        "vgo_assignment",
+        "rule_id IS NULL OR NOT is_exclude",
+        schema=SCHEMA,
+    )
+
+
+def downgrade() -> None:
+    op.alter_column("vgo_assignment", "is_exclude", server_default=sa.text("false"), schema=SCHEMA)
+    op.drop_constraint(
+        "ck_vgo_assignment_rule_not_excluded", "vgo_assignment", type_="check", schema=SCHEMA
+    )

api/alembic/versions/0013_vgo_entry_cascade.py

@@ -0,0 +1,36 @@
+"""Удалять назначения ВГО вместе с исходной строкой ОСВ."""
+
+from alembic import op
+from src.models.base import SCHEMA
+
+revision = "0013_vgo_entry_cascade"
+down_revision = "0012_vgo_rule_not_excluded"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.drop_constraint("fk_vgo_assignment_entry", "vgo_assignment", type_="foreignkey", schema=SCHEMA)
+    op.create_foreign_key(
+        "fk_vgo_assignment_entry",
+        "vgo_assignment",
+        "osv_entry",
+        ["osv_entry_id"],
+        ["id"],
+        source_schema=SCHEMA,
+        referent_schema=SCHEMA,
+        ondelete="CASCADE",
+    )
+
+
+def downgrade() -> None:
+    op.drop_constraint("fk_vgo_assignment_entry", "vgo_assignment", type_="foreignkey", schema=SCHEMA)
+    op.create_foreign_key(
+        "fk_vgo_assignment_entry",
+        "vgo_assignment",
+        "osv_entry",
+        ["osv_entry_id"],
+        ["id"],
+        source_schema=SCHEMA,
+        referent_schema=SCHEMA,
+    )

api/alembic/versions/0014_vgo_check_names.py

@@ -0,0 +1,28 @@
+"""Привести имена CHECK vgo_assignment к naming convention ORM."""
+
+import sqlalchemy as sa
+
+from alembic import op
+from src.models.base import SCHEMA
+
+revision = "0014_vgo_check_names"
+down_revision = "0013_vgo_entry_cascade"
+branch_labels = None
+depends_on = None
+
+CONSTRAINTS = ("action_values", "action_rule_consistency", "rule_not_excluded")
+
+
+def _rename(*, backwards: bool) -> None:
+    for suffix in CONSTRAINTS:
+        name = f"ck_vgo_assignment_{suffix}"
+        old, new = (name, f"ck_vgo_assignment_{name}") if backwards else (f"ck_vgo_assignment_{name}", name)
+        op.execute(sa.text(f'ALTER TABLE "{SCHEMA}".vgo_assignment RENAME CONSTRAINT "{old}" TO "{new}"'))
+
+
+def upgrade() -> None:
+    _rename(backwards=False)
+
+
+def downgrade() -> None:
+    _rename(backwards=True)

api/src/api/dependencies/consolidations.py

@@ -18,7 +18,7 @@ from src.services.corrections import CorrectionService
 from src.services.counterparty_mapping import CounterpartyMappingService
 from src.services.elimination_rules import EliminationRuleService
 from src.services.intercompany_matrix import IntercompanyMatrixService
-from src.services.vgo import VgoService
+from src.services.vgo import VgoAssignmentCrudService, VgoService
 
 
 def get_consolidation_service(
@@ -133,3 +133,16 @@ def get_correction_service(
 
 
 CorrectionServiceDep = Annotated[CorrectionService, Depends(get_correction_service)]
+
+
+def get_vgo_assignment_crud_service(
+    session: SessionDep,
+    audit: Annotated[RequestAudit, Depends(get_request_audit)],
+    scope: Annotated[VisibilityScope, Depends(get_scope)],
+    osv_service: ConsolidatedOsvServiceDep,
+) -> VgoAssignmentCrudService:
+    """Сервис ручной правки назначений ВГО."""
+    return VgoAssignmentCrudService(session, audit, scope, osv_service)
+
+
+VgoAssignmentCrudServiceDep = Annotated[VgoAssignmentCrudService, Depends(get_vgo_assignment_crud_service)]

api/src/api/endpoints/consolidations/vgo_assignment.py

@@ -5,18 +5,63 @@ import uuid
 from fastapi import APIRouter, Depends
 
 from src.api.dependencies.auth import require
-from src.api.dependencies.consolidations import VgoServiceDep
-from src.schemas.consolidations import VgoAssignmentCandidatesOut
+from src.api.dependencies.consolidations import VgoAssignmentCrudServiceDep, VgoServiceDep
+from src.schemas.consolidations import (
+    VgoAssignmentAttachRuleIn,
+    VgoAssignmentCandidatesOut,
+    VgoAssignmentDetachRuleIn,
+    VgoAssignmentExcludeIn,
+    VgoAssignmentOut,
+)
 
 router = APIRouter(prefix="/consolidations/versions")
 
 
-@router.post(
+@router.get(
     "/{version_id}/vgo-assignments/distribute",
-    dependencies=[Depends(require("CONSOLIDATION_EDIT"))],
+    dependencies=[Depends(require("CONSOLIDATION_VIEW"))],
     response_model=VgoAssignmentCandidatesOut,
     summary="Агрегаты ВГО по правилам и компаниям с ручными назначениями",
 )
 async def distribute_articles(version_id: uuid.UUID, service: VgoServiceDep) -> VgoAssignmentCandidatesOut:
     """Возвращает агрегаты и ручные назначения по группам, а также строки без назначения."""
     return await service.assignment_candidates(version_id)
+
+
+@router.put(
+    "/{version_id}/vgo-assignments/rule",
+    dependencies=[Depends(require("CONSOLIDATION_EDIT"))],
+    response_model=list[VgoAssignmentOut],
+    summary="Привязать выбранные строки ОСВ к правилу ВГО",
+)
+async def attach_rule(
+    version_id: uuid.UUID, data: VgoAssignmentAttachRuleIn, service: VgoAssignmentCrudServiceDep
+) -> list[VgoAssignmentOut]:
+    """Обновляет назначения для указанных ID и создаёт недостающие записи."""
+    return await service.attach_rule(version_id, data)
+
+
+@router.delete(
+    "/{version_id}/vgo-assignments/rule",
+    dependencies=[Depends(require("CONSOLIDATION_EDIT"))],
+    response_model=list[VgoAssignmentOut],
+    summary="Снять назначение правила ВГО с выбранных строк ОСВ",
+)
+async def detach_rule(
+    version_id: uuid.UUID, data: VgoAssignmentDetachRuleIn, service: VgoAssignmentCrudServiceDep
+) -> list[VgoAssignmentOut]:
+    """Оставляет или создаёт записи без правила с is_exclude=true."""
+    return await service.detach_rule(version_id, data)
+
+
+@router.put(
+    "/{version_id}/vgo-assignments/exclude",
+    dependencies=[Depends(require("CONSOLIDATION_EDIT"))],
+    response_model=list[VgoAssignmentOut],
+    summary="Исключить или вернуть выбранные строки ОСВ в КФО",
+)
+async def set_exclusion(
+    version_id: uuid.UUID, data: VgoAssignmentExcludeIn, service: VgoAssignmentCrudServiceDep
+) -> list[VgoAssignmentOut]:
+    """Сохраняет is_exclude для выбранных ID; строки с правилом исключать нельзя."""
+    return await service.set_exclusion(version_id, data)

api/src/domain/vgo_distribution.py

@@ -0,0 +1,90 @@
+"""Чистый выбор агрегатов и строк ВГО после поиска встречных сумм."""
+
+import uuid
+from collections import Counter, defaultdict
+from collections.abc import Mapping, Sequence, Set
+from dataclasses import dataclass
+from decimal import Decimal
+
+from src.domain.consolidated_osv import ConsolidatedRow
+
+RowKey = tuple[uuid.UUID, ...]
+ZERO = Decimal("0")
+
+
+@dataclass(frozen=True)
+class Distribution:
+    """Результат выбора: расхождения, строки без пары и остаток для ручного разбора."""
+
+    selected_diff: tuple[int, ...]
+    auto_rules: Mapping[RowKey, uuid.UUID]
+    unassigned: frozenset[RowKey]
+    used_ids: frozenset[uuid.UUID]
+
+
+@dataclass(frozen=True)
+class DistributionInput:
+    rows: Sequence[ConsolidatedRow]
+    exact_ids: Set[uuid.UUID]
+    diff_ids: Sequence[Set[uuid.UUID]]
+    manual_keys: Set[RowKey]
+    owners: Mapping[uuid.UUID, Set[uuid.UUID]]
+    members: Set[uuid.UUID]
+
+
+def account_owners(
+    selections: Sequence[tuple[uuid.UUID, Set[uuid.UUID]]],
+) -> dict[uuid.UUID, frozenset[uuid.UUID]]:
+    """Для каждого раскрытого дочернего счёта найти все правила, в которые он входит."""
+    owners: defaultdict[uuid.UUID, set[uuid.UUID]] = defaultdict(set)
+    for rule_id, account_ids in selections:
+        for account_id in account_ids:
+            owners[account_id].add(rule_id)
+    return {account_id: frozenset(rule_ids) for account_id, rule_ids in owners.items()}
+
+
+def select_exact_aggregates(
+    used_ids: Set[uuid.UUID], candidate_ids: Sequence[Set[uuid.UUID]]
+) -> tuple[tuple[int, ...], frozenset[uuid.UUID]]:
+    """Закрепить точные агрегаты по порядку правил без повторного использования строк."""
+    used = set(used_ids)
+    selected = []
+    for index, ids in enumerate(candidate_ids):
+        if ids and used.isdisjoint(ids):
+            selected.append(index)
+            used.update(ids)
+    return tuple(selected), frozenset(used)
+
+
+def distribute_vgo_rows(data: DistributionInput) -> Distribution:
+    """Выбрать непересекающиеся расхождения, затем однозначные строки без пары и остаток.
+
+    Точные агрегаты уже закреплены по порядку правил. Пересекающиеся расхождения все остаются
+    нераспределёнными; ручные решения не заменяются автоматическими.
+    """
+    candidates = [(index, ids) for index, ids in enumerate(data.diff_ids) if ids and data.exact_ids.isdisjoint(ids)]
+    counts = Counter(entry_id for _, ids in candidates for entry_id in ids)
+    selected = tuple(index for index, ids in candidates if all(counts[entry_id] == 1 for entry_id in ids))
+    used = set(data.exact_ids)
+    for index in selected:
+        used.update(data.diff_ids[index])
+
+    auto_rules: dict[RowKey, uuid.UUID] = {}
+    unassigned: set[RowKey] = set()
+    for row in data.rows:
+        if not used.isdisjoint(row.entry_ids):
+            continue
+        rule_ids = data.owners.get(row.account_id, frozenset()) if row.account_id is not None else frozenset()
+        if (
+            row.entry_ids not in data.manual_keys
+            and len(rule_ids) == 1
+            and (
+                row.legal_entity_id in data.members
+                and row.counterparty_id in data.members
+                and (row.closing_debit not in (None, ZERO) or row.closing_credit not in (None, ZERO))
+            )
+        ):
+            auto_rules[row.entry_ids] = next(iter(rule_ids))
+        elif rule_ids or row.entry_ids in data.manual_keys:
+            unassigned.add(row.entry_ids)
+    return Distribution(selected, auto_rules, frozenset(unassigned), frozenset(used))

api/src/models/kfo.py

@@ -34,6 +34,7 @@ class VgoAssignment(UUIDPrimaryMixin, Base):
             "(action = 'ASSIGN' AND rule_id IS NOT NULL) OR (action = 'UNASSIGNED' AND rule_id IS NULL)",
             name="action_rule_consistency",
         ),
+        CheckConstraint("rule_id IS NULL OR NOT is_exclude", name="rule_not_excluded"),
     )
 
     consolidation_version_id: Mapped[uuid.UUID] = mapped_column(
@@ -41,13 +42,15 @@ class VgoAssignment(UUIDPrimaryMixin, Base):
     )
     statement_type: Mapped[StatementType] = mapped_column(pg_enum("statement_type"), nullable=False)
     osv_entry_id: Mapped[uuid.UUID] = mapped_column(
-        UUID(as_uuid=True), ForeignKey("osv_entry.id", name="fk_vgo_assignment_entry"), nullable=False
+        UUID(as_uuid=True),
+        ForeignKey("osv_entry.id", name="fk_vgo_assignment_entry", ondelete="CASCADE"),
+        nullable=False,
     )
     rule_id: Mapped[uuid.UUID | None] = mapped_column(
         UUID(as_uuid=True), ForeignKey("elimination_rule.id", name="fk_vgo_assignment_rule")
     )
     action: Mapped[str] = mapped_column(String, nullable=False)
-    is_exclude: Mapped[bool] = mapped_column(Boolean, nullable=False, server_default=text("false"))
+    is_exclude: Mapped[bool] = mapped_column(Boolean, nullable=False, server_default=text("true"))
     created_by: Mapped[uuid.UUID | None] = mapped_column(
         UUID(as_uuid=True), ForeignKey("app_user.id", name="fk_vgo_assignment_created_by")
     )

api/src/repositories/vgo_assignments.py

@@ -10,15 +10,23 @@ from src.models.kfo import VgoAssignment
 
 
 async def list_assignments(
-    session: AsyncSession, version_id: uuid.UUID, statement_type: StatementType
+    session: AsyncSession, version_id: uuid.UUID, statement_type: StatementType | None = None
 ) -> list[VgoAssignment]:
-    """Все ручные решения для версии и типа отчётности."""
-    stmt = (
-        select(VgoAssignment)
-        .where(
-            VgoAssignment.consolidation_version_id == version_id,
-            VgoAssignment.statement_type == statement_type,
-        )
-        .order_by(VgoAssignment.created_at, VgoAssignment.id)
-    )
+    """Все ручные решения версии, при необходимости ограниченные типом отчётности."""
+    stmt = select(VgoAssignment).where(VgoAssignment.consolidation_version_id == version_id)
+    if statement_type is not None:
+        stmt = stmt.where(VgoAssignment.statement_type == statement_type)
+    stmt = stmt.order_by(VgoAssignment.created_at, VgoAssignment.id)
     return list((await session.scalars(stmt)).all())
+
+
+async def for_entries(
+    session: AsyncSession, version_id: uuid.UUID, statement_type: StatementType, entry_ids: list[uuid.UUID]
+) -> dict[uuid.UUID, VgoAssignment]:
+    """Решения для конкретных исходных строк ОСВ."""
+    stmt = select(VgoAssignment).where(
+        VgoAssignment.consolidation_version_id == version_id,
+        VgoAssignment.statement_type == statement_type,
+        VgoAssignment.osv_entry_id.in_(entry_ids),
+    )
+    return {item.osv_entry_id: item for item in (await session.scalars(stmt)).all()}

api/src/schemas/consolidations.py

@@ -425,16 +425,37 @@ class EliminationPairsOut(BaseModel):
         return None if value is None else value.quantize(CENT, rounding=ROUND_HALF_UP)
 
 
-class VgoAssignmentRowOut(ConsolidatedRowOut):
-    """Строка свода с данными ручного решения, если оно есть."""
+class VgoAssignmentRowOut(BaseModel):
+    """Поля строки свода, необходимые для распределения ВГО."""
 
-    osv_entry_id: uuid.UUID | None = None
+    entry_ids: list[uuid.UUID]
+    legal_entity_id: uuid.UUID
+    legal_entity_name: str
+    counterparty_id: uuid.UUID | None
+    counterparty_name: str | None
+    account_code: str | None
+    period: str
+    opening_debit: Decimal | None
+    opening_credit: Decimal | None
+    turnover_debit: Decimal | None
+    turnover_credit: Decimal | None
+    closing_debit: Decimal | None
+    closing_credit: Decimal | None
     action: Literal["ASSIGN", "UNASSIGNED"] | None = None
-    is_exclude: bool | None = None
+    is_exclude: bool = True
+    is_auto_assigned: bool = False
+    unassigned_reason: Literal["PARTIAL_MANUAL_ASSIGNMENT"] | None = None
+
+    @field_validator(
+        "opening_debit", "opening_credit", "turnover_debit", "turnover_credit", "closing_debit", "closing_credit"
+    )
+    @classmethod
+    def _two_decimals(cls, value: Decimal | None) -> Decimal | None:
+        return None if value is None else value.quantize(CENT, rounding=ROUND_HALF_UP)
 
 
 class VgoAssignmentGroupOut(BaseModel):
-    """Принятые агрегаты и ручные строки одного правила и направленной пары компаний."""
+    """Принятые агрегаты и строки без встречной пары одного правила и пары компаний."""
 
     rule_id: uuid.UUID
     rule_no: int
@@ -466,6 +487,65 @@ class VgoAssignmentCandidatesOut(BaseModel):
     unassigned: list[VgoAssignmentRowOut]
 
 
+class VgoAssignmentAttachRuleIn(BaseModel):
+    """Привязать все исходные ID выбранных строк сводной ОСВ к правилу."""
+
+    entry_ids: list[uuid.UUID] = Field(min_length=1)
+    rule_id: uuid.UUID
+
+    @field_validator("entry_ids")
+    @classmethod
+    def unique_entries(cls, value: list[uuid.UUID]) -> list[uuid.UUID]:
+        if len(value) != len(set(value)):
+            raise ValueError("entry_ids не должны повторяться")
+        return value
+
+
+class VgoAssignmentExcludeIn(BaseModel):
+    """Включить или исключить выбранные строки сводной ОСВ из КФО."""
+
+    statement_type: StatementType
+    entry_ids: list[uuid.UUID] = Field(min_length=1)
+    is_exclude: bool
+
+    @field_validator("entry_ids")
+    @classmethod
+    def unique_entries(cls, value: list[uuid.UUID]) -> list[uuid.UUID]:
+        if len(value) != len(set(value)):
+            raise ValueError("entry_ids не должны повторяться")
+        return value
+
+
+class VgoAssignmentDetachRuleIn(BaseModel):
+    """Исходные ID выбранных строк сводной ОСВ для снятия правила."""
+
+    statement_type: StatementType
+    entry_ids: list[uuid.UUID] = Field(min_length=1)
+
+    @field_validator("entry_ids")
+    @classmethod
+    def unique_entries(cls, value: list[uuid.UUID]) -> list[uuid.UUID]:
+        if len(value) != len(set(value)):
+            raise ValueError("entry_ids не должны повторяться")
+        return value
+
+
+class VgoAssignmentOut(BaseModel):
+    """Сохранённое решение для одной исходной строки ОСВ."""
+
+    id: uuid.UUID
+    consolidation_version_id: uuid.UUID
+    statement_type: StatementType
+    osv_entry_id: uuid.UUID
+    rule_id: uuid.UUID | None
+    action: Literal["ASSIGN", "UNASSIGNED"]
+    is_exclude: bool
+    created_by: uuid.UUID | None
+    created_at: datetime.datetime
+    updated_by: uuid.UUID | None
+    updated_at: datetime.datetime | None
+
+
 EliminationTotalStatus = Literal["AUTO", "DISCREPANCY", "ONE_SIDED", "MANUAL"]
 
 

api/src/services/vgo.py

@@ -8,22 +8,38 @@ from decimal import Decimal
 
 from sqlalchemy.ext.asyncio import AsyncSession
 
+from src.api.dependencies.audit import RequestAudit
+from src.common.clock import utc_now
 from src.common.exceptions import DomainValidationError
 from src.domain.consolidated_osv import ConsolidatedRow
 from src.domain.elimination_packing import calculate_rule_totals
+from src.domain.policy import VisibilityScope
+from src.domain.vgo_distribution import (
+    Distribution,
+    DistributionInput,
+    account_owners,
+    distribute_vgo_rows,
+    select_exact_aggregates,
+)
 from src.models.accounting import EliminationRule
-from src.models.enums import StatementType
+from src.models.enums import AuditAction, AuditSection, StatementType
 from src.models.kfo import VgoAssignment
+from src.repositories import directories as directories_repo
 from src.repositories import elimination_rules as elimination_rules_repo
 from src.repositories import vgo_assignments as vgo_assignments_repo
 from src.schemas.consolidations import (
     EliminationMatchedPairOut,
+    VgoAssignmentAttachRuleIn,
     VgoAssignmentCandidatesOut,
+    VgoAssignmentDetachRuleIn,
+    VgoAssignmentExcludeIn,
     VgoAssignmentGroupOut,
+    VgoAssignmentOut,
     VgoAssignmentRowOut,
 )
 from src.services.consolidated_osv import ConsolidatedOsvBuildResult, ConsolidatedOsvService
-from src.services.intercompany_matrix import IntercompanyMatrixService
+from src.services.consolidation_guard import lock_editable_version
+from src.services.intercompany_matrix import IntercompanyMatrixService, _make_account_tree
 
 ZERO = Decimal("0")
 GroupKey = tuple[uuid.UUID, uuid.UUID, uuid.UUID]
@@ -52,9 +68,22 @@ def _ids_from_aggregate(aggregate: EliminationMatchedPairOut) -> set[uuid.UUID]:
 
 
 def _assignment_row(row: ConsolidatedRow, assignment: VgoAssignment | None = None) -> VgoAssignmentRowOut:
-    output = VgoAssignmentRowOut.model_validate(row, from_attributes=True)
+    output = VgoAssignmentRowOut(
+        entry_ids=list(row.entry_ids),
+        legal_entity_id=row.legal_entity_id,
+        legal_entity_name=row.legal_entity_name,
+        counterparty_id=row.counterparty_id,
+        counterparty_name=row.counterparty_entity_name or row.counterparty_name,
+        account_code=row.account_code,
+        period=row.period,
+        opening_debit=row.opening_debit,
+        opening_credit=row.opening_credit,
+        turnover_debit=row.turnover_debit,
+        turnover_credit=row.turnover_credit,
+        closing_debit=row.closing_debit,
+        closing_credit=row.closing_credit,
+    )
     if assignment is not None:
-        output.osv_entry_id = assignment.osv_entry_id if len(row.entry_ids) == 1 else None
         output.action = "ASSIGN" if assignment.action == "ASSIGN" else "UNASSIGNED"
         output.is_exclude = assignment.is_exclude
     return output
@@ -80,6 +109,7 @@ class _ManualResult:
     groups: dict[GroupKey, list[ManualItem]]
     by_row: dict[tuple[uuid.UUID, ...], VgoAssignment]
     assigned_ids: set[uuid.UUID]
+    partial_rows: set[tuple[uuid.UUID, ...]]
 
 
 def _manual_assignments(
@@ -90,15 +120,15 @@ def _manual_assignments(
     groups: defaultdict[GroupKey, list[ManualItem]] = defaultdict(list)
     by_row: dict[tuple[uuid.UUID, ...], VgoAssignment] = {}
     assigned_ids: set[uuid.UUID] = set()
+    partial_rows: set[tuple[uuid.UUID, ...]] = set()
     for row in rows:
         found = [by_entry[entry_id] for entry_id in row.entry_ids if entry_id in by_entry]
         if not found:
             continue
         if len(found) != len(row.entry_ids):
-            raise DomainValidationError(
-                "AMBIGUOUS_VGO_ASSIGNMENT",
-                "Для части исходных строк сводной ОСВ есть ручное назначение, а для остальных нет",
-            )
+            by_row[row.entry_ids] = found[0]
+            partial_rows.add(row.entry_ids)
+            continue
         assignment = found[0]
         if any(
             (item.action, item.rule_id, item.is_exclude)
@@ -115,23 +145,46 @@ def _manual_assignments(
             key, amount_a, amount_b = _manual_key_amount(row, rule)
             groups[key].append((_assignment_row(row, assignment), amount_a, amount_b))
             assigned_ids.update(row.entry_ids)
-    return _ManualResult(groups, by_row, assigned_ids)
+    return _ManualResult(groups, by_row, assigned_ids, partial_rows)
+
+
+def _assign_unpaired(
+    rows: Sequence[ConsolidatedRow],
+    manual: _ManualResult,
+    auto_rules: Mapping[tuple[uuid.UUID, ...], uuid.UUID],
+    rules: Mapping[uuid.UUID, EliminationRule],
+) -> None:
+    """Отнести оставшиеся без агрегата строки к единственному правилу их счёта."""
+    for row in rows:
+        rule_id = auto_rules.get(row.entry_ids)
+        if rule_id is None:
+            continue
+        rule = rules[rule_id]
+        key, amount_a, amount_b = _manual_key_amount(row, rule)
+        output = _assignment_row(row)
+        output.action = "ASSIGN"
+        output.is_exclude = False
+        output.is_auto_assigned = True
+        manual.groups.setdefault(key, []).append((output, amount_a, amount_b))
+        manual.assigned_ids.update(row.entry_ids)
 
 
 def _unassigned_rows(
-    all_rows: Sequence[ConsolidatedRow],
-    candidates: Sequence[ConsolidatedRow],
+    rows: Sequence[ConsolidatedRow],
     manual: _ManualResult,
-    used_ids: set[uuid.UUID],
+    distribution: Distribution,
 ) -> list[VgoAssignmentRowOut]:
-    rows = [*candidates]
-    candidate_keys = {row.entry_ids for row in candidates}
-    rows.extend(row for row in all_rows if row.entry_ids not in candidate_keys and row.entry_ids in manual.by_row)
-    return [
-        _assignment_row(row, manual.by_row.get(row.entry_ids))
-        for row in rows
-        if used_ids.isdisjoint(row.entry_ids) and manual.assigned_ids.isdisjoint(row.entry_ids)
-    ]
+    result = []
+    for row in rows:
+        if row.entry_ids not in distribution.unassigned or not manual.assigned_ids.isdisjoint(row.entry_ids):
+            continue
+        if row.entry_ids in manual.partial_rows:
+            item = _assignment_row(row)
+            item.unassigned_reason = "PARTIAL_MANUAL_ASSIGNMENT"
+        else:
+            item = _assignment_row(row, manual.by_row.get(row.entry_ids))
+        result.append(item)
+    return result
 
 
 @dataclass
@@ -145,21 +198,15 @@ class _AggregateSearchResult:
     used_ids: set[uuid.UUID] = field(default_factory=set)
 
 
-def _select_diff_aggregates(search: _AggregateSearchResult) -> dict[GroupKey, list[EliminationMatchedPairOut]]:
-    """Принять только расхождения с единственным вариантом для набора исходных строк."""
-    by_ids: defaultdict[tuple[uuid.UUID, ...], list[EliminationMatchedPairOut]] = defaultdict(list)
-    for aggregate in search.diff_aggregates:
-        ids = _ids_from_aggregate(aggregate)
-        if ids and search.used_ids.isdisjoint(ids):
-            by_ids[tuple(sorted(ids))].append(aggregate)
-
+def _select_diff_aggregates(
+    search: _AggregateSearchResult, distribution: Distribution
+) -> dict[GroupKey, list[EliminationMatchedPairOut]]:
+    """Сгруппировать выбранные доменной функцией расхождения."""
     diff_aggregates: defaultdict[GroupKey, list[EliminationMatchedPairOut]] = defaultdict(list)
-    for options in by_ids.values():
-        if len(options) == 1:
-            aggregate = options[0]
-            key = (aggregate.rule_id, aggregate.company_a_id, aggregate.company_b_id)
-            diff_aggregates[key].append(aggregate)
-            search.used_ids.update(_ids_from_aggregate(aggregate))
+    for index in distribution.selected_diff:
+        aggregate = search.diff_aggregates[index]
+        key = (aggregate.rule_id, aggregate.company_a_id, aggregate.company_b_id)
+        diff_aggregates[key].append(aggregate)
     return diff_aggregates
 
 
@@ -182,8 +229,8 @@ def _build_groups(
         manual_rows = manual.get(key, [])
         amount_a = sum((aggregate.amount_a for aggregate in aggregates), ZERO)
         amount_b = sum((aggregate.amount_b for aggregate in aggregates), ZERO)
-        amount_a += sum((amount_a for row, amount_a, _ in manual_rows if not row.is_exclude), ZERO)
-        amount_b += sum((amount_b for row, _, amount_b in manual_rows if not row.is_exclude), ZERO)
+        amount_a += sum((item_amount_a for _, item_amount_a, _ in manual_rows), ZERO)
+        amount_b += sum((item_amount_b for _, _, item_amount_b in manual_rows), ZERO)
         totals = calculate_rule_totals(amount_a, amount_b, manual=False)
         groups.append(
             VgoAssignmentGroupOut(
@@ -218,6 +265,20 @@ class VgoService:
         self._matrix_service = matrix_service
         self.session = session
 
+    async def _rule_accounts(
+        self, osv_result: ConsolidatedOsvBuildResult, rules: Sequence[EliminationRule]
+    ) -> dict[uuid.UUID, frozenset[uuid.UUID]]:
+        """Раскрыть дочерние счета и определить их принадлежность правилам."""
+        accounts = await directories_repo.list_accounts(self.session)
+        by_id, children = _make_account_tree(accounts)
+        selections = []
+        for rule in rules:
+            selection, _ = await self._matrix_service._resolve_preset_selection(
+                rule.id, osv_result.version, by_id, children
+            )
+            selections.append((rule.id, selection.company_1_account_ids | selection.company_2_account_ids))
+        return account_owners(selections)
+
     async def _find_aggregates(
         self,
         version_id: uuid.UUID,
@@ -227,21 +288,23 @@ class VgoService:
     ) -> _AggregateSearchResult:
         """Перебрать правила и компании, сразу закрепляя только точные агрегаты."""
         search = _AggregateSearchResult()
-        for company_a in osv_result.members:
-            for company_b in osv_result.members:
-                if company_a == company_b:
-                    continue
-                for rule in rules:
+        for rule in rules:
+            for company_a in osv_result.members:
+                for company_b in osv_result.members:
+                    if company_a == company_b:
+                        continue
                     free_rows = [row for row in rows if search.used_ids.isdisjoint(row.entry_ids)]
                     result = await self._matrix_service.build_pairs(
                         version_id, rule.id, company_a, company_b, free_rows, osv_result=osv_result
                     )
-                    for aggregate in result.matched:
-                        ids = _ids_from_aggregate(aggregate)
-                        if search.used_ids.isdisjoint(ids):
-                            search.used_ids.update(ids)
-                            key = (aggregate.rule_id, aggregate.company_a_id, aggregate.company_b_id)
-                            search.matched_aggregates[key].append(aggregate)
+                    accepted, used_ids = select_exact_aggregates(
+                        search.used_ids, [_ids_from_aggregate(aggregate) for aggregate in result.matched]
+                    )
+                    search.used_ids.update(used_ids)
+                    for index in accepted:
+                        aggregate = result.matched[index]
+                        key = (aggregate.rule_id, aggregate.company_a_id, aggregate.company_b_id)
+                        search.matched_aggregates[key].append(aggregate)
                     search.diff_aggregates.extend(result.with_discrepancy)
         return search
 
@@ -249,15 +312,35 @@ class VgoService:
         """Найти агрегаты и общие для всех правил строки без назначения, не записывая их в БД."""
         osv_result = await self._osv_service.build_with_members(version_id)
         all_rules = await elimination_rules_repo.list_rules(self.session, osv_result.version.elimination_rules_version)
-        rules = [
-            rule for rule in all_rules if rule.statement_type == StatementType.balance and rule.measure == "closing"
-        ]
+        rules = sorted(
+            (rule for rule in all_rules if rule.statement_type == StatementType.balance and rule.measure == "closing"),
+            key=lambda rule: rule.rule_no,
+        )
         rows = assignment_candidates(osv_result.consolidated.rows)
         assignments = await vgo_assignments_repo.list_assignments(self.session, version_id, StatementType.balance)
         manual = _manual_assignments(osv_result.consolidated.rows, assignments, rules)
         free_rows = [row for row in rows if row.entry_ids not in manual.by_row]
+        owners = await self._rule_accounts(osv_result, rules) if any(row.account_id is not None for row in rows) else {}
         search = await self._find_aggregates(version_id, osv_result, rules, free_rows)
-        diff_aggregates = _select_diff_aggregates(search)
+        candidate_keys = {row.entry_ids for row in rows}
+        decision_rows = [*rows]
+        decision_rows.extend(
+            row
+            for row in osv_result.consolidated.rows
+            if row.entry_ids not in candidate_keys and row.entry_ids in manual.by_row
+        )
+        distribution = distribute_vgo_rows(
+            DistributionInput(
+                rows=decision_rows,
+                exact_ids=search.used_ids,
+                diff_ids=[_ids_from_aggregate(aggregate) for aggregate in search.diff_aggregates],
+                manual_keys=set(manual.by_row),
+                owners=owners,
+                members=set(osv_result.members),
+            )
+        )
+        diff_aggregates = _select_diff_aggregates(search, distribution)
+        _assign_unpaired(free_rows, manual, distribution.auto_rules, {rule.id: rule for rule in rules})
 
         names = {company_id: entity.name for company_id, entity in osv_result.members.items()}
         for row in osv_result.consolidated.rows:
@@ -267,5 +350,145 @@ class VgoService:
                     names.setdefault(row.counterparty_id, name)
 
         groups = _build_groups(search, diff_aggregates, rules, manual.groups, names)
-        unassigned = _unassigned_rows(osv_result.consolidated.rows, rows, manual, search.used_ids)
+        unassigned = _unassigned_rows(decision_rows, manual, distribution)
         return VgoAssignmentCandidatesOut(groups=groups, unassigned=unassigned)
+
+
+class VgoAssignmentCrudService:
+    """Ручная правка назначений исходных строк ОСВ одной версии."""
+
+    def __init__(
+        self, session: AsyncSession, audit: RequestAudit, scope: VisibilityScope, osv_service: ConsolidatedOsvService
+    ) -> None:
+        self.session = session
+        self.audit = audit
+        self.scope = scope
+        self.osv_service = osv_service
+
+    async def _check_rows(self, version_id: uuid.UUID, entry_ids: list[uuid.UUID]) -> None:
+        """Проверить, что ID образуют одну или несколько полных строк свода."""
+        osv_result = await self.osv_service.build_with_members(version_id)
+        requested = set(entry_ids)
+        matched_rows = [row for row in osv_result.consolidated.rows if requested.intersection(row.entry_ids)]
+        matched_ids = {entry_id for row in matched_rows for entry_id in row.entry_ids}
+        if requested != matched_ids:
+            raise DomainValidationError(
+                "INVALID_OSV_ENTRY_IDS", "entry_ids должны содержать все исходные ID выбранных строк сводной ОСВ"
+            )
+
+    async def attach_rule(self, version_id: uuid.UUID, data: VgoAssignmentAttachRuleIn) -> list[VgoAssignmentOut]:
+        """Привязать выбранные полные строки сводной ОСВ версии к правилу."""
+        version, _ = await lock_editable_version(self.session, self.scope, version_id, self.audit.user_id)
+        rule = await self.session.get(EliminationRule, data.rule_id)
+        if rule is None or rule.version_no != version.elimination_rules_version:
+            raise DomainValidationError("INVALID_VGO_RULE", "правило не относится к версии консолидации")
+        await self._check_rows(version_id, data.entry_ids)
+        existing = await vgo_assignments_repo.for_entries(self.session, version_id, rule.statement_type, data.entry_ids)
+        return await self._persist_rule(version_id, rule, data.entry_ids, existing)
+
+    async def set_exclusion(self, version_id: uuid.UUID, data: VgoAssignmentExcludeIn) -> list[VgoAssignmentOut]:
+        """Установить признак исключения, не меняя назначенное правило."""
+        await lock_editable_version(self.session, self.scope, version_id, self.audit.user_id)
+        await self._check_rows(version_id, data.entry_ids)
+        existing = await vgo_assignments_repo.for_entries(self.session, version_id, data.statement_type, data.entry_ids)
+        if data.is_exclude and any(item.rule_id is not None for item in existing.values()):
+            raise DomainValidationError("VGO_RULE_ASSIGNED", "строку с правилом нельзя исключить из КФО")
+
+        before = {str(entry_id): _assignment_snapshot(item) for entry_id, item in existing.items()}
+        result = []
+        for entry_id in data.entry_ids:
+            item = existing.get(entry_id)
+            if item is None:
+                item = VgoAssignment(
+                    consolidation_version_id=version_id,
+                    statement_type=data.statement_type,
+                    osv_entry_id=entry_id,
+                    action="UNASSIGNED",
+                    is_exclude=data.is_exclude,
+                    created_by=self.audit.user_id,
+                )
+                self.session.add(item)
+            else:
+                item.is_exclude = data.is_exclude
+                item.updated_by = self.audit.user_id
+                item.updated_at = utc_now()
+            result.append(item)
+        await self.session.flush()
+        await self._audit(version_id, before, {str(item.osv_entry_id): _assignment_snapshot(item) for item in result})
+        return [VgoAssignmentOut.model_validate(item, from_attributes=True) for item in result]
+
+    async def detach_rule(self, version_id: uuid.UUID, data: VgoAssignmentDetachRuleIn) -> list[VgoAssignmentOut]:
+        """Снять правило и создать отсутствующие записи как исключённые."""
+        await lock_editable_version(self.session, self.scope, version_id, self.audit.user_id)
+        await self._check_rows(version_id, data.entry_ids)
+        existing = await vgo_assignments_repo.for_entries(self.session, version_id, data.statement_type, data.entry_ids)
+
+        before = {str(entry_id): _assignment_snapshot(item) for entry_id, item in existing.items()}
+        result = []
+        for entry_id in data.entry_ids:
+            item = existing.get(entry_id)
+            if item is None:
+                item = VgoAssignment(
+                    consolidation_version_id=version_id,
+                    statement_type=data.statement_type,
+                    osv_entry_id=entry_id,
+                    created_by=self.audit.user_id,
+                )
+                self.session.add(item)
+            else:
+                item.updated_by = self.audit.user_id
+                item.updated_at = utc_now()
+            item.action = "UNASSIGNED"
+            item.rule_id = None
+            item.is_exclude = True
+            result.append(item)
+        await self.session.flush()
+        await self._audit(version_id, before, {str(item.osv_entry_id): _assignment_snapshot(item) for item in result})
+        return [VgoAssignmentOut.model_validate(item, from_attributes=True) for item in result]
+
+    async def _persist_rule(
+        self,
+        version_id: uuid.UUID,
+        rule: EliminationRule,
+        entry_ids: list[uuid.UUID],
+        existing: dict[uuid.UUID, VgoAssignment],
+    ) -> list[VgoAssignmentOut]:
+        """Сохранить привязку к правилу и аудит для проверенных ID."""
+
+        before = {str(entry_id): _assignment_snapshot(item) for entry_id, item in existing.items()}
+        result = []
+        for entry_id in entry_ids:
+            item = existing.get(entry_id)
+            if item is None:
+                item = VgoAssignment(
+                    consolidation_version_id=version_id,
+                    statement_type=rule.statement_type,
+                    osv_entry_id=entry_id,
+                    created_by=self.audit.user_id,
+                )
+                self.session.add(item)
+            else:
+                item.updated_by = self.audit.user_id
+                item.updated_at = utc_now()
+            item.action = "ASSIGN"
+            item.rule_id = rule.id
+            item.is_exclude = False
+            result.append(item)
+        await self.session.flush()
+        await self._audit(version_id, before, {str(item.osv_entry_id): _assignment_snapshot(item) for item in result})
+        return [VgoAssignmentOut.model_validate(item, from_attributes=True) for item in result]
+
+    async def _audit(self, version_id: uuid.UUID, before: Mapping[str, object], after: Mapping[str, object]) -> None:
+        await self.audit.record(
+            action=AuditAction.ASSIGNMENT_CHANGE,
+            section=AuditSection.KFO,
+            object_type="vgo_assignment",
+            object_id=version_id,
+            consolidation_version_id=version_id,
+            before_value={"entries": before},
+            after_value={"entries": after},
+        )
+
+
+def _assignment_snapshot(item: VgoAssignment) -> dict[str, object]:
+    return {"action": item.action, "rule_id": item.rule_id, "is_exclude": item.is_exclude}

api/tests/test_api_vgo_assignments.py

@@ -0,0 +1,325 @@
+"""Ручные операции над назначениями ВГО для полных строк сводной ОСВ."""
+
+import uuid
+from http import HTTPStatus
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, Mock
+
+import pytest
+from pydantic import ValidationError
+from sqlalchemy import delete, select, text
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from src.api.endpoints.consolidations.vgo_assignment import router
+from src.common.clock import utc_now
+from src.common.exceptions import DomainValidationError
+from src.models.access import AppUser
+from src.models.accounting import EliminationRule
+from src.models.base import SCHEMA
+from src.models.consolidation import ConsolidationDocument
+from src.models.directories import RegionalOffice
+from src.models.enums import ConsolidationStatus, PeriodQuarter
+from src.models.kfo import VgoAssignment
+from src.models.osv import OsvEntry
+from src.schemas.consolidations import VgoAssignmentAttachRuleIn, VgoAssignmentDetachRuleIn, VgoAssignmentExcludeIn
+from src.services import vgo
+from src.services.vgo import VgoAssignmentCrudService
+from tests.conftest import AuthClientFactory
+from tests.factories import make_company_group, make_confirmed_osv, make_consolidation, make_legal_entity, make_lock
+
+USER_EMAIL = "ka_rf@armka.local"
+
+
+def test_assignment_routes_are_limited_to_requested_actions() -> None:
+    routes = {(method, route.path) for route in router.routes for method in route.methods}
+    prefix = "/consolidations/versions/{version_id}/vgo-assignments"
+    assert routes == {
+        ("GET", f"{prefix}/distribute"),
+        ("PUT", f"{prefix}/rule"),
+        ("DELETE", f"{prefix}/rule"),
+        ("PUT", f"{prefix}/exclude"),
+    }
+
+
+def test_inputs_require_unique_ids() -> None:
+    entry_id = uuid.uuid4()
+    with pytest.raises(ValidationError, match="entry_ids не должны повторяться"):
+        VgoAssignmentAttachRuleIn(entry_ids=[entry_id, entry_id], rule_id=uuid.uuid4())
+    with pytest.raises(ValidationError, match="entry_ids не должны повторяться"):
+        VgoAssignmentDetachRuleIn(statement_type="balance", entry_ids=[entry_id, entry_id])
+    with pytest.raises(ValidationError, match="entry_ids не должны повторяться"):
+        VgoAssignmentExcludeIn(statement_type="balance", entry_ids=[entry_id, entry_id], is_exclude=True)
+
+
+async def test_write_requires_complete_consolidated_row() -> None:
+    entry_a, entry_b = uuid.uuid4(), uuid.uuid4()
+    row = SimpleNamespace(entry_ids=(entry_a, entry_b))
+    osv_result = SimpleNamespace(consolidated=SimpleNamespace(rows=[row]))
+
+    service = VgoAssignmentCrudService(
+        session=AsyncMock(),
+        audit=AsyncMock(),
+        scope=SimpleNamespace(),
+        osv_service=SimpleNamespace(build_with_members=AsyncMock(return_value=osv_result)),
+    )
+    version_id = uuid.uuid4()
+    with pytest.raises(DomainValidationError, match="все исходные ID"):
+        await service._check_rows(version_id, [entry_a])
+    await service._check_rows(version_id, [entry_a, entry_b])
+
+
+async def test_cannot_exclude_entries_with_rule(monkeypatch: pytest.MonkeyPatch) -> None:
+    entry_id = uuid.uuid4()
+    session = AsyncMock()
+    service = VgoAssignmentCrudService(
+        session=session,
+        audit=SimpleNamespace(user_id=uuid.uuid4()),
+        scope=SimpleNamespace(),
+        osv_service=SimpleNamespace(
+            build_with_members=AsyncMock(
+                return_value=SimpleNamespace(
+                    consolidated=SimpleNamespace(rows=[SimpleNamespace(entry_ids=(entry_id,))])
+                )
+            )
+        ),
+    )
+    monkeypatch.setattr(vgo, "lock_editable_version", AsyncMock())
+    monkeypatch.setattr(
+        vgo.vgo_assignments_repo,
+        "for_entries",
+        AsyncMock(return_value={entry_id: SimpleNamespace(rule_id=uuid.uuid4())}),
+    )
+    data = VgoAssignmentExcludeIn(statement_type="balance", entry_ids=[entry_id], is_exclude=True)
+    with pytest.raises(DomainValidationError, match="нельзя исключить из КФО"):
+        await service.set_exclusion(uuid.uuid4(), data)
+    session.flush.assert_not_awaited()
+
+
+async def test_detach_rule_keeps_assignment(monkeypatch: pytest.MonkeyPatch) -> None:
+    version_id, entry_id, user_id = uuid.uuid4(), uuid.uuid4(), uuid.uuid4()
+    item = VgoAssignment(
+        id=uuid.uuid4(),
+        consolidation_version_id=version_id,
+        statement_type="balance",
+        osv_entry_id=entry_id,
+        action="ASSIGN",
+        rule_id=uuid.uuid4(),
+        is_exclude=False,
+        created_by=user_id,
+        created_at=utc_now(),
+    )
+    session = AsyncMock()
+    service = VgoAssignmentCrudService(
+        session=session,
+        audit=SimpleNamespace(user_id=user_id),
+        scope=SimpleNamespace(),
+        osv_service=SimpleNamespace(
+            build_with_members=AsyncMock(
+                return_value=SimpleNamespace(
+                    consolidated=SimpleNamespace(rows=[SimpleNamespace(entry_ids=(entry_id,))])
+                )
+            )
+        ),
+    )
+    service._audit = AsyncMock()
+    monkeypatch.setattr(vgo, "lock_editable_version", AsyncMock())
+    monkeypatch.setattr(vgo.vgo_assignments_repo, "for_entries", AsyncMock(return_value={entry_id: item}))
+    result = await service.detach_rule(
+        version_id, VgoAssignmentDetachRuleIn(statement_type="balance", entry_ids=[entry_id])
+    )
+    assert result[0].id == item.id
+    assert result[0].action == "UNASSIGNED"
+    assert result[0].rule_id is None
+    assert result[0].is_exclude is True
+    session.delete.assert_not_awaited()
+    session.flush.assert_awaited_once()
+
+
+async def test_detach_rule_creates_missing_assignment(monkeypatch: pytest.MonkeyPatch) -> None:
+    version_id, entry_id, user_id = uuid.uuid4(), uuid.uuid4(), uuid.uuid4()
+    session = SimpleNamespace(add=Mock(), flush=AsyncMock())
+    service = VgoAssignmentCrudService(
+        session=session,
+        audit=SimpleNamespace(user_id=user_id),
+        scope=SimpleNamespace(),
+        osv_service=SimpleNamespace(
+            build_with_members=AsyncMock(
+                return_value=SimpleNamespace(
+                    consolidated=SimpleNamespace(rows=[SimpleNamespace(entry_ids=(entry_id,))])
+                )
+            )
+        ),
+    )
+    service._audit = AsyncMock()
+    monkeypatch.setattr(vgo, "lock_editable_version", AsyncMock())
+    monkeypatch.setattr(vgo.vgo_assignments_repo, "for_entries", AsyncMock(return_value={}))
+    monkeypatch.setattr(vgo.VgoAssignmentOut, "model_validate", lambda item, from_attributes: item)
+
+    result = await service.detach_rule(
+        version_id, VgoAssignmentDetachRuleIn(statement_type="balance", entry_ids=[entry_id])
+    )
+    item = result[0]
+    session.add.assert_called_once_with(item)
+    assert item.osv_entry_id == entry_id
+    assert item.rule_id is None
+    assert item.action == "UNASSIGNED"
+    assert item.is_exclude is True
+    session.flush.assert_awaited_once()
+
+
+async def test_exclusion_creates_missing_assignment(monkeypatch: pytest.MonkeyPatch) -> None:
+    version_id, entry_id, user_id = uuid.uuid4(), uuid.uuid4(), uuid.uuid4()
+    session = SimpleNamespace(add=Mock(), flush=AsyncMock())
+    service = VgoAssignmentCrudService(
+        session=session,
+        audit=SimpleNamespace(user_id=user_id),
+        scope=SimpleNamespace(),
+        osv_service=SimpleNamespace(
+            build_with_members=AsyncMock(
+                return_value=SimpleNamespace(
+                    consolidated=SimpleNamespace(rows=[SimpleNamespace(entry_ids=(entry_id,))])
+                )
+            )
+        ),
+    )
+    service._audit = AsyncMock()
+    monkeypatch.setattr(vgo, "lock_editable_version", AsyncMock())
+    monkeypatch.setattr(vgo.vgo_assignments_repo, "for_entries", AsyncMock(return_value={}))
+    monkeypatch.setattr(vgo.VgoAssignmentOut, "model_validate", lambda item, from_attributes: item)
+
+    result = await service.set_exclusion(
+        version_id, VgoAssignmentExcludeIn(statement_type="balance", entry_ids=[entry_id], is_exclude=True)
+    )
+    item = result[0]
+    session.add.assert_called_once_with(item)
+    assert item.osv_entry_id == entry_id
+    assert item.rule_id is None
+    assert item.action == "UNASSIGNED"
+    assert item.is_exclude is True
+    session.flush.assert_awaited_once()
+
+
+async def test_vgo_assignment_integration(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
+    constraints = (
+        await seeded.execute(
+            text(
+                "SELECT conname, contype::text, confdeltype::text "
+                "FROM pg_constraint WHERE conrelid = to_regclass(:relation)"
+            ),
+            {"relation": f"{SCHEMA}.vgo_assignment"},
+        )
+    ).all()
+    names = {name for name, _, _ in constraints}
+    assert {
+        "ck_vgo_assignment_action_values",
+        "ck_vgo_assignment_action_rule_consistency",
+        "ck_vgo_assignment_rule_not_excluded",
+    } <= names
+    assert ("fk_vgo_assignment_entry", "f", "c") in constraints
+
+    user = (await seeded.scalars(select(AppUser).where(AppUser.email == USER_EMAIL))).one()
+    office = (await seeded.scalars(select(RegionalOffice).where(RegionalOffice.id == user.regional_office_id))).one()
+    group = await make_company_group(seeded, office=office)
+    company_a = await make_legal_entity(seeded, name="Компания A", group=group)
+    company_b = await make_legal_entity(seeded, name="Компания B", group=group)
+    _, version = await make_consolidation(
+        seeded,
+        group=group,
+        year=2024,
+        quarter=PeriodQuarter.Q2,
+        members=[company_a, company_b],
+        initiator=user,
+        status=ConsolidationStatus.DATA_CORRECTION,
+    )
+    await make_lock(seeded, version=version, user=user)
+    documents = []
+    for quarter in (PeriodQuarter.Q1, PeriodQuarter.Q2):
+        document = await make_confirmed_osv(
+            seeded,
+            entity=company_a,
+            account="60",
+            year=2024,
+            quarter=quarter,
+            is_cumulative=False,
+            rows=[{"name": "Компания B", "counterparty_id": company_b.id, "closing_debit": 10}],
+        )
+        documents.append(document)
+        seeded.add(ConsolidationDocument(consolidation_version_id=version.id, document_id=document.id))
+    await seeded.flush()
+    entry_ids = list(
+        (
+            await seeded.scalars(
+                select(OsvEntry.id).where(OsvEntry.document_id.in_([document.id for document in documents]))
+            )
+        ).all()
+    )
+    client = await auth_client(USER_EMAIL)
+    await _check_assignment_actions(client, seeded, version, entry_ids)
+    distribute_url = f"/consolidations/versions/{version.id}/vgo-assignments/distribute"
+    view_only = await auth_client("curator_go@armka.local")
+    distribution = await view_only.get(distribute_url)
+    assert distribution.status_code == HTTPStatus.OK, distribution.text
+    assert "match_stored" not in distribution.text
+    forbidden = await (await auth_client("admin@armka.local")).get(distribute_url)
+    assert forbidden.status_code == HTTPStatus.FORBIDDEN
+    await seeded.execute(delete(OsvEntry).where(OsvEntry.id == entry_ids[0]))
+    assert await seeded.scalar(select(VgoAssignment.id).where(VgoAssignment.osv_entry_id == entry_ids[0])) is None
+    assert await seeded.scalar(select(VgoAssignment.id).where(VgoAssignment.osv_entry_id == entry_ids[1])) is not None
+
+
+async def _check_assignment_actions(client, seeded, version, entry_ids) -> None:
+    url = f"/consolidations/versions/{version.id}/vgo-assignments"
+    entry_id_strings = [str(entry_id) for entry_id in entry_ids]
+    rule = (
+        await seeded.scalars(
+            select(EliminationRule).where(
+                EliminationRule.version_no == version.elimination_rules_version,
+                EliminationRule.statement_type == "balance",
+            )
+        )
+    ).first()
+    assert rule is not None
+    attach_url = f"{url}/rule"
+    attach_body = {"entry_ids": entry_id_strings, "rule_id": str(rule.id)}
+    exclude_url = f"{url}/exclude"
+    exclude_body = {"statement_type": "balance", "entry_ids": entry_id_strings, "is_exclude": True}
+
+    partial = await client.put(exclude_url, json={**exclude_body, "entry_ids": entry_id_strings[:1]})
+    assert partial.status_code == HTTPStatus.UNPROCESSABLE_ENTITY
+    excluded = await client.put(exclude_url, json=exclude_body)
+    assert excluded.status_code == HTTPStatus.OK, excluded.text
+    assert {item["osv_entry_id"] for item in excluded.json()} == set(entry_id_strings)
+    assert all(item["is_exclude"] and item["rule_id"] is None for item in excluded.json())
+    included = await client.put(exclude_url, json={**exclude_body, "is_exclude": False})
+    assert included.status_code == HTTPStatus.OK, included.text
+    assert all(not item["is_exclude"] for item in included.json())
+
+    attached = await client.put(attach_url, json=attach_body)
+    assert attached.status_code == HTTPStatus.OK, attached.text
+    assert attached.json()[0]["action"] == "ASSIGN"
+    assert attached.json()[0]["rule_id"] == str(rule.id)
+    assert attached.json()[0]["is_exclude"] is False
+    assert {item["id"] for item in attached.json()} == {item["id"] for item in excluded.json()}
+    assert len(attached.json()) == len(entry_ids)
+
+    detached = await client.request(
+        "DELETE", attach_url, json={"statement_type": "balance", "entry_ids": entry_id_strings}
+    )
+    assert detached.status_code == HTTPStatus.OK, detached.text
+    assert {item["id"] for item in detached.json()} == {item["id"] for item in attached.json()}
+    assert all(item["action"] == "UNASSIGNED" and item["rule_id"] is None for item in detached.json())
+    assert all(item["is_exclude"] for item in detached.json())
+    excluded_after_detach = await client.put(exclude_url, json=exclude_body)
+    assert excluded_after_detach.status_code == HTTPStatus.OK, excluded_after_detach.text
+    attached = await client.put(attach_url, json=attach_body)
+    assert attached.status_code == HTTPStatus.OK, attached.text
+    assert all(not item["is_exclude"] for item in attached.json())
+
+    forbidden = await client.put(exclude_url, json=exclude_body)
+    assert forbidden.status_code == HTTPStatus.UNPROCESSABLE_ENTITY
+    assert forbidden.json()["detail"]["code"] == "VGO_RULE_ASSIGNED"
+    partial = await client.put(attach_url, json={"entry_ids": [str(entry_ids[0])], "rule_id": str(rule.id)})
+    assert partial.status_code == HTTPStatus.UNPROCESSABLE_ENTITY
+
+    invalid_id = await client.put(attach_url, json={"entry_ids": [str(uuid.uuid4())], "rule_id": str(rule.id)})
+    assert invalid_id.status_code == HTTPStatus.UNPROCESSABLE_ENTITY

api/tests/test_vgo.py

@@ -9,6 +9,7 @@ import pytest
 
 from src.common.exceptions import DomainValidationError
 from src.domain.consolidated_osv import CounterpartyType
+from src.domain.vgo_distribution import DistributionInput, distribute_vgo_rows
 from src.models.enums import StatementType
 from src.schemas.consolidations import EliminationMatchedPairOut
 from src.services import vgo
@@ -43,6 +44,7 @@ def _row(entry_id: uuid.UUID, company_a: uuid.UUID, company_b: uuid.UUID, index:
         legal_entity_id=company_a,
         legal_entity_name="A",
         legal_entity_inn="1234567890",
+        account_id=None,
         account_code="76",
         account_root="76",
         period="2026 Q1",
@@ -70,6 +72,82 @@ def _members(company_a: uuid.UUID, company_b: uuid.UUID) -> dict[uuid.UUID, Simp
     return {company_a: SimpleNamespace(name="A"), company_b: SimpleNamespace(name="B")}
 
 
+def test_partially_overlapping_diff_aggregates_remain_unassigned() -> None:
+    rule_6, rule_7, company_a, company_b, a_76, b_60, b_76, extra_a, extra_b = (uuid.uuid4() for _ in range(9))
+    first = _pair(rule_6, company_a, company_b, a_76, b_60, amount_b="9")
+    second = _pair(rule_7, company_a, company_b, a_76, b_76, amount_b="9")
+    independent = _pair(rule_6, company_a, company_b, extra_a, extra_b, amount_b="9")
+    distribution = distribute_vgo_rows(
+        DistributionInput(
+            rows=[],
+            exact_ids=set(),
+            diff_ids=[vgo._ids_from_aggregate(pair) for pair in (first, second, independent)],
+            manual_keys=set(),
+            owners={},
+            members=set(),
+        )
+    )
+
+    assert distribution.selected_diff == (2,)
+    assert distribution.used_ids == {extra_a, extra_b}
+
+
+async def test_unpaired_row_with_unique_rule_account_is_auto_assigned(monkeypatch: pytest.MonkeyPatch) -> None:
+    version_id, company_a, company_b, rule_1, rule_2, unique_account, shared_account = (uuid.uuid4() for _ in range(7))
+    unique_row = _row(uuid.uuid4(), company_a, company_b, 1)
+    unique_row.account_id = unique_account
+    unique_row.closing_debit = Decimal("7")
+    shared_row = _row(uuid.uuid4(), company_a, company_b, 2)
+    shared_row.account_id = shared_account
+    shared_row.closing_debit = Decimal("3")
+    outside_row = _row(uuid.uuid4(), company_a, company_b, 3)
+    outside_row.account_id = uuid.uuid4()
+    rules = [
+        SimpleNamespace(
+            id=rule_id,
+            rule_no=number,
+            statement_type=StatementType.balance,
+            measure="closing",
+            side_1="D",
+            article_1_line="A",
+            article_2_line="B",
+        )
+        for number, rule_id in enumerate((rule_1, rule_2), start=1)
+    ]
+    osv_result = SimpleNamespace(
+        version=SimpleNamespace(id=version_id, elimination_rules_version=2),
+        consolidated=SimpleNamespace(rows=[unique_row, shared_row, outside_row]),
+        members=_members(company_a, company_b),
+    )
+    monkeypatch.setattr(vgo.elimination_rules_repo, "list_rules", AsyncMock(return_value=rules))
+    monkeypatch.setattr(vgo.vgo_assignments_repo, "list_assignments", AsyncMock(return_value=[]))
+    monkeypatch.setattr(vgo.directories_repo, "list_accounts", AsyncMock(return_value=[]))
+
+    async def resolve(rule_id: uuid.UUID, *_args: object) -> tuple[SimpleNamespace, SimpleNamespace]:
+        account_ids = {shared_account, unique_account} if rule_id == rule_1 else {shared_account}
+        selection = SimpleNamespace(company_1_account_ids=frozenset(account_ids), company_2_account_ids=frozenset())
+        return selection, rules[0]
+
+    matrix = SimpleNamespace(
+        _resolve_preset_selection=AsyncMock(side_effect=resolve),
+        build_pairs=AsyncMock(return_value=SimpleNamespace(matched=[], with_discrepancy=[])),
+    )
+    service = vgo.VgoService(
+        AsyncMock(), SimpleNamespace(build_with_members=AsyncMock(return_value=osv_result)), matrix
+    )
+
+    response = await service.assignment_candidates(version_id)
+
+    assert len(response.groups) == 1
+    assert response.groups[0].rule_id == rule_1
+    assert response.groups[0].amount_a == Decimal("7.00")
+    assert response.groups[0].manual[0].entry_ids == [unique_row.entry_ids[0]]
+    assert response.groups[0].manual[0].is_auto_assigned is True
+    assert response.groups[0].manual[0].is_exclude is False
+    assert [row.entry_ids for row in response.unassigned] == [[shared_row.entry_ids[0]]]
+    assert "match_stored" not in response.unassigned[0].model_dump()
+
+
 async def test_pairs_are_grouped_and_entry_ids_are_used_once(monkeypatch: pytest.MonkeyPatch) -> None:
     version_id, company_a, company_b, rule_1, rule_2 = (uuid.uuid4() for _ in range(5))
     entry_ids = [uuid.uuid4() for _ in range(11)]
@@ -97,7 +175,7 @@ async def test_pairs_are_grouped_and_entry_ids_are_used_once(monkeypatch: pytest
             article_2_line="Статья B2",
         ),
     ]
-    monkeypatch.setattr(vgo.elimination_rules_repo, "list_rules", AsyncMock(return_value=rules))
+    monkeypatch.setattr(vgo.elimination_rules_repo, "list_rules", AsyncMock(return_value=list(reversed(rules))))
     monkeypatch.setattr(vgo.vgo_assignments_repo, "list_assignments", AsyncMock(return_value=[]))
     exact_1 = _pair(rule_1, company_a, company_b, entry_ids[0], entry_ids[1])
     exact_1_more = _pair(rule_1, company_a, company_b, entry_ids[9], entry_ids[10])
@@ -126,7 +204,18 @@ async def test_pairs_are_grouped_and_entry_ids_are_used_once(monkeypatch: pytest
         SimpleNamespace(build_with_members=AsyncMock(return_value=osv_result)),
         SimpleNamespace(build_pairs=AsyncMock(side_effect=build_pairs)),
     )
+    account_id = uuid.uuid4()
+    for row in rows:
+        row.account_id = account_id
+    service._rule_accounts = AsyncMock(return_value={account_id: frozenset({rule_1, rule_2})})
     response = await service.assignment_candidates(version_id)
+    calls = service._matrix_service.build_pairs.await_args_list
+    assert [(call.args[1], call.args[2], call.args[3]) for call in calls] == [
+        (rule_1, company_a, company_b),
+        (rule_1, company_b, company_a),
+        (rule_2, company_a, company_b),
+        (rule_2, company_b, company_a),
+    ]
     groups = response.groups
 
     assert [(group.rule_id, group.company_a_id, group.company_b_id) for group in groups] == [
@@ -161,7 +250,7 @@ async def test_pairs_are_grouped_and_entry_ids_are_used_once(monkeypatch: pytest
     ]
 
 
-async def test_manual_assignments_override_auto_and_keep_exclusion_flag(monkeypatch: pytest.MonkeyPatch) -> None:
+async def test_manual_assignments_override_auto_and_show_exclusion_flag(monkeypatch: pytest.MonkeyPatch) -> None:
     version_id, company_a, company_b, rule_id = (uuid.uuid4() for _ in range(4))
     entry_ids = [uuid.uuid4() for _ in range(4)]
     rows = [_row(entry_id, company_a, company_b, index) for index, entry_id in enumerate(entry_ids)]
@@ -183,7 +272,7 @@ async def test_manual_assignments_override_auto_and_keep_exclusion_flag(monkeypa
     )
     assignments = [
         SimpleNamespace(osv_entry_id=entry_ids[0], rule_id=rule_id, action="ASSIGN", is_exclude=False),
-        SimpleNamespace(osv_entry_id=entry_ids[1], rule_id=rule_id, action="ASSIGN", is_exclude=True),
+        SimpleNamespace(osv_entry_id=entry_ids[1], rule_id=rule_id, action="ASSIGN", is_exclude=False),
         SimpleNamespace(osv_entry_id=entry_ids[2], rule_id=None, action="UNASSIGNED", is_exclude=True),
     ]
     monkeypatch.setattr(vgo.elimination_rules_repo, "list_rules", AsyncMock(return_value=[rule]))
@@ -208,38 +297,45 @@ async def test_manual_assignments_override_auto_and_keep_exclusion_flag(monkeypa
         company_a,
         company_b,
     )
-    assert (response.groups[0].amount_a, response.groups[0].amount_b) == (Decimal("10.00"), Decimal("0.00"))
-    assert [(row.osv_entry_id, row.is_exclude) for row in response.groups[0].manual] == [
-        (entry_ids[0], False),
-        (entry_ids[1], True),
+    assert (response.groups[0].amount_a, response.groups[0].amount_b) == (Decimal("30.00"), Decimal("0.00"))
+    assert [(row.entry_ids, row.is_exclude) for row in response.groups[0].manual] == [
+        ([entry_ids[0]], False),
+        ([entry_ids[1]], False),
     ]
     assert "manual" in response.model_dump()["groups"][0]
     assert [(row.entry_ids, row.action, row.is_exclude) for row in response.unassigned] == [
         ([entry_ids[2]], "UNASSIGNED", True),
-        ([entry_ids[3]], None, None),
     ]
 
 
-async def test_glued_row_requires_assignments_for_all_source_entries(monkeypatch: pytest.MonkeyPatch) -> None:
+async def test_partial_manual_assignment_stays_unassigned_and_is_not_searched(monkeypatch: pytest.MonkeyPatch) -> None:
     version_id, company_a, company_b, rule_id, entry_a, entry_b = (uuid.uuid4() for _ in range(6))
     row = _row(entry_a, company_a, company_b, 1)
     row.entry_ids = (entry_a, entry_b)
+    free_row = _row(uuid.uuid4(), company_a, company_b, 2)
     osv_result = SimpleNamespace(
         version=SimpleNamespace(elimination_rules_version=2),
-        consolidated=SimpleNamespace(rows=[row]),
+        consolidated=SimpleNamespace(rows=[row, free_row]),
         members=_members(company_a, company_b),
     )
     assignment = SimpleNamespace(osv_entry_id=entry_a, rule_id=rule_id, action="ASSIGN", is_exclude=False)
-    monkeypatch.setattr(vgo.elimination_rules_repo, "list_rules", AsyncMock(return_value=[]))
+    rule = SimpleNamespace(id=rule_id, rule_no=1, statement_type=StatementType.balance, measure="closing")
+    monkeypatch.setattr(vgo.elimination_rules_repo, "list_rules", AsyncMock(return_value=[rule]))
     monkeypatch.setattr(vgo.vgo_assignments_repo, "list_assignments", AsyncMock(return_value=[assignment]))
+    matrix = SimpleNamespace(build_pairs=AsyncMock(return_value=SimpleNamespace(matched=[], with_discrepancy=[])))
     service = vgo.VgoService(
         AsyncMock(),
         SimpleNamespace(build_with_members=AsyncMock(return_value=osv_result)),
-        SimpleNamespace(build_pairs=AsyncMock()),
+        matrix,
     )
 
-    with pytest.raises(DomainValidationError, match="для остальных нет"):
-        await service.assignment_candidates(version_id)
+    response = await service.assignment_candidates(version_id)
+
+    assert response.groups == []
+    assert [item.entry_ids for item in response.unassigned] == [[entry_a, entry_b]]
+    assert response.unassigned[0].unassigned_reason == "PARTIAL_MANUAL_ASSIGNMENT"
+    assert response.unassigned[0].action is None
+    assert all(call.args[4] == [free_row] for call in matrix.build_pairs.await_args_list)
 
 
 async def test_glued_row_with_same_assignments_is_counted_once(monkeypatch: pytest.MonkeyPatch) -> None:
@@ -279,7 +375,7 @@ async def test_glued_row_with_same_assignments_is_counted_once(monkeypatch: pyte
     assert response.groups[0].amount_a == Decimal("10.00")
     assert len(response.groups[0].manual) == 1
     assert response.groups[0].manual[0].entry_ids == [entry_a, entry_b]
-    assert response.groups[0].manual[0].osv_entry_id is None
+    assert "osv_entry_id" not in response.groups[0].manual[0].model_dump()
     assert response.unassigned == []
     assert all(call.args[4] == [] for call in matrix_service.build_pairs.await_args_list)