""" Cloud analysis API package — Phase 14 router aggregator. Provides the /analysis sub-router tree under /api/cloud. All routes are owner-scoped and blocked for admin accounts (get_regular_user enforced at each leaf router). Route family (Plan 03 cache + Plan 04 jobs/controls): GET /analysis/cache — cache usage/settings PATCH /analysis/cache/settings — update settings POST /analysis/connections/{id}/estimate — scope estimate (no bytes) POST /analysis/connections/{id}/jobs — enqueue analysis job GET /analysis/jobs — list jobs GET /analysis/jobs/{job_id} — job status POST /analysis/jobs/{job_id}/cancel — cancel batch POST /analysis/jobs/{job_id}/items/{item_id}/cancel — cancel single item POST /analysis/jobs/{job_id}/items/{item_id}/skip — skip single item POST /analysis/jobs/{job_id}/items/{item_id}/retry — retry failed item Security invariants (T-14-01, T-14-02, T-14-03): - All routes use get_regular_user — admin accounts blocked. - All routes are owner-scoped: connection_id and job_id resolved against user. - Response schemas never include credentials_enc, object_key, or provider URLs. - No provider mutation methods are called in this module. Module is importable as api.cloud.analysis for tests and callers that verify router registration before endpoint implementation is complete. """ from __future__ import annotations import uuid from typing import List, Optional from fastapi import APIRouter, Depends, HTTPException, Query, Request, status from fastapi.responses import JSONResponse from sqlalchemy.ext.asyncio import AsyncSession from api.cloud.cache import router as cache_router from api.cloud.schemas import ( AnalysisControlOut, AnalysisEnqueueOut, AnalysisEnqueueRequest, AnalysisEstimateOut, AnalysisEstimateRequest, AnalysisJobItemOut, AnalysisJobOut, ) from db.models import User from deps.auth import get_regular_user from deps.db import get_db from services.cloud_analysis import ( AnalysisItemNotFound, AnalysisJobNotFound, cancel_job, cancel_job_item, enqueue_analysis_job, estimate_scope, get_analysis_job, list_analysis_jobs, list_job_items, retry_job_item, skip_job_item, InvalidJobState, ) from services.cloud_items import ConnectionNotFound from services.rate_limiting import account_limiter router = APIRouter() router.include_router(cache_router) # ── Estimate ─────────────────────────────────────────────────────────────────── @router.post( "/analysis/connections/{connection_id}/estimate", response_model=AnalysisEstimateOut, ) @account_limiter.limit("60/minute") async def estimate_analysis_scope( request: Request, connection_id: uuid.UUID, body: AnalysisEstimateRequest, session: AsyncSession = Depends(get_db), current_user: User = Depends(get_regular_user), ) -> AnalysisEstimateOut: """Estimate analysis scope without downloading provider bytes. Returns counts and byte totals for the requested scope. No bytes are fetched from the provider. No provider mutations are made. T-14-04: bytes never downloaded during estimate. T-14-03: no provider mutations. T-14-01: connection must be owned by current_user. """ request.state.current_user = current_user try: result = await estimate_scope( session, user_id=current_user.id, connection_id=connection_id, scope=body.scope, provider_item_ids=body.provider_item_ids, recursive=body.recursive, ) except ConnectionNotFound as exc: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=str(exc), ) from exc except ValueError as exc: raise HTTPException( status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=str(exc), ) from exc return AnalysisEstimateOut( supported_count=result.supported_count, unsupported_count=result.unsupported_count, total_provider_bytes=result.total_provider_bytes, recursive=result.recursive, is_partial=result.is_partial, scope_kind=result.scope_kind, ) # ── Enqueue ──────────────────────────────────────────────────────────────────── @router.post( "/analysis/connections/{connection_id}/jobs", response_model=AnalysisEnqueueOut, status_code=202, ) @account_limiter.limit("30/minute") async def enqueue_job( request: Request, connection_id: uuid.UUID, body: AnalysisEnqueueRequest, session: AsyncSession = Depends(get_db), current_user: User = Depends(get_regular_user), ) -> AnalysisEnqueueOut: """Enqueue an analysis job for the requested scope. Creates a CloudAnalysisJob and populates job items. Already-current items are skipped without downloading bytes. Returns a stable job_id for polling. Accepts failure_behavior override for this job (D-11). T-14-01: connection must be owned by current_user. T-14-03: no provider mutations. T-14-04: no bytes downloaded for already-current checks. """ request.state.current_user = current_user # Validate failure_behavior before calling the service valid_fb = {"pause_batch", "continue_item"} if body.failure_behavior not in valid_fb: raise HTTPException( status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=f"Invalid failure_behavior {body.failure_behavior!r}. Valid values: {sorted(valid_fb)}", ) try: result = await enqueue_analysis_job( session, user_id=current_user.id, connection_id=connection_id, scope=body.scope, provider_item_ids=body.provider_item_ids, recursive=body.recursive, failure_behavior=body.failure_behavior, ) except ConnectionNotFound as exc: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=str(exc), ) from exc except ValueError as exc: raise HTTPException( status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=str(exc), ) from exc await session.commit() return AnalysisEnqueueOut( job_id=str(result.job_id), status="queued", total_count=result.total_count, queued_count=result.queued_count, already_current_count=result.already_current_count, unsupported_count=result.unsupported_count, ) # ── Job status / list ────────────────────────────────────────────────────────── def _build_job_out(job, *, detail: bool = False) -> AnalysisJobOut: """Build a credential-free AnalysisJobOut from a CloudAnalysisJob row. Simple mode aggregates internal statuses into human-friendly labels: waiting = queued working = downloading + extracting + classifying done = indexed + already_current skipped = cancelled + unsupported failed = failed Detailed mode includes per-stage counts. """ # Simple aggregate labels waiting_count = job.queued_count or 0 working_count = ( (job.downloading_count or 0) + (job.extracting_count or 0) + (job.classifying_count or 0) ) done_count = (job.indexed_count or 0) + (job.already_current_count or 0) skipped_count = (job.cancelled_count or 0) + (job.unsupported_count or 0) failed_count = job.failed_count or 0 out = AnalysisJobOut( job_id=str(job.id), connection_id=str(job.connection_id), scope_kind=job.scope_kind, status=job.status, failure_behavior=job.failure_behavior, recursive=job.recursive, total_count=job.total_count or 0, waiting_count=waiting_count, working_count=working_count, done_count=done_count, skipped_count=skipped_count, failed_count=failed_count, # Per-stage counts always included (0 when not in detailed mode) queued_count=job.queued_count or 0, downloading_count=job.downloading_count or 0, extracting_count=job.extracting_count or 0, classifying_count=job.classifying_count or 0, indexed_count=job.indexed_count or 0, already_current_count=job.already_current_count or 0, cancelled_count=job.cancelled_count or 0, unsupported_count=job.unsupported_count or 0, created_at=job.created_at, started_at=job.started_at, finished_at=job.finished_at, ) return out @router.get( "/analysis/jobs", response_model=List[AnalysisJobOut], ) @account_limiter.limit("120/minute") async def list_jobs( request: Request, connection_id: Optional[uuid.UUID] = Query(default=None), job_status: Optional[str] = Query(default=None, alias="status"), detail: bool = Query(default=False), session: AsyncSession = Depends(get_db), current_user: User = Depends(get_regular_user), ) -> List[AnalysisJobOut]: """List analysis jobs owned by the current user. Optional filters: connection_id, status. Use ?detail=true for per-stage counts (ANALYZE-04). """ request.state.current_user = current_user jobs = await list_analysis_jobs( session, user_id=current_user.id, connection_id=connection_id, status=job_status, ) return [_build_job_out(j, detail=detail) for j in jobs] @router.get( "/analysis/jobs/{job_id}", response_model=AnalysisJobOut, ) @account_limiter.limit("120/minute") async def get_job_status( request: Request, job_id: uuid.UUID, detail: bool = Query(default=False), session: AsyncSession = Depends(get_db), current_user: User = Depends(get_regular_user), ) -> AnalysisJobOut: """Return job status for a single analysis job. Use ?detail=true for per-stage item counts (ANALYZE-04). Response never includes credentials, object_key, or raw provider data. """ request.state.current_user = current_user try: job = await get_analysis_job(session, job_id=job_id, user_id=current_user.id) except AnalysisJobNotFound as exc: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=str(exc), ) from exc return _build_job_out(job, detail=detail) # ── Cancel job ───────────────────────────────────────────────────────────────── @router.post( "/analysis/jobs/{job_id}/cancel", response_model=AnalysisControlOut, ) @account_limiter.limit("30/minute") async def cancel_analysis_job( request: Request, job_id: uuid.UUID, session: AsyncSession = Depends(get_db), current_user: User = Depends(get_regular_user), ) -> AnalysisControlOut: """Cancel a queued or running analysis batch. Queued items transition to cancelled immediately. Running items are marked for cooperative cancellation. Returns a typed result — not HTTPException — so kind/reason appear at response top level (Phase 13 mutation response style). """ request.state.current_user = current_user try: await cancel_job(session, job_id=job_id, user_id=current_user.id) except AnalysisJobNotFound as exc: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=str(exc), ) from exc except InvalidJobState: # Already terminal — still return 200 with typed reason return AnalysisControlOut( kind="cancelled", reason="already_terminal", job_id=str(job_id), ) await session.commit() return AnalysisControlOut( kind="cancelled", reason="batch_cancelled", job_id=str(job_id), ) # ── Cancel item ──────────────────────────────────────────────────────────────── @router.post( "/analysis/jobs/{job_id}/items/{item_id}/cancel", response_model=AnalysisControlOut, ) @account_limiter.limit("60/minute") async def cancel_analysis_item( request: Request, job_id: uuid.UUID, item_id: uuid.UUID, session: AsyncSession = Depends(get_db), current_user: User = Depends(get_regular_user), ) -> AnalysisControlOut: """Cancel a single queued or in-flight job item. item_id is the cloud_item_id (DocuVault stable UUID). """ request.state.current_user = current_user try: await cancel_job_item( session, job_id=job_id, item_id=item_id, user_id=current_user.id, ) except AnalysisJobNotFound as exc: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=str(exc), ) from exc except AnalysisItemNotFound as exc: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=str(exc), ) from exc except InvalidJobState as exc: return JSONResponse( status_code=status.HTTP_409_CONFLICT, content={"kind": "invalid_state", "reason": str(exc)}, ) await session.commit() return AnalysisControlOut( kind="cancelled", reason="item_cancelled", job_id=str(job_id), item_id=str(item_id), ) # ── Skip item ────────────────────────────────────────────────────────────────── @router.post( "/analysis/jobs/{job_id}/items/{item_id}/skip", response_model=AnalysisControlOut, ) @account_limiter.limit("60/minute") async def skip_analysis_item( request: Request, job_id: uuid.UUID, item_id: uuid.UUID, session: AsyncSession = Depends(get_db), current_user: User = Depends(get_regular_user), ) -> AnalysisControlOut: """Skip a queued or failed job item. Marks the item as cancelled (skip is a user-initiated cancel variant). item_id is the cloud_item_id (DocuVault stable UUID). """ request.state.current_user = current_user try: await skip_job_item( session, job_id=job_id, item_id=item_id, user_id=current_user.id, ) except AnalysisJobNotFound as exc: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=str(exc), ) from exc except AnalysisItemNotFound as exc: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=str(exc), ) from exc except InvalidJobState as exc: return JSONResponse( status_code=status.HTTP_409_CONFLICT, content={"kind": "invalid_state", "reason": str(exc)}, ) await session.commit() return AnalysisControlOut( kind="skipped", reason="item_skipped", job_id=str(job_id), item_id=str(item_id), ) # ── Retry item ───────────────────────────────────────────────────────────────── @router.post( "/analysis/jobs/{job_id}/items/{item_id}/retry", response_model=AnalysisControlOut, ) @account_limiter.limit("30/minute") async def retry_analysis_item( request: Request, job_id: uuid.UUID, item_id: uuid.UUID, session: AsyncSession = Depends(get_db), current_user: User = Depends(get_regular_user), ) -> AnalysisControlOut: """Retry a failed job item from the beginning. Resets the item to queued status, increments retry_count, clears error fields. item_id is the cloud_item_id (DocuVault stable UUID). """ request.state.current_user = current_user try: await retry_job_item( session, job_id=job_id, item_id=item_id, user_id=current_user.id, ) except AnalysisJobNotFound as exc: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=str(exc), ) from exc except AnalysisItemNotFound as exc: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=str(exc), ) from exc except InvalidJobState as exc: return JSONResponse( status_code=status.HTTP_409_CONFLICT, content={"kind": "invalid_state", "reason": str(exc)}, ) await session.commit() return AnalysisControlOut( kind="retried", reason="item_requeued", job_id=str(job_id), item_id=str(item_id), )