Add promo offer broadcast endpoint

app/webapi/routes/promo_offers.py

@@ -11,6 +11,7 @@ from app.database.crud.discount_offer import (
     list_discount_offers,
     upsert_discount_offer,
 )
+from app.handlers.admin.messages import get_custom_users, get_target_users
 from app.database.crud.promo_offer_log import list_promo_offer_logs
 from app.database.crud.promo_offer_template import (
     get_promo_offer_template_by_id,
@@ -22,6 +23,8 @@ from app.database.models import DiscountOffer, PromoOfferLog, PromoOfferTemplate
 
 from ..dependencies import get_db_session, require_api_token
 from ..schemas.promo_offers import (
+    PromoOfferBroadcastRequest,
+    PromoOfferBroadcastResponse,
     PromoOfferCreateRequest,
     PromoOfferListResponse,
     PromoOfferLogListResponse,
@@ -138,6 +141,14 @@ def _build_log_response(entry: PromoOfferLog) -> PromoOfferLogResponse:
     )
 
 
+async def _resolve_target_users(db: AsyncSession, target: str) -> list[User]:
+    normalized = target.strip().lower()
+    if normalized.startswith("custom_"):
+        criteria = normalized[len("custom_"):]
+        return await get_custom_users(db, criteria)
+    return await get_target_users(db, normalized)
+
+
 @router.get("", response_model=PromoOfferListResponse)
 async def list_promo_offers(
     _: Any = Security(require_api_token),
@@ -250,6 +261,101 @@ async def create_promo_offer(
     return _serialize_offer(offer)
 
 
+@router.post(
+    "/broadcast",
+    response_model=PromoOfferBroadcastResponse,
+    status_code=status.HTTP_201_CREATED,
+)
+async def broadcast_promo_offers(
+    payload: PromoOfferBroadcastRequest,
+    _: Any = Security(require_api_token),
+    db: AsyncSession = Depends(get_db_session),
+) -> PromoOfferBroadcastResponse:
+    if payload.discount_percent < 0:
+        raise HTTPException(status.HTTP_400_BAD_REQUEST, "discount_percent must be non-negative")
+    if payload.bonus_amount_kopeks < 0:
+        raise HTTPException(status.HTTP_400_BAD_REQUEST, "bonus_amount_kopeks must be non-negative")
+    if payload.valid_hours <= 0:
+        raise HTTPException(status.HTTP_400_BAD_REQUEST, "valid_hours must be positive")
+    if not payload.notification_type.strip():
+        raise HTTPException(status.HTTP_400_BAD_REQUEST, "notification_type must not be empty")
+    if not payload.effect_type.strip():
+        raise HTTPException(status.HTTP_400_BAD_REQUEST, "effect_type must not be empty")
+
+    recipients: dict[int, User] = {}
+
+    target = payload.target
+    if target:
+        users = await _resolve_target_users(db, target)
+        recipients.update({user.id: user for user in users if user and user.id})
+
+    target_user_id = payload.user_id
+    user: Optional[User] = None
+    if payload.telegram_id is not None:
+        user = await get_user_by_telegram_id(db, payload.telegram_id)
+        if not user:
+            raise HTTPException(status.HTTP_404_NOT_FOUND, "User not found")
+
+        if target_user_id and target_user_id != user.id:
+            raise HTTPException(
+                status.HTTP_400_BAD_REQUEST,
+                "Provided user_id does not match telegram_id",
+            )
+
+        target_user_id = user.id
+
+    if target_user_id is not None:
+        if user is None:
+            user = await db.get(User, target_user_id)
+        if not user:
+            raise HTTPException(status.HTTP_404_NOT_FOUND, "User not found")
+        recipients[target_user_id] = user
+
+    if not recipients:
+        raise HTTPException(
+            status.HTTP_400_BAD_REQUEST,
+            "Пустая аудитория: укажите target или конкретного пользователя",
+        )
+
+    if payload.subscription_id is not None:
+        if len(recipients) > 1:
+            raise HTTPException(
+                status.HTTP_400_BAD_REQUEST,
+                "subscription_id можно использовать только при отправке одному пользователю",
+            )
+        sole_user = next(iter(recipients.values()))
+        subscription = await db.get(Subscription, payload.subscription_id)
+        if not subscription:
+            raise HTTPException(status.HTTP_404_NOT_FOUND, "Subscription not found")
+        if subscription.user_id != sole_user.id:
+            raise HTTPException(
+                status.HTTP_400_BAD_REQUEST,
+                "Subscription does not belong to the user",
+            )
+
+    created_offers = 0
+    for user in recipients.values():
+        offer = await upsert_discount_offer(
+            db,
+            user_id=user.id,
+            subscription_id=payload.subscription_id,
+            notification_type=payload.notification_type.strip(),
+            discount_percent=payload.discount_percent,
+            bonus_amount_kopeks=payload.bonus_amount_kopeks,
+            valid_hours=payload.valid_hours,
+            effect_type=payload.effect_type,
+            extra_data=payload.extra_data,
+        )
+        if offer:
+            created_offers += 1
+
+    return PromoOfferBroadcastResponse(
+        created_offers=created_offers,
+        user_ids=list(recipients.keys()),
+        target=payload.target,
+    )
+
+
 @router.get("/logs", response_model=PromoOfferLogListResponse)
 async def get_promo_offer_logs(
     _: Any = Security(require_api_token),

app/webapi/schemas/promo_offers.py

@@ -1,9 +1,9 @@
 from __future__ import annotations
 
 from datetime import datetime
-from typing import Any, Dict, List, Optional
+from typing import Any, ClassVar, Dict, List, Optional
 
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, Field, validator
 
 
 class PromoOfferUserInfo(BaseModel):
@@ -61,6 +61,65 @@ class PromoOfferCreateRequest(BaseModel):
     extra_data: Dict[str, Any] = Field(default_factory=dict)
 
 
+class PromoOfferBroadcastRequest(PromoOfferCreateRequest):
+    target: Optional[str] = Field(
+        None,
+        description=(
+            "Категория пользователей для рассылки. Поддерживает те же сегменты, что "
+            "и API рассылок (all, active, trial, custom_today и т.д.)."
+        ),
+    )
+
+    _ALLOWED_TARGETS: ClassVar[set[str]] = {
+        "all",
+        "active",
+        "trial",
+        "no",
+        "expiring",
+        "expired",
+        "active_zero",
+        "trial_zero",
+        "zero",
+    }
+    _CUSTOM_TARGETS: ClassVar[set[str]] = {
+        "today",
+        "week",
+        "month",
+        "active_today",
+        "inactive_week",
+        "inactive_month",
+        "referrals",
+        "direct",
+    }
+    _TARGET_ALIASES: ClassVar[dict[str, str]] = {
+        "no_sub": "no",
+    }
+
+    @validator("target")
+    def validate_target(cls, value: Optional[str]) -> Optional[str]:
+        if value is None:
+            return None
+
+        normalized = value.strip().lower()
+        normalized = cls._TARGET_ALIASES.get(normalized, normalized)
+
+        if normalized in cls._ALLOWED_TARGETS:
+            return normalized
+
+        if normalized.startswith("custom_"):
+            criteria = normalized[len("custom_"):]
+            if criteria in cls._CUSTOM_TARGETS:
+                return normalized
+
+        raise ValueError("Unsupported target value")
+
+
+class PromoOfferBroadcastResponse(BaseModel):
+    created_offers: int
+    user_ids: List[int]
+    target: Optional[str] = None
+
+
 class PromoOfferTemplateResponse(BaseModel):
     id: int
     name: str