sysops
231 lines
7.2 KiB
Python
231 lines
7.2 KiB
Python
"""Revenue Cockpit — page + API."""
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
from typing import Any, Optional
|
|
|
|
from fastapi import APIRouter, HTTPException, Request
|
|
from fastapi.responses import HTMLResponse
|
|
from fastapi.templating import Jinja2Templates
|
|
from pydantic import BaseModel, Field
|
|
|
|
from app.services import revenue_cockpit as svc
|
|
|
|
BASE_DIR = Path(__file__).resolve().parent.parent.parent
|
|
templates = Jinja2Templates(directory=str(BASE_DIR / "templates"))
|
|
|
|
router = APIRouter(tags=["revenue-cockpit"])
|
|
api = APIRouter(prefix="/api/revenue-cockpit", tags=["revenue-cockpit-api"])
|
|
|
|
|
|
class ProjectUpdateBody(BaseModel):
|
|
name: Optional[str] = None
|
|
category: Optional[str] = None
|
|
margin_month: Optional[float] = None
|
|
margin_year: Optional[float] = None
|
|
target_revenue: Optional[float] = None
|
|
next_steps: Optional[str] = None
|
|
status: Optional[str] = None
|
|
priority: Optional[str] = None
|
|
row_style: Optional[str] = None
|
|
|
|
|
|
class GoalsUpdateBody(BaseModel):
|
|
vision_text: Optional[str] = None
|
|
horizon_text: Optional[str] = None
|
|
mid_text: Optional[str] = None
|
|
tagline: Optional[str] = None
|
|
|
|
|
|
class ObjectiveBody(BaseModel):
|
|
title: str = Field(..., min_length=1)
|
|
description: Optional[str] = None
|
|
priority: str = "normal"
|
|
due_date: Optional[str] = None
|
|
|
|
|
|
class ObjectiveUpdateBody(BaseModel):
|
|
title: Optional[str] = None
|
|
description: Optional[str] = None
|
|
status: Optional[str] = None
|
|
priority: Optional[str] = None
|
|
due_date: Optional[str] = None
|
|
|
|
|
|
class AssignTaskBody(BaseModel):
|
|
agent_name: str = Field(..., min_length=1)
|
|
title: str = Field(..., min_length=1)
|
|
description: Optional[str] = None
|
|
objective_id: Optional[int] = None
|
|
priority: str = "normal"
|
|
delegate_herman: bool = True
|
|
|
|
|
|
class DelegateBody(BaseModel):
|
|
message: str = Field(..., min_length=1)
|
|
|
|
|
|
class ImportBody(BaseModel):
|
|
path: str = "Succes Sheet .xlsx"
|
|
sheet: Optional[str] = "Projects next steps revenue"
|
|
replace: bool = True
|
|
|
|
|
|
@router.get("/revenue-cockpit", response_class=HTMLResponse)
|
|
async def revenue_cockpit_page(request: Request):
|
|
return templates.TemplateResponse(
|
|
"revenue_cockpit.html",
|
|
{"request": request, "page_title": "Revenue Cockpit"},
|
|
)
|
|
|
|
|
|
@api.get("/live")
|
|
async def live_from_excel(path: str = "Succes Sheet .xlsx", sheet: Optional[str] = "Projects next steps revenue"):
|
|
"""Leading data source: fresh parse from NAS Excel."""
|
|
try:
|
|
parsed = await svc.fetch_excel_parse_async(path, sheet)
|
|
except Exception as exc:
|
|
raise HTTPException(502, str(exc)) from exc
|
|
projects = parsed.get("projects") or []
|
|
margin_month = sum(p.get("margin_month") or 0 for p in projects)
|
|
margin_year = sum(p.get("margin_year") or 0 for p in projects)
|
|
by_style: dict[str, int] = {}
|
|
by_category: dict[str, int] = {}
|
|
for p in projects:
|
|
by_style[p.get("row_style") or "white"] = by_style.get(p.get("row_style") or "white", 0) + 1
|
|
by_category[p.get("category") or "deal"] = by_category.get(p.get("category") or "deal", 0) + 1
|
|
return {
|
|
"ok": True,
|
|
"source": "excel",
|
|
**parsed,
|
|
"aggregates": {
|
|
"total_margin_month": margin_month,
|
|
"total_margin_year": margin_year,
|
|
"with_margin": sum(1 for p in projects if p.get("margin_month")),
|
|
"by_style": by_style,
|
|
"by_category": by_category,
|
|
},
|
|
}
|
|
|
|
|
|
@api.get("/dashboard")
|
|
async def dashboard():
|
|
return {"ok": True, "stats": svc.dashboard_stats(), "projects": svc.list_projects()}
|
|
|
|
|
|
@api.get("/projects")
|
|
def list_projects(status: Optional[str] = None):
|
|
return {"ok": True, "items": svc.list_projects(status)}
|
|
|
|
|
|
@api.get("/projects/{project_id}")
|
|
def get_project(project_id: int):
|
|
p = svc.get_project(project_id)
|
|
if not p:
|
|
raise HTTPException(404, "Project not found")
|
|
return {"ok": True, "project": p}
|
|
|
|
|
|
@api.patch("/projects/{project_id}")
|
|
def patch_project(project_id: int, body: ProjectUpdateBody):
|
|
data = body.model_dump(exclude_unset=True)
|
|
row_style = data.pop("row_style", None)
|
|
if row_style is not None:
|
|
svc.set_project_row_style(project_id, row_style)
|
|
p = svc.update_project(project_id, data)
|
|
if not p:
|
|
raise HTTPException(404, "Project not found")
|
|
svc.take_snapshot()
|
|
return {"ok": True, "project": p}
|
|
|
|
|
|
@api.patch("/goals")
|
|
def patch_goals(body: GoalsUpdateBody):
|
|
data = body.model_dump(exclude_unset=True)
|
|
g = svc.update_goals(data)
|
|
if not g:
|
|
raise HTTPException(404, "Goals not found")
|
|
return {"ok": True, "goals": g}
|
|
|
|
|
|
@api.post("/projects/{project_id}/objectives")
|
|
def add_objective(project_id: int, body: ObjectiveBody):
|
|
try:
|
|
obj = svc.create_objective(project_id, body.title, body.description, body.priority)
|
|
except Exception as exc:
|
|
raise HTTPException(400, str(exc)) from exc
|
|
svc.take_snapshot()
|
|
return {"ok": True, "objective": obj}
|
|
|
|
|
|
@api.patch("/objectives/{objective_id}")
|
|
def patch_objective(objective_id: int, body: ObjectiveUpdateBody):
|
|
data = body.model_dump(exclude_unset=True)
|
|
obj = svc.update_objective(objective_id, data)
|
|
if not obj:
|
|
raise HTTPException(404, "Objective not found")
|
|
svc.take_snapshot()
|
|
return {"ok": True, "objective": obj}
|
|
|
|
|
|
@api.post("/projects/{project_id}/assign-task")
|
|
def assign_task(project_id: int, body: AssignTaskBody):
|
|
try:
|
|
task = svc.assign_agent_task(
|
|
project_id,
|
|
body.agent_name,
|
|
body.title,
|
|
body.description,
|
|
body.objective_id,
|
|
body.priority,
|
|
body.delegate_herman,
|
|
)
|
|
except ValueError as exc:
|
|
raise HTTPException(404, str(exc)) from exc
|
|
svc.take_snapshot()
|
|
return {"ok": True, "task": task}
|
|
|
|
|
|
@api.post("/projects/{project_id}/delegate")
|
|
async def delegate_project(project_id: int, body: DelegateBody):
|
|
try:
|
|
result = await svc.delegate_via_herman(project_id, body.message)
|
|
except ValueError as exc:
|
|
raise HTTPException(404, str(exc)) from exc
|
|
return result
|
|
|
|
|
|
@api.get("/tasks")
|
|
def list_tasks(limit: int = 50):
|
|
return {"ok": True, "items": svc.list_agent_tasks(limit)}
|
|
|
|
|
|
@api.get("/snapshots")
|
|
def snapshots(limit: int = 90):
|
|
return {"ok": True, "items": svc.list_snapshots(limit)}
|
|
|
|
|
|
@api.post("/snapshot")
|
|
def create_snapshot():
|
|
snap = svc.take_snapshot()
|
|
return {"ok": True, "snapshot": snap}
|
|
|
|
|
|
@api.post("/import-from-excel")
|
|
async def import_from_excel(body: ImportBody):
|
|
try:
|
|
parsed = await svc.fetch_excel_parse_async(body.path, body.sheet)
|
|
result = svc.import_from_parsed(parsed, imported_by="ceo", replace=body.replace)
|
|
return {"ok": True, **result, "preview": {"project_count": parsed.get("project_count"), "sheet": parsed.get("sheet_name")}}
|
|
except Exception as exc:
|
|
raise HTTPException(502, f"Import failed: {exc}") from exc
|
|
|
|
|
|
@api.get("/preview-excel")
|
|
async def preview_excel(path: str = "Succes Sheet .xlsx", sheet: Optional[str] = "Projects next steps revenue"):
|
|
try:
|
|
parsed = await svc.fetch_excel_parse_async(path, sheet)
|
|
return {"ok": True, **parsed}
|
|
except Exception as exc:
|
|
raise HTTPException(502, str(exc)) from exc
|