123 lines
3.9 KiB
Python
123 lines
3.9 KiB
Python
"""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
|