-
Notifications
You must be signed in to change notification settings - Fork 0
/
main.py
95 lines (74 loc) · 3.51 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
import logging
from os import environ
from dotenv import load_dotenv
import datetime
from functions import get_random_video
from telegram import __version__ as TG_VER
from telegram import __version_info__
from telegram import ForceReply, Update
from telegram.ext import Application, CommandHandler, ContextTypes, MessageHandler, filters, CallbackContext, JobQueue, Updater
# Enable logging
logging.basicConfig(
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", level=logging.INFO
)
logger = logging.getLogger(__name__)
# code extracted and adapted from https://docs.python-telegram-bot.org/en/stable/examples.timerbot.html
# Define a few command handlers. These usually take the two arguments update and
# context.
async def start(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
"""Sends explanation on how to use the bot."""
await update.message.reply_text(
"Hi!" +
"\nUse /set to receive every day the daily song of the day." +
"\nUse /unset to stop receiving the daily song of the day." +
"\nUse /song to receive a random song."
)
def remove_job_if_exists(name: str, context: ContextTypes.DEFAULT_TYPE) -> bool:
"""Remove job with given name. Returns whether job was removed."""
current_jobs = context.job_queue.jobs()
if not current_jobs:
return False
for job in current_jobs:
if name == str(job.chat_id):
job.schedule_removal()
return True
async def callback_daily(context: CallbackContext):
job = context.job
random_video = get_random_video()
await context.bot.send_message(chat_id=job.chat_id,
text=f'Song of the day: {random_video}')
async def set_timer(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
"""Add a job to the queue."""
chat_id = update.effective_message.chat_id
job_removed = remove_job_if_exists(str(chat_id), context)
# context.job_queue.run_repeating(callback_minute, chat_id=chat_id, interval=5, first=1)
# context.job_queue.run_once(callback_minute, due, chat_id=chat_id, name=str(chat_id), data=due)
context.job_queue.run_daily(callback_daily, time=datetime.time(hour=0, minute=0), days=(0, 1, 2, 3, 4, 5, 6), chat_id=chat_id)
text = "Daily song successfully set!"
if job_removed:
text += " Old one was removed."
await update.effective_message.reply_text(text)
async def unset(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
"""Remove the job if the user changed their mind."""
chat_id = update.message.chat_id
job_removed = remove_job_if_exists(str(chat_id), context)
text = "Daily song successfully unset!" if job_removed else "You have no active daily song."
await update.message.reply_text(text)
async def random_song(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
"""Sends a random song."""
random_video = get_random_video()
await update.message.reply_text(random_video)
def main() -> None:
"""Start the bot."""
load_dotenv(".env")
API_KEY = environ['BOT_TOKEN']
application = Application.builder().token(API_KEY).build()
# on different commands - answer in Telegram
application.add_handler(CommandHandler(["start", "help"], start))
application.add_handler(CommandHandler("set", set_timer))
application.add_handler(CommandHandler("unset", unset))
application.add_handler(CommandHandler("song", random_song))
# Run the bot until the user presses Ctrl-C
application.run_polling()
if __name__ == "__main__":
main()