steffo/greed
steffo/greed
1
Fork 0
mirror of https://github.com/Steffo99/greed.git synced 2024-11-24 14:54:18 +00:00
Code Activity

Merge pull request #54 from Steffo99/i18n

Browse source 
Closes #51: Better strings and localization support
This commit is contained in:
Steffo 2020-06-09 02:40:28 +02:00 committed by GitHub
commit 038b0e574d
Signed by: github
GPG key ID: 4AEE18F83AFDEB23
10 changed files with 435 additions and 275 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

21
config/template_config.ini
View file

@ -5,16 +5,23 @@
# Config file parameters
[Config]
; Config file version. DO NOT EDIT THIS!
version = 17
version = 18
; Set this to no when you are done editing the file
is_template = yes
; Language code for string file
# Language parameters
[Language]
; Available languages:
; it_IT - Italian, by Steffo
; en_US - English, by https://github.com/DarrenWestwood (incomplete, please improve it!)
; ua_UK - Ukrainian, by https://github.com/pzhuk
; ru_RU - Russian, by https://github.com/pzhuk
language = it_IT
; it - Italian, by https://github.com/Steffo99
; en - English, by https://github.com/DarrenWestwood
; uk - Ukrainian, by https://github.com/pzhuk
; ru - Russian, by https://github.com/pzhuk
; The lanugages that messages can be displayed in
enabled_languages = it | en | uk | ru
; The default language to be set for users whose language cannot be autodetected or whose language is not enabled
default_language = it
; The language to fallback to if a string is missing in a specific language
fallback_language = en
# Telegram bot parameters
[Telegram]

24
core.py
View file

@ -4,7 +4,7 @@ import worker
import configloader
import utils
import threading
import importlib
import localization
import logging
try:
@ -12,9 +12,6 @@ try:
except ImportError:
coloredlogs = None
language = configloader.config["Config"]["language"]
strings = importlib.import_module("strings." + language)
def main():
"""The core code of the program. Should be run only in the main process!"""
@ -48,6 +45,11 @@ def main():
sys.exit(1)
log.debug("Bot token is valid!")
# Finding default language
default_language = configloader.config["Language"]["default_language"]
# Creating localization object
default_loc = localization.Localization(language=default_language, fallback=default_language)
# Create a dictionary linking the chat ids to the Worker objects
# {"1234": <Worker>}
chat_workers = {}
@ -72,7 +74,7 @@ def main():
if update.message.chat.type != "private":
log.debug(f"Received a message from a non-private chat: {update.message.chat.id}")
# Notify the chat
bot.send_message(update.message.chat.id, strings.error_nonprivate_chat)
bot.send_message(update.message.chat.id, default_loc.get("error_nonprivate_chat"))
# Skip the update
continue
# If the message is a start command...
@ -85,7 +87,9 @@ def main():
log.debug(f"Received request to stop {old_worker.name}")
old_worker.stop("request")
# Initialize a new worker for the chat
new_worker = worker.Worker(bot, update.message.chat)
new_worker = worker.Worker(bot=bot,
chat=update.message.chat,
telegram_user=update.message.from_user)
# Start the worker
log.debug(f"Starting {new_worker.name}")
new_worker.start()
@ -99,12 +103,12 @@ def main():
if receiving_worker is None or not receiving_worker.is_alive():
log.debug(f"Received a message in a chat without worker: {update.message.chat.id}")
# Suggest that the user restarts the chat with /start
bot.send_message(update.message.chat.id, strings.error_no_worker_for_chat,
bot.send_message(update.message.chat.id, default_loc.get("error_no_worker_for_chat"),
reply_markup=telegram.ReplyKeyboardRemove())
# Skip the update
continue
# If the message contains the "Cancel" string defined in the strings file...
if update.message.text == strings.menu_cancel:
if update.message.text == receiving_worker.loc.get("menu_cancel"):
log.debug(f"Forwarding CancelSignal to {receiving_worker}")
# Send a CancelSignal to the worker instead of the update
receiving_worker.queue.put(worker.CancelSignal())
@ -120,7 +124,7 @@ def main():
if receiving_worker is None:
log.debug(f"Received a callback query in a chat without worker: {update.callback_query.from_user.id}")
# Suggest that the user restarts the chat with /start
bot.send_message(update.callback_query.from_user.id, strings.error_no_worker_for_chat)
bot.send_message(update.callback_query.from_user.id, default_loc.get("error_no_worker_for_chat"))
# Skip the update
continue
# Check if the pressed inline key is a cancel button
@ -146,7 +150,7 @@ def main():
try:
bot.answer_pre_checkout_query(update.pre_checkout_query.id,
ok=False,
error_message=strings.error_invoice_expired)
error_message=default_loc.get("error_invoice_expired"))
except telegram.error.BadRequest:
log.error("pre-checkout query expired before an answer could be sent!")
# Go to the next update

85
database.py
View file

@ -7,10 +7,10 @@ import configloader
import telegram
import requests
import utils
import importlib
import localization
import logging
language = configloader.config["Config"]["language"]
strings = importlib.import_module("strings." + language)
log = logging.getLogger(__name__)
# Create a (lazy) database engine
engine = create_engine(configloader.config["Database"]["engine"])
@ -31,6 +31,7 @@ class User(TableDeclarativeBase):
first_name = Column(String, nullable=False)
last_name = Column(String)
username = Column(String)
language = Column(String, nullable=False)
# Current wallet credit
credit = Column(Integer, nullable=False)
@ -38,14 +39,16 @@ class User(TableDeclarativeBase):
# Extra table parameters
__tablename__ = "users"
def __init__(self, telegram_chat: telegram.Chat, **kwargs):
def __init__(self, telegram_user: telegram.User, **kwargs):
# Initialize the super
super().__init__(**kwargs)
# Get the data from telegram
self.user_id = telegram_chat.id
self.first_name = telegram_chat.first_name
self.last_name = telegram_chat.last_name
self.username = telegram_chat.username
self.user_id = telegram_user.id
self.first_name = telegram_user.first_name
self.last_name = telegram_user.last_name
self.username = telegram_user.username
self.language = telegram_user.language_code if telegram_user.language_code else configloader.config["Language"][
"default_language"]
# The starting wallet value is 0
self.credit = 0
@ -74,6 +77,13 @@ class User(TableDeclarativeBase):
valid_transactions: typing.List[Transaction] = [t for t in self.transactions if not t.refunded]
self.credit = sum(map(lambda t: t.value, valid_transactions))
@property
def full_name(self):
if self.last_name:
return f"{self.first_name} {self.last_name}"
else:
return self.first_name
def __repr__(self):
return f"<User {self} having {self.credit} credit>"
@ -99,21 +109,18 @@ class Product(TableDeclarativeBase):
# No __init__ is needed, the default one is sufficient
def __str__(self):
return self.text()
def text(self, style: str="full", cart_qty: int=None):
def text(self, *, loc: localization.Localization, style: str = "full", cart_qty: int = None):
"""Return the product details formatted with Telegram HTML. The image is omitted."""
if style == "short":
return f"{cart_qty}x {utils.telegram_html_escape(self.name)} - {str(utils.Price(self.price) * cart_qty)}"
return f"{cart_qty}x {utils.telegram_html_escape(self.name)} - {str(utils.Price(self.price, loc) * cart_qty)}"
elif style == "full":
if cart_qty is not None:
cart = strings.in_cart_format_string.format(quantity=cart_qty)
cart = loc.get("in_cart_format_string", quantity=cart_qty)
else:
cart = ''
return strings.product_format_string.format(name=utils.telegram_html_escape(self.name),
return loc.get("product_format_string", name=utils.telegram_html_escape(self.name),
description=utils.telegram_html_escape(self.description),
price=str(utils.Price(self.price)),
price=str(utils.Price(self.price, loc)),
cart=cart)
else:
raise ValueError("style is not an accepted value")
@ -121,18 +128,18 @@ class Product(TableDeclarativeBase):
def __repr__(self):
return f"<Product {self.name}>"
def send_as_message(self, chat_id: int) -> dict:
def send_as_message(self, loc: localization.Localization, chat_id: int) -> dict:
"""Send a message containing the product data."""
if self.image is None:
r = requests.get(f"https://api.telegram.org/bot{configloader.config['Telegram']['token']}/sendMessage",
params={"chat_id": chat_id,
"text": self.text(),
"text": self.text(loc=loc),
"parse_mode": "HTML"})
else:
r = requests.post(f"https://api.telegram.org/bot{configloader.config['Telegram']['token']}/sendPhoto",
files={"photo": self.image},
params={"chat_id": chat_id,
"caption": self.text(),
"caption": self.text(loc=loc),
"parse_mode": "HTML"})
return r.json()
@ -181,10 +188,10 @@ class Transaction(TableDeclarativeBase):
__tablename__ = "transactions"
__table_args__ = (UniqueConstraint("provider", "provider_charge_id"),)
def __str__(self):
string = f"<b>T{self.transaction_id}</b> | {str(self.user)} | {utils.Price(self.value)}"
def text(self, *, loc: localization.Localization):
string = f"<b>T{self.transaction_id}</b> | {str(self.user)} | {utils.Price(self.value, loc)}"
if self.refunded:
string += f" | {strings.emoji_refunded}"
string += f" | {loc.get('emoji_refunded')}"
if self.provider:
string += f" | {self.provider}"
if self.notes:
@ -247,36 +254,38 @@ class Order(TableDeclarativeBase):
def __repr__(self):
return f"<Order {self.order_id} placed by User {self.user_id}>"
def get_text(self, session, user=False):
def text(self, *, loc: localization.Localization, session, user=False):
joined_self = session.query(Order).filter_by(order_id=self.order_id).join(Transaction).one()
items = ""
for item in self.items:
items += str(item) + "\n"
if self.delivery_date is not None:
status_emoji = strings.emoji_completed
status_text = strings.text_completed
status_emoji = loc.get("emoji_completed")
status_text = loc.get("text_completed")
elif self.refund_date is not None:
status_emoji = strings.emoji_refunded
status_text = strings.text_refunded
status_emoji = loc.get("emoji_refunded")
status_text = loc.get("text_refunded")
else:
status_emoji = strings.emoji_not_processed
status_text = strings.text_not_processed
status_emoji = loc.get("emoji_not_processed")
status_text = loc.get("text_not_processed")
if user and configloader.config["Appearance"]["full_order_info"] == "no":
return strings.user_order_format_string.format(status_emoji=status_emoji,
return loc.get("user_order_format_string",
status_emoji=status_emoji,
status_text=status_text,
items=items,
notes=self.notes,
value=str(utils.Price(-joined_self.transaction.value))) + \
(strings.refund_reason.format(reason=self.refund_reason) if self.refund_date is not None else "")
value=str(utils.Price(-joined_self.transaction.value, loc))) + \
(loc.get("refund_reason", reason=self.refund_reason) if self.refund_date is not None else "")
else:
return status_emoji + " " + \
strings.order_number.format(id=self.order_id) + "\n" + \
strings.order_format_string.format(user=self.user.mention(),
loc.get("order_number", id=self.order_id) + "\n" + \
loc.get("order_format_string",
user=self.user.mention(),
date=self.creation_date.isoformat(),
items=items,
notes=self.notes if self.notes is not None else "",
value=str(utils.Price(-joined_self.transaction.value))) + \
(strings.refund_reason.format(reason=self.refund_reason) if self.refund_date is not None else "")
value=str(utils.Price(-joined_self.transaction.value, loc))) + \
(loc.get("refund_reason", reason=self.refund_reason) if self.refund_date is not None else "")
class OrderItem(TableDeclarativeBase):
@ -293,8 +302,8 @@ class OrderItem(TableDeclarativeBase):
# Extra table parameters
__tablename__ = "orderitems"
def __str__(self):
return f"{self.product.name} - {str(utils.Price(self.product.price))}"
def text(self, *, loc: localization.Localization):
return f"{self.product.name} - {str(utils.Price(self.product.price, loc))}"
def __repr__(self):
return f"<OrderItem {self.item_id}>"

59
localization.py Normal file
View file

@ -0,0 +1,59 @@
from typing import *
import importlib
import types
import logging
import json
log = logging.getLogger(__name__)
class IgnoreDict(dict):
"""A dictionary that if passed to format_map, ignores the missing replacement fields."""
def __missing__(self, key):
return "{" + key + "}"
class Localization:
def __init__(self, language: str, *, fallback: str, replacements: Dict[str, str] = None):
log.debug(f"Creating localization for {language}")
self.language: str = language
log.debug(f"Importing strings.{language}")
self.module: types.ModuleType = importlib.import_module(f"strings.{language}")
if language != fallback:
log.debug(f"Importing strings.{fallback} as fallback")
self.fallback_language: str = fallback
self.fallback_module = importlib.import_module(f"strings.{fallback}") if fallback else None
else:
log.debug("Language is the same as the default, not importing any fallback")
self.fallback_language = None
self.fallback_module = None
self.replacements: Dict[str, str] = replacements if replacements else {}
def get(self, key: str, **kwargs) -> str:
try:
log.debug(f"Getting localized string with key {key}")
string = self.module.__getattribute__(key)
except AttributeError:
if self.fallback_module:
log.warning(f"Missing localized string with key {key}, using default")
string = self.fallback_module.__getattribute__(key)
else:
raise
assert isinstance(string, str)
formatter = IgnoreDict(**self.replacements, **kwargs)
return string.format_map(formatter)
def boolmoji(self, boolean: bool) -> str:
return self.get("emoji_yes") if boolean else self.get("emoji_no")
def create_json_localization_file_from_strings(language: str):
module: types.ModuleType = importlib.import_module(f"strings.{language}")
raw = module.__dict__
clean = {}
for key in raw:
if not (key.startswith("__") and key.endswith("__")):
clean[key] = raw[key]
with open(f"locale/{language}.json", "w") as file:
json.dump(clean, file)

6
strings/en_US.py → strings/en.py
View file

@ -118,6 +118,9 @@ conversation_open_help_menu = "What kind of help do you need?"
conversation_confirm_admin_promotion = "Are you sure you want to promote this user to 💼 Manager?\n" \
"It is an irreversible action!"
# Conversation: language select menu header
conversation_language_select = "Select a language:"
# Conversation: switching to user mode
conversation_switch_to_user_mode = " You are switching to 👤 Customer mode.\n" \
"If you want to go back to the 💼 Manager menu, restart the conversation with /start."
@ -214,6 +217,9 @@ menu_csv = "📄 .csv"
# Menu: edit admins list
menu_edit_admins = "🏵 Edit Managers"
# Menu: language
menu_language = "🇬🇧 Language"
# Emoji: unprocessed order
emoji_not_processed = "*️⃣"

6
strings/it_IT.py → strings/it.py
View file

@ -118,6 +118,9 @@ conversation_open_help_menu = "Che tipo di assistenza desideri ricevere?"
conversation_confirm_admin_promotion = "Sei sicuro di voler promuovere questo utente a 💼 Gestore?\n" \
"E' un'azione irreversibile!"
# Conversation: language select menu header
conversation_language_select = "Scegli una lingua:"
# Conversation: switching to user mode
conversation_switch_to_user_mode = "Stai passando alla modalità 👤 Cliente.\n" \
"Se vuoi riassumere il ruolo di 💼 Gestore, riavvia la conversazione con /start."
@ -214,6 +217,9 @@ menu_csv = "📄 .csv"
# Menu: edit admins list
menu_edit_admins = "🏵 Modifica gestori"
# Menu: language
menu_language = "🇮🇹 Lingua"
# Emoji: unprocessed order
emoji_not_processed = "*️⃣"

0
strings/ru_RU.py → strings/ru.py
View file

0
strings/uk_UA.py → strings/uk.py
View file

46
utils.py
View file

@ -8,18 +8,12 @@ import sys
import importlib
import logging
import traceback
import localization
log = logging.getLogger(__name__)
language = config["Config"]["language"]
try:
strings = importlib.import_module("strings." + language)
except ModuleNotFoundError:
print("The strings file you specified in the config file does not exist.")
sys.exit(1)
if config["Error Reporting"]["sentry_token"] != \
"https://00000000000000000000000000000000:00000000000000000000000000000000@sentry.io/0000000":
import raven
@ -39,7 +33,9 @@ class Price:
"""The base class for the prices in greed.
Its int value is in minimum units, while its float and str values are in decimal format.int("""
def __init__(self, value: typing.Union[int, float, str, "Price"] = 0):
def __init__(self, value: typing.Union[int, float, str, "Price"], loc: localization.Localization):
# Keep a reference to the localization file
self.loc = loc
if isinstance(value, int):
# Keep the value as it is
self.value = int(value)
@ -57,9 +53,9 @@ class Price:
return f"<Price of value {self.value}>"
def __str__(self):
return strings.currency_format_string.format(symbol=(config["Payments"]["currency_symbol"] or strings.currency_symbol),
value="{0:.2f}".format(
self.value / (10 ** int(config["Payments"]["currency_exp"]))))
return self.loc.get("currency_format_string",
symbol=config["Payments"]["currency_symbol"],
value="{0:.2f}".format(self.value / (10 ** int(config["Payments"]["currency_exp"]))))
def __int__(self):
return self.value
@ -68,48 +64,48 @@ class Price:
return self.value / (10 ** int(config["Payments"]["currency_exp"]))
def __ge__(self, other):
return self.value >= Price(other).value
return self.value >= Price(other, self.loc).value
def __le__(self, other):
return self.value <= Price(other).value
return self.value <= Price(other, self.loc).value
def __eq__(self, other):
return self.value == Price(other).value
return self.value == Price(other, self.loc).value
def __gt__(self, other):
return self.value > Price(other).value
return self.value > Price(other, self.loc).value
def __lt__(self, other):
return self.value < Price(other).value
return self.value < Price(other, self.loc).value
def __add__(self, other):
return Price(self.value + Price(other).value)
return Price(self.value + Price(other, self.loc).value, self.loc)
def __sub__(self, other):
return Price(self.value - Price(other).value)
return Price(self.value - Price(other, self.loc).value, self.loc)
def __mul__(self, other):
return Price(int(self.value * other))
return Price(int(self.value * other), self.loc)
def __floordiv__(self, other):
return Price(int(self.value // other))
return Price(int(self.value // other), self.loc)
def __radd__(self, other):
return self.__add__(other)
def __rsub__(self, other):
return Price(Price(other).value - self.value)
return Price(Price(other, self.loc).value - self.value, self.loc)
def __rmul__(self, other):
return self.__mul__(other)
def __iadd__(self, other):
self.value += Price(other).value
self.value += Price(other, self.loc).value
return self
def __isub__(self, other):
self.value -= Price(other).value
self.value -= Price(other, self.loc).value
return self
def __imul__(self, other):
@ -235,7 +231,3 @@ class DuckBot:
return self.bot.send_document(*args, **kwargs)
# More methods can be added here
def boolmoji(boolean: bool):
return strings.emoji_yes if boolean else strings.emoji_no

439
worker.py
View file

@ -13,14 +13,11 @@ import os
import traceback
from html import escape
import requests
import importlib
import logging
import localization
log = logging.getLogger(__name__)
language = configloader.config["Config"]["language"]
strings = importlib.import_module("strings." + language)
class StopSignal:
"""A data class that should be sent to the worker when the conversation has to be stopped abnormally."""
@ -36,12 +33,13 @@ class CancelSignal:
class Worker(threading.Thread):
"""A worker for a single conversation. A new one is created every time the /start command is sent."""
def __init__(self, bot: utils.DuckBot, chat: telegram.Chat, *args, **kwargs):
def __init__(self, bot: utils.DuckBot, chat: telegram.Chat, telegram_user: telegram.User, *args, **kwargs):
# Initialize the thread
super().__init__(name=f"Worker {chat.id}", *args, **kwargs)
# Store the bot and chat info inside the class
self.bot: utils.DuckBot = bot
self.chat: telegram.Chat = chat
self.telegram_user: telegram.User = telegram_user
# Open a new database session
log.debug(f"Opening new database session for {self.name}")
self.session = db.Session()
@ -52,6 +50,8 @@ class Worker(threading.Thread):
self.queue = queuem.Queue()
# The current active invoice payload; reject all invoices with a different payload
self.invoice_payload = None
# The localization strings for this user
self.loc = None
# The Sentry client for reporting errors encountered by the user
if configloader.config["Error Reporting"]["sentry_token"] != \
"https://00000000000000000000000000000000:00000000000000000000000000000000@sentry.io/0000000":
@ -69,9 +69,7 @@ class Worker(threading.Thread):
def run(self):
"""The conversation code."""
# Welcome the user to the bot
log.debug("Starting conversation")
self.bot.send_message(self.chat.id, strings.conversation_after_start)
# Get the user db data from the users and admin tables
self.user = self.session.query(db.User).filter(db.User.user_id == self.chat.id).one_or_none()
self.admin = self.session.query(db.Admin).filter(db.Admin.user_id == self.chat.id).one_or_none()
@ -80,7 +78,7 @@ class Worker(threading.Thread):
# Check if there are other registered users: if there aren't any, the first user will be owner of the bot
will_be_owner = (self.session.query(db.Admin).first() is None)
# Create the new record
self.user = db.User(self.chat)
self.user = db.User(self.telegram_user)
# Add the new record to the db
self.session.add(self.user)
# Flush the session to get an userid
@ -102,9 +100,13 @@ class Worker(threading.Thread):
log.info(f"Created new user: {self.user}")
if will_be_owner:
log.warning(f"User was auto-promoted to Admin as no other admins existed: {self.user}")
# Create the localization object
self.__create_localization()
# Capture exceptions that occour during the conversation
# noinspection PyBroadException
try:
# Welcome the user to the bot
self.bot.send_message(self.chat.id, self.loc.get("conversation_after_start"))
# If the user is not an admin, send him to the user menu
if self.admin is None:
self.__user_menu()
@ -120,7 +122,7 @@ class Worker(threading.Thread):
# Try to notify the user of the exception
# noinspection PyBroadException
try:
self.bot.send_message(self.chat.id, strings.fatal_conversation_exception)
self.bot.send_message(self.chat.id, self.loc.get("fatal_conversation_exception"))
except Exception:
pass
# If the Sentry integration is enabled, log the exception
@ -309,7 +311,7 @@ class Worker(threading.Thread):
# Find all the users in the database
users = self.session.query(db.User).order_by(db.User.user_id).all()
# Create a list containing all the keyboard button strings
keyboard_buttons = [[strings.menu_cancel]]
keyboard_buttons = [[self.loc.get("menu_cancel")]]
# Add to the list all the users
for user in users:
keyboard_buttons.append([user.identifiable_str()])
@ -318,7 +320,7 @@ class Worker(threading.Thread):
# Keep asking until a result is returned
while True:
# Send the keyboard
self.bot.send_message(self.chat.id, strings.conversation_admin_select_user, reply_markup=keyboard)
self.bot.send_message(self.chat.id, self.loc.get("conversation_admin_select_user"), reply_markup=keyboard)
# Wait for a reply
reply = self.__wait_for_regex("user_([0-9]+)", cancellable=True)
# Propagate CancelSignals
@ -328,7 +330,7 @@ class Worker(threading.Thread):
user = self.session.query(db.User).filter_by(user_id=int(reply)).one_or_none()
# Ensure the user exists
if not user:
self.bot.send_message(self.chat.id, strings.error_user_does_not_exist)
self.bot.send_message(self.chat.id, self.loc.get("error_user_does_not_exist"))
continue
return user
@ -339,38 +341,50 @@ class Worker(threading.Thread):
# Loop used to returning to the menu after executing a command
while True:
# Create a keyboard with the user main menu
keyboard = [[telegram.KeyboardButton(strings.menu_order)],
[telegram.KeyboardButton(strings.menu_order_status)],
[telegram.KeyboardButton(strings.menu_add_credit)],
[telegram.KeyboardButton(strings.menu_help), telegram.KeyboardButton(strings.menu_bot_info)]]
keyboard = [[telegram.KeyboardButton(self.loc.get("menu_order"))],
[telegram.KeyboardButton(self.loc.get("menu_order_status"))],
[telegram.KeyboardButton(self.loc.get("menu_add_credit"))],
[telegram.KeyboardButton(self.loc.get("menu_language"))],
[telegram.KeyboardButton(self.loc.get("menu_help")),
telegram.KeyboardButton(self.loc.get("menu_bot_info"))]]
# Send the previously created keyboard to the user (ensuring it can be clicked only 1 time)
self.bot.send_message(self.chat.id,
strings.conversation_open_user_menu.format(credit=utils.Price(self.user.credit)),
self.loc.get("conversation_open_user_menu",
credit=utils.Price(self.user.credit, self.loc)),
reply_markup=telegram.ReplyKeyboardMarkup(keyboard, one_time_keyboard=True))
# Wait for a reply from the user
selection = self.__wait_for_specific_message([strings.menu_order, strings.menu_order_status,
strings.menu_add_credit, strings.menu_bot_info,
strings.menu_help])
selection = self.__wait_for_specific_message([
self.loc.get("menu_order"),
self.loc.get("menu_order_status"),
self.loc.get("menu_add_credit"),
self.loc.get("menu_language"),
self.loc.get("menu_help"),
self.loc.get("menu_bot_info"),
])
# After the user reply, update the user data
self.update_user()
# If the user has selected the Order option...
if selection == strings.menu_order:
if selection == self.loc.get("menu_order"):
# Open the order menu
self.__order_menu()
# If the user has selected the Order Status option...
elif selection == strings.menu_order_status:
elif selection == self.loc.get("menu_order_status"):
# Display the order(s) status
self.__order_status()
# If the user has selected the Add Credit option...
elif selection == strings.menu_add_credit:
elif selection == self.loc.get("menu_add_credit"):
# Display the add credit menu
self.__add_credit_menu()
# If the user has selected the Language option...
elif selection == self.loc.get("menu_language"):
# Display the language menu
self.__language_menu()
# If the user has selected the Bot Info option...
elif selection == strings.menu_bot_info:
elif selection == self.loc.get("menu_bot_info"):
# Display information about the bot
self.__bot_info()
# If the user has selected the Help option...
elif selection == strings.menu_help:
elif selection == self.loc.get("menu_help"):
# Go to the Help menu
self.__help_menu()
@ -392,24 +406,27 @@ class Worker(threading.Thread):
# Add the product to the cart
cart[message['result']['message_id']] = [product, 0]
# Create the inline keyboard to add the product to the cart
inline_keyboard = telegram.InlineKeyboardMarkup([[telegram.InlineKeyboardButton(strings.menu_add_to_cart,
callback_data="cart_add")]])
inline_keyboard = telegram.InlineKeyboardMarkup(
[[telegram.InlineKeyboardButton(self.loc.get("menu_add_to_cart"), callback_data="cart_add")]]
)
# Edit the sent message and add the inline keyboard
if product.image is None:
self.bot.edit_message_text(chat_id=self.chat.id,
message_id=message['result']['message_id'],
text=product.text(),
text=product.text(loc=self.loc),
reply_markup=inline_keyboard)
else:
self.bot.edit_message_caption(chat_id=self.chat.id,
message_id=message['result']['message_id'],
caption=product.text(),
caption=product.text(loc=self.loc),
reply_markup=inline_keyboard)
# Create the keyboard with the cancel button
inline_keyboard = telegram.InlineKeyboardMarkup([[telegram.InlineKeyboardButton(strings.menu_cancel,
inline_keyboard = telegram.InlineKeyboardMarkup([[telegram.InlineKeyboardButton(self.loc.get("menu_cancel"),
callback_data="cart_cancel")]])
# Send a message containing the button to cancel or pay
final_msg = self.bot.send_message(self.chat.id, strings.conversation_cart_actions, reply_markup=inline_keyboard)
final_msg = self.bot.send_message(self.chat.id,
self.loc.get("conversation_cart_actions"),
reply_markup=inline_keyboard)
# Wait for user input
while True:
callback = self.__wait_for_inlinekeyboard_callback()
@ -430,31 +447,36 @@ class Worker(threading.Thread):
# Create the product inline keyboard
product_inline_keyboard = telegram.InlineKeyboardMarkup(
[
[telegram.InlineKeyboardButton(strings.menu_add_to_cart, callback_data="cart_add"),
telegram.InlineKeyboardButton(strings.menu_remove_from_cart, callback_data="cart_remove")]
[telegram.InlineKeyboardButton(self.loc.get("menu_add_to_cart"),
callback_data="cart_add"),
telegram.InlineKeyboardButton(self.loc.get("menu_remove_from_cart"),
callback_data="cart_remove")]
])
# Create the final inline keyboard
final_inline_keyboard = telegram.InlineKeyboardMarkup(
[
[telegram.InlineKeyboardButton(strings.menu_cancel, callback_data="cart_cancel")],
[telegram.InlineKeyboardButton(strings.menu_done, callback_data="cart_done")]
[telegram.InlineKeyboardButton(self.loc.get("menu_cancel"), callback_data="cart_cancel")],
[telegram.InlineKeyboardButton(self.loc.get("menu_done"), callback_data="cart_done")]
])
# Edit both the product and the final message
if product.image is None:
self.bot.edit_message_text(chat_id=self.chat.id,
message_id=callback.message.message_id,
text=product.text(cart_qty=cart[callback.message.message_id][1]),
text=product.text(loc=self.loc,
cart_qty=cart[callback.message.message_id][1]),
reply_markup=product_inline_keyboard)
else:
self.bot.edit_message_caption(chat_id=self.chat.id,
message_id=callback.message.message_id,
caption=product.text(cart_qty=cart[callback.message.message_id][1]),
caption=product.text(loc=self.loc,
cart_qty=cart[callback.message.message_id][1]),
reply_markup=product_inline_keyboard)
self.bot.edit_message_text(
chat_id=self.chat.id,
message_id=final_msg.message_id,
text=strings.conversation_confirm_cart.format(product_list=self.__get_cart_summary(cart),
text=self.loc.get("conversation_confirm_cart",
product_list=self.__get_cart_summary(cart),
total_cost=str(self.__get_cart_value(cart))),
reply_markup=final_inline_keyboard)
# If the Remove from cart button has been pressed...
@ -470,35 +492,39 @@ class Worker(threading.Thread):
else:
continue
# Create the product inline keyboard
product_inline_list = [[telegram.InlineKeyboardButton(strings.menu_add_to_cart,
product_inline_list = [[telegram.InlineKeyboardButton(self.loc.get("menu_add_to_cart"),
callback_data="cart_add")]]
if cart[callback.message.message_id][1] > 0:
product_inline_list[0].append(telegram.InlineKeyboardButton(strings.menu_remove_from_cart,
product_inline_list[0].append(telegram.InlineKeyboardButton(self.loc.get("menu_remove_from_cart"),
callback_data="cart_remove"))
product_inline_keyboard = telegram.InlineKeyboardMarkup(product_inline_list)
# Create the final inline keyboard
final_inline_list = [[telegram.InlineKeyboardButton(strings.menu_cancel, callback_data="cart_cancel")]]
final_inline_list = [[telegram.InlineKeyboardButton(self.loc.get("menu_cancel"),
callback_data="cart_cancel")]]
for product_id in cart:
if cart[product_id][1] > 0:
final_inline_list.append([telegram.InlineKeyboardButton(strings.menu_done,
final_inline_list.append([telegram.InlineKeyboardButton(self.loc.get("menu_done"),
callback_data="cart_done")])
break
final_inline_keyboard = telegram.InlineKeyboardMarkup(final_inline_list)
# Edit the product message
if product.image is None:
self.bot.edit_message_text(chat_id=self.chat.id, message_id=callback.message.message_id,
text=product.text(cart_qty=cart[callback.message.message_id][1]),
text=product.text(loc=self.loc,
cart_qty=cart[callback.message.message_id][1]),
reply_markup=product_inline_keyboard)
else:
self.bot.edit_message_caption(chat_id=self.chat.id,
message_id=callback.message.message_id,
caption=product.text(cart_qty=cart[callback.message.message_id][1]),
caption=product.text(loc=self.loc,
cart_qty=cart[callback.message.message_id][1]),
reply_markup=product_inline_keyboard)
self.bot.edit_message_text(
chat_id=self.chat.id,
message_id=final_msg.message_id,
text=strings.conversation_confirm_cart.format(product_list=self.__get_cart_summary(cart),
text=self.loc.get("conversation_confirm_cart",
product_list=self.__get_cart_summary(cart),
total_cost=str(self.__get_cart_value(cart))),
reply_markup=final_inline_keyboard)
# If the done button has been pressed...
@ -506,10 +532,10 @@ class Worker(threading.Thread):
# End the loop
break
# Create an inline keyboard with a single skip button
cancel = telegram.InlineKeyboardMarkup([[telegram.InlineKeyboardButton(strings.menu_skip,
cancel = telegram.InlineKeyboardMarkup([[telegram.InlineKeyboardButton(self.loc.get("menu_skip"),
callback_data="cmd_cancel")]])
# Ask if the user wants to add notes to the order
self.bot.send_message(self.chat.id, strings.ask_order_notes, reply_markup=cancel)
self.bot.send_message(self.chat.id, self.loc.get("ask_order_notes"), reply_markup=cancel)
# Wait for user input
notes = self.__wait_for_regex(r"(.*)", cancellable=True)
# Create a new Order
@ -530,14 +556,14 @@ class Worker(threading.Thread):
credit_required = self.__get_cart_value(cart) - self.user.credit
# Notify user in case of insufficient credit
if credit_required > 0:
self.bot.send_message(self.chat.id, strings.error_not_enough_credit)
self.bot.send_message(self.chat.id, self.loc.get("error_not_enough_credit"))
# Suggest payment for missing credit value if configuration allows refill
if configloader.config["Credit Card"]["credit_card_token"] != "" \
and configloader.config["Appearance"]["refill_on_checkout"] == 'yes' \
and utils.Price(int(configloader.config["Credit Card"]["min_amount"])) <= \
and utils.Price(int(configloader.config["Credit Card"]["min_amount"]), self.loc) <= \
credit_required <= \
utils.Price(int(configloader.config["Credit Card"]["max_amount"])):
self.__make_payment(utils.Price(credit_required))
utils.Price(int(configloader.config["Credit Card"]["max_amount"]), self.loc):
self.__make_payment(utils.Price(credit_required, self.loc))
# If afer requested payment credit is still insufficient (either payment failure or cancel)
if self.user.credit < self.__get_cart_value(cart):
# Rollback all the changes
@ -546,21 +572,21 @@ class Worker(threading.Thread):
# User has credit and valid order, perform transaction now
self.__order_transaction(order=order, value=-int(self.__get_cart_value(cart)))
@staticmethod
def __get_cart_value(cart):
def __get_cart_value(self, cart):
# Calculate total items value in cart
value = utils.Price(0)
value = utils.Price(0, self.loc)
for product in cart:
value += cart[product][0].price * cart[product][1]
return value
@staticmethod
def __get_cart_summary(cart):
def __get_cart_summary(self, cart):
# Create the cart summary
product_list = ""
for product_id in cart:
if cart[product_id][1] > 0:
product_list += cart[product_id][0].text(style="short", cart_qty=cart[product_id][1]) + "\n"
product_list += cart[product_id][0].text(loc=self.loc,
style="short",
cart_qty=cart[product_id][1]) + "\n"
return product_list
def __order_transaction(self, order, value):
@ -580,20 +606,22 @@ class Worker(threading.Thread):
def __order_notify_admins(self, order):
# Notify the user of the order result
self.bot.send_message(self.chat.id, strings.success_order_created.format(order=order.get_text(self.session,
self.bot.send_message(self.chat.id, self.loc.get("success_order_created", order=order.text(loc=self.loc,
session=self.session,
user=True)))
# Notify the admins (in Live Orders mode) of the new order
admins = self.session.query(db.Admin).filter_by(live_mode=True).all()
# Create the order keyboard
order_keyboard = telegram.InlineKeyboardMarkup(
[
[telegram.InlineKeyboardButton(strings.menu_complete, callback_data="order_complete")],
[telegram.InlineKeyboardButton(strings.menu_refund, callback_data="order_refund")]
[telegram.InlineKeyboardButton(self.loc.get("menu_complete"), callback_data="order_complete")],
[telegram.InlineKeyboardButton(self.loc.get("menu_refund"), callback_data="order_refund")]
])
# Notify them of the new placed order
for admin in admins:
self.bot.send_message(admin.user_id,
f"{strings.notification_order_placed.format(order=order.get_text(self.session))}",
self.loc.get('notification_order_placed',
order=order.text(loc=self.loc, session=self.session)),
reply_markup=order_keyboard)
def __order_status(self):
@ -607,10 +635,10 @@ class Worker(threading.Thread):
.all()
# Ensure there is at least one order to display
if len(orders) == 0:
self.bot.send_message(self.chat.id, strings.error_no_orders)
self.bot.send_message(self.chat.id, self.loc.get("error_no_orders"))
# Display the order status to the user
for order in orders:
self.bot.send_message(self.chat.id, order.get_text(self.session, user=True))
self.bot.send_message(self.chat.id, order.text(loc=self.loc, session=self.session, user=True))
# TODO: maybe add a page displayer instead of showing the latest 5 orders
def __add_credit_menu(self):
@ -620,25 +648,25 @@ class Worker(threading.Thread):
keyboard = list()
# Add the supported payment methods to the keyboard
# Cash
keyboard.append([telegram.KeyboardButton(strings.menu_cash)])
keyboard.append([telegram.KeyboardButton(self.loc.get("menu_cash"))])
# Telegram Payments
if configloader.config["Credit Card"]["credit_card_token"] != "":
keyboard.append([telegram.KeyboardButton(strings.menu_credit_card)])
keyboard.append([telegram.KeyboardButton(self.loc.get("menu_credit_card"))])
# Keyboard: go back to the previous menu
keyboard.append([telegram.KeyboardButton(strings.menu_cancel)])
keyboard.append([telegram.KeyboardButton(self.loc.get("menu_cancel"))])
# Send the keyboard to the user
self.bot.send_message(self.chat.id, strings.conversation_payment_method,
self.bot.send_message(self.chat.id, self.loc.get("conversation_payment_method"),
reply_markup=telegram.ReplyKeyboardMarkup(keyboard, one_time_keyboard=True))
# Wait for a reply from the user
selection = self.__wait_for_specific_message([strings.menu_cash, strings.menu_credit_card, strings.menu_cancel],
selection = self.__wait_for_specific_message([self.loc.get("menu_cash"), self.loc.get("menu_credit_card"), self.loc.get("menu_cancel")],
cancellable=True)
# If the user has selected the Cash option...
if selection == strings.menu_cash:
if selection == self.loc.get("menu_cash"):
# Go to the pay with cash function
self.bot.send_message(self.chat.id,
strings.payment_cash.format(user_cash_id=self.user.identifiable_str()))
self.loc.get("payment_cash", user_cash_id=self.user.identifiable_str()))
# If the user has selected the Credit Card option...
elif selection == strings.menu_credit_card:
elif selection == self.loc.get("menu_credit_card"):
# Go to the pay with credit card function
self.__add_credit_cc()
# If the user has selected the Cancel option...
@ -651,36 +679,32 @@ class Worker(threading.Thread):
log.debug("Displaying __add_credit_cc")
# Create a keyboard to be sent later
presets = list(map(lambda s: s.strip(" "), configloader.config["Credit Card"]["payment_presets"].split('|')))
keyboard = [[telegram.KeyboardButton(str(utils.Price(preset)))] for preset in presets]
keyboard.append([telegram.KeyboardButton(strings.menu_cancel)])
keyboard = [[telegram.KeyboardButton(str(utils.Price(preset, self.loc)))] for preset in presets]
keyboard.append([telegram.KeyboardButton(self.loc.get("menu_cancel"))])
# Boolean variable to check if the user has cancelled the action
cancelled = False
# Loop used to continue asking if there's an error during the input
while not cancelled:
# Send the message and the keyboard
self.bot.send_message(self.chat.id, strings.payment_cc_amount,
self.bot.send_message(self.chat.id, self.loc.get("payment_cc_amount"),
reply_markup=telegram.ReplyKeyboardMarkup(keyboard, one_time_keyboard=True))
# Wait until a valid amount is sent
selection = self.__wait_for_regex(r"([0-9]+(?:[.,][0-9]+)?|" + strings.menu_cancel + r")", cancellable=True)
selection = self.__wait_for_regex(r"([0-9]+(?:[.,][0-9]+)?|" + self.loc.get("menu_cancel") + r")", cancellable=True)
# If the user cancelled the action
if isinstance(selection, CancelSignal):
# Exit the loop
cancelled = True
continue
# Convert the amount to an integer
value = utils.Price(selection)
value = utils.Price(selection, self.loc)
# Ensure the amount is within the range
if value > utils.Price(int(configloader.config["Credit Card"]["max_amount"])):
if value > utils.Price(int(configloader.config["Credit Card"]["max_amount"]), self.loc):
self.bot.send_message(self.chat.id,
strings.error_payment_amount_over_max.format(
max_amount=utils.Price(configloader.config["Credit Card"]["max_amount"]))
)
self.loc.get("error_payment_amount_over_max", max_amount=utils.Price(configloader.config["Credit Card"]["max_amount"], self.loc)))
continue
elif value < utils.Price(int(configloader.config["Credit Card"]["min_amount"])):
elif value < utils.Price(int(configloader.config["Credit Card"]["min_amount"]), self.loc):
self.bot.send_message(self.chat.id,
strings.error_payment_amount_under_min.format(
min_amount=utils.Price(configloader.config["Credit Card"]["min_amount"]))
)
self.loc.get("error_payment_amount_under_min", min_amount=utils.Price(configloader.config["Credit Card"]["min_amount"], self.loc)))
continue
break
# If the user cancelled the action...
@ -694,20 +718,20 @@ class Worker(threading.Thread):
# Set the invoice active invoice payload
self.invoice_payload = str(uuid.uuid4())
# Create the price array
prices = [telegram.LabeledPrice(label=strings.payment_invoice_label, amount=int(amount))]
prices = [telegram.LabeledPrice(label=self.loc.get("payment_invoice_label"), amount=int(amount))]
# If the user has to pay a fee when using the credit card, add it to the prices list
fee = int(self.__get_total_fee(amount))
if fee > 0:
prices.append(telegram.LabeledPrice(label=strings.payment_invoice_fee_label,
prices.append(telegram.LabeledPrice(label=self.loc.get("payment_invoice_fee_label"),
amount=fee))
# Create the invoice keyboard
inline_keyboard = telegram.InlineKeyboardMarkup([[telegram.InlineKeyboardButton(strings.menu_pay, pay=True)],
[telegram.InlineKeyboardButton(strings.menu_cancel,
inline_keyboard = telegram.InlineKeyboardMarkup([[telegram.InlineKeyboardButton(self.loc.get("menu_pay"), pay=True)],
[telegram.InlineKeyboardButton(self.loc.get("menu_cancel"),
callback_data="cmd_cancel")]])
# The amount is valid, send the invoice
self.bot.send_invoice(self.chat.id,
title=strings.payment_invoice_title,
description=strings.payment_invoice_description.format(amount=str(amount)),
title=self.loc.get("payment_invoice_title"),
description=self.loc.get("payment_invoice_description", amount=str(amount)),
payload=self.invoice_payload,
provider_token=configloader.config["Credit Card"]["credit_card_token"],
start_parameter="tempdeeplink",
@ -757,7 +781,7 @@ class Worker(threading.Thread):
def __bot_info(self):
"""Send information about the bot."""
log.debug("Displaying __bot_info")
self.bot.send_message(self.chat.id, strings.bot_info)
self.bot.send_message(self.chat.id, self.loc.get("bot_info"))
def __admin_menu(self):
"""Function called from the run method when the user is an administrator.
@ -768,52 +792,52 @@ class Worker(threading.Thread):
# Create a keyboard with the admin main menu based on the admin permissions specified in the db
keyboard = []
if self.admin.edit_products:
keyboard.append([strings.menu_products])
keyboard.append([self.loc.get("menu_products")])
if self.admin.receive_orders:
keyboard.append([strings.menu_orders])
keyboard.append([self.loc.get("menu_orders")])
if self.admin.create_transactions:
keyboard.append([strings.menu_edit_credit])
keyboard.append([strings.menu_transactions, strings.menu_csv])
keyboard.append([self.loc.get("menu_edit_credit")])
keyboard.append([self.loc.get("menu_transactions"), self.loc.get("menu_csv")])
if self.admin.is_owner:
keyboard.append([strings.menu_edit_admins])
keyboard.append([strings.menu_user_mode])
keyboard.append([self.loc.get("menu_edit_admins")])
keyboard.append([self.loc.get("menu_user_mode")])
# Send the previously created keyboard to the user (ensuring it can be clicked only 1 time)
self.bot.send_message(self.chat.id, strings.conversation_open_admin_menu,
self.bot.send_message(self.chat.id, self.loc.get("conversation_open_admin_menu"),
reply_markup=telegram.ReplyKeyboardMarkup(keyboard, one_time_keyboard=True),
)
# Wait for a reply from the user
selection = self.__wait_for_specific_message([strings.menu_products, strings.menu_orders,
strings.menu_user_mode, strings.menu_edit_credit,
strings.menu_transactions, strings.menu_csv,
strings.menu_edit_admins])
selection = self.__wait_for_specific_message([self.loc.get("menu_products"), self.loc.get("menu_orders"),
self.loc.get("menu_user_mode"), self.loc.get("menu_edit_credit"),
self.loc.get("menu_transactions"), self.loc.get("menu_csv"),
self.loc.get("menu_edit_admins")])
# If the user has selected the Products option...
if selection == strings.menu_products:
if selection == self.loc.get("menu_products"):
# Open the products menu
self.__products_menu()
# If the user has selected the Orders option...
elif selection == strings.menu_orders:
elif selection == self.loc.get("menu_orders"):
# Open the orders menu
self.__orders_menu()
# If the user has selected the Transactions option...
elif selection == strings.menu_edit_credit:
elif selection == self.loc.get("menu_edit_credit"):
# Open the edit credit menu
self.__create_transaction()
# If the user has selected the User mode option...
elif selection == strings.menu_user_mode:
elif selection == self.loc.get("menu_user_mode"):
# Tell the user how to go back to admin menu
self.bot.send_message(self.chat.id, strings.conversation_switch_to_user_mode)
self.bot.send_message(self.chat.id, self.loc.get("conversation_switch_to_user_mode"))
# Start the bot in user mode
self.__user_menu()
# If the user has selected the Add Admin option...
elif selection == strings.menu_edit_admins:
elif selection == self.loc.get("menu_edit_admins"):
# Open the edit admin menu
self.__add_admin()
# If the user has selected the Transactions option...
elif selection == strings.menu_transactions:
elif selection == self.loc.get("menu_transactions"):
# Open the transaction pages
self.__transaction_pages()
# If the user has selected the .csv option...
elif selection == strings.menu_csv:
elif selection == self.loc.get("menu_csv"):
# Generate the .csv file
self.__transactions_file()
@ -825,13 +849,13 @@ class Worker(threading.Thread):
# Create a list of product names
product_names = [product.name for product in products]
# Insert at the start of the list the add product option, the remove product option and the Cancel option
product_names.insert(0, strings.menu_cancel)
product_names.insert(1, strings.menu_add_product)
product_names.insert(2, strings.menu_delete_product)
product_names.insert(0, self.loc.get("menu_cancel"))
product_names.insert(1, self.loc.get("menu_add_product"))
product_names.insert(2, self.loc.get("menu_delete_product"))
# Create a keyboard using the product names
keyboard = [[telegram.KeyboardButton(product_name)] for product_name in product_names]
# Send the previously created keyboard to the user (ensuring it can be clicked only 1 time)
self.bot.send_message(self.chat.id, strings.conversation_admin_select_product,
self.bot.send_message(self.chat.id, self.loc.get("conversation_admin_select_product"),
reply_markup=telegram.ReplyKeyboardMarkup(keyboard, one_time_keyboard=True))
# Wait for a reply from the user
selection = self.__wait_for_specific_message(product_names, cancellable=True)
@ -840,11 +864,11 @@ class Worker(threading.Thread):
# Exit the menu
return
# If the user has selected the Add Product option...
elif selection == strings.menu_add_product:
elif selection == self.loc.get("menu_add_product"):
# Open the add product menu
self.__edit_product_menu()
# If the user has selected the Remove Product option...
elif selection == strings.menu_delete_product:
elif selection == self.loc.get("menu_delete_product"):
# Open the delete product menu
self.__delete_product_menu()
# If the user has selected a product
@ -858,15 +882,15 @@ class Worker(threading.Thread):
"""Add a product to the database or edit an existing one."""
log.debug("Displaying __edit_product_menu")
# Create an inline keyboard with a single skip button
cancel = telegram.InlineKeyboardMarkup([[telegram.InlineKeyboardButton(strings.menu_skip,
cancel = telegram.InlineKeyboardMarkup([[telegram.InlineKeyboardButton(self.loc.get("menu_skip"),
callback_data="cmd_cancel")]])
# Ask for the product name until a valid product name is specified
while True:
# Ask the question to the user
self.bot.send_message(self.chat.id, strings.ask_product_name)
self.bot.send_message(self.chat.id, self.loc.get("ask_product_name"))
# Display the current name if you're editing an existing product
if product:
self.bot.send_message(self.chat.id, strings.edit_current_value.format(value=escape(product.name)),
self.bot.send_message(self.chat.id, self.loc.get("edit_current_value", value=escape(product.name)),
reply_markup=cancel)
# Wait for an answer
name = self.__wait_for_regex(r"(.*)", cancellable=bool(product))
@ -875,24 +899,24 @@ class Worker(threading.Thread):
self.session.query(db.Product).filter_by(name=name, deleted=False).one_or_none() in [None, product]:
# Exit the loop
break
self.bot.send_message(self.chat.id, strings.error_duplicate_name)
self.bot.send_message(self.chat.id, self.loc.get("error_duplicate_name"))
# Ask for the product description
self.bot.send_message(self.chat.id, strings.ask_product_description)
self.bot.send_message(self.chat.id, self.loc.get("ask_product_description"))
# Display the current description if you're editing an existing product
if product:
self.bot.send_message(self.chat.id,
strings.edit_current_value.format(value=escape(product.description)),
self.loc.get("edit_current_value", value=escape(product.description)),
reply_markup=cancel)
# Wait for an answer
description = self.__wait_for_regex(r"(.*)", cancellable=bool(product))
# Ask for the product price
self.bot.send_message(self.chat.id,
strings.ask_product_price)
self.loc.get("ask_product_price"))
# Display the current name if you're editing an existing product
if product:
self.bot.send_message(self.chat.id,
strings.edit_current_value.format(
value=(str(utils.Price(product.price))
self.loc.get("edit_current_value",
value=(str(utils.Price(product.price, self.loc))
if product.price is not None else 'Non in vendita')),
reply_markup=cancel)
# Wait for an answer
@ -904,9 +928,9 @@ class Worker(threading.Thread):
elif price.lower() == "x":
price = None
else:
price = utils.Price(price)
price = utils.Price(price, self.loc)
# Ask for the product image
self.bot.send_message(self.chat.id, strings.ask_product_image, reply_markup=cancel)
self.bot.send_message(self.chat.id, self.loc.get("ask_product_image"), reply_markup=cancel)
# Wait for an answer
photo_list = self.__wait_for_photo(cancellable=True)
# If a new product is being added...
@ -935,14 +959,14 @@ class Worker(threading.Thread):
# Get the file object associated with the photo
photo_file = self.bot.get_file(largest_photo.file_id)
# Notify the user that the bot is downloading the image and might be inactive for a while
self.bot.send_message(self.chat.id, strings.downloading_image)
self.bot.send_message(self.chat.id, self.loc.get("downloading_image"))
self.bot.send_chat_action(self.chat.id, action="upload_photo")
# Set the image for that product
product.set_image(photo_file)
# Commit the session changes
self.session.commit()
# Notify the user
self.bot.send_message(self.chat.id, strings.success_product_edited)
self.bot.send_message(self.chat.id, self.loc.get("success_product_edited"))
def __delete_product_menu(self):
log.debug("Displaying __delete_product_menu")
@ -951,11 +975,11 @@ class Worker(threading.Thread):
# Create a list of product names
product_names = [product.name for product in products]
# Insert at the start of the list the Cancel button
product_names.insert(0, strings.menu_cancel)
product_names.insert(0, self.loc.get("menu_cancel"))
# Create a keyboard using the product names
keyboard = [[telegram.KeyboardButton(product_name)] for product_name in product_names]
# Send the previously created keyboard to the user (ensuring it can be clicked only 1 time)
self.bot.send_message(self.chat.id, strings.conversation_admin_select_product_to_delete,
self.bot.send_message(self.chat.id, self.loc.get("conversation_admin_select_product_to_delete"),
reply_markup=telegram.ReplyKeyboardMarkup(keyboard, one_time_keyboard=True))
# Wait for a reply from the user
selection = self.__wait_for_specific_message(product_names, cancellable=True)
@ -969,22 +993,22 @@ class Worker(threading.Thread):
product.deleted = True
self.session.commit()
# Notify the user
self.bot.send_message(self.chat.id, strings.success_product_deleted)
self.bot.send_message(self.chat.id, self.loc.get("success_product_deleted"))
def __orders_menu(self):
"""Display a live flow of orders."""
log.debug("Displaying __orders_menu")
# Create a cancel and a stop keyboard
stop_keyboard = telegram.InlineKeyboardMarkup([[telegram.InlineKeyboardButton(strings.menu_stop,
stop_keyboard = telegram.InlineKeyboardMarkup([[telegram.InlineKeyboardButton(self.loc.get("menu_stop"),
callback_data="cmd_cancel")]])
cancel_keyboard = telegram.InlineKeyboardMarkup([[telegram.InlineKeyboardButton(strings.menu_cancel,
cancel_keyboard = telegram.InlineKeyboardMarkup([[telegram.InlineKeyboardButton(self.loc.get("menu_cancel"),
callback_data="cmd_cancel")]])
# Send a small intro message on the Live Orders mode
self.bot.send_message(self.chat.id, strings.conversation_live_orders_start, reply_markup=stop_keyboard)
self.bot.send_message(self.chat.id, self.loc.get("conversation_live_orders_start"), reply_markup=stop_keyboard)
# Create the order keyboard
order_keyboard = telegram.InlineKeyboardMarkup([[telegram.InlineKeyboardButton(strings.menu_complete,
order_keyboard = telegram.InlineKeyboardMarkup([[telegram.InlineKeyboardButton(self.loc.get("menu_complete"),
callback_data="order_complete")],
[telegram.InlineKeyboardButton(strings.menu_refund,
[telegram.InlineKeyboardButton(self.loc.get("menu_refund"),
callback_data="order_refund")]])
# Display the past pending orders
orders = self.session.query(db.Order) \
@ -995,7 +1019,7 @@ class Worker(threading.Thread):
# Create a message for every one of them
for order in orders:
# Send the created message
self.bot.send_message(self.chat.id, order.get_text(session=self.session),
self.bot.send_message(self.chat.id, order.text(loc=self.loc, session=self.session),
reply_markup=order_keyboard)
# Set the Live mode flag to True
self.admin.live_mode = True
@ -1010,12 +1034,12 @@ class Worker(threading.Thread):
self.admin.live_mode = False
break
# Find the order
order_id = re.search(strings.order_number.replace("{id}", "([0-9]+)"), update.message.text).group(1)
order_id = re.search(self.loc.get("order_number").replace("{id}", "([0-9]+)"), update.message.text).group(1)
order = self.session.query(db.Order).filter(db.Order.order_id == order_id).one()
# Check if the order hasn't been already cleared
if order.delivery_date is not None or order.refund_date is not None:
# Notify the admin and skip that order
self.bot.edit_message_text(self.chat.id, strings.error_order_already_cleared)
self.bot.edit_message_text(self.chat.id, self.loc.get("error_order_already_cleared"))
break
# If the user pressed the complete order button, complete the order
if update.data == "order_complete":
@ -1024,16 +1048,15 @@ class Worker(threading.Thread):
# Commit the transaction
self.session.commit()
# Update order message
self.bot.edit_message_text(order.get_text(session=self.session), chat_id=self.chat.id,
self.bot.edit_message_text(order.text(loc=self.loc, session=self.session), chat_id=self.chat.id,
message_id=update.message.message_id)
# Notify the user of the completition
self.bot.send_message(order.user_id,
strings.notification_order_completed.format(order=order.get_text(self.session,
user=True)))
self.loc.get("notification_order_completed", order=order.text(loc=self.loc, session=self.session, user=True)))
# If the user pressed the refund order button, refund the order...
elif update.data == "order_refund":
# Ask for a refund reason
reason_msg = self.bot.send_message(self.chat.id, strings.ask_refund_reason,
reason_msg = self.bot.send_message(self.chat.id, self.loc.get("ask_refund_reason"),
reply_markup=cancel_keyboard)
# Wait for a reply
reply = self.__wait_for_regex("(.*)", cancellable=True)
@ -1053,15 +1076,16 @@ class Worker(threading.Thread):
# Commit the changes
self.session.commit()
# Update the order message
self.bot.edit_message_text(order.get_text(session=self.session),
self.bot.edit_message_text(order.text(loc=self.loc, session=self.session),
chat_id=self.chat.id,
message_id=update.message.message_id)
# Notify the user of the refund
self.bot.send_message(order.user_id,
strings.notification_order_refunded.format(order=order.get_text(self.session,
self.loc.get("notification_order_refunded", order=order.text(loc=self.loc,
session=self.session,
user=True)))
# Notify the admin of the refund
self.bot.send_message(self.chat.id, strings.success_order_refunded.format(order_id=order.order_id))
self.bot.send_message(self.chat.id, self.loc.get("success_order_refunded", order_id=order.order_id))
def __create_transaction(self):
"""Edit manually the credit of an user."""
@ -1072,19 +1096,19 @@ class Worker(threading.Thread):
if isinstance(user, CancelSignal):
return
# Create an inline keyboard with a single cancel button
cancel = telegram.InlineKeyboardMarkup([[telegram.InlineKeyboardButton(strings.menu_cancel,
cancel = telegram.InlineKeyboardMarkup([[telegram.InlineKeyboardButton(self.loc.get("menu_cancel"),
callback_data="cmd_cancel")]])
# Request from the user the amount of money to be credited manually
self.bot.send_message(self.chat.id, strings.ask_credit, reply_markup=cancel)
self.bot.send_message(self.chat.id, self.loc.get("ask_credit"), reply_markup=cancel)
# Wait for an answer
reply = self.__wait_for_regex(r"(-? ?[0-9]{1,3}(?:[.,][0-9]{1,2})?)", cancellable=True)
# Allow the cancellation of the operation
if isinstance(reply, CancelSignal):
return
# Convert the reply to a price object
price = utils.Price(reply)
price = utils.Price(reply, self.loc)
# Ask the user for notes
self.bot.send_message(self.chat.id, strings.ask_transaction_notes, reply_markup=cancel)
self.bot.send_message(self.chat.id, self.loc.get("ask_transaction_notes"), reply_markup=cancel)
# Wait for an answer
reply = self.__wait_for_regex(r"(.*)", cancellable=True)
# Allow the cancellation of the operation
@ -1102,36 +1126,36 @@ class Worker(threading.Thread):
self.session.commit()
# Notify the user of the credit/debit
self.bot.send_message(user.user_id,
strings.notification_transaction_created.format(transaction=str(transaction)))
self.loc.get("notification_transaction_created", transaction=str(transaction)))
# Notify the admin of the success
self.bot.send_message(self.chat.id, strings.success_transaction_created.format(transaction=str(transaction)))
self.bot.send_message(self.chat.id, self.loc.get("success_transaction_created", transaction=str(transaction)))
def __help_menu(self):
"""Help menu. Allows the user to ask for assistance, get a guide or see some info about the bot."""
log.debug("Displaying __help_menu")
# Create a keyboard with the user help menu
keyboard = [[telegram.KeyboardButton(strings.menu_guide)],
[telegram.KeyboardButton(strings.menu_contact_shopkeeper)],
[telegram.KeyboardButton(strings.menu_cancel)]]
keyboard = [[telegram.KeyboardButton(self.loc.get("menu_guide"))],
[telegram.KeyboardButton(self.loc.get("menu_contact_shopkeeper"))],
[telegram.KeyboardButton(self.loc.get("menu_cancel"))]]
# Send the previously created keyboard to the user (ensuring it can be clicked only 1 time)
self.bot.send_message(self.chat.id,
strings.conversation_open_help_menu,
self.loc.get("conversation_open_help_menu"),
reply_markup=telegram.ReplyKeyboardMarkup(keyboard, one_time_keyboard=True))
# Wait for a reply from the user
selection = self.__wait_for_specific_message([strings.menu_guide, strings.menu_contact_shopkeeper,
strings.menu_cancel])
selection = self.__wait_for_specific_message([self.loc.get("menu_guide"), self.loc.get("menu_contact_shopkeeper"),
self.loc.get("menu_cancel")])
# If the user has selected the Guide option...
if selection == strings.menu_guide:
if selection == self.loc.get("menu_guide"):
# Send them the bot guide
self.bot.send_message(self.chat.id, strings.help_msg)
self.bot.send_message(self.chat.id, self.loc.get("help_msg"))
# If the user has selected the Order Status option...
elif selection == strings.menu_contact_shopkeeper:
elif selection == self.loc.get("menu_contact_shopkeeper"):
# Find the list of available shopkeepers
shopkeepers = self.session.query(db.Admin).filter_by(display_on_help=True).join(db.User).all()
# Create the string
shopkeepers_string = "\n".join([admin.user.mention() for admin in shopkeepers])
# Send the message to the user
self.bot.send_message(self.chat.id, strings.contact_shopkeeper.format(shopkeepers=shopkeepers_string))
self.bot.send_message(self.chat.id, self.loc.get("contact_shopkeeper", shopkeepers=shopkeepers_string))
# If the user has selected the Cancel option the function will return immediately
def __transaction_pages(self):
@ -1140,7 +1164,7 @@ class Worker(threading.Thread):
# Page number
page = 0
# Create and send a placeholder message to be populated
message = self.bot.send_message(self.chat.id, strings.loading_transactions)
message = self.bot.send_message(self.chat.id, self.loc.get("loading_transactions"))
# Loop used to move between pages
while True:
# Retrieve the 10 transactions in that page
@ -1155,22 +1179,21 @@ class Worker(threading.Thread):
if page != 0:
# Add a previous page button
inline_keyboard_list[0].append(
telegram.InlineKeyboardButton(strings.menu_previous, callback_data="cmd_previous")
telegram.InlineKeyboardButton(self.loc.get("menu_previous"), callback_data="cmd_previous")
)
# Don't add a next page button if this is the last page
if len(transactions) == 10:
# Add a next page button
inline_keyboard_list[0].append(
telegram.InlineKeyboardButton(strings.menu_next, callback_data="cmd_next")
telegram.InlineKeyboardButton(self.loc.get("menu_next"), callback_data="cmd_next")
)
# Add a Done button
inline_keyboard_list.append([telegram.InlineKeyboardButton(strings.menu_done, callback_data="cmd_done")])
inline_keyboard_list.append([telegram.InlineKeyboardButton(self.loc.get("menu_done"), callback_data="cmd_done")])
# Create the inline keyboard markup
inline_keyboard = telegram.InlineKeyboardMarkup(inline_keyboard_list)
# Create the message text
transactions_string = "\n".join([str(transaction) for transaction in transactions])
text = strings.transactions_page.format(page=page + 1,
transactions=transactions_string)
text = self.loc.get("transactions_page", page=page + 1, transactions=transactions_string)
# Update the previously sent message
self.bot.edit_message_text(chat_id=self.chat.id, message_id=message.message_id, text=text,
reply_markup=inline_keyboard)
@ -1224,7 +1247,7 @@ class Worker(threading.Thread):
f"{transaction.payment_email if transaction.payment_email is not None else ''};"
f"{transaction.refunded if transaction.refunded is not None else ''}\n")
# Describe the file to the user
self.bot.send_message(self.chat.id, strings.csv_caption)
self.bot.send_message(self.chat.id, self.loc.get("csv_caption"))
# Reopen the file for reading
with open(f"transactions_{self.chat.id}.csv") as file:
# Send the file via a manual request to Telegram
@ -1247,13 +1270,13 @@ class Worker(threading.Thread):
admin = self.session.query(db.Admin).filter_by(user_id=user.user_id).one_or_none()
if admin is None:
# Create the keyboard to be sent
keyboard = telegram.ReplyKeyboardMarkup([[strings.emoji_yes, strings.emoji_no]], one_time_keyboard=True)
keyboard = telegram.ReplyKeyboardMarkup([[self.loc.get("emoji_yes"), self.loc.get("emoji_no")]], one_time_keyboard=True)
# Ask for confirmation
self.bot.send_message(self.chat.id, strings.conversation_confirm_admin_promotion, reply_markup=keyboard)
self.bot.send_message(self.chat.id, self.loc.get("conversation_confirm_admin_promotion"), reply_markup=keyboard)
# Wait for an answer
selection = self.__wait_for_specific_message([strings.emoji_yes, strings.emoji_no])
selection = self.__wait_for_specific_message([self.loc.get("emoji_yes"), self.loc.get("emoji_no")])
# Proceed only if the answer is yes
if selection == strings.emoji_no:
if selection == self.loc.get("emoji_no"):
return
# Create a new admin
admin = db.Admin(user=user,
@ -1264,22 +1287,22 @@ class Worker(threading.Thread):
display_on_help=False)
self.session.add(admin)
# Send the empty admin message and record the id
message = self.bot.send_message(self.chat.id, strings.admin_properties.format(name=str(admin.user)))
message = self.bot.send_message(self.chat.id, self.loc.get("admin_properties", name=str(admin.user)))
# Start accepting edits
while True:
# Create the inline keyboard with the admin status
inline_keyboard = telegram.InlineKeyboardMarkup([
[telegram.InlineKeyboardButton(f"{utils.boolmoji(admin.edit_products)} {strings.prop_edit_products}",
[telegram.InlineKeyboardButton(f"{self.loc.boolmoji(admin.edit_products)} {self.loc.get('prop_edit_products')}",
callback_data="toggle_edit_products")],
[telegram.InlineKeyboardButton(f"{utils.boolmoji(admin.receive_orders)} {strings.prop_receive_orders}",
[telegram.InlineKeyboardButton(f"{self.loc.boolmoji(admin.receive_orders)} {self.loc.get('prop_receive_orders')}",
callback_data="toggle_receive_orders")],
[telegram.InlineKeyboardButton(
f"{utils.boolmoji(admin.create_transactions)} {strings.prop_create_transactions}",
f"{self.loc.boolmoji(admin.create_transactions)} {self.loc.get('prop_create_transactions')}",
callback_data="toggle_create_transactions")],
[telegram.InlineKeyboardButton(
f"{utils.boolmoji(admin.display_on_help)} {strings.prop_display_on_help}",
f"{self.loc.boolmoji(admin.display_on_help)} {self.loc.get('prop_display_on_help')}",
callback_data="toggle_display_on_help")],
[telegram.InlineKeyboardButton(strings.menu_done, callback_data="cmd_done")]
[telegram.InlineKeyboardButton(self.loc.get('menu_done'), callback_data="cmd_done")]
])
# Update the inline keyboard
self.bot.edit_message_reply_markup(message_id=message.message_id,
@ -1300,13 +1323,67 @@ class Worker(threading.Thread):
break
self.session.commit()
def __language_menu(self):
"""Select a language."""
log.debug("Displaying __language_menu")
keyboard = []
options: Dict[str, str] = {}
# https://en.wikipedia.org/wiki/List_of_language_names
if "it" in configloader.config["Language"]["enabled_languages"]:
lang = "🇮🇹 Italiano"
keyboard.append([telegram.KeyboardButton(lang)])
options[lang] = "it"
if "en" in configloader.config["Language"]["enabled_languages"]:
lang = "🇬🇧 English"
keyboard.append([telegram.KeyboardButton(lang)])
options[lang] = "en"
if "ru" in configloader.config["Language"]["enabled_languages"]:
lang = "🇷🇺 Русский"
keyboard.append([telegram.KeyboardButton(lang)])
options[lang] = "ru"
if "uk" in configloader.config["Language"]["enabled_languages"]:
lang = "🇺🇦 Українська"
keyboard.append([telegram.KeyboardButton(lang)])
options[lang] = "uk"
# Send the previously created keyboard to the user (ensuring it can be clicked only 1 time)
self.bot.send_message(self.chat.id,
self.loc.get("conversation_language_select"),
reply_markup=telegram.ReplyKeyboardMarkup(keyboard, one_time_keyboard=True))
# Wait for an answer
response = self.__wait_for_specific_message(list(options.keys()))
# Set the language to the corresponding value
self.user.language = options[response]
# Commit the edit to the database
self.session.commit()
# Recreate the localization object
self.__create_localization()
def __create_localization(self):
# Check if the user's language is enabled; if it isn't, change it to the default
if self.user.language not in configloader.config["Language"]["enabled_languages"]:
log.debug(f"User's language '{self.user.language}' is not enabled, changing it to the default")
self.user.language = configloader.config["Language"]["default_language"]
self.session.commit()
# Create a new Localization object
self.loc = localization.Localization(
language=self.user.language,
fallback=configloader.config["Language"]["fallback_language"],
replacements={
"user_string": str(self.user),
"user_mention": self.user.mention(),
"user_full_name": self.user.full_name,
"user_first_name": self.user.first_name,
"today": datetime.datetime.now().strftime("%a %d %b %Y"),
}
)
def __graceful_stop(self, stop_trigger: StopSignal):
"""Handle the graceful stop of the thread."""
log.debug("Gracefully stopping the conversation")
# If the session has expired...
if stop_trigger.reason == "timeout":
# Notify the user that the session has expired and remove the keyboard
self.bot.send_message(self.chat.id, strings.conversation_expired,
self.bot.send_message(self.chat.id, self.loc.get('conversation_expired'),
reply_markup=telegram.ReplyKeyboardRemove())
# If a restart has been requested...
# Do nothing.