hexdev/malenia
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(autorenewal): implement automatic subscription renewal system

Browse Source
This commit is contained in:
hexdev
2026-05-18 17:32:04 +07:00
parent 281a460974
commit 56f10d19c5
14 changed files with 411 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
.env.example
View File

@@ -37,3 +37,9 @@ WHITELIST_DEVICE_THRESHOLD=6
# Squads
GENERAL_SQUAD=2de0b5ea-49b9-4181-916f-ae67b9c83b80
WHITELIST_SQUAD=e4435baa-64d3-4b29-a18e-81ad1c60d977
# Autorenewal
AUTORENEW_DAYS_BEFORE=1
# Autorenewal
AUTORENEW_DAYS_BEFORE=1

40
alembic/versions/0d71262de0d5_add_autorenew_to_balancetxtype.py Normal file
View File

@@ -0,0 +1,40 @@
"""add_autorenew_to_balancetxtype
Revision ID: 0d71262de0d5
Revises: f5fbc1975807
Create Date: 2026-05-18 17:26:23.302066
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision: str = '0d71262de0d5'
down_revision: Union[str, Sequence[str], None] = 'f5fbc1975807'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
op.execute("ALTER TYPE balancetxtype ADD VALUE 'autorenew'")
def downgrade() -> None:
# PostgreSQL не поддерживает удаление значений из ENUM напрямую.
# Для даунгрейда нужно пересоздать тип, как в миграции f0668f8e7310.
op.execute("""
CREATE TYPE balancetxtype_old AS ENUM (
'deposit', 'purchase', 'refund', 'referral', 'manual'
)
""")
op.execute("ALTER TABLE balance_transactions ALTER COLUMN tx_type DROP DEFAULT")
op.execute("""
ALTER TABLE balance_transactions
ALTER COLUMN tx_type TYPE balancetxtype_old
USING (tx_type::text::balancetxtype_old)
""")
op.execute("DROP TYPE balancetxtype")
op.execute("ALTER TYPE balancetxtype_old RENAME TO balancetxtype")

28
alembic/versions/bb4199d71f9e_add_autorenew_to_balancetxtype.py Normal file
View File

@@ -0,0 +1,28 @@
"""add_autorenew_to_balancetxtype
Revision ID: bb4199d71f9e
Revises: 0d71262de0d5
Create Date: 2026-05-18 17:26:23.454401
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision: str = 'bb4199d71f9e'
down_revision: Union[str, Sequence[str], None] = '0d71262de0d5'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
"""Upgrade schema."""
pass
def downgrade() -> None:
"""Downgrade schema."""
pass

42
alembic/versions/f5fbc1975807_id_unique_deleted_deprecated_tables.py Normal file
View File

@@ -0,0 +1,42 @@
"""id->unique, deleted deprecated tables
Revision ID: f5fbc1975807
Revises: 92dfa9b572f1
Create Date: 2026-05-18 17:08:56.792369
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
from sqlalchemy.dialects import postgresql
# revision identifiers, used by Alembic.
revision: str = 'f5fbc1975807'
down_revision: Union[str, Sequence[str], None] = '92dfa9b572f1'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
"""Upgrade schema."""
# ### commands auto generated by Alembic - please adjust! ###
op.drop_table('tickets')
op.create_unique_constraint(None, 'bills', ['id'])
# ### end Alembic commands ###
def downgrade() -> None:
"""Downgrade schema."""
# ### commands auto generated by Alembic - please adjust! ###
op.drop_constraint(None, 'bills', type_='unique')
op.create_table('tickets',
sa.Column('id', sa.INTEGER(), autoincrement=True, nullable=False),
sa.Column('creator_id', sa.BIGINT(), autoincrement=False, nullable=False),
sa.Column('thread_id', sa.BIGINT(), autoincrement=False, nullable=True),
sa.Column('status', postgresql.ENUM('OPEN', 'CLOSED', 'PENDING', name='ticketstatus'), autoincrement=False, nullable=False),
sa.PrimaryKeyConstraint('id', name=op.f('tickets_pkey')),
sa.UniqueConstraint('id', name=op.f('tickets_id_key'), postgresql_include=[], postgresql_nulls_not_distinct=False),
sa.UniqueConstraint('thread_id', name=op.f('tickets_thread_id_key'), postgresql_include=[], postgresql_nulls_not_distinct=False)
)
# ### end Alembic commands ###

22
bot/handlers/client/menus.py
View File

@@ -166,9 +166,6 @@ async def toggle_autorenewal(
cb: CallbackQuery, state: FSMContext, session: AsyncSession, deps: DependenciesDTO
):
await state.clear()
await cb.answer(NOTHING_PLACEHOLDER + " В разработке.")
return
await cb.message.edit_text(GENERAL_PROCESSING)
sub = await deps.subscriptions_repository.get_subscription_by_user_id(session, cb.from_user.id)
@@ -186,7 +183,24 @@ async def toggle_autorenewal(
session, cb.from_user.id, deps.rw_sdk
)
await cb.answer(GENERAL_SUCCESS)
status_text = "включено" if new_autorenew else "отключено"
await cb.answer(f"Автопродление {status_text}")
await cb.message.edit_text(
deps.user_service.format_subscription_message(
rw_user, link=user.subscription_link, autorenew=sub.autorenew
),
reply_markup=subscription_controls(
show_autorenew=bool(rw_user), current_autorenewal_status=sub.autorenew
),
)
rw_user = await get_user_by_telegram_id(deps.rw_sdk, cb.from_user.id)
user: User | None = await deps.user_service.update_user_link(
session, cb.from_user.id, deps.rw_sdk
)
status_text = "включено" if new_autorenew else "отключено"
await cb.answer(f"Автопродление {status_text}")
await cb.message.edit_text(
deps.user_service.format_subscription_message(
rw_user, link=user.subscription_link, autorenew=sub.autorenew

27
bot/main.py
View File

@@ -8,6 +8,7 @@ from remnawave import RemnawaveSDK
from bot.handlers import admin_routers, client_routers, system_routers
from bot.middlewares import DIMiddleware
from bot.scheduler import start_scheduler
from config import settings
from db.session import async_session
from misc.pally import PallyClient
@@ -16,6 +17,7 @@ from repositories.subscriptions import SubscriptionRepository
from repositories.users import UserRepository
from schemas.di import DependenciesDTO
from services.billing_service import BillingService
from services.renewal_service import RenewalService
from services.user_service import UserService
logging.basicConfig(level=logging.DEBUG)
@@ -40,6 +42,17 @@ async def main():
token=settings.remnawave_token,
)
aiohttp_session = AiohttpSession(proxy=settings.proxy) if settings.proxy else None
default = DefaultBotProperties(parse_mode="HTML")
bot = Bot(token=settings.bot_token, default=default, session=aiohttp_session)
renewal_service = RenewalService(
subscription_repository=subscriptions_repository,
user_repository=user_repository,
rw_sdk=rw_sdk,
bot=bot,
)
deps = DependenciesDTO(
user_repository=user_repository,
user_service=user_service,
@@ -48,13 +61,10 @@ async def main():
bills_repository=bills_repository,
billing_service=billing_service,
subscriptions_repository=subscriptions_repository,
renewal_service=renewal_service,
)
dp.update.middleware.register(DIMiddleware(deps, async_session))
aiohttp_session = AiohttpSession(proxy=settings.proxy) if settings.proxy else None
default = DefaultBotProperties(parse_mode="HTML")
bot = Bot(token=settings.bot_token, default=default, session=aiohttp_session)
for r in admin_routers:
dp.include_router(r)
for r in client_routers:
@@ -62,7 +72,16 @@ async def main():
for r in system_routers:
dp.include_router(r)
scheduler_task = asyncio.create_task(start_scheduler(renewal_service, async_session))
try:
await dp.start_polling(bot)
finally:
scheduler_task.cancel()
try:
await scheduler_task
except asyncio.CancelledError:
pass
if __name__ == "__main__":

29
bot/scheduler.py Normal file
View File

@@ -0,0 +1,29 @@
import asyncio
import logging
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
from services.renewal_service import RenewalService
logger = logging.getLogger(__name__)
async def run_autorenewal(renewal_service: RenewalService, session_factory: async_sessionmaker):
async with session_factory() as session:
try:
results = await renewal_service.process_autorenewals(session)
logger.info("Autorenewal job completed: %s", results)
except Exception:
logger.exception("Autorenewal job failed")
await session.rollback()
async def start_scheduler(
renewal_service: RenewalService,
session_factory: async_sessionmaker,
interval_hours: int = 24,
):
logger.info("Starting autorenewal scheduler with interval %dh", interval_hours)
while True:
await run_autorenewal(renewal_service, session_factory)
await asyncio.sleep(interval_hours * 3600)

9
bot/texts.py
View File

@@ -219,3 +219,12 @@ ADM_VERIFY_PROMOTION = (
"<b>❓ Отправить рассылку? <i>Это действие будет невозможно отменить.</i></b>"
)
SUCCESSFULLY_SENT_PROMO = "<b>💚 Успешно отправлена рассылка.</b>\n<i>⚡ Успешных отправок: <code>{successful}</code>.</i>"
AUTORENEWAL_INSUFFICIENT_FUNDS = (
"<b>⚠️ Не удалось продлить подписку автоматически.</b>\n"
"Недостаточно средств на балансе (нужно <code>{price}₽</code>, <code>доступно {balance}₽</code>). Автопродление отключено."
)
AUTORENEWAL_SUCCESS = (
"<b>✅ Подписка продлена автоматически. Списано <code>{price}₽</code>.</b>"
"Новая дата окончания: <code>{end_date}</code>."
)

4
config.py
View File

@@ -34,5 +34,9 @@ class Settings(BaseSettings):
general_squad: str = Field("2de0b5ea-49b9-4181-916f-ae67b9c83b80")
whitelist_squad: str = Field("e4435baa-64d3-4b29-a18e-81ad1c60d977")
autorenew_days_before: int = Field(1, alias="AUTORENEW_DAYS_BEFORE")
autorenew_days_before: int = Field(1, alias="AUTORENEW_DAYS_BEFORE")
settings = Settings() # type: ignore

1
db/models/transactions.py
View File

@@ -17,6 +17,7 @@ class BalanceTxType(StrEnum):
REFUND = "refund"
REFERRAL_BONUS = "referral"
MANUAL = "manual"
AUTORENEW = "autorenew"
class BalanceTransaction(Base):

20
repositories/subscriptions.py
View File

@@ -1,4 +1,4 @@
from datetime import datetime
from datetime import datetime, timezone
from sqlalchemy import select, update
from sqlalchemy.ext.asyncio import AsyncSession
@@ -7,6 +7,24 @@ from db.models.subscriptions import Subscription, SubscriptionStatus
class SubscriptionRepository:
async def get_subscriptions_for_autorenewal(
self, session: AsyncSession, *, days_before_expiry: int = 1
) -> list[Subscription]:
from datetime import timedelta
threshold = datetime.now(timezone.utc) + timedelta(days=days_before_expiry)
stmt = (
select(Subscription)
.where(
Subscription.autorenew == True,
Subscription.status == SubscriptionStatus.ACTIVE,
Subscription.end_date <= threshold,
)
.with_for_update()
)
result = await session.execute(stmt)
return list(result.scalars().all())
async def get_subscription_by_user_id(
self, session: AsyncSession, user_id: int
) -> Subscription | None:

2
repositories/users.py
View File

@@ -112,7 +112,7 @@ class UserRepository:
amount=amount,
tx_type=tx_type,
balance_before=user.balance,
balance_after=user.balance + amount,
balance_after=user.balance - amount,
description=description,
created_at=created_at,
)

3
schemas/di.py
View File

@@ -1,5 +1,6 @@
from dataclasses import dataclass
from aiogram import Bot
from remnawave import RemnawaveSDK
from misc.pally import PallyClient
@@ -7,6 +8,7 @@ from repositories.bills import BillsRepository
from repositories.subscriptions import SubscriptionRepository
from repositories.users import UserRepository
from services.billing_service import BillingService
from services.renewal_service import RenewalService
from services.user_service import UserService
@@ -19,6 +21,7 @@ class DependenciesDTO:
subscriptions_repository: SubscriptionRepository
pally_client: PallyClient
rw_sdk: RemnawaveSDK
renewal_service: RenewalService
@dataclass

186
services/renewal_service.py Normal file
View File

@@ -0,0 +1,186 @@
import datetime
import logging
from aiogram import Bot
from remnawave import RemnawaveSDK
from sqlalchemy.ext.asyncio import AsyncSession
from bot.texts import AUTORENEWAL_INSUFFICIENT_FUNDS, AUTORENEWAL_SUCCESS
from config import settings
from db.models.subscriptions import Subscription
from db.models.transactions import BalanceTxType
from misc import utils
from repositories.subscriptions import SubscriptionRepository
from repositories.users import UserRepository
from schemas.subscriptions import InternalSquads, SubscriptionPlan, SubscriptionStatus
logger = logging.getLogger(__name__)
class RenewalServiceError(Exception):
"""Base exception for renewal service errors."""
class InsufficientBalanceError(RenewalServiceError):
"""Raised when user has insufficient balance for renewal."""
class RenewalService:
def __init__(
self,
subscription_repository: SubscriptionRepository,
user_repository: UserRepository,
rw_sdk: RemnawaveSDK,
bot: Bot,
) -> None:
self.subscription_repository = subscription_repository
self.user_repository = user_repository
self.rw_sdk = rw_sdk
self.bot = bot
async def _calculate_renewal_price(self, subscription: Subscription) -> int:
duration = utils.convert_int_to_duration(subscription.duration_days)
plan = SubscriptionPlan(
devices=subscription.devices,
duration=duration,
whitelists=subscription.whitelists,
)
return utils.calculate_price(plan)
async def _renew_subscription_in_rw(self, subscription: Subscription) -> None:
from services import rw as rw_service
rw_user = await rw_service.get_user_by_telegram_id(self.rw_sdk, subscription.user_id)
if not rw_user:
raise RenewalServiceError(f"RW user not found for user_id={subscription.user_id}")
duration = datetime.timedelta(days=subscription.duration_days)
squads = [InternalSquads.DEFAULT]
if subscription.whitelists:
squads.append(InternalSquads.WHITELISTS)
if rw_user.expire_at > datetime.datetime.now(datetime.UTC):
new_expire = rw_user.expire_at + duration
else:
new_expire = datetime.datetime.now(datetime.UTC) + duration
await rw_service.add_days(self.rw_sdk, rw_user.uuid, subscription.duration_days)
await rw_service.update_squads(self.rw_sdk, rw_user.uuid, squads)
await rw_service.set_hwid_limit(self.rw_sdk, rw_user.uuid, subscription.devices)
async def _notify_user(self, user_id: int, message: str) -> None:
try:
await self.bot.send_message(user_id, message)
except Exception:
logger.warning("Failed to notify user %d about renewal", user_id, exc_info=True)
async def _process_single_renewal(
self, session: AsyncSession, subscription: Subscription
) -> bool:
user_id = subscription.user_id
price = await self._calculate_renewal_price(subscription)
user = await self.user_repository.get_user_by_id(session, user_id)
if not user:
logger.error("User %d not found for autorenewal", user_id)
return False
if user.balance < price:
logger.info(
"Insufficient balance for user %d: balance=%d, price=%d",
user_id,
user.balance,
price,
)
await self.subscription_repository.update_subscription_autorenew(
session, user_id, False
)
await self._notify_user(
user_id,
AUTORENEWAL_INSUFFICIENT_FUNDS
)
return False
result = await self.user_repository.decrease_user_balance(
session,
user_id,
price,
tx_type=BalanceTxType.AUTORENEW,
description=f"autorenewal: devices={subscription.devices}, duration={subscription.duration_days}d, whitelists={subscription.whitelists}",
)
if not result:
logger.error("Failed to decrease balance for user %d", user_id)
return False
try:
await self._renew_subscription_in_rw(subscription)
except RenewalServiceError:
logger.exception("Failed to renew subscription in RW for user %d", user_id)
await self.user_repository.increase_user_balance(
session,
user_id,
price,
tx_type=BalanceTxType.REFUND,
description=f"refund for failed autorenewal: devices={subscription.devices}, duration={subscription.duration_days}d",
)
return False
new_end_date = subscription.end_date + datetime.timedelta(days=subscription.duration_days)
await self.subscription_repository.update_sub(
session,
user_id,
end_date=new_end_date,
autorenew=True,
status=SubscriptionStatus.ACTIVE,
devices=subscription.devices,
whitelists=subscription.whitelists,
duration_days=subscription.duration_days,
plan_price_paid=price,
)
await self._notify_user(
user_id,
AUTORENEWAL_SUCCESS.format(
price=price,
end_date=new_end_date.strftime('%d.%m.%Y'),
),
)
logger.info("Successfully renewed subscription for user %d", user_id)
return True
async def process_autorenewals(self, session: AsyncSession) -> dict:
subscriptions = await self.subscription_repository.get_subscriptions_for_autorenewal(
session, days_before_expiry=settings.autorenew_days_before
)
logger.info("Found %d subscriptions for autorenewal", len(subscriptions))
results = {"success": 0, "failed": 0, "insufficient_balance": 0}
for subscription in subscriptions:
try:
success = await self._process_single_renewal(session, subscription)
if success:
results["success"] += 1
else:
user = await self.user_repository.get_user_by_id(session, subscription.user_id)
if user and user.balance < await self._calculate_renewal_price(subscription):
results["insufficient_balance"] += 1
else:
results["failed"] += 1
except Exception:
logger.exception(
"Error processing autorenewal for user %d", subscription.user_id
)
results["failed"] += 1
logger.info(
"Autorenewal completed: success=%d, failed=%d, insufficient_balance=%d",
results["success"],
results["failed"],
results["insufficient_balance"],
)
return results