高级功能 - Telegram 机器人
目录
- 内联模式
- 支付(Telegram Stars)
- 迷你应用(WebApps)
- 对话处理器(FSM)
- 贴纸
- 游戏
- Passport
- 企业机器人(Business Bots)
- 消息草稿(流式输出)
内联模式
允许用户在任何聊天中输入@你的机器人 查询来使用机器人。
启用:与 @BotFather 对话并发送/setinline。
fromtelegramimportUpdate,InlineQueryResultArticle,InputTextMessageContentfromtelegram.extimportInlineQueryHandlerasyncdefinline_query(update:Update,context):query=update.inline_query.queryifnotquery:returnresults=[InlineQueryResultArticle(id="1",title=f"Resultado para:{query}",input_message_content=InputTextMessageContent(message_text=f"Voce buscou:{query}"),description="Clique para enviar"),InlineQueryResultArticle(id="2",title="Busca em maiusculas",input_message_content=InputTextMessageContent(message_text=query.upper()))]awaitupdate.inline_query.answer(results,cache_time=10)app.add_handler(InlineQueryHandler(inline_query))内联结果类型
InlineQueryResultArticle- 通用文本InlineQueryResultPhoto- 带预览的照片InlineQueryResultGif- GIFInlineQueryResultVideo- 视频InlineQueryResultAudio- 音频InlineQueryResultDocument- 文档InlineQueryResultLocation- 位置InlineQueryResultVenue- 地点/场所InlineQueryResultContact- 联系人InlineQueryResultCachedPhoto- 已在 Telegram 服务器上的照片
支付(Telegram Stars)
Telegram 允许通过 Stars(内部货币)为数字商品进行支付。
fromtelegramimportLabeledPrice# Enviar invoiceawaitbot.send_invoice(chat_id=chat_id,title="Assinatura Premium",description="Acesso premium por 30 dias",payload="premium_30days",currency="XTR",# XTR = Telegram Starsprices=[LabeledPrice("Assinatura Premium",100)],# 100 Stars)# Handler de pre-checkoutasyncdefprecheckout(update:Update,context):query=update.pre_checkout_queryifquery.invoice_payload=="premium_30days":awaitquery.answer(ok=True)else:awaitquery.answer(ok=False,error_message="Payload invalido")# Handler de pagamento concluidoasyncdefsuccessful_payment(update:Update,context):payment=update.message.successful_paymentawaitupdate.message.reply_text(f"Pagamento recebido!{payment.total_amount}Stars. "f"Seu acesso premium foi ativado.")app.add_handler(PreCheckoutQueryHandler(precheckout))app.add_handler(MessageHandler(filters.SUCCESSFUL_PAYMENT,successful_payment))外部支付提供商
对于实物商品,使用 Stripe、YooMoney 等提供商:
awaitbot.send_invoice(chat_id=chat_id,title="Camiseta Bot Telegram",description="Camiseta tamanho M, algodao",payload="tshirt_m",provider_token="SEU_PROVIDER_TOKEN",# do BotFathercurrency="BRL",prices=[LabeledPrice("Camiseta",5990),# R$ 59.90 (em centavos)LabeledPrice("Frete",1500)# R$ 15.00],need_shipping_address=True,need_name=True,need_phone_number=True)迷你应用(WebApps)
迷你应用是在 Telegram 内运行的 Web 应用。
fromtelegramimportWebAppInfo,InlineKeyboardButton,InlineKeyboardMarkup# Botao que abre Mini Appkeyboard=InlineKeyboardMarkup([[InlineKeyboardButton("Abrir App",web_app=WebAppInfo(url="https://seu-app.com"))]])awaitbot.send_message(chat_id,"Clique para abrir:",reply_markup=keyboard)# Via Reply KeyboardfromtelegramimportReplyKeyboardMarkup,KeyboardButton keyboard=ReplyKeyboardMarkup([[KeyboardButton("Abrir App",web_app=WebAppInfo(url="https://seu-app.com"))]])接收迷你应用的数据
asyncdefweb_app_data(update:Update,context):data=update.effective_message.web_app_data.data# data e uma string JSON enviada pelo Mini Appimportjson parsed=json.loads(data)awaitupdate.message.reply_text(f"Recebi do app:{parsed}")app.add_handler(MessageHandler(filters.StatusUpdate.WEB_APP_DATA,web_app_data))对话处理器(FSM)
用于多步骤对话(表单、向导):
fromtelegram.extimportConversationHandler,CommandHandler,MessageHandler,filters# EstadosNAME,AGE,CONFIRM=range(3)asyncdefstart_form(update:Update,context):awaitupdate.message.reply_text("Qual seu nome?")returnNAMEasyncdefget_name(update:Update,context):context.user_data['name']=update.message.textawaitupdate.message.reply_text("Qual sua idade?")returnAGEasyncdefget_age(update:Update,context):context.user_data['age']=update.message.text name=context.user_data['name']age=context.user_data['age']keyboard=InlineKeyboardMarkup([[InlineKeyboardButton("Confirmar",callback_data="confirm"),InlineKeyboardButton("Cancelar",callback_data="cancel")]])awaitupdate.message.reply_text(f"Nome:{name}\nIdade:{age}\n\nConfirma?",reply_markup=keyboard)returnCONFIRMasyncdefconfirm(update:Update,context):query=update.callback_queryawaitquery.answer()ifquery.data=="confirm":awaitquery.edit_message_text("Cadastro realizado com sucesso!")else:awaitquery.edit_message_text("Cadastro cancelado.")returnConversationHandler.ENDasyncdefcancel(update:Update,context):awaitupdate.message.reply_text("Operacao cancelada.")returnConversationHandler.END conv_handler=ConversationHandler(entry_points=[CommandHandler('cadastro',start_form)],states={NAME:[MessageHandler(filters.TEXT&~filters.COMMAND,get_name)],AGE:[MessageHandler(filters.TEXT&~filters.COMMAND,get_age)],CONFIRM:[CallbackQueryHandler(confirm)],},fallbacks=[CommandHandler('cancel',cancel)],conversation_timeout=300# 5 minutos timeout)app.add_handler(conv_handler)贴纸
# Enviar sticker por file_idawaitbot.send_sticker(chat_id,sticker="CAACAgIAAxkBAAI...")# Obter sticker setsticker_set=awaitbot.get_sticker_set("set_name")forstickerinsticker_set.stickers:print(f"Emoji:{sticker.emoji}, ID:{sticker.file_id}")# Criar sticker set (requer imagem 512x512 PNG/WEBP)awaitbot.create_new_sticker_set(user_id=user_id,name="meupack_by_meubot",title="Meu Sticker Pack",stickers=[InputSticker(sticker=open("sticker.webp","rb"),emoji_list=["😀"],format="static")])# Adicionar sticker ao setawaitbot.add_sticker_to_set(user_id=user_id,name="meupack_by_meubot",sticker=InputSticker(sticker=open("sticker2.webp","rb"),emoji_list=["😎"],format="static"))游戏
# Enviar jogo (precisa registrar no BotFather com /newgame)awaitbot.send_game(chat_id,game_short_name="meu_jogo")# Handler de callback para jogoasyncdefgame_callback(update:Update,context):query=update.callback_queryawaitquery.answer(url="https://seu-jogo.com/?user_id="+str(query.from_user.id))# Salvar scoreawaitbot.set_game_score(user_id=user_id,score=150,chat_id=chat_id,message_id=game_message_id)# Obter high scoresscores=awaitbot.get_game_high_scores(user_id=user_id,chat_id=chat_id,message_id=game_message_id)企业机器人(Business Bots)
用于 Telegram 企业账户的机器人:
# Receber conexao businessasyncdefbusiness_connection(update:Update,context):conn=update.business_connectionifconn.is_enabled:print(f"Bot conectado ao business de{conn.user.first_name}")else:print(f"Bot desconectado do business de{conn.user.first_name}")# Receber mensagens businessasyncdefbusiness_message(update:Update,context):msg=update.business_message# Responder em nome do businessawaitcontext.bot.send_message(chat_id=msg.chat.id,text="Obrigado pela mensagem! Responderemos em breve.",business_connection_id=msg.business_connection_id)app.add_handler(BusinessConnectionHandler(business_connection))app.add_handler(BusinessMessagesHandler(business_message))消息草稿(流式输出)
对于较长的回复(如 AI 的回复),使用草稿提供实时反馈:
importrequests TOKEN="SEU_TOKEN"BASE=f"https://api.telegram.org/bot{TOKEN}"defstream_response(chat_id,full_text):"""Simula streaming enviando drafts parciais."""words=full_text.split()partial=""# Enviar draft inicialfori,wordinenumerate(words):partial+=word+" "ifi%5==0:# Atualizar a cada 5 palavrasrequests.post(f"{BASE}/sendMessageDraft",json={"chat_id":chat_id,"text":partial.strip()+"..."})# Enviar mensagem finalrequests.post(f"{BASE}/sendMessage",json={"chat_id":chat_id,"text":full_text})注意:sendMessageDraft是一个较新的方法。请在你使用的 API 版本中检查其可用性。