test2/Amnezia-Web-Panel-main
Template
1
2
Fork 1
Code Issues Pull Requests Packages Projects Releases 18 Wiki Activity

Add optional expiration that starts on first config use.

Browse Source 
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
orohi
2026-07-26 05:07:50 +03:00
co-authored by Cursor
parent 16d14a7256
commit 745b5e5d8c
9 changed files with 245 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files
telegram_bot.py
+14 -2
View File
@@ -499,7 +499,7 @@ async def _handle_refresh(api: TelegramAPI, chat_id: int, message_id: int, callb
await api.edit_message(chat_id, message_id, f"<b>Your connections</b> ({len(conns)}) — tap to get config:", reply_markup=kb)
async def _handle_get_config(api: TelegramAPI, chat_id: int, message_id: int, callback_id: str, conn_id: str, tg_id: str, load_data_fn: Callable, generate_vpn_link_fn: Callable):
async def _handle_get_config(api: TelegramAPI, chat_id: int, message_id: int, callback_id: str, conn_id: str, tg_id: str, load_data_fn: Callable, generate_vpn_link_fn: Callable, save_data_fn: Optional[Callable] = None):
await api.answer_callback(callback_id, "Fetching config...")
panel_user = _find_user(load_data_fn, tg_id)
@@ -508,6 +508,18 @@ async def _handle_get_config(api: TelegramAPI, chat_id: int, message_id: int, ca
return
data = load_data_fn()
try:
from managers.user_expiration import maybe_start_user_expiration, user_is_expired
owner = next((u for u in data.get("users", []) if u.get("id") == panel_user.get("id")), panel_user)
if user_is_expired(owner):
await api.send_message(chat_id, "❌ Subscription expired.")
return
if save_data_fn and maybe_start_user_expiration(data, owner.get("id")):
save_data_fn(data)
data = load_data_fn()
except Exception:
logger.exception("Bot: failed to start expiration on first use")
conn = next((c for c in data.get("user_connections", []) if c.get("id") == conn_id and (_is_admin(panel_user) or c.get("user_id") == panel_user.get("id"))), None)
if not conn:
await api.send_message(chat_id, "❌ Connection not found.")
@@ -1072,7 +1084,7 @@ async def _dispatch(api: TelegramAPI, update: dict, load_data_fn: Callable, gene
await _handle_refresh(api, chat_id, message_id, callback_id, tg_id, load_data_fn)
return
if data_str.startswith("cfg:"):
await _handle_get_config(api, chat_id, message_id, callback_id, data_str[4:], tg_id, load_data_fn, generate_vpn_link_fn)
await _handle_get_config(api, chat_id, message_id, callback_id, data_str[4:], tg_id, load_data_fn, generate_vpn_link_fn, save_data_fn)
return
panel_user = _require_admin(load_data_fn, tg_id)