94 lines
3.1 KiB
Python
94 lines
3.1 KiB
Python
"""Impl-tracker HTTP router — progress evaluation and manual override endpoints."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import asdict
|
|
from datetime import datetime, timezone
|
|
|
|
from fastapi import APIRouter
|
|
from fastapi.responses import JSONResponse
|
|
from pydantic import BaseModel
|
|
|
|
from app.modules.impl_tracker.application.services import ImplTrackerService
|
|
from app.modules.impl_tracker.domain.entities import ImplProgress
|
|
from app.modules.project.application.services import ProjectService
|
|
from app.modules.scanner.application.services import ScanService
|
|
|
|
router = APIRouter(prefix="/projects/{project_id}/impl-progress", tags=["impl-tracker"])
|
|
|
|
_project_service: ProjectService | None = None
|
|
_scan_service: ScanService | None = None
|
|
_impl_tracker_service: ImplTrackerService | None = None
|
|
|
|
|
|
def init_router(
|
|
project_service: ProjectService,
|
|
scan_service: ScanService,
|
|
impl_tracker_service: ImplTrackerService,
|
|
) -> None:
|
|
global _project_service, _scan_service, _impl_tracker_service
|
|
_project_service = project_service
|
|
_scan_service = scan_service
|
|
_impl_tracker_service = impl_tracker_service
|
|
|
|
|
|
class ManualProgressRequest(BaseModel):
|
|
percentage: float
|
|
|
|
|
|
def _progress_to_dict(p) -> dict:
|
|
d = asdict(p)
|
|
d["evaluated_at"] = p.evaluated_at.isoformat()
|
|
return d
|
|
|
|
|
|
@router.post("")
|
|
def evaluate_progress(project_id: str):
|
|
"""Evaluate implementation progress for all modules."""
|
|
project = _project_service.get_project(project_id)
|
|
scan_result = _scan_service.get_latest_scan(project_id)
|
|
if scan_result is None:
|
|
return JSONResponse(
|
|
status_code=404,
|
|
content={"detail": "No scan available. Run POST /scan first."},
|
|
)
|
|
progress = _impl_tracker_service.evaluate(project, scan_result)
|
|
return [_progress_to_dict(p) for p in progress]
|
|
|
|
|
|
@router.get("")
|
|
def get_progress(project_id: str):
|
|
"""Get cached implementation progress."""
|
|
_project_service.get_project(project_id) # Ensure project exists (raises 404)
|
|
progress = _impl_tracker_service.get_progress(project_id)
|
|
if progress is None:
|
|
return JSONResponse(
|
|
status_code=404,
|
|
content={"detail": "No progress evaluated yet. Run POST /impl-progress first."},
|
|
)
|
|
return [_progress_to_dict(p) for p in progress]
|
|
|
|
|
|
@router.put("/{module_id}")
|
|
def set_manual_progress(project_id: str, module_id: str, body: ManualProgressRequest):
|
|
"""Set manual progress override for a module."""
|
|
_project_service.get_project(project_id) # Ensure project exists (raises 404)
|
|
_impl_tracker_service.set_manual_progress(project_id, module_id, body.percentage)
|
|
|
|
# Return the updated progress entry
|
|
progress = _impl_tracker_service.get_progress(project_id)
|
|
if progress:
|
|
for p in progress:
|
|
if p.module_id == module_id:
|
|
return _progress_to_dict(p)
|
|
|
|
# If no cached progress, return a constructed response
|
|
return _progress_to_dict(
|
|
ImplProgress(
|
|
module_id=module_id,
|
|
percentage=body.percentage,
|
|
source="manual",
|
|
evaluated_at=datetime.now(timezone.utc),
|
|
)
|
|
)
|