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)
python类location()的实例源码
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 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 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()
