RSNBusBot/main.py at main · geraldnyeo/RSNBusBot · GitHub

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
"""
RSNBusBot v1.3.0
"""

### IMPORTS
from telegram import (
    Update,
)
from telegram.ext import (
    Application,
    CommandHandler,
    CallbackQueryHandler,
    MessageHandler,
    filters
)

from contextlib import asynccontextmanager
from fastapi import FastAPI, Request, Response
from http import HTTPStatus

import os
from datetime import time
import pytz

from setup import * # Ensure setup.py in same directory
from handlers import * # Ensure handlers.py in same directory

### CONSTANTS
# Environment Variables
TOKEN = os.environ['TOKEN']
BOT_USERNAME = os.environ['BOT_USERNAME']
TIMEZONE = os.environ['TIMEZONE']

### MAIN
"""
There are two applications running:
 - The Python Telegram Bot application, which handles receiving, processing and sending Telegram updates.
 - The FastAPI application, which sets up the webhook endpoint for Telegram to send updates to.
"""
setup_db()

print('Starting bot...') # Logging

# Create the PTB application
ptb = (
    Application.builder()
    .token(TOKEN)
    .concurrent_updates(False)
    .build()
)

@asynccontextmanager
async def lifespan(app: FastAPI):
    """This code only runs once, before the application starts up and starts receiving requests."""

    await ptb.bot.setWebhook(url="https://rsnbusbot.onrender.com/webhook",
                            certificate=None) # Sets up webhook

    # Allows ptb and fastapi applications to run together
    async with ptb:
        await ptb.start()
        yield
        await ptb.stop()

# Create the FastAPI application
app = FastAPI(lifespan=lifespan) # Do not run FastAPI code for local dev using polling

@app.get("/")
async def index():
    """Landing page for the bot."""
    # TODO FUTURE: Add a basic single static page here to explain the bot!
    return "Hello"

@app.post("/webhook")
async def process_update(request: Request):
    """Updates PTB application when post request received at webhook"""
    req = await request.json()
    update = Update.de_json(req, ptb.bot)
    await ptb.process_update(update)
    return Response(status_code = HTTPStatus.OK)

# Set up PTB handlers
# Commands (General)
ptb.add_handler(CommandHandler('start', start_command))
ptb.add_handler(CommandHandler('reset', reset_command))
ptb.add_handler(CommandHandler('view_settings', view_settings_command))
ptb.add_handler(settings_handler)
ptb.add_handler(CommandHandler('help', help_command))

# Commands (Booking)
ptb.add_handler(CommandHandler('book', book_command))
ptb.add_handler(CallbackQueryHandler(booking_cb_handler, r"^(book|cancel)$")) # Handles callbacks for book command
ptb.add_handler(manage_book_handler)
ptb.add_handler(CommandHandler('cancel_book', cancel_book_command))
ptb.add_handler(CommandHandler('uncancel_book', uncancel_book_command))

# Commands (Scheduling)
ptb.add_handler(view_schedule_handler)
ptb.add_handler(schedule_handler)

# Commands (Broadcasting)
ptb.add_handler(broadcast_handler)
ptb.add_handler(CommandHandler('notify_late', lambda u, c: notify_late(u, c)))
ptb.add_handler(CommandHandler('notify_late_all', lambda u, c: notify_late(u, c, all_chats=True)))

# Commands (Data)
ptb.add_handler(CommandHandler('view_data_summary', view_data_summary_command))
ptb.add_handler(edit_db_handler)

# Messages
# For later development

# Other Events
ptb.add_handler(MessageHandler(filters.StatusUpdate.MIGRATE, migrate_chat))

# Errors
ptb.add_error_handler(error)

# Automatic Processes
ptb.job_queue.run_daily(daily_booking,
                        time=time(hour=17, minute=30, second=0, tzinfo=pytz.timezone(TIMEZONE)),
                        days=(0, 1, 2, 3, 4, 5, 6)) # MUST be 0 to 6 to work
ptb.job_queue.run_daily(end_book_job,
                        time=time(hour=23, minute=59, second=59, tzinfo=pytz.timezone(TIMEZONE)),
                        days=(0, 1, 2, 3, 4, 5, 6))

# Polling, for dev purposes
# print('Polling...')
# ptb.run_polling(poll_interval=1, allowed_updates=Update.ALL_TYPES)