-
Notifications
You must be signed in to change notification settings - Fork 0
/
main.py
153 lines (118 loc) · 5.62 KB
/
main.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
import os
from telegram import Update, InlineKeyboardButton, InlineKeyboardMarkup
from telegram.ext import Updater, CommandHandler, MessageHandler, Filters, CallbackContext, CallbackQueryHandler
import requests
import json
from telegram.ext import JobQueue
TELEGRAM_BOT_TOKEN = 'YOUR_TELEGRAM_BOT_TOKEN'
OPENWEATHERMAP_API_KEY = 'YOUR_OPENWEATHERMAP_API_KEY'
def start(update: Update, context: CallbackContext) -> None:
keyboard = [
[InlineKeyboardButton("Current Weather", callback_data='current_weather')],
[InlineKeyboardButton("Forecast", callback_data='forecast')],
[InlineKeyboardButton("Set Location", callback_data='set_location')]
]
reply_markup = InlineKeyboardMarkup(keyboard)
update.message.reply_text('Welcome to Aesthetic Weather Bot! Use the buttons below:', reply_markup=reply_markup)
def button(update: Update, context: CallbackContext) -> None:
query = update.callback_query
query.answer()
if query.data == 'current_weather':
weather(query, context)
elif query.data == 'forecast':
forecast(query, context)
elif query.data == 'set_location':
query.edit_message_text('Please use /setlocation <city> to set your location.')
def check_weather(context: CallbackContext) -> None:
job = context.job
location = job.context['location']
chat_id = job.context['chat_id']
weather_data = get_weather(location)
if weather_data.get('cod') != 200:
context.bot.send_message(chat_id, text='Failed to get weather data.')
return
weather_desc = weather_data['weather'][0]['description']
if 'rain' in weather_desc:
context.bot.send_message(chat_id, text='It\'s going to rain today! ☔')
def set_alert(update: Update, context: CallbackContext) -> None:
chat_id = update.message.chat_id
location = context.user_data.get('location')
if not location:
update.message.reply_text('Please set your location using /setlocation <city>.')
return
context.job_queue.run_repeating(check_weather, interval=3600, first=0, context={'chat_id': chat_id, 'location': location})
update.message.reply_text('Weather alerts have been set!')
def set_location(update: Update, context: CallbackContext) -> None:
user = update.message.from_user
location = ' '.join(context.args)
context.user_data['location'] = location
update.message.reply_text(f'Set location to {location}. Use /weather to get the current weather.')
def get_weather(location: str) -> dict:
url = f'http://api.openweathermap.org/data/2.5/weather?q={location}&appid={OPENWEATHERMAP_API_KEY}&units=metric'
response = requests.get(url)
return response.json()
def weather(update: Update, context: CallbackContext) -> None:
location = context.user_data.get('location')
if not location:
update.message.reply_text('Please set your location using /setlocation <city>.')
return
weather_data = get_weather(location)
if weather_data.get('cod') != 200:
update.message.reply_text('Failed to get weather data. Please check your location and try again.')
return
weather_desc = weather_data['weather'][0]['description']
temp = weather_data['main']['temp']
humidity = weather_data['main']['humidity']
wind_speed = weather_data['wind']['speed']
icon = weather_data['weather'][0]['icon']
weather_icons = {
"01d": "☀️", "01n": "🌙",
"02d": "🌤️", "02n": "☁️",
"03d": "🌥️", "03n": "☁️",
"04d": "☁️", "04n": "☁️",
"09d": "🌧️", "09n": "🌧️",
"10d": "🌦️", "10n": "🌧️",
"11d": "⛈️", "11n": "⛈️",
"13d": "❄️", "13n": "❄️",
"50d": "🌫️", "50n": "🌫️"
}
weather_icon = weather_icons.get(icon, "🌈")
weather_message = (
f"Weather in *{location}*:\n"
f"{weather_icon} *{weather_desc.capitalize()}*\n"
f"🌡️ *Temperature*: {temp}°C\n"
f"💧 *Humidity*: {humidity}%\n"
f"💨 *Wind Speed*: {wind_speed} m/s"
)
update.message.reply_text(weather_message, parse_mode='Markdown')
def forecast(update: Update, context: CallbackContext) -> None:
location = context.user_data.get('location')
if not location:
update.message.reply_text('Please set your location using /setlocation <city>.')
return
url = f'http://api.openweathermap.org/data/2.5/forecast?q={location}&appid={OPENWEATHERMAP_API_KEY}&units=metric'
response = requests.get(url)
forecast_data = response.json()
if forecast_data.get('cod') != '200':
update.message.reply_text('Failed to get forecast data. Please check your location and try again.')
return
forecast_message = f"Weather forecast for *{location}*:\n"
for item in forecast_data['list'][:8]:
time = item['dt_txt']
desc = item['weather'][0]['description']
temp = item['main']['temp']
forecast_message += f"`{time}`: *{desc.capitalize()}*, {temp}°C\n"
update.message.reply_text(forecast_message, parse_mode='Markdown')
def main() -> None:
updater = Updater(TELEGRAM_BOT_TOKEN, use_context=True)
dispatcher = updater.dispatcher
dispatcher.add_handler(CommandHandler('start', start))
dispatcher.add_handler(CommandHandler('setlocation', set_location))
dispatcher.add_handler(CommandHandler('weather', weather))
dispatcher.add_handler(CommandHandler('forecast', forecast))
dispatcher.add_handler(CommandHandler('alert', set_alert))
dispatcher.add_handler(CallbackQueryHandler(button))
updater.start_polling()
updater.idle()
if __name__ == '__main__':
main()