252 lines
10 KiB
Python
252 lines
10 KiB
Python
"""Manual revision application service."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
|
|
from fastapi import HTTPException, status
|
|
from sqlalchemy import func, select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.api.schemas.revision import (
|
|
RegisterRevisionRequest,
|
|
RegisterRevisionResponse,
|
|
RevisionChainItem,
|
|
RevisionChainResponse,
|
|
)
|
|
from app.domain.enums import AssetType, OrderStatus, ReviewTaskStatus, WorkflowStepName
|
|
from app.infra.db.models.asset import AssetORM
|
|
from app.infra.db.models.order import OrderORM
|
|
from app.infra.db.models.review_task import ReviewTaskORM
|
|
from app.infra.db.models.workflow_run import WorkflowRunORM
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class RevisionSnapshot:
|
|
"""Revision summary used by the order, queue, and workflow responses."""
|
|
|
|
current_revision_asset_id: int | None
|
|
current_revision_version: int | None
|
|
latest_revision_asset_id: int | None
|
|
latest_revision_version: int | None
|
|
revision_count: int
|
|
review_task_status: ReviewTaskStatus | None
|
|
pending_manual_confirm: bool
|
|
root_asset_id: int | None
|
|
|
|
|
|
class RevisionService:
|
|
"""Application service for manual revision registration and lookup."""
|
|
|
|
async def get_revision_snapshot(self, session: AsyncSession, order_id: int) -> RevisionSnapshot:
|
|
"""Return the current revision summary for an order."""
|
|
|
|
await self._require_order(session, order_id)
|
|
|
|
revision_result = await session.execute(
|
|
select(AssetORM)
|
|
.where(
|
|
AssetORM.order_id == order_id,
|
|
AssetORM.asset_type == AssetType.MANUAL_REVISION,
|
|
)
|
|
.order_by(AssetORM.version_no.asc(), AssetORM.created_at.asc(), AssetORM.id.asc())
|
|
)
|
|
revisions = revision_result.scalars().all()
|
|
latest_revision = revisions[-1] if revisions else None
|
|
|
|
review_result = await session.execute(
|
|
select(ReviewTaskORM)
|
|
.where(ReviewTaskORM.order_id == order_id)
|
|
.order_by(ReviewTaskORM.created_at.desc(), ReviewTaskORM.id.desc())
|
|
)
|
|
latest_review_task = review_result.scalars().first()
|
|
|
|
return RevisionSnapshot(
|
|
current_revision_asset_id=latest_revision.id if latest_revision else None,
|
|
current_revision_version=latest_revision.version_no if latest_revision else None,
|
|
latest_revision_asset_id=latest_revision.id if latest_revision else None,
|
|
latest_revision_version=latest_revision.version_no if latest_revision else None,
|
|
revision_count=len(revisions),
|
|
review_task_status=latest_review_task.status if latest_review_task else None,
|
|
pending_manual_confirm=bool(
|
|
latest_review_task and latest_review_task.status == ReviewTaskStatus.REVISION_UPLOADED
|
|
),
|
|
root_asset_id=latest_revision.root_asset_id if latest_revision else None,
|
|
)
|
|
|
|
async def register_revision(
|
|
self,
|
|
session: AsyncSession,
|
|
order_id: int,
|
|
payload: RegisterRevisionRequest,
|
|
) -> RegisterRevisionResponse:
|
|
"""Register an offline manual revision asset for a waiting-review order."""
|
|
|
|
order = await self._require_order(session, order_id)
|
|
if order.status != OrderStatus.WAITING_REVIEW:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_409_CONFLICT,
|
|
detail="Order is not waiting for review",
|
|
)
|
|
|
|
workflow_run = await self._require_workflow_run(session, order_id)
|
|
review_task = await self._get_active_review_task(session, order_id)
|
|
|
|
parent_asset = await session.get(AssetORM, payload.parent_asset_id)
|
|
if parent_asset is None or parent_asset.order_id != order_id:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="Parent asset does not belong to the order",
|
|
)
|
|
|
|
root_asset_id = parent_asset.root_asset_id or parent_asset.id
|
|
version_result = await session.execute(
|
|
select(func.max(AssetORM.version_no)).where(
|
|
AssetORM.order_id == order_id,
|
|
AssetORM.asset_type == AssetType.MANUAL_REVISION,
|
|
AssetORM.root_asset_id == root_asset_id,
|
|
)
|
|
)
|
|
latest_version = version_result.scalar_one()
|
|
if parent_asset.asset_type != AssetType.MANUAL_REVISION and latest_version:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_409_CONFLICT,
|
|
detail="Parent asset is not the current revision version",
|
|
)
|
|
if parent_asset.asset_type == AssetType.MANUAL_REVISION and not parent_asset.is_current_version:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_409_CONFLICT,
|
|
detail="Parent asset is not the current revision version",
|
|
)
|
|
|
|
version_no = (latest_version or 0) + 1
|
|
current_version_result = await session.execute(
|
|
select(AssetORM).where(
|
|
AssetORM.order_id == order_id,
|
|
AssetORM.asset_type == AssetType.MANUAL_REVISION,
|
|
AssetORM.root_asset_id == root_asset_id,
|
|
AssetORM.is_current_version.is_(True),
|
|
)
|
|
)
|
|
current_version_asset = current_version_result.scalar_one_or_none()
|
|
if current_version_asset is not None:
|
|
current_version_asset.is_current_version = False
|
|
|
|
asset = AssetORM(
|
|
order_id=order_id,
|
|
asset_type=AssetType.MANUAL_REVISION,
|
|
step_name=WorkflowStepName.REVIEW,
|
|
parent_asset_id=payload.parent_asset_id,
|
|
root_asset_id=root_asset_id,
|
|
version_no=version_no,
|
|
is_current_version=True,
|
|
uri=payload.uploaded_uri,
|
|
metadata_json={
|
|
"reviewer_id": payload.reviewer_id,
|
|
"comment": payload.comment,
|
|
"source": "manual_revision",
|
|
},
|
|
)
|
|
session.add(asset)
|
|
await session.flush()
|
|
|
|
review_task.status = ReviewTaskStatus.REVISION_UPLOADED
|
|
review_task.latest_revision_asset_id = asset.id
|
|
review_task.resume_asset_id = asset.id
|
|
review_task.selected_asset_id = asset.id
|
|
review_task.reviewer_id = payload.reviewer_id
|
|
review_task.comment = payload.comment
|
|
await session.commit()
|
|
|
|
revision_count = await self.count_manual_revisions(session, order_id)
|
|
return RegisterRevisionResponse(
|
|
order_id=order_id,
|
|
workflow_id=workflow_run.workflow_id,
|
|
asset_id=asset.id,
|
|
parent_asset_id=asset.parent_asset_id or payload.parent_asset_id,
|
|
root_asset_id=asset.root_asset_id or root_asset_id,
|
|
version_no=asset.version_no,
|
|
review_task_status=review_task.status,
|
|
latest_revision_asset_id=asset.id,
|
|
revision_count=revision_count,
|
|
)
|
|
|
|
async def list_revision_chain(
|
|
self,
|
|
session: AsyncSession,
|
|
order_id: int,
|
|
) -> RevisionChainResponse:
|
|
"""List all manual revisions for an order in version order."""
|
|
|
|
await self._require_order(session, order_id)
|
|
|
|
result = await session.execute(
|
|
select(AssetORM)
|
|
.where(
|
|
AssetORM.order_id == order_id,
|
|
AssetORM.asset_type == AssetType.MANUAL_REVISION,
|
|
)
|
|
.order_by(AssetORM.version_no.asc(), AssetORM.created_at.asc(), AssetORM.id.asc())
|
|
)
|
|
assets = result.scalars().all()
|
|
snapshot = await self.get_revision_snapshot(session, order_id)
|
|
|
|
return RevisionChainResponse(
|
|
order_id=order_id,
|
|
latest_revision_asset_id=snapshot.latest_revision_asset_id,
|
|
revision_count=len(assets),
|
|
items=[
|
|
RevisionChainItem(
|
|
asset_id=asset.id,
|
|
order_id=asset.order_id,
|
|
parent_asset_id=asset.parent_asset_id,
|
|
root_asset_id=asset.root_asset_id,
|
|
version_no=asset.version_no,
|
|
is_current_version=asset.is_current_version,
|
|
uri=asset.uri,
|
|
created_at=asset.created_at,
|
|
)
|
|
for asset in assets
|
|
],
|
|
)
|
|
|
|
async def count_manual_revisions(self, session: AsyncSession, order_id: int) -> int:
|
|
"""Return the number of manual revision assets for an order."""
|
|
|
|
result = await session.execute(
|
|
select(func.count(AssetORM.id)).where(
|
|
AssetORM.order_id == order_id,
|
|
AssetORM.asset_type == AssetType.MANUAL_REVISION,
|
|
)
|
|
)
|
|
return result.scalar_one()
|
|
|
|
async def _require_order(self, session: AsyncSession, order_id: int) -> OrderORM:
|
|
order = await session.get(OrderORM, order_id)
|
|
if order is None:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Order not found")
|
|
return order
|
|
|
|
async def _require_workflow_run(self, session: AsyncSession, order_id: int) -> WorkflowRunORM:
|
|
result = await session.execute(select(WorkflowRunORM).where(WorkflowRunORM.order_id == order_id))
|
|
workflow_run = result.scalar_one_or_none()
|
|
if workflow_run is None:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Workflow not found")
|
|
return workflow_run
|
|
|
|
async def _get_active_review_task(self, session: AsyncSession, order_id: int) -> ReviewTaskORM:
|
|
result = await session.execute(
|
|
select(ReviewTaskORM)
|
|
.where(
|
|
ReviewTaskORM.order_id == order_id,
|
|
ReviewTaskORM.status.in_(
|
|
[ReviewTaskStatus.PENDING, ReviewTaskStatus.REVISION_UPLOADED]
|
|
),
|
|
)
|
|
.order_by(ReviewTaskORM.created_at.desc(), ReviewTaskORM.id.desc())
|
|
)
|
|
review_task = result.scalars().first()
|
|
if review_task is None:
|
|
raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="No active review task")
|
|
return review_task
|