我们从Python开源项目中，提取了以下19个代码示例，用于说明如何使用telegram.ext.Filters.location()。
def get_location(bot, update): print(update.message.location.latitude) print(update.message.location.longitude) target_latitude = update.message.location.latitude target_longitude = update.message.location.longitude address_1, address_2, address_3 = map_coordinate_api.get_map_address(target_longitude, target_latitude) weather_coordinate = map_coordinate_api.get_weather_map_coordinate(address_1, address_2, address_3) weather_data = weather_api.get_weather_api(weather_coordinate['x'], weather_coordinate['y']) weather_SKY = weather_data['SKY'] weather_REH = weather_data['REH'] weather_T3H = weather_data['T3H'] weather_POP = weather_data['POP'] send2telegram_weather = "?? ?? : {}\n" \ "?? ?? : {}\n" \ "???? : {}%\n" \ "?? ?? : {}".format(weather_T3H, weather_REH, weather_POP, weather_SKY) update.message.reply_text(send2telegram_weather)
def get_location_from_baidu(latitude: Union[float, str], longitude: Union[float, str]): params = ( ('callback', 'renderReverse'), ('location', str(latitude) + ',' + str(longitude)), ('output', 'json'), ('pois', '1'), ('ak', BAIDU_API), ) result = requests.get('http://api.map.baidu.com/geocoder/v2/', params=params) result = result.text.replace('renderReverse&&renderReverse(', '')[:-1] result_json = json.loads(result) if result_json['status'] == 0: return result_json['result']['formatted_address'] else: return 'Baidu API returned an error code: ' + str(result_json['status'])
def location_from_telegram(bot: telegram.Bot, update: telegram.Update): message: telegram.Message = update.message tg_group_id = message.chat_id # telegram group id forward_index = get_forward_index(tg_group_id=tg_group_id) latitude = message.location.latitude longitude = message.location.longitude reply_entity = list() reply_entity.append({ 'type': 'text', 'data': {'text': '????????' + get_location_from_baidu(latitude, longitude)} }) qq_message_id = send_from_tg_to_qq(forward_index, message=reply_entity, tg_group_id=tg_group_id, tg_user=message.from_user, tg_forward_from=message, tg_reply_to=message.reply_to_message) global_vars.mdb.append_message(qq_message_id, message.message_id, forward_index, 0)
def bot_hook(): """Entry point for the Telegram connection.""" bot = telegram.Bot(botdata['BotToken']) dispatcher = Dispatcher(bot, None, workers=0) dispatcher.add_handler(CommandHandler('Abfahrten', abfahrten, pass_args=True)) dispatcher.add_handler(CommandHandler('abfahrten', abfahrten, pass_args=True)) dispatcher.add_handler(CommandHandler('Abfahrt', abfahrten, pass_args=True)) dispatcher.add_handler(CommandHandler('abfahrt', abfahrten, pass_args=True)) dispatcher.add_handler(CommandHandler('A', abfahrten, pass_args=True)) dispatcher.add_handler(CommandHandler('a', abfahrten, pass_args=True)) dispatcher.add_handler(CommandHandler('Hilfe', hilfe)) dispatcher.add_handler(CommandHandler('hilfe', hilfe)) dispatcher.add_handler(CommandHandler('help', hilfe)) dispatcher.add_handler(MessageHandler(Filters.location, nearest_stations)) update = telegram.update.Update.de_json(request.json, bot) dispatcher.process_update(update) return 'OK'
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))
def cmd_help(bot, update): chat_id = update.message.chat_id logger.info('[%s] Sending help text.' % (chat_id)) text = "/help /start \n" + \ "/add <#pokedexID> \n" + \ "/add <#pokedexID1> <#pokedexID2> ... \n" + \ "/addbyrarity <#rarity> - With 1 uncommon to 5 ultrarare \n" + \ "/clear \n" + \ "/rem <#pokedexID> \n" + \ "/rem <#pokedexID1> <#pokedexID2> ... \n" + \ "Send <location> - Search a location \n" +\ "/location <s> - Send a location as text \n" +\ "/radius <m> - Search radius in m \n" +\ "/remloc - Clear location data\n" +\ "/list \n" + \ "/save \n" + \ "/load \n" + \ "/lang en" bot.sendMessage(chat_id, text) tmp = '' for key in pokemon_name: tmp += "%s, " % (key) tmp = tmp[:-2] bot.sendMessage(chat_id, text= '/lang [%s]' % (tmp))
def cmd_location(bot, update): chat_id = update.message.chat_id pref = prefs.get(chat_id) if chat_id not in jobs: bot.sendMessage(chat_id, text='You have no active scanner.') return user_location = update.message.location # We set the location from the users sent location. pref.set('location', [user_location.latitude, user_location.longitude, location_radius]) logger.info('[%s] Setting scan location to Lat %s, Lon %s, R %s' % (chat_id, pref['location'][0], pref['location'][1], pref['location'][2])) # Send confirmation nessage bot.sendMessage(chat_id, text="Setting scan location to: %f / %f with radius %.2f m" % (pref['location'][0], pref['location'][1], 1000*pref['location'][2]))
def bot_start(bot, update): send2telegram = "weather notie telegram bot\n" \ "send location, reply weather\n" \ "source code : https://github.com/Bill-Park/pycon2017_tutorial" update.message.reply_text(send2telegram)
def help(bot, update): send2telegram = "weather notie telegram bot\n" \ "send location, reply weather\n" \ "source code : https://github.com/Bill-Park/pycon2017_tutorial" update.message.reply_text(send2telegram)
def get_location(bot, update, user_data): print(update.message.location) update.message.reply_text('??????? {}'.format(get_avatar(user_data)))
def start_bot(): my_bot = Updater(TELEGRAM_API_KEY) dp = my_bot.dispatcher dp.add_handler(CommandHandler("start", reply_to_start_command, pass_user_data=True)) dp.add_handler(CommandHandler("cat", send_cat, pass_user_data=True)) dp.add_handler(RegexHandler("^(???????? ??????)$", send_cat, pass_user_data=True)) dp.add_handler(RegexHandler("^(??????? ????????)$", change_avatar_step1, pass_user_data=True)) dp.add_handler(CommandHandler("avatar", change_avatar_step2, pass_args=True, pass_user_data=True)) dp.add_handler(MessageHandler(Filters.contact, get_contact, pass_user_data=True)) dp.add_handler(MessageHandler(Filters.location, get_location, pass_user_data=True)) my_bot.start_polling() my_bot.idle()
def get_location(bot, update, user_data): user = get_user(update.effective_user, user_data) print(update.message.location) update.message.reply_text('??????? {}'.format(get_avatar(user_data)))
def cmd_location_str(bot, update,args): chat_id = update.message.chat_id pref = prefs.get(chat_id) if chat_id not in jobs: bot.sendMessage(chat_id, text='You have no active scanner.') return if len(args) <= 0: bot.sendMessage(chat_id, text='You have not supplied a location') return try: user_location = geolocator.geocode(' '.join(args)) except Exception as e: logger.error('[%s] %s' % (chat_id, repr(e))) bot.sendMessage(chat_id, text='Location not found, or openstreetmap is down.') return # We set the location from the users sent location. pref.set('location', [user_location.latitude, user_location.longitude, location_radius]) logger.info('[%s] Setting scan location to Lat %s, Lon %s, R %s' % (chat_id, pref['location'][0], pref.preferences['location'][1], pref.preferences['location'][2])) # Send confirmation nessage bot.sendMessage(chat_id, text="Setting scan location to: %f / %f with radius %.2f m" % (pref['location'][0], pref['location'][1], 1000*pref['location'][2]))
def cmd_radius(bot, update, args): chat_id = update.message.chat_id pref = prefs.get(chat_id) if chat_id not in jobs: bot.sendMessage(chat_id, text='You have no active scanner.') return # Check if user has set a location user_location = pref.get('location') if user_location[0] is None: bot.sendMessage(chat_id, text="You have not sent a location. Do that first!") return # Get the users location logger.info('[%s] Retrieved Location as Lat %s, Lon %s, R %s (Km)' % ( chat_id, user_location[0], user_location[1], user_location[2])) if len(args) < 1: bot.sendMessage(chat_id, text="Current scan location is: %f / %f with radius %.2f m" % (user_location[0], user_location[1], user_location[2])) return # Change the radius pref.set('location', [user_location[0], user_location[1], float(args[0])/1000]) logger.info('[%s] Set Location as Lat %s, Lon %s, R %s (Km)' % (chat_id, pref['location'][0], pref['location'][1], pref['location'][2])) # Send confirmation bot.sendMessage(chat_id, text="Setting scan location to: %f / %f with radius %.2f m" % (pref['location'][0], pref['location'][1], 1000*pref['location'][2]))
def cmd_clearlocation(bot, update): chat_id = update.message.chat_id pref = prefs.get(chat_id) pref.set('location', [None, None, None]) bot.sendMessage(chat_id, text='Your location has been removed.') logger.info('[%s] Location has been unset' % chat_id) ## Functions
def update_location(bot: Bot, update: Update): """ Handle location being sent in the conversation :param bot: :param update: :return: """ location = update.message.location set_location(bot, update, [location.latitude, location.longitude])
def set_location(bot: Bot, update: Update, args): """ Handles /location command :param bot: :param update: :param args: :return: """ global data chat_id = update.message.chat_id if chat_id in data.conversations: if len(args) < 1: send_help(bot, chat_id, "set_location", "Please indicate GPS coordinates or the name of a place") return else: send_chat_action(bot=bot, chat_id=chat_id, action=ChatAction.FIND_LOCATION) r = requests.get("{}{}?format=json&limit=1&bounded=0" .format(location_search_url, ' '.join([str(x) for x in args]))) try: conversation = data.conversations[chat_id] latitude = r.json()[0]["lat"] longitude = r.json()[0]["lon"] conversation.session.update_location(latitude, longitude) send_message(bot, chat_id, "location_updated") conversation.refresh_users() send_location(latitude=latitude, longitude=longitude, bot=bot, chat_id=chat_id) except AttributeError: send_help(bot, chat_id, "set_location", "Facebook token needs to be set first") else: send_error(bot=bot, chat_id=chat_id, name="account_not_setup")
def main(): db.db.connect() try: db.db.create_tables([db.Conversation, db.User, db.IsMod, db.Vote]) except pw.OperationalError: pass updater = Updater(settings.KEY) dispatcher = updater.dispatcher dispatcher.add_handler(CommandHandler('start', start)) dispatcher.add_handler(CommandHandler('help', send_help_message)) dispatcher.add_handler(CommandHandler('auto', set_auto)) dispatcher.add_handler(CommandHandler('location', set_location, pass_args=True)) dispatcher.add_handler(CommandHandler('set_account', set_account)) dispatcher.add_handler(CommandHandler('unlink', unlink)) dispatcher.add_handler(CommandHandler('matches', send_matches)) dispatcher.add_handler(CallbackQueryHandler(inline.do_press_inline_button, pass_job_queue=True)) dispatcher.add_handler(MessageHandler(Filters.text, message_handler, pass_job_queue=True)) dispatcher.add_handler(MessageHandler(Filters.location, update_location)) dispatcher.add_handler(CommandHandler('new_vote', start_vote_session, pass_job_queue=True)) dispatcher.add_handler(CommandHandler('timeout', set_timeout, pass_args=True)) dispatcher.add_handler(CommandHandler('about', send_about)) # Chat functionality dispatcher.add_handler(CommandHandler('poll_msgs', chat.poll_messages, pass_args=True)) dispatcher.add_handler(CommandHandler('poll_unanswered', chat.poll_unanswered_messages, pass_args=True)) dispatcher.add_handler(CommandHandler('unblock', chat.unblock)) # Settings dispatcher.add_handler(CommandHandler('set_setting', admin.set_setting, pass_args=True)) dispatcher.add_handler(CommandHandler('list_settings', admin.list_settings)) dispatcher.add_handler(CommandHandler('help_settings', admin.help_settings)) # Moderators dispatcher.add_handler(CommandHandler('make_me_a_mod', admin.make_me_a_mod)) inline_caps_handler = InlineQueryHandler(inline.inline_preview) dispatcher.add_handler(inline_caps_handler) dispatcher.add_handler(MessageHandler(Filters.command, custom_command_handler)) dispatcher.add_error_handler(error_callback) updater.start_polling() updater.idle()
