""" USER PAYROLL Models """ from sqlalchemy import Column, String, Boolean, Integer, Text, Date, DateTime, Numeric, ForeignKey, CheckConstraint, UniqueConstraint from sqlalchemy.dialects.postgresql import UUID, JSONB from sqlalchemy.orm import relationship from datetime import datetime, date from typing import Optional from decimal import Decimal from app.models.base import BaseModel class UserPayroll(BaseModel): """ User Payroll (Worker Compensation & Payment Tracking) Tracks all work done by users and their compensation. Used for all workers (casual, hourly, salaried). Key Features: - Flexible pay periods (weekly, bi-weekly, monthly) - Aggregated work summary from timesheets (tickets, days) - Simple compensation: base_earnings + bonus - deductions - Payment tracking with reference numbers - Optimistic locking for concurrent updates Business Rules: - One payroll record per user per project per pay period - period_end_date >= period_start_date - All amounts must be non-negative - total_amount = base_earnings + bonus - deductions - Can only be paid once (is_paid flag) - Version column prevents race conditions during recalculation Compensation Calculation: - base_earnings calculated from project_role compensation_type: * FIXED_RATE: days_worked × base_rate (or hours × base_rate for hourly) * PER_UNIT: tickets_closed × per_unit_rate * COMMISSION: ticket_value × commission_percentage * FIXED_PLUS_COMMISSION: (days × base_rate) + (ticket_value × commission%) Generation: - Automatically generated weekly (Monday morning for previous Mon-Sun) - Recalculated if timesheets are corrected - Manual generation supported for ad-hoc periods """ __tablename__ = "user_payroll" # Links user_id = Column(UUID(as_uuid=True), ForeignKey("users.id", ondelete="RESTRICT"), nullable=False, index=True) project_id = Column(UUID(as_uuid=True), ForeignKey("projects.id", ondelete="CASCADE"), nullable=False, index=True) project_team_id = Column(UUID(as_uuid=True), ForeignKey("project_team.id", ondelete="SET NULL"), nullable=True) # Pay Period period_start_date = Column(Date, nullable=False, index=True) period_end_date = Column(Date, nullable=False, index=True) # Work Summary (aggregated from timesheets) tickets_closed = Column(Integer, default=0, nullable=False) days_worked = Column(Integer, default=0, nullable=False) # Compensation Breakdown base_earnings = Column(Numeric(12, 2), default=0, nullable=False) # Calculated from compensation_type bonus_amount = Column(Numeric(12, 2), default=0, nullable=False) # Performance bonuses (manual) deductions = Column(Numeric(12, 2), default=0, nullable=False) # Deductions (advances, etc.) total_amount = Column(Numeric(12, 2), nullable=False) # Total compensation # Calculation Details calculation_notes = Column(Text, nullable=True) # Payment Status is_paid = Column(Boolean, default=False, nullable=False, index=True) paid_at = Column(DateTime(timezone=True), nullable=True) payment_method = Column(String, nullable=True) # 'mobile_money', 'bank_transfer', 'cash' payment_reference = Column(String, nullable=True) # M-Pesa transaction ID, etc. paid_by_user_id = Column(UUID(as_uuid=True), ForeignKey("users.id", ondelete="SET NULL"), nullable=True) # Metadata additional_metadata = Column(JSONB, default={}, nullable=False) # Concurrency Control (Optimistic Locking) version = Column(Integer, default=1, nullable=False, index=True) # Relationships user = relationship("User", foreign_keys=[user_id]) project = relationship("Project", back_populates="payrolls") project_team = relationship("ProjectTeam") paid_by = relationship("User", foreign_keys=[paid_by_user_id]) timesheets = relationship("Timesheet", back_populates="payroll", foreign_keys="[Timesheet.payroll_id]") # Constraints __table_args__ = ( UniqueConstraint('user_id', 'project_id', 'period_start_date', 'period_end_date', name='uq_user_payroll_period'), CheckConstraint('period_end_date >= period_start_date', name='chk_valid_period'), CheckConstraint('tickets_closed >= 0', name='chk_positive_tickets'), CheckConstraint('days_worked >= 0', name='chk_positive_days'), CheckConstraint('base_earnings >= 0', name='chk_positive_base_earnings'), CheckConstraint('bonus_amount >= 0', name='chk_positive_bonus'), CheckConstraint('deductions >= 0', name='chk_positive_deductions'), CheckConstraint('total_amount >= 0', name='chk_positive_total'), ) def __repr__(self): return f"" @property def is_current_week(self) -> bool: """Check if this payroll is for the current week""" today = date.today() return self.period_start_date <= today <= self.period_end_date @property def can_recalculate(self) -> bool: """Check if payroll can be recalculated (not yet paid)""" return not self.is_paid @property def net_earnings(self) -> Decimal: """Calculate net earnings (before deductions)""" return Decimal(self.base_earnings or 0) + Decimal(self.bonus_amount or 0) def calculate_total(self) -> Decimal: """Calculate total compensation amount""" net = self.net_earnings total = net - Decimal(self.deductions or 0) return max(total, Decimal('0')) # Never negative def mark_as_paid( self, paid_by_user_id: UUID, payment_method: str, payment_reference: Optional[str] = None ) -> None: """Mark payroll as paid""" if self.is_paid: raise ValueError("Payroll already marked as paid") self.is_paid = True self.paid_at = datetime.utcnow() self.paid_by_user_id = paid_by_user_id self.payment_method = payment_method self.payment_reference = payment_reference self.updated_at = datetime.utcnow() def recalculate_from_data( self, tickets_closed: int, days_worked: int, base_earnings: Decimal, bonus: Decimal = Decimal('0'), deductions: Decimal = Decimal('0'), calculation_notes: Optional[str] = None ) -> None: """Recalculate payroll from fresh data (used when timesheets are corrected)""" if self.is_paid: raise ValueError("Cannot recalculate paid payroll") self.tickets_closed = tickets_closed self.days_worked = days_worked self.base_earnings = base_earnings self.bonus_amount = bonus self.deductions = deductions self.total_amount = self.calculate_total() if calculation_notes: self.calculation_notes = calculation_notes # Increment version for optimistic locking self.version += 1 self.updated_at = datetime.utcnow() @property def period_label(self) -> str: """Human-readable period label""" return f"{self.period_start_date.strftime('%b %d')} - {self.period_end_date.strftime('%b %d, %Y')}"