""" Discount rule service for business logic and discount calculation engine. """ from decimal import Decimal from uuid import UUID from sqlalchemy.ext.asyncio import AsyncSession from app.core.exceptions import AlreadyExistsException, NotFoundException, ValidationException from app.core.money import CENT, npr from app.models.audit import AuditCategory from app.models.billing import DiscountRule, DiscountType from app.models.user import User from app.repositories.discount_repository import DiscountRepository from app.schemas.billing import DiscountRuleCreate, DiscountRuleUpdate from app.services.audit_service import AuditService class DiscountService: """Service handling discount rule management and calculation.""" def __init__( self, repository: DiscountRepository | None = None, audit_service: AuditService | None = None, ) -> None: self.repo = repository or DiscountRepository() self.audit = audit_service or AuditService() async def _record_admin_event( self, db: AsyncSession, *, action: str, description: str, lab_id: UUID, target_id: UUID, actor: User | None, details: dict[str, object] | None = None, ) -> None: role = getattr(actor.role, "value", actor.role) if actor else "system" await self.audit.record_event( db, action=action, description=description, category=AuditCategory.BILLING.value, actor_name=actor.name if actor else "System", actor_role=str(role), actor_id=actor.id if actor else None, target_type="discount_scheme", target_id=str(target_id), ip_address=None, details=details, lab_id=lab_id, ) async def create_discount_rule( self, db: AsyncSession, lab_id: UUID, data: DiscountRuleCreate, actor: User | None = None, ) -> DiscountRule: """ Create a new discount rule scoped to a lab branch. Validates that the rule name is unique within the branch. """ existing = await self.repo.get_by_name(db, lab_id, data.name) if existing: raise AlreadyExistsException( f"Discount rule with name '{data.name.strip()}' already exists in this branch." ) rule = await self.repo.create(db, lab_id, data, commit=False) await self._record_admin_event( db, action="DISCOUNT_SCHEME_CREATED", description="Created a discount scheme.", lab_id=lab_id, target_id=rule.id, actor=actor, details={"discount_type": data.discount_type.value, "code_present": data.code is not None}, ) await db.commit() await db.refresh(rule) return rule async def get_discount_rule( self, db: AsyncSession, rule_id: UUID, lab_id: UUID | None = None, ) -> DiscountRule: """Retrieve a discount rule by ID, optionally enforcing branch scoping.""" rule = await self.repo.get_by_id(db, rule_id, lab_id) if not rule: raise NotFoundException(f"Discount rule with ID '{rule_id}' was not found.") return rule async def list_discount_rules( self, db: AsyncSession, lab_id: UUID, is_active: bool | None = None, search: str | None = None, page: int = 1, limit: int = 50, ) -> tuple[list[DiscountRule], int]: """List discount rules scoped to a lab branch with pagination.""" offset = (page - 1) * limit return await self.repo.list_rules( db, lab_id=lab_id, is_active=is_active, search=search, offset=offset, limit=limit, ) async def update_discount_rule( self, db: AsyncSession, rule_id: UUID, lab_id: UUID, data: DiscountRuleUpdate, actor: User | None = None, ) -> DiscountRule: """Update a discount rule, enforcing name uniqueness if name is changed.""" rule = await self.get_discount_rule(db, rule_id, lab_id) if data.name and data.name.strip().lower() != rule.name.lower(): existing = await self.repo.get_by_name(db, lab_id, data.name) if existing and existing.id != rule.id: raise AlreadyExistsException( f"Discount rule with name '{data.name.strip()}' already exists in this branch." ) final_type = data.discount_type or rule.discount_type final_value = data.discount_value if data.discount_value is not None else rule.discount_value if final_type == DiscountType.PERCENTAGE and final_value > Decimal("100.00"): raise ValidationException("Percentage discount cannot exceed 100.00") updated_rule = await self.repo.update(db, rule, data, commit=False) await self._record_admin_event( db, action="DISCOUNT_SCHEME_UPDATED", description="Updated a discount scheme.", lab_id=lab_id, target_id=updated_rule.id, actor=actor, details={"changed_fields": list(data.model_dump(exclude_unset=True))}, ) await db.commit() await db.refresh(updated_rule) return updated_rule async def delete_discount_rule( self, db: AsyncSession, rule_id: UUID, lab_id: UUID, actor: User | None = None, ) -> None: """Soft-delete a discount rule.""" rule = await self.get_discount_rule(db, rule_id, lab_id) await self.repo.soft_delete(db, rule, commit=False) await self._record_admin_event( db, action="DISCOUNT_SCHEME_REMOVED", description="Removed a discount scheme from the active directory.", lab_id=lab_id, target_id=rule.id, actor=actor, ) await db.commit() def calculate_discount( self, subtotal: Decimal, rule: DiscountRule, ) -> Decimal: """ Pure calculation engine: evaluate discount amount for a given subtotal. - If rule is inactive, raises ValidationException. - If subtotal <= 0, discount is 0.00. - PERCENTAGE: (subtotal * value / 100) quantized to paisa (HALF_UP). - FIXED: min(value, subtotal) quantized to paisa. - Guaranteed to be between 0.00 and subtotal. """ if not rule.is_active: raise ValidationException(f"Discount rule '{rule.name}' is inactive and cannot be applied.") if subtotal <= CENT: return CENT if rule.discount_type == DiscountType.PERCENTAGE: discount = npr(subtotal * (rule.discount_value / Decimal("100.00"))) elif rule.discount_type == DiscountType.FIXED: discount = npr(min(rule.discount_value, subtotal)) else: discount = CENT # Clamp between 0 and subtotal return max(CENT, min(discount, npr(subtotal)))