|
| 1 | +# app/billing/time_tracking.py |
| 2 | +""" |
| 3 | +Time Tracking and Billing System |
| 4 | +Track time spent on projects/tasks, generate invoices, and manage billing. |
| 5 | +""" |
| 6 | + |
| 7 | +from dataclasses import dataclass, field |
| 8 | +from datetime import datetime, timedelta |
| 9 | +from typing import Dict, List, Optional |
| 10 | +from enum import Enum |
| 11 | +import uuid |
| 12 | + |
| 13 | + |
| 14 | +class TimeEntryStatus(Enum): |
| 15 | + """Time entry status.""" |
| 16 | + DRAFT = "draft" |
| 17 | + SUBMITTED = "submitted" |
| 18 | + APPROVED = "approved" |
| 19 | + INVOICED = "invoiced" |
| 20 | + |
| 21 | + |
| 22 | +class BillingStatus(Enum): |
| 23 | + """Invoice status.""" |
| 24 | + DRAFT = "draft" |
| 25 | + ISSUED = "issued" |
| 26 | + SENT = "sent" |
| 27 | + PAID = "paid" |
| 28 | + OVERDUE = "overdue" |
| 29 | + |
| 30 | + |
| 31 | +@dataclass |
| 32 | +class TimeEntry: |
| 33 | + """Time tracking entry.""" |
| 34 | + entry_id: str = field(default_factory=lambda: str(uuid.uuid4())) |
| 35 | + user_id: str = "" |
| 36 | + project_id: str = "" |
| 37 | + task_id: str = "" |
| 38 | + date: datetime = field(default_factory=datetime.utcnow) |
| 39 | + hours: float = 0.0 |
| 40 | + minutes: int = 0 |
| 41 | + description: str = "" |
| 42 | + status: TimeEntryStatus = TimeEntryStatus.DRAFT |
| 43 | + hourly_rate: float = 0.0 |
| 44 | + |
| 45 | + @property |
| 46 | + def total_time_minutes(self) -> int: |
| 47 | + return int(self.hours * 60) + self.minutes |
| 48 | + |
| 49 | + @property |
| 50 | + def cost(self) -> float: |
| 51 | + total_hours = self.hours + (self.minutes / 60) |
| 52 | + return total_hours * self.hourly_rate |
| 53 | + |
| 54 | + def to_dict(self) -> Dict: |
| 55 | + return { |
| 56 | + 'entry_id': self.entry_id, |
| 57 | + 'user_id': self.user_id, |
| 58 | + 'project_id': self.project_id, |
| 59 | + 'task_id': self.task_id, |
| 60 | + 'date': self.date.isoformat(), |
| 61 | + 'hours': round(self.hours, 2), |
| 62 | + 'minutes': self.minutes, |
| 63 | + 'description': self.description, |
| 64 | + 'status': self.status.value, |
| 65 | + 'cost': round(self.cost, 2) |
| 66 | + } |
| 67 | + |
| 68 | + |
| 69 | +@dataclass |
| 70 | +class BillingCycle: |
| 71 | + """Billing cycle.""" |
| 72 | + cycle_id: str = field(default_factory=lambda: str(uuid.uuid4())) |
| 73 | + project_id: str = "" |
| 74 | + start_date: datetime = field(default_factory=datetime.utcnow) |
| 75 | + end_date: datetime = field(default_factory=lambda: datetime.utcnow() + timedelta(days=30)) |
| 76 | + status: str = "open" # open, closed, invoiced |
| 77 | + total_hours: float = 0.0 |
| 78 | + total_cost: float = 0.0 |
| 79 | + |
| 80 | + def to_dict(self) -> Dict: |
| 81 | + return { |
| 82 | + 'cycle_id': self.cycle_id, |
| 83 | + 'project_id': self.project_id, |
| 84 | + 'start_date': self.start_date.isoformat(), |
| 85 | + 'end_date': self.end_date.isoformat(), |
| 86 | + 'status': self.status, |
| 87 | + 'total_hours': round(self.total_hours, 2), |
| 88 | + 'total_cost': round(self.total_cost, 2) |
| 89 | + } |
| 90 | + |
| 91 | + |
| 92 | +@dataclass |
| 93 | +class BillingInvoice: |
| 94 | + """Billing invoice.""" |
| 95 | + invoice_id: str = field(default_factory=lambda: str(uuid.uuid4())) |
| 96 | + project_id: str = "" |
| 97 | + client_id: str = "" |
| 98 | + cycle_id: str = "" |
| 99 | + amount: float = 0.0 |
| 100 | + tax_amount: float = 0.0 |
| 101 | + total_amount: float = 0.0 |
| 102 | + status: BillingStatus = BillingStatus.DRAFT |
| 103 | + issued_date: Optional[datetime] = None |
| 104 | + due_date: Optional[datetime] = None |
| 105 | + paid_date: Optional[datetime] = None |
| 106 | + |
| 107 | + def to_dict(self) -> Dict: |
| 108 | + return { |
| 109 | + 'invoice_id': self.invoice_id, |
| 110 | + 'project_id': self.project_id, |
| 111 | + 'amount': round(self.amount, 2), |
| 112 | + 'tax_amount': round(self.tax_amount, 2), |
| 113 | + 'total_amount': round(self.total_amount, 2), |
| 114 | + 'status': self.status.value, |
| 115 | + 'issued_date': self.issued_date.isoformat() if self.issued_date else None, |
| 116 | + 'due_date': self.due_date.isoformat() if self.due_date else None, |
| 117 | + 'paid_date': self.paid_date.isoformat() if self.paid_date else None |
| 118 | + } |
| 119 | + |
| 120 | + |
| 121 | +@dataclass |
| 122 | +class ExpenseReport: |
| 123 | + """Expense report.""" |
| 124 | + report_id: str = field(default_factory=lambda: str(uuid.uuid4())) |
| 125 | + user_id: str = "" |
| 126 | + project_id: str = "" |
| 127 | + submitted_date: datetime = field(default_factory=datetime.utcnow) |
| 128 | + status: str = "draft" # draft, submitted, approved, reimbursed |
| 129 | + total_amount: float = 0.0 |
| 130 | + |
| 131 | + def to_dict(self) -> Dict: |
| 132 | + return { |
| 133 | + 'report_id': self.report_id, |
| 134 | + 'user_id': self.user_id, |
| 135 | + 'status': self.status, |
| 136 | + 'total_amount': round(self.total_amount, 2), |
| 137 | + 'submitted_date': self.submitted_date.isoformat() |
| 138 | + } |
| 139 | + |
| 140 | + |
| 141 | +class TimeTrackingAndBillingManager: |
| 142 | + """ |
| 143 | + Manages time tracking, billing cycles, and invoicing. |
| 144 | + """ |
| 145 | + |
| 146 | + def __init__(self): |
| 147 | + """Initialize time tracking and billing manager.""" |
| 148 | + self.time_entries: Dict[str, TimeEntry] = {} |
| 149 | + self.billing_cycles: Dict[str, BillingCycle] = {} |
| 150 | + self.invoices: Dict[str, BillingInvoice] = {} |
| 151 | + self.expense_reports: Dict[str, ExpenseReport] = {} |
| 152 | + self.stats = { |
| 153 | + 'total_hours_tracked': 0.0, |
| 154 | + 'total_invoiced': 0.0, |
| 155 | + 'unpaid_invoices': 0, |
| 156 | + 'overdue_invoices': 0 |
| 157 | + } |
| 158 | + |
| 159 | + def create_time_entry(self, user_id: str, project_id: str, task_id: str, |
| 160 | + hours: float, minutes: int = 0, description: str = "", |
| 161 | + hourly_rate: float = 0.0) -> TimeEntry: |
| 162 | + """Create time entry.""" |
| 163 | + entry = TimeEntry( |
| 164 | + user_id=user_id, |
| 165 | + project_id=project_id, |
| 166 | + task_id=task_id, |
| 167 | + hours=hours, |
| 168 | + minutes=minutes, |
| 169 | + description=description, |
| 170 | + hourly_rate=hourly_rate, |
| 171 | + status=TimeEntryStatus.DRAFT |
| 172 | + ) |
| 173 | + |
| 174 | + self.time_entries[entry.entry_id] = entry |
| 175 | + self.stats['total_hours_tracked'] += hours + (minutes / 60) |
| 176 | + |
| 177 | + return entry |
| 178 | + |
| 179 | + def get_time_entry(self, entry_id: str) -> Optional[TimeEntry]: |
| 180 | + """Get time entry.""" |
| 181 | + return self.time_entries.get(entry_id) |
| 182 | + |
| 183 | + def submit_time_entry(self, entry_id: str) -> bool: |
| 184 | + """Submit time entry for approval.""" |
| 185 | + if entry_id not in self.time_entries: |
| 186 | + return False |
| 187 | + |
| 188 | + self.time_entries[entry_id].status = TimeEntryStatus.SUBMITTED |
| 189 | + return True |
| 190 | + |
| 191 | + def approve_time_entry(self, entry_id: str) -> bool: |
| 192 | + """Approve time entry.""" |
| 193 | + if entry_id not in self.time_entries: |
| 194 | + return False |
| 195 | + |
| 196 | + self.time_entries[entry_id].status = TimeEntryStatus.APPROVED |
| 197 | + return True |
| 198 | + |
| 199 | + def get_user_time_entries(self, user_id: str, start_date: datetime = None, |
| 200 | + end_date: datetime = None) -> List[Dict]: |
| 201 | + """Get time entries for user.""" |
| 202 | + entries = [e for e in self.time_entries.values() if e.user_id == user_id] |
| 203 | + |
| 204 | + if start_date: |
| 205 | + entries = [e for e in entries if e.date >= start_date] |
| 206 | + if end_date: |
| 207 | + entries = [e for e in entries if e.date <= end_date] |
| 208 | + |
| 209 | + return [e.to_dict() for e in entries] |
| 210 | + |
| 211 | + def get_project_time_entries(self, project_id: str) -> List[Dict]: |
| 212 | + """Get time entries for project.""" |
| 213 | + entries = [e for e in self.time_entries.values() if e.project_id == project_id |
| 214 | + and e.status == TimeEntryStatus.APPROVED] |
| 215 | + |
| 216 | + return [e.to_dict() for e in entries] |
| 217 | + |
| 218 | + def create_billing_cycle(self, project_id: str, start_date: datetime, |
| 219 | + end_date: datetime) -> BillingCycle: |
| 220 | + """Create billing cycle.""" |
| 221 | + cycle = BillingCycle( |
| 222 | + project_id=project_id, |
| 223 | + start_date=start_date, |
| 224 | + end_date=end_date, |
| 225 | + status="open" |
| 226 | + ) |
| 227 | + |
| 228 | + # Calculate totals from approved time entries |
| 229 | + entries = [e for e in self.time_entries.values() |
| 230 | + if e.project_id == project_id |
| 231 | + and start_date <= e.date <= end_date |
| 232 | + and e.status == TimeEntryStatus.APPROVED] |
| 233 | + |
| 234 | + for entry in entries: |
| 235 | + cycle.total_hours += entry.hours + (entry.minutes / 60) |
| 236 | + cycle.total_cost += entry.cost |
| 237 | + |
| 238 | + self.billing_cycles[cycle.cycle_id] = cycle |
| 239 | + return cycle |
| 240 | + |
| 241 | + def create_invoice(self, project_id: str, client_id: str, cycle_id: str, |
| 242 | + amount: float, tax_rate: float = 0.0) -> BillingInvoice: |
| 243 | + """Create invoice from billing cycle.""" |
| 244 | + tax_amount = amount * (tax_rate / 100) |
| 245 | + total = amount + tax_amount |
| 246 | + |
| 247 | + invoice = BillingInvoice( |
| 248 | + project_id=project_id, |
| 249 | + client_id=client_id, |
| 250 | + cycle_id=cycle_id, |
| 251 | + amount=amount, |
| 252 | + tax_amount=tax_amount, |
| 253 | + total_amount=total, |
| 254 | + status=BillingStatus.DRAFT, |
| 255 | + due_date=datetime.utcnow() + timedelta(days=30) |
| 256 | + ) |
| 257 | + |
| 258 | + self.invoices[invoice.invoice_id] = invoice |
| 259 | + self.stats['unpaid_invoices'] += 1 |
| 260 | + |
| 261 | + return invoice |
| 262 | + |
| 263 | + def issue_invoice(self, invoice_id: str) -> bool: |
| 264 | + """Issue invoice.""" |
| 265 | + if invoice_id not in self.invoices: |
| 266 | + return False |
| 267 | + |
| 268 | + invoice = self.invoices[invoice_id] |
| 269 | + invoice.status = BillingStatus.ISSUED |
| 270 | + invoice.issued_date = datetime.utcnow() |
| 271 | + |
| 272 | + return True |
| 273 | + |
| 274 | + def mark_invoice_paid(self, invoice_id: str) -> bool: |
| 275 | + """Mark invoice as paid.""" |
| 276 | + if invoice_id not in self.invoices: |
| 277 | + return False |
| 278 | + |
| 279 | + invoice = self.invoices[invoice_id] |
| 280 | + invoice.status = BillingStatus.PAID |
| 281 | + invoice.paid_date = datetime.utcnow() |
| 282 | + |
| 283 | + self.stats['unpaid_invoices'] = max(0, self.stats['unpaid_invoices'] - 1) |
| 284 | + self.stats['total_invoiced'] += invoice.total_amount |
| 285 | + |
| 286 | + return True |
| 287 | + |
| 288 | + def get_billing_summary(self, project_id: str) -> Dict: |
| 289 | + """Get billing summary for project.""" |
| 290 | + cycles = [c for c in self.billing_cycles.values() if c.project_id == project_id] |
| 291 | + invoices = [i for i in self.invoices.values() if i.project_id == project_id] |
| 292 | + |
| 293 | + return { |
| 294 | + 'project_id': project_id, |
| 295 | + 'total_hours': sum(c.total_hours for c in cycles), |
| 296 | + 'total_cost': sum(c.total_cost for c in cycles), |
| 297 | + 'total_invoices': len(invoices), |
| 298 | + 'paid_invoices': len([i for i in invoices if i.status == BillingStatus.PAID]), |
| 299 | + 'unpaid_invoices': len([i for i in invoices if i.status != BillingStatus.PAID]) |
| 300 | + } |
| 301 | + |
| 302 | + def get_stats(self) -> Dict: |
| 303 | + """Get time tracking and billing statistics.""" |
| 304 | + return { |
| 305 | + 'total_hours_tracked': round(self.stats['total_hours_tracked'], 1), |
| 306 | + 'total_invoiced': round(self.stats['total_invoiced'], 2), |
| 307 | + 'unpaid_invoices': self.stats['unpaid_invoices'], |
| 308 | + 'overdue_invoices': self.stats['overdue_invoices'] |
| 309 | + } |
| 310 | + |
| 311 | + |
| 312 | +# Global time tracking and billing manager |
| 313 | +time_tracking_manager = TimeTrackingAndBillingManager() |
0 commit comments