-
Notifications
You must be signed in to change notification settings - Fork 6k
Expand file tree
/
Copy pathpaymentbot.py
More file actions
167 lines (137 loc) · 6.12 KB
/
paymentbot.py
File metadata and controls
167 lines (137 loc) · 6.12 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
#!/usr/bin/env python
# pylint: disable=unused-argument
# This program is dedicated to the public domain under the CC0 license.
"""Basic example for a bot that can receive payments from users."""
import logging
from telegram import LabeledPrice, ShippingOption, Update
from telegram.ext import (
Application,
CommandHandler,
ContextTypes,
MessageHandler,
PreCheckoutQueryHandler,
ShippingQueryHandler,
filters,
)
# Enable logging
logging.basicConfig(
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", level=logging.INFO
)
# set higher logging level for httpx to avoid all GET and POST requests being logged
logging.getLogger("httpx").setLevel(logging.WARNING)
logger = logging.getLogger(__name__)
# Insert the token from your payment provider.
# In order to get a provider_token see https://core.telegram.org/bots/payments#getting-a-token
PAYMENT_PROVIDER_TOKEN = "PAYMENT_PROVIDER_TOKEN"
async def start_callback(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
"""Provides instructions on how to use the bot."""
msg = (
"Use /shipping to receive an invoice with shipping included, or /noshipping for an "
"invoice without shipping."
)
await update.message.reply_text(msg)
async def start_with_shipping_callback(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
"""Sends an invoice which triggers a shipping query."""
chat_id = update.message.chat_id
title = "Payment Example"
description = "Example of a payment process using the python-telegram-bot library."
# Unique payload to identify this payment request as being from your bot
payload = "Custom-Payload"
# Set up the currency.
# List of supported currencies: https://core.telegram.org/bots/payments#supported-currencies
currency = "USD"
# Price in dollars
price = 1
# Convert price to cents from dollars.
prices = [LabeledPrice("Test", price * 100)]
# Optional parameters like need_shipping_address and is_flexible trigger extra user prompts
# https://docs.python-telegram-bot.org/en/stable/telegram.bot.html#telegram.Bot.send_invoice
await context.bot.send_invoice(
chat_id,
title,
description,
payload,
currency,
prices,
provider_token=PAYMENT_PROVIDER_TOKEN,
need_name=True,
need_phone_number=True,
need_email=True,
need_shipping_address=True,
is_flexible=True,
)
async def start_without_shipping_callback(
update: Update, context: ContextTypes.DEFAULT_TYPE
) -> None:
"""Sends an invoice without requiring shipping details."""
chat_id = update.message.chat_id
title = "Payment Example"
description = "Example of a payment process using the python-telegram-bot library."
# Unique payload to identify this payment request as being from your bot
payload = "Custom-Payload"
currency = "USD"
# Price in dollars
price = 1
# Convert price to cents from dollars.
prices = [LabeledPrice("Test", price * 100)]
# optionally pass need_name=True, need_phone_number=True,
# need_email=True, need_shipping_address=True, is_flexible=True
await context.bot.send_invoice(
chat_id,
title,
description,
payload,
currency,
prices,
provider_token=PAYMENT_PROVIDER_TOKEN,
)
async def shipping_callback(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
"""Handles the ShippingQuery with available shipping options."""
query = update.shipping_query
# Verify if the payload matches, ensure it's from your bot
if query.invoice_payload != "Custom-Payload":
# If not, respond with an error
await query.answer(ok=False, error_message="Something went wrong...")
return
# Define available shipping options
# First option with a single price entry
options = [ShippingOption("1", "Shipping Option A", [LabeledPrice("A", 100)])]
# Second option with multiple price entries
price_list = [LabeledPrice("B1", 150), LabeledPrice("B2", 200)]
options.append(ShippingOption("2", "Shipping Option B", price_list))
await query.answer(ok=True, shipping_options=options)
# After (optional) shipping, process the pre-checkout step
async def precheckout_callback(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
"""Responds to the PreCheckoutQuery as the final confirmation for checkout."""
query = update.pre_checkout_query
# Verify if the payload matches, ensure it's from your bot
if query.invoice_payload != "Custom-Payload":
# If not, respond with an error
await query.answer(ok=False, error_message="Something went wrong...")
else:
await query.answer(ok=True)
# Final callback after successful payment
async def successful_payment_callback(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
"""Acknowledges successful payment and thanks the user."""
await update.message.reply_text("Thank you for your payment.")
def main() -> None:
"""Starts the bot and sets up handlers."""
# Create the Application and pass it your bot's token.
application = Application.builder().token("TOKEN").build()
# Start command to display usage instructions
application.add_handler(CommandHandler("start", start_callback))
# Command handlers for starting the payment process
application.add_handler(CommandHandler("shipping", start_with_shipping_callback))
application.add_handler(CommandHandler("noshipping", start_without_shipping_callback))
# Handler for shipping query (if product requires shipping)
application.add_handler(ShippingQueryHandler(shipping_callback))
# Pre-checkout handler for verifying payment details.
application.add_handler(PreCheckoutQueryHandler(precheckout_callback))
# Handler for successful payment. Notify the user that the payment was successful.
application.add_handler(
MessageHandler(filters.SUCCESSFUL_PAYMENT, successful_payment_callback)
)
# Start polling for updates until interrupted (CTRL+C)
application.run_polling(allowed_updates=Update.ALL_TYPES)
if __name__ == "__main__":
main()