128 lines
3.9 KiB
Python
128 lines
3.9 KiB
Python
"""
|
|
Сезонные deep link и callback для подписок.
|
|
|
|
Формат текущего сезона (SEASON=zk2026):
|
|
start=zk2026_post_{tg_message_id}
|
|
start=zk2026_event_{tg_message_id}
|
|
callback: subscribe_zk2026_{id} / unsubscribe_zk2026_{id}
|
|
callback: subscribe_evt_zk2026_{id} / unsubscribe_evt_zk2026_{id}
|
|
|
|
Старые форматы (post_123, event_123, subscribe_123, …) и чужой сезон
|
|
считаются завершённым прошлым сезоном.
|
|
"""
|
|
|
|
import os
|
|
import re
|
|
from typing import Optional, Tuple
|
|
|
|
from dotenv import load_dotenv
|
|
|
|
_ENV_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), ".env")
|
|
load_dotenv(_ENV_PATH)
|
|
load_dotenv()
|
|
|
|
SEASON = (os.getenv("SEASON") or "zk2026").strip().strip('"').strip("'")
|
|
|
|
# start: zk2026_post_123 / zk2026_event_123
|
|
_START_RE = re.compile(
|
|
rf"^(?P<season>[A-Za-z0-9]+)_((?P<kind>post|event)_)(?P<id>\d+)$"
|
|
)
|
|
# legacy start: post_123 / event_123
|
|
_START_LEGACY_RE = re.compile(r"^(?P<kind>post|event)_(?P<id>\d+)$")
|
|
|
|
# callback posts: subscribe_zk2026_123
|
|
_CB_POST_RE = re.compile(
|
|
r"^(?P<action>subscribe|unsubscribe)_(?P<season>[A-Za-z0-9]+)_(?P<id>\d+)$"
|
|
)
|
|
# legacy: subscribe_123
|
|
_CB_POST_LEGACY_RE = re.compile(r"^(?P<action>subscribe|unsubscribe)_(?P<id>\d+)$")
|
|
|
|
# callback events: subscribe_evt_zk2026_123
|
|
_CB_EVT_RE = re.compile(
|
|
r"^(?P<action>subscribe|unsubscribe)_evt_(?P<season>[A-Za-z0-9]+)_(?P<id>\d+)$"
|
|
)
|
|
# legacy: subscribe_evt_123
|
|
_CB_EVT_LEGACY_RE = re.compile(r"^(?P<action>subscribe|unsubscribe)_evt_(?P<id>\d+)$")
|
|
|
|
|
|
def subscription_start_link(bot_name: str, kind: str, message_id) -> str:
|
|
"""HTTPS deep link на бота: kind = 'post' | 'event'."""
|
|
payload = f"{SEASON}_{kind}_{message_id}"
|
|
return f"https://t.me/{bot_name}?start={payload}"
|
|
|
|
|
|
def callback_subscribe_post(message_id) -> str:
|
|
return f"subscribe_{SEASON}_{message_id}"
|
|
|
|
|
|
def callback_unsubscribe_post(message_id) -> str:
|
|
return f"unsubscribe_{SEASON}_{message_id}"
|
|
|
|
|
|
def callback_subscribe_event(message_id) -> str:
|
|
return f"subscribe_evt_{SEASON}_{message_id}"
|
|
|
|
|
|
def callback_unsubscribe_event(message_id) -> str:
|
|
return f"unsubscribe_evt_{SEASON}_{message_id}"
|
|
|
|
|
|
def parse_start_payload(payload: str) -> Tuple[Optional[str], Optional[str], bool]:
|
|
"""
|
|
Разбор аргумента /start.
|
|
|
|
Returns:
|
|
(kind, id, is_current_season)
|
|
kind: 'post' | 'event' | None
|
|
is_current_season: False → показать «сезон завершён»
|
|
"""
|
|
if not payload:
|
|
return None, None, True
|
|
|
|
m = _START_RE.match(payload)
|
|
if m:
|
|
return m.group("kind"), m.group("id"), m.group("season") == SEASON
|
|
|
|
m = _START_LEGACY_RE.match(payload)
|
|
if m:
|
|
return m.group("kind"), m.group("id"), False
|
|
|
|
# Неизвестный payload — считаем устаревшим/чужим, без действия по id
|
|
return None, None, False
|
|
|
|
|
|
def parse_post_callback(data: str) -> Tuple[Optional[str], Optional[str], bool]:
|
|
"""
|
|
Returns:
|
|
(action, id, is_current_season) для кнопок подписки на посты.
|
|
action: 'subscribe' | 'unsubscribe' | None
|
|
"""
|
|
if data.startswith("subscribe_evt_") or data.startswith("unsubscribe_evt_"):
|
|
return None, None, True
|
|
|
|
m = _CB_POST_RE.match(data)
|
|
if m:
|
|
return m.group("action"), m.group("id"), m.group("season") == SEASON
|
|
|
|
m = _CB_POST_LEGACY_RE.match(data)
|
|
if m:
|
|
return m.group("action"), m.group("id"), False
|
|
|
|
return None, None, True
|
|
|
|
|
|
def parse_event_callback(data: str) -> Tuple[Optional[str], Optional[str], bool]:
|
|
"""
|
|
Returns:
|
|
(action, id, is_current_season) для кнопок подписки на события.
|
|
"""
|
|
m = _CB_EVT_RE.match(data)
|
|
if m:
|
|
return m.group("action"), m.group("id"), m.group("season") == SEASON
|
|
|
|
m = _CB_EVT_LEGACY_RE.match(data)
|
|
if m:
|
|
return m.group("action"), m.group("id"), False
|
|
|
|
return None, None, True
|