67 lines
2.2 KiB
Python
67 lines
2.2 KiB
Python
"""Dashboard service: KPIs and activity feed."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import UTC, datetime
|
|
from decimal import Decimal
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.activity import Activity
|
|
from app.models.deal import Deal, DealStage
|
|
from app.services._base import OrgScopedQuery
|
|
|
|
|
|
async def get_kpis(db: AsyncSession, *, org_id: int) -> dict[str, object]:
|
|
"""Compute headline KPIs for the org's dashboard."""
|
|
scoped = OrgScopedQuery(Deal, db, org_id=org_id)
|
|
open_stages = (DealStage.lead, DealStage.qualified, DealStage.proposal, DealStage.negotiation)
|
|
|
|
all_deals = await scoped.list(skip=0, limit=10_000, order_by=Deal.id)
|
|
open_deals = [d for d in all_deals if d.stage in open_stages]
|
|
open_deals_count = len(open_deals)
|
|
|
|
pipeline_value = float(
|
|
sum((d.value for d in open_deals), Decimal("0"))
|
|
)
|
|
|
|
now = datetime.now(UTC)
|
|
# SQLite returns naive datetimes; strip tz for comparison
|
|
month_start = now.replace(tzinfo=None, day=1, hour=0, minute=0, second=0, microsecond=0)
|
|
won_this_month = [
|
|
d for d in all_deals
|
|
if d.stage == DealStage.won and d.created_at >= month_start
|
|
]
|
|
won_count = len(won_this_month)
|
|
|
|
won_total = sum(1 for d in all_deals if d.stage == DealStage.won)
|
|
lost_total = sum(1 for d in all_deals if d.stage == DealStage.lost)
|
|
if won_total + lost_total == 0:
|
|
conversion_rate = 0.0
|
|
else:
|
|
conversion_rate = round((won_total / (won_total + lost_total)) * 100, 2)
|
|
|
|
return {
|
|
"open_deals_count": open_deals_count,
|
|
"pipeline_value": pipeline_value,
|
|
"won_this_month": won_count,
|
|
"conversion_rate": conversion_rate,
|
|
}
|
|
|
|
|
|
async def get_activity_feed(
|
|
db: AsyncSession, *, org_id: int, limit: int = 20
|
|
) -> list[Activity]:
|
|
"""Return the most recent activities, ordered by created_at desc."""
|
|
result = await db.execute(
|
|
select(Activity)
|
|
.where(Activity.org_id == org_id, Activity.deleted_at.is_(None))
|
|
.order_by(Activity.created_at.desc())
|
|
.limit(limit)
|
|
)
|
|
return list(result.scalars().all())
|
|
|
|
|
|
__all__ = ["get_activity_feed", "get_kpis"]
|