"""Phase 7.4 Campaign Engine validators — lifecycle transitions, window, and grant math.""" from __future__ import annotations import math from datetime import date from shared.exceptions import AppError from app.models.types import CampaignStatus # Canonical transitions for the Campaign Engine lifecycle actions. ALLOWED_TRANSITIONS: dict[str, set[CampaignStatus]] = { "schedule": {CampaignStatus.DRAFT}, "activate": { CampaignStatus.DRAFT, CampaignStatus.SCHEDULED, CampaignStatus.PAUSED, }, "pause": {CampaignStatus.ACTIVE}, "resume": {CampaignStatus.PAUSED}, "expire": { CampaignStatus.ACTIVE, CampaignStatus.PAUSED, CampaignStatus.SCHEDULED, }, } TARGET_STATUS: dict[str, CampaignStatus] = { "schedule": CampaignStatus.SCHEDULED, "activate": CampaignStatus.ACTIVE, "pause": CampaignStatus.PAUSED, "resume": CampaignStatus.ACTIVE, "expire": CampaignStatus.EXPIRED, } def ensure_campaign_transition(*, action: str, current: CampaignStatus) -> None: allowed = ALLOWED_TRANSITIONS.get(action, set()) if current not in allowed: raise AppError( "انتقال وضعیت کمپین مجاز نیست", status_code=409, error_code="invalid_campaign_transition", details={ "from_status": current.value, "action": action, "allowed_from": sorted(s.value for s in allowed), }, ) def target_campaign_status(action: str) -> CampaignStatus: return TARGET_STATUS[action] def ensure_schedule_requires_start_date(starts_on: date | None) -> None: if starts_on is None: raise AppError( "برای زمان‌بندی کمپین، تاریخ شروع الزامی است", status_code=422, error_code="campaign_starts_on_required", ) def ensure_campaign_active(status: CampaignStatus) -> None: if status != CampaignStatus.ACTIVE: raise AppError( "کمپین فعال نیست", status_code=409, error_code="campaign_not_active", ) def ensure_campaign_window( starts_on: date | None, ends_on: date | None, *, today: date ) -> None: if starts_on is not None and today < starts_on: raise AppError( "کمپین هنوز شروع نشده است", status_code=409, error_code="campaign_not_started", ) if ends_on is not None and today > ends_on: raise AppError( "کمپین به پایان رسیده است", status_code=409, error_code="campaign_ended", ) def ensure_max_applications_per_member( current_count: int, max_applications: int | None ) -> None: if max_applications is not None and current_count >= max_applications: raise AppError( "سقف اعمال این کمپین برای این عضو تکمیل شده است", status_code=409, error_code="campaign_max_applications_exceeded", ) def compute_grant_amount(rules: dict | None, base_amount: int | None) -> int: rules = rules or {} bonus = int(rules.get("earn_bonus_points", 0) or 0) if bonus < 0: bonus = 0 if base_amount is not None: multiplier = float(rules.get("earn_multiplier", 1.0) or 1.0) grant = math.floor(base_amount * multiplier) + bonus else: grant = bonus return max(0, int(grant)) def ensure_positive_grant(grant: int) -> int: if grant <= 0: raise AppError( "مقدار امتیاز قابل اعطا باید بزرگ‌تر از صفر باشد", status_code=422, error_code="campaign_grant_invalid", ) return grant