"""
recap_tg_bot.py — Recap Studio Telegram Bot
Uses requests library (urllib SSL fails on HuggingFace)
Env vars:
TELEGRAM_BOT_TOKEN — @BotFather
RECAP_WEBAPP_URL — https://recap.psonline.shop
ADMIN_TELEGRAM_CHAT_ID — numeric Telegram ID
ADMIN_USERNAME — backend admin username
"""
import os, json, time, logging, requests
logging.basicConfig(level=logging.INFO,
format='%(asctime)s [%(levelname)s] %(message)s')
log = logging.getLogger(__name__)
BOT_TOKEN = os.getenv('TELEGRAM_BOT_TOKEN', '')
WEBAPP_URL = os.getenv('RECAP_WEBAPP_URL', 'https://recap.psonline.shop')
ADMIN_ID = os.getenv('ADMIN_TELEGRAM_CHAT_ID', '')
ADMIN_U = os.getenv('ADMIN_USERNAME', '')
API_BASE = f'https://api.telegram.org/bot{BOT_TOKEN}'
if not BOT_TOKEN:
raise RuntimeError('TELEGRAM_BOT_TOKEN not set!')
# ── API helpers ───────────────────────────────────────────────────────────────
def _tg(method, **kw):
try:
r = requests.post(f'{API_BASE}/{method}', json=kw, timeout=30)
data = r.json()
if not data.get('ok'):
log.warning('[BOT] %s not ok: %s', method, data.get('description', ''))
return data
except Exception as e:
log.error('[BOT] %s error: %s', method, e)
return {'ok': False}
def send_msg(chat_id, text, markup=None, parse_mode='HTML'):
kw = {'chat_id': chat_id, 'text': text, 'parse_mode': parse_mode,
'disable_web_page_preview': True}
if markup:
kw['reply_markup'] = markup
r = _tg('sendMessage', **kw)
return r.get('result', {}).get('message_id')
def edit_msg(chat_id, msg_id, text, parse_mode='HTML'):
_tg('editMessageText', chat_id=chat_id, message_id=msg_id,
text=text, parse_mode=parse_mode)
# ── Keyboards ─────────────────────────────────────────────────────────────────
def inline_kb():
return {'inline_keyboard': [[
{'text': '🎬 Recap Studio ဖွင့်မည်', 'web_app': {'url': WEBAPP_URL}}
]]}
def reply_kb():
return {
'keyboard': [[{'text': '🎬 Recap Studio', 'web_app': {'url': WEBAPP_URL}}]],
'resize_keyboard': True, 'persistent': True,
}
# ── Handlers ──────────────────────────────────────────────────────────────────
def on_start(msg):
chat_id = msg['chat']['id']
fname = msg['from'].get('first_name', 'User')
send_msg(chat_id,
f'👋 မင်္ဂလာပါ {fname}!\n\n'
'🎬 Recap Studio မှ ကြိုဆိုပါသည်။\n\n'
'AI ဖြင့် မြန်မာဘာသာ Movie Recap ဗီဒီယိုများ '
'အလိုအလျောက် ထုတ်လုပ်ပေးသော app ဖြစ်သည်။\n\n'
'⬇️ Button နှိပ်၍ ဖွင့်ပါ။',
markup=reply_kb()
)
log.info('/start chat=%s user=%s', chat_id, msg['from'].get('username', fname))
def on_help(msg):
send_msg(msg['chat']['id'],
'📖 Recap Studio — သုံးနည်း\n\n'
'1️⃣ /start နှိပ်ပါ\n'
'2️⃣ 🎬 Recap Studio button နှိပ်ပါ\n'
'3️⃣ Video URL ထည့်ပါ\n'
'4️⃣ Settings ချိန်ညှိပြီး Auto Process နှိပ်ပါ\n'
'5️⃣ App ပိတ်၍ progress + video ဤ chat ထဲ ရောက်မည်\n\n'
'💰 Coins:\n'
'• Process တစ်ခု = 1 Coin\n'
'• App ထဲ Buy Coins နှိပ်ဝယ်ပါ\n\n'
'Commands:\n'
'/start — App ဖွင့်မည်\n'
'/coins — Coin လက်ကျန် စစ်မည်\n'
'/help — ဤ message',
markup=inline_kb()
)
def on_coins(msg):
chat_id = msg['chat']['id']
tg_id = str(msg['from']['id'])
try:
r = requests.get(f'{WEBAPP_URL}/api/coins_by_tgid',
params={'tg_id': tg_id}, timeout=10)
d = r.json()
except Exception:
d = {'ok': False}
if d.get('ok'):
send_msg(chat_id,
f"🪙 {d.get('username','')}\n\n"
f"လက်ကျန် Coins: {d.get('coins', 0)}\n\n"
'Coins ဝယ်ရန် App ဖွင့်ပါ 👇',
markup=inline_kb()
)
else:
send_msg(chat_id,
'❌ Account မတွေ့ပါ။\n/start နှိပ်၍ app ကနေ login ဝင်ပါ။',
markup=inline_kb()
)
def on_broadcast(msg):
chat_id = msg['chat']['id']
if str(chat_id) != str(ADMIN_ID):
send_msg(chat_id, '❌ Admin only.')
return
parts = (msg.get('text') or '').split(None, 1)
if len(parts) < 2:
send_msg(chat_id, '❌ Usage: /broadcast ')
return
try:
r = requests.get(f'{WEBAPP_URL}/api/admin/tg_users',
params={'caller': ADMIN_U}, timeout=15)
tg_ids = r.json().get('tg_ids', [])
except Exception:
tg_ids = []
sent = 0
for tid in tg_ids:
if send_msg(tid, f'📢 Recap Studio\n\n{parts[1]}'):
sent += 1
time.sleep(0.05)
send_msg(chat_id, f'✅ Sent to {sent}/{len(tg_ids)} users.')
def on_unknown(msg):
send_msg(msg['chat']['id'],
'🎬 Recap Studio ဖွင့်ရန်:',
markup=inline_kb())
# ── Dispatcher ────────────────────────────────────────────────────────────────
def dispatch(update):
msg = update.get('message')
if not msg:
return
text = (msg.get('text') or '').strip()
if not text:
return
cmd = text.split()[0].split('@')[0].lower() if text.startswith('/') else ''
log.info('MSG chat=%s cmd=%r', msg['chat']['id'], cmd or text[:20])
if cmd == '/start': on_start(msg)
elif cmd == '/help': on_help(msg)
elif cmd == '/coins': on_coins(msg)
elif cmd == '/broadcast': on_broadcast(msg)
else: on_unknown(msg)
# ── Startup: clear webhook ────────────────────────────────────────────────────
def clear_webhook():
for attempt in range(1, 4):
try:
r = requests.post(f'{API_BASE}/deleteWebhook',
json={'drop_pending_updates': True}, timeout=15)
if r.json().get('ok'):
log.info('Webhook cleared (attempt %d)', attempt)
return
log.warning('deleteWebhook not ok: %s', r.json())
except Exception as e:
log.warning('deleteWebhook attempt %d failed: %s', attempt, e)
time.sleep(3)
log.error('Could not clear webhook — polling may conflict')
# ── Polling loop ──────────────────────────────────────────────────────────────
def run_polling():
log.info('Recap Studio Bot — clearing webhook...')
clear_webhook()
time.sleep(2)
log.info('Recap Studio Bot — polling started')
offset = 0
while True:
try:
r = requests.get(
f'{API_BASE}/getUpdates',
params={'offset': offset, 'timeout': 25,
'allowed_updates': ['message']},
timeout=30
)
data = r.json()
if not data.get('ok'):
desc = data.get('description', '')
log.warning('getUpdates not ok: %s', desc)
time.sleep(15 if 'conflict' in desc.lower() else 5)
continue
for upd in data.get('result', []):
offset = upd['update_id'] + 1
try:
dispatch(upd)
except Exception as e:
log.error('dispatch error: %s', e)
except requests.exceptions.ReadTimeout:
pass
except requests.exceptions.ConnectionError as e:
log.error('Connection error: %s — retry 10s', e)
time.sleep(10)
except Exception as e:
log.error('Polling error: %s', e)
time.sleep(5)
if __name__ == '__main__':
run_polling()