Fix admin notification service f-string syntax

This commit is contained in:
Egor
2025-10-31 22:52:11 +03:00
parent ca63b4d758
commit 7df64d8edf
3 changed files with 50 additions and 18 deletions
+30 -9
View File
@@ -77,6 +77,20 @@ class AdminNotificationService:
)
return None
def _get_user_display(self, user: User) -> str:
first_name = getattr(user, "first_name", "") or ""
if first_name:
return first_name
username = getattr(user, "username", "") or ""
if username:
return username
telegram_id = getattr(user, "telegram_id", None)
if telegram_id is None:
return "IDUnknown"
return f"ID{telegram_id}"
def _format_promo_group_discounts(self, promo_group: PromoGroup) -> List[str]:
discount_lines: List[str] = []
@@ -185,6 +199,7 @@ class AdminNotificationService:
referrer_info = await self._get_referrer_info(db, user.referred_by_id)
promo_group = await self._get_user_promo_group(db, user)
promo_block = self._format_promo_group_block(promo_group)
user_display = self._get_user_display(user)
trial_device_limit = subscription.device_limit
if trial_device_limit is None:
@@ -196,7 +211,7 @@ class AdminNotificationService:
message = f"""🎯 <b>АКТИВАЦИЯ ТРИАЛА</b>
👤 <b>Пользователь:</b> {getattr(user, 'first_name', '') or getattr(user, 'username', '') or f"ID{getattr(user, 'telegram_id', 'Unknown')}"}
👤 <b>Пользователь:</b> {user_display}
🆔 <b>Telegram ID:</b> <code>{user.telegram_id}</code>
📱 <b>Username:</b> @{getattr(user, 'username', None) or 'отсутствует'}
👥 <b>Статус:</b> {user_status}
@@ -235,26 +250,27 @@ class AdminNotificationService:
try:
event_type = "🔄 КОНВЕРСИЯ ИЗ ТРИАЛА" if was_trial_conversion else "💎 ПОКУПКА ПОДПИСКИ"
if was_trial_conversion:
user_status = "🎯 Конверсия из триала"
elif user.has_had_paid_subscription:
user_status = "🔄 Продление/Обновление"
else:
user_status = "🆕 Первая покупка"
servers_info = await self._get_servers_info(subscription.connected_squads)
payment_method = self._get_payment_method_display(transaction.payment_method) if transaction else "Баланс"
referrer_info = await self._get_referrer_info(db, user.referred_by_id)
promo_group = await self._get_user_promo_group(db, user)
promo_block = self._format_promo_group_block(promo_group)
user_display = self._get_user_display(user)
total_amount = amount_kopeks if amount_kopeks is not None else (transaction.amount_kopeks if transaction else 0)
transaction_id = transaction.id if transaction else ""
message = f"""💎 <b>{event_type}</b>
👤 <b>Пользователь:</b> {getattr(user, 'first_name', '') or getattr(user, 'username', '') or f"ID{getattr(user, 'telegram_id', 'Unknown')}"}
👤 <b>Пользователь:</b> {user_display}
🆔 <b>Telegram ID:</b> <code>{user.telegram_id}</code>
📱 <b>Username:</b> @{getattr(user, 'username', None) or 'отсутствует'}
👥 <b>Статус:</b> {user_status}
@@ -372,10 +388,11 @@ class AdminNotificationService:
subscription_status = self._get_subscription_status(subscription)
promo_block = self._format_promo_group_block(promo_group)
timestamp = datetime.now().strftime('%d.%m.%Y %H:%M:%S')
user_display = self._get_user_display(user)
return f"""💰 <b>ПОПОЛНЕНИЕ БАЛАНСА</b>
👤 <b>Пользователь:</b> {getattr(user, 'first_name', '') or getattr(user, 'username', '') or f"ID{getattr(user, 'telegram_id', 'Unknown')}"}
👤 <b>Пользователь:</b> {user_display}
🆔 <b>Telegram ID:</b> <code>{user.telegram_id}</code>
📱 <b>Username:</b> @{getattr(user, 'username', None) or 'отсутствует'}
💳 <b>Статус:</b> {topup_status}
@@ -548,13 +565,14 @@ class AdminNotificationService:
servers_info = await self._get_servers_info(subscription.connected_squads)
promo_group = await self._get_user_promo_group(db, user)
promo_block = self._format_promo_group_block(promo_group)
user_display = self._get_user_display(user)
current_end_date = new_end_date or subscription.end_date
current_balance = balance_after if balance_after is not None else user.balance_kopeks
message = f"""⏰ <b>ПРОДЛЕНИЕ ПОДПИСКИ</b>
👤 <b>Пользователь:</b> {getattr(user, 'first_name', '') or getattr(user, 'username', '') or f"ID{getattr(user, 'telegram_id', 'Unknown')}"}
👤 <b>Пользователь:</b> {user_display}
🆔 <b>Telegram ID:</b> <code>{user.telegram_id}</code>
📱 <b>Username:</b> @{getattr(user, 'username', None) or 'отсутствует'}
@@ -600,11 +618,12 @@ class AdminNotificationService:
promo_block = self._format_promo_group_block(promo_group)
type_display = self._get_promocode_type_display(promocode_data.get("type"))
usage_info = f"{promocode_data.get('current_uses', 0)}/{promocode_data.get('max_uses', 0)}"
user_display = self._get_user_display(user)
message_lines = [
"🎫 <b>АКТИВАЦИЯ ПРОМОКОДА</b>",
"",
f"👤 <b>Пользователь:</b> {getattr(user, 'first_name', '') or getattr(user, 'username', '') or f"ID{getattr(user, 'telegram_id', 'Unknown')}"}",
f"👤 <b>Пользователь:</b> {user_display}",
f"🆔 <b>Telegram ID:</b> <code>{user.telegram_id}</code>",
f"📱 <b>Username:</b> @{getattr(user, 'username', None) or 'отсутствует'}",
"",
@@ -727,11 +746,12 @@ class AdminNotificationService:
)
elif automatic:
initiator_line = "🤖 Автоматическое назначение"
user_display = self._get_user_display(user)
message_lines = [
f"{title}",
"",
f"👤 <b>Пользователь:</b> {getattr(user, 'first_name', '') or getattr(user, 'username', '') or f"ID{getattr(user, 'telegram_id', 'Unknown')}"}",
f"👤 <b>Пользователь:</b> {user_display}",
f"🆔 <b>Telegram ID:</b> <code>{user.telegram_id}</code>",
f"📱 <b>Username:</b> @{getattr(user, 'username', None) or 'отсутствует'}",
"",
@@ -1103,6 +1123,7 @@ class AdminNotificationService:
referrer_info = await self._get_referrer_info(db, user.referred_by_id)
promo_group = await self._get_user_promo_group(db, user)
promo_block = self._format_promo_group_block(promo_group)
user_display = self._get_user_display(user)
update_types = {
"traffic": ("📊 ИЗМЕНЕНИЕ ТРАФИКА", "трафик"),
@@ -1115,7 +1136,7 @@ class AdminNotificationService:
message_lines = [
f"{title}",
"",
f"👤 <b>Пользователь:</b> {getattr(user, 'first_name', '') or getattr(user, 'username', '') or f"ID{getattr(user, 'telegram_id', 'Unknown')}"}",
f"👤 <b>Пользователь:</b> {user_display}",
f"🆔 <b>Telegram ID:</b> <code>{user.telegram_id}</code>",
f"📱 <b>Username:</b> @{getattr(user, 'username', None) or 'отсутствует'}",
"",
+15 -6
View File
@@ -52,12 +52,12 @@ def _serialize(group: PromoGroup, members_count: int = 0) -> PromoGroupResponse:
apply_discounts_to_addons=group.apply_discounts_to_addons,
is_default=group.is_default,
members_count=members_count,
created_at=group.created_at,
updated_at=group.updated_at,
created_at=getattr(group, "created_at", None),
updated_at=getattr(group, "updated_at", None),
)
@router.get("", response_model=PromoGroupListResponse)
@router.get("", response_model=PromoGroupListResponse, response_model_exclude_none=True)
async def list_promo_groups(
_: Any = Security(require_api_token),
db: AsyncSession = Depends(get_db_session),
@@ -79,7 +79,7 @@ async def list_promo_groups(
)
@router.get("/{group_id}", response_model=PromoGroupResponse)
@router.get("/{group_id}", response_model=PromoGroupResponse, response_model_exclude_none=True)
async def get_promo_group(
group_id: int,
_: Any = Security(require_api_token),
@@ -93,7 +93,12 @@ async def get_promo_group(
return _serialize(group, members_count=members_count)
@router.post("", response_model=PromoGroupResponse, status_code=status.HTTP_201_CREATED)
@router.post(
"",
response_model=PromoGroupResponse,
response_model_exclude_none=True,
status_code=status.HTTP_201_CREATED,
)
async def create_promo_group_endpoint(
payload: PromoGroupCreateRequest,
_: Any = Security(require_api_token),
@@ -120,7 +125,11 @@ async def create_promo_group_endpoint(
return _serialize(group, members_count=0)
@router.patch("/{group_id}", response_model=PromoGroupResponse)
@router.patch(
"/{group_id}",
response_model=PromoGroupResponse,
response_model_exclude_none=True,
)
async def update_promo_group_endpoint(
group_id: int,
payload: PromoGroupUpdateRequest,
+5 -3
View File
@@ -3,7 +3,7 @@ from __future__ import annotations
from datetime import datetime
from typing import Dict, Optional
from pydantic import BaseModel, Field, validator
from pydantic import BaseModel, ConfigDict, Field, validator
def _normalize_period_discounts(value: Optional[Dict[object, object]]) -> Optional[Dict[int, int]]:
@@ -23,6 +23,8 @@ def _normalize_period_discounts(value: Optional[Dict[object, object]]) -> Option
class PromoGroupResponse(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: int
name: str
server_discount_percent: int
@@ -33,8 +35,8 @@ class PromoGroupResponse(BaseModel):
apply_discounts_to_addons: bool
is_default: bool
members_count: int = 0
created_at: datetime
updated_at: datetime
created_at: Optional[datetime] = None
updated_at: Optional[datetime] = None
class _PromoGroupBase(BaseModel):