def nearest_stations(bot, update, count=5):
with open('allstations.csv', newline='', encoding='utf-8') as infile:
csv_reader = csv.reader(infile, delimiter=';')
stations = [(int(row[0]), float(row[1]), float(row[2]), row[3]) for row in csv_reader]
# distance sorting based on http://stackoverflow.com/a/28368926 by Sergey Ivanov
coord = (float(update.message.location.latitude), float(update.message.location.longitude))
pts = [geopy.Point(p[1], p[2], p[0]) for p in stations]
sts = [p[3] for p in stations]
onept = geopy.Point(coord[0], coord[1])
alldist = [(p, geopy.distance.distance(p, onept).m) for p in pts]
nearest = sorted(alldist, key=lambda x: (x[1]))[:count]
nearest_points = [n[0] for n in nearest]
nearest_distances = [n[1] for n in nearest]
nearest_sts = [sts[int(n.altitude)] for n in nearest_points]
msg = 'Nächstgelegene Stationen:'
for s, d, p in zip(nearest_sts, nearest_distances, nearest_points):
msg += '\n{} (<a href="https://www.google.de/maps?q={},{}">{:.0f}m</a>)'.format(s, p.latitude,
p.longitude, d)
reply_keyboard = [[telegram.KeyboardButton(text='/Abfahrten {}'.format(n))] for n in nearest_sts]
bot.sendMessage(chat_id=update.message.chat_id, text=msg, parse_mode='HTML',
reply_markup=telegram.ReplyKeyboardMarkup(reply_keyboard, one_time_keyboard=True))
python类KeyboardButton()的实例源码
def execute(self, chat_id, bot: Bot, update: Update):
handled = False
if update.message.text in self._buildings_dict:
handled = True
building_identifier = self._buildings_dict[update.message.text]
classrooms = self.get_classroom_source().get_classrooms_in_building(building_identifier)
keyboard_button = []
counter = 0
row = -1
for classroom in classrooms:
if counter == 0:
keyboard_button.append([])
row += 1
keyboard_button[row].append(KeyboardButton("/"+classroom.get_name().lower()))
counter += 1
if counter == 3:
counter = 0
keyboard_button.append(["/buildings"])
reply_keyboard = ReplyKeyboardMarkup(keyboard_button)
bot.send_message(chat_id=chat_id, text="Available classrooms", reply_markup=reply_keyboard)
return handled
def trade_cmd(bot, update):
reply_msg = "Buy or sell?"
buttons = [
KeyboardButton(KeyboardEnum.BUY.clean()),
KeyboardButton(KeyboardEnum.SELL.clean())
]
cancel_btn = [
KeyboardButton(KeyboardEnum.CANCEL.clean())
]
reply_mrk = ReplyKeyboardMarkup(build_menu(buttons, n_cols=2, footer_buttons=cancel_btn))
update.message.reply_text(reply_msg, reply_markup=reply_mrk)
return WorkflowEnum.TRADE_BUY_SELL
# Save if BUY or SELL order and choose the currency to trade
def trade_buy_sell(bot, update, chat_data):
chat_data["buysell"] = update.message.text
reply_msg = "Choose currency"
cancel_btn = [
KeyboardButton(KeyboardEnum.CANCEL.clean())
]
# If SELL chosen, then include button 'ALL' to sell everything
if chat_data["buysell"].upper() == KeyboardEnum.SELL.clean():
cancel_btn.insert(0, KeyboardButton(KeyboardEnum.ALL.clean()))
reply_mrk = ReplyKeyboardMarkup(build_menu(coin_buttons(), n_cols=3, footer_buttons=cancel_btn))
update.message.reply_text(reply_msg, reply_markup=reply_mrk)
return WorkflowEnum.TRADE_CURRENCY
# Show confirmation to sell all assets
def trade_price(bot, update, chat_data):
chat_data["price"] = update.message.text
reply_msg = "How to enter the volume?"
buttons = [
KeyboardButton(config["trade_to_currency"].upper()),
KeyboardButton(KeyboardEnum.VOLUME.clean())
]
cancel_btn = [
KeyboardButton(KeyboardEnum.ALL.clean()),
KeyboardButton(KeyboardEnum.CANCEL.clean())
]
reply_mrk = ReplyKeyboardMarkup(build_menu(buttons, n_cols=2, footer_buttons=cancel_btn))
update.message.reply_text(reply_msg, reply_markup=reply_mrk)
return WorkflowEnum.TRADE_VOL_TYPE
# Save volume type decision and enter volume
def orders_choose_order(bot, update):
buttons = list()
# Go through all open orders and create a button
if orders:
for order in orders:
order_id = next(iter(order), None)
buttons.append(KeyboardButton(order_id))
else:
update.message.reply_text("No open orders")
return ConversationHandler.END
msg = "Which order to close?"
close_btn = [
KeyboardButton(KeyboardEnum.CANCEL.clean())
]
reply_mrk = ReplyKeyboardMarkup(build_menu(buttons, n_cols=1, footer_buttons=close_btn))
update.message.reply_text(msg, reply_markup=reply_mrk)
return WorkflowEnum.ORDERS_CLOSE_ORDER
# Close all open orders
def bot_cmd(bot, update):
reply_msg = "What do you want to do?"
buttons = [
KeyboardButton(KeyboardEnum.UPDATE_CHECK.clean()),
KeyboardButton(KeyboardEnum.UPDATE.clean()),
KeyboardButton(KeyboardEnum.RESTART.clean()),
KeyboardButton(KeyboardEnum.SHUTDOWN.clean()),
KeyboardButton(KeyboardEnum.SETTINGS.clean()),
KeyboardButton(KeyboardEnum.CANCEL.clean())
]
reply_mrk = ReplyKeyboardMarkup(build_menu(buttons, n_cols=2))
update.message.reply_text(reply_msg, reply_markup=reply_mrk)
return WorkflowEnum.BOT_SUB_CMD
# Execute chosen sub-cmd of 'bot' cmd
def chart_cmd(bot, update):
reply_msg = "Choose currency"
buttons = list()
for coin, url in config["coin_charts"].items():
buttons.append(KeyboardButton(coin))
cancel_btn = [
KeyboardButton(KeyboardEnum.CANCEL.clean())
]
reply_mrk = ReplyKeyboardMarkup(build_menu(buttons, n_cols=3, footer_buttons=cancel_btn))
update.message.reply_text(reply_msg, reply_markup=reply_mrk)
return WorkflowEnum.CHART_CURRENCY
# Get chart URL for every coin in config
def settings_cmd(bot, update):
settings = str()
buttons = list()
# Go through all settings in config file
for key, value in config.items():
settings += key + " = " + str(value) + "\n\n"
buttons.append(KeyboardButton(key.upper()))
# Send message with all current settings (key & value)
update.message.reply_text(settings)
cancel_btn = [
KeyboardButton(KeyboardEnum.CANCEL.clean())
]
msg = "Choose key to change value"
reply_mrk = ReplyKeyboardMarkup(build_menu(buttons, n_cols=2, footer_buttons=cancel_btn))
update.message.reply_text(msg, reply_markup=reply_mrk)
return WorkflowEnum.SETTINGS_CHANGE
# Change setting
def keyboard_cmds():
command_buttons = [
KeyboardButton("/trade"),
KeyboardButton("/orders"),
KeyboardButton("/balance"),
KeyboardButton("/price"),
KeyboardButton("/value"),
KeyboardButton("/chart"),
KeyboardButton("/history"),
KeyboardButton("/funding"),
KeyboardButton("/bot")
]
return ReplyKeyboardMarkup(build_menu(command_buttons, n_cols=3))
# Generic custom keyboard that shows YES and NO
def start(bot, update):
"""
Shows an welcome message and help info about the available commands.
"""
me = bot.get_me()
# Welcome message
msg = _("Hello!\n")
msg += _("I'm {0} and I came here to help you.\n").format(me.first_name)
msg += _("What would you like to do?\n\n")
msg += _("/support - Opens a new support ticket\n")
msg += _("/settings - Settings of your account\n\n")
# Commands menu
main_menu_keyboard = [[telegram.KeyboardButton('/support')],
[telegram.KeyboardButton('/settings')]]
reply_kb_markup = telegram.ReplyKeyboardMarkup(main_menu_keyboard,
resize_keyboard=True,
one_time_keyboard=True)
# Send the message with menu
bot.send_message(chat_id=update.message.chat_id,
text=msg,
reply_markup=reply_kb_markup)
def settings(bot, update):
"""
Configure the messages language using a custom keyboard.
"""
# Languages message
msg = _("Please, choose a language:\n")
msg += "en_US - English (US)\n"
msg += "pt_BR - Português (Brasil)\n"
# Languages menu
languages_keyboard = [
[telegram.KeyboardButton('en_US - English (US)')],
[telegram.KeyboardButton('pt_BR - Português (Brasil)')]
]
reply_kb_markup = telegram.ReplyKeyboardMarkup(languages_keyboard,
resize_keyboard=True,
one_time_keyboard=True)
# Sends message with languages menu
bot.send_message(chat_id=update.message.chat_id,
text=msg,
reply_markup=reply_kb_markup)
def __init__(self, token, host, port, cert, cert_key, working_dir):
self.token = token
self.host = host
self.port = port
self.cert = cert
self.cert_key = cert_key
self.bot = telegram.Bot(self.token)
self.app = Flask(__name__)
self.context = (self.cert, self.cert_key)
self.working_dir = working_dir
self.kb = [[telegram.KeyboardButton('Offer me a coffee'), telegram.KeyboardButton('Source Code'), telegram.KeyboardButton('Vote Me')]]
self.kb_markup = telegram.ReplyKeyboardMarkup(self.kb, resize_keyboard=True)
def get_keyboard():
contact_button = KeyboardButton('?????????? ??????', request_contact=True)
location_button = KeyboardButton('??????????', request_location=True)
reply_keyboard = [['???????? ??????', '??????? ????????'], [contact_button, location_button]]
return reply_keyboard
def get_keyboard():
contact_button = KeyboardButton('?????????? ??????', request_contact=True)
button2 = KeyboardButton('??????????', request_location=True)
reply_keyboard = [['???????? ??????', '??????? ????????'], [contact_button, button2]]
return reply_keyboard
def start(bot, update):
hiKey = KeyboardButton("Hi", callback_data="Bonjour")
howRU = KeyboardButton("How are you?")
bye = KeyboardButton("Bye")
introduce = KeyboardButton("I'm Gutsy")
keyboard = [
[hiKey, howRU],
[bye, introduce]
]
reply_markup = ReplyKeyboardMarkup(keyboard)
update.message.reply_text('Please choose:', reply_markup=reply_markup)
def createKeyboard(self, strKeys):
keyboard = []
for row in strKeys:
newRow = map(KeyboardButton, row)
keyboard.append(newRow)
return keyboard
def __init__(self, booking: Booking):
classrooms_source = booking.get_classroom_source()
keys = []
self._keyboard = None
for building in classrooms_source.get_all_buildings():
building_command = "/" + building.get_name().lower().replace(" ", "_")
keys.append([KeyboardButton(building_command)])
if len(keys) > 0:
self._keyboard = ReplyKeyboardMarkup(keys)
super().__init__(booking, ["buildings"], False, exact_match=True)
def value_cmd(bot, update):
reply_msg = "Choose currency"
footer_btns = [
KeyboardButton(KeyboardEnum.ALL.clean()),
KeyboardButton(KeyboardEnum.CANCEL.clean())
]
reply_mrk = ReplyKeyboardMarkup(build_menu(coin_buttons(), n_cols=3, footer_buttons=footer_btns))
update.message.reply_text(reply_msg, reply_markup=reply_mrk)
return WorkflowEnum.VALUE_CURRENCY
# Choose for which currency you want to know the current value
def funding_cmd(bot, update):
reply_msg = "Choose currency"
cancel_btn = [
KeyboardButton(KeyboardEnum.CANCEL.clean())
]
reply_mrk = ReplyKeyboardMarkup(build_menu(coin_buttons(), n_cols=3, footer_buttons=cancel_btn))
update.message.reply_text(reply_msg, reply_markup=reply_mrk)
return WorkflowEnum.FUNDING_CURRENCY
# Choose withdraw or deposit
def keyboard_confirm():
buttons = [
KeyboardButton(KeyboardEnum.YES.clean()),
KeyboardButton(KeyboardEnum.NO.clean())
]
return ReplyKeyboardMarkup(build_menu(buttons, n_cols=2))
# Create a list with a button for every coin in config
def coin_buttons():
buttons = list()
for coin in config["used_coins"]:
buttons.append(KeyboardButton(coin))
return buttons
# Check order state and send message if order closed
def downloader(self, message, chat_id, user):
try:
curr_dir = self.working_dir + '/' + str(chat_id).replace("-", "")
os.chdir(curr_dir)
RequestUrl = 'http://www.youtubeinmp3.com/fetch/index.php?format=json&video=%s' % (message)
JsonVideoData = json.load(urlopen(RequestUrl))
self.bot.sendMessage(chat_id=chat_id, text = 'Downloading ' + JsonVideoData['title'] + '...', reply_markup = self.kb_markup);
f = urlopen(JsonVideoData['link'], JsonVideoData['title'].replace('/', '').encode('utf-8').strip() + '.mp3')
fulltitle = (curr_dir + '/' + JsonVideoData['title'].replace('/', '') + '.mp3')
fulltitle = unicodedata.normalize('NFKD', fulltitle).encode('utf-8').strip()
with open(fulltitle, 'wb') as mp3:
mp3.write(f.read());
with open(fulltitle) as mp3:
self.bot.sendAudio(chat_id=chat_id, audio=mp3)
os.remove(fulltitle);
with open('log', 'a') as f:
f.write(message + " " + str(datetime.now().time()))
f.write("\n");
f.close()
url_data = urlparse.urlparse(message)
if url_data.hostname == 'youtu.be':
video = url_data.path[1:]
else:
query = urlparse.parse_qs(url_data.query)
video = query["v"][0]
VideosUrl = 'https://www.googleapis.com/youtube/v3/search?part=snippet&relatedToVideoId=%s&type=video&key=xx' % video
VideosJson = json.load(urlopen(VideosUrl))
videokb = [[telegram.KeyboardButton('https://www.youtube.com/watch?v='+VideosJson['items'][0]['id']['videoId'])], [telegram.KeyboardButton('https://www.youtube.com/watch?v='+VideosJson['items'][1]['id']['videoId'])], [telegram.KeyboardButton('https://www.youtube.com/watch?v='+VideosJson['items'][2]['id']['videoId'])], [telegram.KeyboardButton('Offer me a coffee'), telegram.KeyboardButton('Source Code'), telegram.KeyboardButton('Vote Me')] ]
videokb_markup = telegram.ReplyKeyboardMarkup(videokb, resize_keyboard=True)
self.bot.sendMessage(chat_id=chat_id, text = 'Here on the keyboard there are some related videos, in order:\n\n1) ' + VideosJson['items'][0]['snippet']['title'] + '\n\n2) ' + VideosJson['items'][1]['snippet']['title'] + '\n\n3) ' + VideosJson['items'][2]['snippet']['title'], reply_markup = videokb_markup);
except ValueError as e:
self.bot.sendMessage(chat_id=chat_id, text = 'Video not found on the database of www.youtubeinmp3.com, but you can download the mp3 (and update the database) by using this link: ');
RequestUrl = 'http://www.youtubeinmp3.com/download/?video=%s' % (message)
self.bot.sendMessage(chat_id=chat_id, text = RequestUrl)
except Exception as e:
if str(e) != "101":
self.bot.sendMessage(chat_id=chat_id, text = 'Something went wrong, maybe the video does not exists,\nif the error persist send the code in the next message to the developer, Telegram: @aakagoree')
self.bot.sendMessage(chat_id=chat_id, text = str(chat_id))
#print str(e)
with open('log', 'a') as f:
f.write("!!EXCEPTION!!! " + message + " " + str(datetime.now().time()) + " " + str(e))
f.write("\n");
f.close()
finally:
os.chdir(self.working_dir)
def orders_cmd(bot, update):
update.message.reply_text("Retrieving orders...")
# Send request to Kraken to get open orders
res_data = kraken_api("OpenOrders", private=True)
# If Kraken replied with an error, show it
if res_data["error"]:
error = btfy(res_data["error"][0])
update.message.reply_text(error)
logger.error(error)
return
# Reset global orders list
global orders
orders = list()
# Go through all open orders and show them to the user
if res_data["result"]["open"]:
for order_id, order_details in res_data["result"]["open"].items():
# Add order to global order list so that it can be used later
# without requesting data from Kraken again
orders.append({order_id: order_details})
order_desc = trim_zeros(order_details["descr"]["order"])
update.message.reply_text(bold(order_id + "\n" + order_desc), parse_mode=ParseMode.MARKDOWN)
else:
update.message.reply_text("No open orders")
return ConversationHandler.END
reply_msg = "What do you want to do?"
buttons = [
KeyboardButton(KeyboardEnum.CLOSE_ORDER.clean()),
KeyboardButton(KeyboardEnum.CLOSE_ALL.clean())
]
close_btn = [
KeyboardButton(KeyboardEnum.CANCEL.clean())
]
reply_mrk = ReplyKeyboardMarkup(build_menu(buttons, n_cols=2, footer_buttons=close_btn))
update.message.reply_text(reply_msg, reply_markup=reply_mrk)
return WorkflowEnum.ORDERS_CLOSE
# Choose what to do with the open orders
def history_cmd(bot, update):
update.message.reply_text("Retrieving history data...")
# Send request to Kraken to get trades history
res_trades = kraken_api("TradesHistory", private=True)
# If Kraken replied with an error, show it
if res_trades["error"]:
error = btfy(res_trades["error"][0])
update.message.reply_text(error)
logger.error(error)
return
# Reset global trades list
global trades
trades = list()
# Add all trades to global list
for trade_id, trade_details in res_trades["result"]["trades"].items():
trades.append(trade_details)
if trades:
# Sort global list with trades - on executed time
trades = sorted(trades, key=lambda k: k['time'], reverse=True)
buttons = [
KeyboardButton(KeyboardEnum.NEXT.clean()),
KeyboardButton(KeyboardEnum.CANCEL.clean())
]
# Get number of first items in list (latest trades)
for items in range(config["history_items"]):
newest_trade = next(iter(trades), None)
total_value = "{0:.2f}".format(float(newest_trade["price"]) * float(newest_trade["vol"]))
msg = get_trade_str(newest_trade) + " (Value: " + total_value + " EUR)"
reply_mrk = ReplyKeyboardMarkup(build_menu(buttons, n_cols=2))
update.message.reply_text(bold(msg), reply_markup=reply_mrk, parse_mode=ParseMode.MARKDOWN)
# Remove the first item in the trades list
trades.remove(newest_trade)
return WorkflowEnum.HISTORY_NEXT
else:
update.message.reply_text("No item in trade history", reply_markup=keyboard_cmds())
return ConversationHandler.END
# Save if BUY, SELL or ALL trade history and choose how many entries to list