Spaces:
Sleeping
Sleeping
File size: 5,141 Bytes
c0cb280 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 | import logging
from typing import Any, Dict, List
from service_delivery.models import Contract, Project, ProjectTask
from sqlalchemy import func
from sqlalchemy.orm import Session
from core.database import get_db_session
from core.models import User
logger = logging.getLogger(__name__)
class MarginCalculatorService:
"""
Service for calculating project and product margins based on labor costs.
"""
def calculate_project_labor_cost(self, project_id: str, db: Session = None) -> float:
"""Sum of (actual_hours * hourly_cost_rate) for all tasks in a project."""
if db is None:
with get_db_session() as db:
return self._calculate_project_labor_cost_impl(project_id, db)
else:
return self._calculate_project_labor_cost_impl(project_id, db)
def _calculate_project_labor_cost_impl(self, project_id: str, db: Session) -> float:
"""Implementation of labor cost calculation."""
tasks = db.query(ProjectTask).filter(ProjectTask.project_id == project_id).all()
total_cost = 0.0
for task in tasks:
if task.assigned_to and task.actual_hours:
user = db.query(User).filter(User.id == task.assigned_to).first()
if user and user.hourly_cost_rate:
total_cost += (task.actual_hours * user.hourly_cost_rate)
return round(total_cost, 2)
def get_project_margin(self, project_id: str, db: Session = None) -> Dict[str, Any]:
"""Returns Project Revenue - Labor Cost and margin percentage."""
if db is None:
with get_db_session() as db:
return self._get_project_margin_impl(project_id, db)
else:
return self._get_project_margin_impl(project_id, db)
def _get_project_margin_impl(self, project_id: str, db: Session) -> Dict[str, Any]:
"""Implementation of project margin calculation."""
project = db.query(Project).filter(Project.id == project_id).first()
if not project:
return {"error": "Project not found"}
revenue = project.budget_amount or 0.0
labor_cost = self._calculate_project_labor_cost_impl(project_id, db)
margin_absolute = revenue - labor_cost
margin_percentage = (margin_absolute / revenue * 100) if revenue > 0 else 0.0
return {
"project_id": project_id,
"project_name": project.name,
"revenue": revenue,
"labor_cost": labor_cost,
"gross_margin": round(margin_absolute, 2),
"margin_percentage": round(margin_percentage, 2)
}
def get_product_margins(self, workspace_id: str, db: Session = None) -> List[Dict[str, Any]]:
"""Aggregates margins across all projects for each BusinessProductService."""
if db is None:
with get_db_session() as db:
return self._get_product_margins_impl(workspace_id, db)
else:
return self._get_product_margins_impl(workspace_id, db)
def _get_product_margins_impl(self, workspace_id: str, db: Session) -> List[Dict[str, Any]]:
"""Implementation of product margins aggregation."""
from core.models import BusinessProductService
products = db.query(BusinessProductService).filter(BusinessProductService.workspace_id == workspace_id).all()
results = []
for product in products:
# Find all contracts for this product
contracts = db.query(Contract).filter(Contract.product_service_id == product.id).all()
contract_ids = [c.id for c in contracts]
# Find projects for these contracts
projects = db.query(Project).filter(Project.contract_id.in_(contract_ids)).all()
total_revenue = 0.0
total_cost = 0.0
for project in projects:
total_revenue += (project.budget_amount or 0.0)
total_cost += self._calculate_project_labor_cost_impl(project.id, db)
# Also include tangible product sales cost if linked to orders
from ecommerce.models import EcommerceOrder, EcommerceOrderItem
order_items = db.query(EcommerceOrderItem).join(EcommerceOrder).filter(
EcommerceOrderItem.product_id == product.id,
EcommerceOrder.workspace_id == workspace_id
).all()
for item in order_items:
total_revenue += (item.price * item.quantity)
total_cost += (product.unit_cost * item.quantity)
margin_abs = total_revenue - total_cost
margin_pct = (margin_abs / total_revenue * 100) if total_revenue > 0 else 0.0
results.append({
"product_id": product.id,
"product_name": product.name,
"total_revenue": round(total_revenue, 2),
"total_labor_cost": round(total_cost, 2),
"gross_margin": round(margin_abs, 2),
"margin_percentage": round(margin_pct, 2)
})
return results
margin_calculator = MarginCalculatorService()
|