from decimal import Decimal from typing import Any from app.core.exceptions import ValidationError from app.flows.base import BaseFlow, FlowContext, Option, Step from app.models import Order, Product from app.utils.helpers import money from app.utils.validators import clean_text class PremiumFlow(BaseFlow): name = "premium" steps = [Step(key="username", type="username", prompt="input_username", config={"self": True})] async def process(self, ctx: FlowContext, order: Order) -> dict[str, Any]: config = order.product_snapshot.get("data", {}) months = int(config.get("months", 0)) if not months: raise ValidationError("missing_months") return await ctx.fragment.buy_premium(order.input_data["username"], months) class StarsFlow(BaseFlow): name = "stars" steps = [Step(key="username", type="username", prompt="input_username", config={"self": True})] async def process(self, ctx: FlowContext, order: Order) -> dict[str, Any]: config = order.product_snapshot.get("data", {}) count = int(config.get("count", 0)) * order.quantity if not count: raise ValidationError("missing_stars_count") return await ctx.fragment.buy_stars(order.input_data["username"], count) class TonFlow(BaseFlow): """Buy TON coins delivered to any wallet address.""" name = "ton" steps = [ Step( key="amount", type="amount", prompt="input_ton_amount", config={"min": "0.1", "balance_calc": True}, ), Step(key="wallet", type="ton_address", prompt="input_ton_wallet"), Step(key="memo", type="text", prompt="input_memo", optional=True, config={"max_length": 120}), ] async def base_price( self, ctx: FlowContext, product: Product | None, input_data: dict[str, Any] ) -> Decimal: rate = await ctx.settings.get_decimal("ton_rate_irt") if rate <= 0: raise ValidationError("ton_rate_not_configured") return money(Decimal(str(input_data["amount"])) * rate) async def process(self, ctx: FlowContext, order: Order) -> dict[str, Any]: return await ctx.ton.transfer( address=order.input_data["wallet"], amount=Decimal(str(order.input_data["amount"])), memo=order.input_data.get("memo"), ) class TonBalanceFlow(BaseFlow): """Charge the TON balance of a Telegram account.""" name = "ton_balance" steps = [ Step( key="amount", type="amount", prompt="input_ton_amount", config={"min": "0.1", "balance_calc": True}, ), Step(key="username", type="username", prompt="input_username", config={"self": True}), ] async def base_price( self, ctx: FlowContext, product: Product | None, input_data: dict[str, Any] ) -> Decimal: rate = await ctx.settings.get_decimal("ton_rate_irt") if rate <= 0: raise ValidationError("ton_rate_not_configured") return money(Decimal(str(input_data["amount"])) * rate) async def process(self, ctx: FlowContext, order: Order) -> dict[str, Any]: return await ctx.ton.charge_balance( username=order.input_data["username"], amount=Decimal(str(order.input_data["amount"])), ) class GiftFlow(BaseFlow): name = "gift" steps = [Step(key="username", type="username", prompt="input_username", config={"self": True})] options = [ Option(key="hide", type="toggle", label="option_hide", default=True), Option( key="comment", type="choice_text", label="option_comment", default="normal", config={ "choices": [ {"value": "normal", "label": "comment_normal", "price_setting": None}, { "value": "premium", "label": "comment_premium", "price_setting": "gift_premium_comment_price", }, ], "max_length": 200, }, ), ] def get_options(self, product: Product | None) -> list[Option]: config = (product.data or {}) if product else {} options = [] for option in self.options: if option.key == "hide" and not config.get("allow_hide", True): continue if option.key == "comment" and not config.get("allow_comment", True): continue options.append(option) return options def validate_all(self, product: Product | None, input_data: dict[str, Any]) -> dict[str, Any]: cleaned = super().validate_all(product, input_data) if cleaned.get("comment"): cleaned["comment"] = clean_text(str(cleaned["comment"]), 200) if cleaned.get("comment_type") not in (None, "normal", "premium"): raise ValidationError("invalid_comment_type") if "hide" in cleaned: cleaned["hide"] = bool(cleaned["hide"]) return cleaned async def extra_price( self, ctx: FlowContext, product: Product | None, input_data: dict[str, Any] ) -> Decimal: if input_data.get("comment_type") == "premium": return await ctx.settings.get_decimal("gift_premium_comment_price") return Decimal("0") def summary(self, order: Order) -> dict[str, Any]: rows = super().summary(order) data = order.input_data or {} rows["hide"] = bool(data.get("hide", True)) if data.get("comment"): rows["comment"] = data["comment"] rows["comment_type"] = data.get("comment_type", "normal") return rows async def process(self, ctx: FlowContext, order: Order) -> dict[str, Any]: config = order.product_snapshot.get("data", {}) stars = int(config.get("stars", 0)) or int(config.get("count", 0)) if not stars: raise ValidationError("missing_gift_stars") return await ctx.fragment.send_gift( username=order.input_data["username"], stars=stars, hide=bool(order.input_data.get("hide", True)), comment=order.input_data.get("comment"), comment_type=order.input_data.get("comment_type", "normal"), ) class BoostFlow(BaseFlow): name = "boost" steps = [Step(key="channel", type="channel", prompt="input_channel")] async def process(self, ctx: FlowContext, order: Order) -> dict[str, Any]: return await ctx.fragment.boost_channel( channel=order.input_data["channel"], count=order.quantity, config=order.product_snapshot.get("data", {}), ) class GiveawayFlow(BaseFlow): name = "giveaway" steps = [Step(key="channel", type="channel", prompt="input_channel")] async def process(self, ctx: FlowContext, order: Order) -> dict[str, Any]: return await ctx.fragment.create_giveaway( channel=order.input_data["channel"], config=order.product_snapshot.get("data", {}), quantity=order.quantity, ) class ReactionFlow(BaseFlow): name = "reaction" steps = [Step(key="channel", type="channel", prompt="input_channel")] async def process(self, ctx: FlowContext, order: Order) -> dict[str, Any]: return await ctx.fragment.send_reactions( channel=order.input_data["channel"], config=order.product_snapshot.get("data", {}), quantity=order.quantity, ) class TopupFlow(BaseFlow): """Wallet top-up. Uses the same checkout, but credits the wallet instead.""" name = "topup" productless = True steps = [Step(key="amount", type="amount", prompt="input_topup_amount")] async def base_price( self, ctx: FlowContext, product: Product | None, input_data: dict[str, Any] ) -> Decimal: return money(Decimal(str(input_data["amount"]))) async def process(self, ctx: FlowContext, order: Order) -> dict[str, Any]: # Crediting happens in PaymentService (wallet deposit); nothing to deliver. return {"topup": True} ALL_FLOWS: list[type[BaseFlow]] = [ PremiumFlow, StarsFlow, TonFlow, TonBalanceFlow, GiftFlow, BoostFlow, GiveawayFlow, ReactionFlow, TopupFlow, ]