Add webhook support for production environment

.env.toml

@@ -0,0 +1,12 @@
+[bot]
+token = ""
+environment = ""
+
+[server]
+host = ""
+port = 443
+listen = "0.0.0.0"
+
+[ssl]
+cert = ""
+priv = ""

.gitignore

@@ -2,3 +2,4 @@
 **~
 **/.venv**/**
 **/.pytest_cache/**
+env.toml

app/services/audio_service.py

@@ -8,7 +8,7 @@
 import ffmpeg
 import math
 
-from configs import config
+from configs import dirs
 
 SAMPLE_RATE = 8000
 FREQ_HIGH = 800
@@ -31,7 +31,7 @@ def sine_sample(freq, volume, x):
 # Generate a noise
 def generate_random_noise():
     name = uuid.uuid4().hex
-    audio_path = config.AUDIO_DIR+"/noise_"+name+".wav"
+    audio_path = dirs.AUDIO_DIR+"/noise_"+name+".wav"
     audio = wave.open(audio_path, "wb")
     audio.setparams((2, 2, 24000, 0, 'NONE', 'not compressed'))
     len = random.randint(20, 60)
@@ -54,7 +54,7 @@ def generate_random_noise():
 # Generate a sine wave audio
 def generate_random_sine():
     name = uuid.uuid4().hex
-    audio_path = config.AUDIO_DIR+"/sine_"+name+".wav"
+    audio_path = dirs.AUDIO_DIR+"/sine_"+name+".wav"
     audio = wave.open(audio_path, "wb")
     audio.setparams((2, 2, 24000, 0, 'NONE', 'not compressed'))
 
@@ -80,7 +80,7 @@ def generate_random_sine():
 def generate_random_mix():
     # filename
     name = uuid.uuid4().hex
-    audio_path = config.AUDIO_DIR+"/mix_"+name+".wav"
+    audio_path = dirs.AUDIO_DIR+"/mix_"+name+".wav"
     # open file in write mode
     audio = wave.open(audio_path, "wb")
     audio.setparams((2, 2, 24000, 0, 'NONE', 'not compressed'))

bot.py

@@ -1,59 +1,31 @@
 # the randomology telegram bot
 
 # config
-from configs import config
-
-# handlers
-from app.handlers.info_handler import get_info
-from app.handlers.text_handler import get_random_text
-from app.handlers.text_handler import get_random_text_mono
-from app.handlers.audio_handler import get_random_noise
-from app.handlers.audio_handler import get_random_sine
-from app.handlers.audio_handler import get_random_mix
-from app.handlers.audio_handler import get_random_voice
-from app.handlers.member_handler import get_welcome
-from app.handlers.inline_handler import inline_dispatch
+from configs import env 
+
+from routes import commands
+from routes import handlers
+
+from configs import env
+
+import server
+
+from telebot import TeleBot
 
 # pyTelegramBotAPI
 import telebot
-from telebot import types
-
-bot = telebot.TeleBot(config.BOT_TOKEN)
-
-def commands():
-    bot.set_my_commands([
-        telebot.types.BotCommand("help", "Get info"),
-        telebot.types.BotCommand("text", "Random text"),
-        telebot.types.BotCommand("mono", "Monospace random text"),
-        telebot.types.BotCommand("noise", "Random noise"),
-        telebot.types.BotCommand("sine", "Random sine wave audio"),
-        telebot.types.BotCommand("mix", "Random mix"),
-        telebot.types.BotCommand("voice", "Random voice")
-    ])
-
-commands()
-
-def handlers():
-    # info
-    bot.register_message_handler(get_info, commands=['start', 'help'], pass_bot=True)
-    # random text
-    bot.register_message_handler(get_random_text, commands=['text'], pass_bot=True)
-    bot.register_message_handler(get_random_text_mono, commands=['mono'], pass_bot=True)
-    # random audio
-    bot.register_message_handler(get_random_noise, commands=['noise'], pass_bot=True)
-    bot.register_message_handler(get_random_sine, commands=['sine'], pass_bot=True)
-    bot.register_message_handler(get_random_mix, commands=['mix'], pass_bot=True)
-    # random voice (a mix of noise and sine wave sent as voice)
-    bot.register_message_handler(get_random_voice, commands=['voice'], pass_bot=True)
-    # chat member change
-    bot.register_message_handler(get_welcome, content_types=['new_chat_members'], pass_bot=True)
-    # inline
-    bot.register_inline_handler(inline_dispatch, lambda query: query, pass_bot=True)
-
-handlers()
-
-def run():
-    bot.infinity_polling()
-
-# Here we go !!!!!!!!
-run()
+
+# bot initialize 
+bot = telebot.TeleBot(env.BOT_TOKEN)
+
+# register commands
+commands.register(bot)
+
+# register handlers
+handlers.register(bot)
+
+# here we go !!!
+if env.BOT_ENVIRONMENT == "prod":
+    server.run(bot)
+else:
+    server.run_dev(bot)

configs/config_utils.py

@@ -0,0 +1,13 @@
+# config_utils.py
+
+import tomllib
+
+from configs import dirs
+
+def env(section, key, default):
+    with open(dirs.ROOT_DIR+"/env.toml", "rb") as f:
+        data = tomllib.load(f)
+        if data[section][key] == "":
+            return default
+        else:
+            return data[section][key]

configs/config.py → configs/dirs.py

@@ -1,9 +1,7 @@
-# config
+# dir.py
 
-# for getting environment variable
 import os
 
-BOT_TOKEN = os.environ["BOT_TOKEN"]
 ROOT_DIR = os.path.dirname(os.path.dirname(__file__))
 STORAGE_DIR = os.path.join(ROOT_DIR, "storage")
 AUDIO_DIR = os.path.join(STORAGE_DIR, "audio")

configs/env.py

@@ -0,0 +1,37 @@
+# config
+from configs.config_utils import env
+
+# the bot token from telegran bot father
+BOT_TOKEN = env(section = "bot",
+                key = "token",
+                default="")
+
+# environment, dev or prod
+BOT_ENVIRONMENT = env(section = "bot",
+                      key = "environment",
+                      default="prod")
+
+# your host
+SERVER_HOST = env(section = "server",
+                  key = "host",
+                  default = "")
+
+# port, default 443
+SERVER_PORT = env(section = "server",
+                  key = "port",
+                  default = "443")
+
+# some vps need this, default 0.0.0.0
+SERVER_LISTEN = env(section = "server",
+                    key = "listen",
+                    default = "0.0.0.0")
+
+# path to your ssl cert file
+SSL_CERT = env(section = "ssl",
+               key = "cert",
+               default = "")
+
+# path to your ssl private key
+SSL_PRIV = env(section = "ssl",
+               key = "priv",
+               default = "")

requirements.txt

@@ -1,12 +1,29 @@
+annotated-types==0.6.0
+anyio==4.3.0
 certifi==2024.2.2
 charset-normalizer==3.3.2
+click==8.1.7
+fastapi==0.110.0
 ffmpeg-python==0.2.0
 future==1.0.0
+h11==0.14.0
+httptools==0.6.1
 idna==3.6
 iniconfig==2.0.0
 packaging==24.0
 pluggy==1.4.0
+pydantic==2.6.4
+pydantic_core==2.16.3
 pyTelegramBotAPI==4.16.1
 pytest==8.1.1
+python-dotenv==1.0.1
+PyYAML==6.0.1
 requests==2.31.0
+sniffio==1.3.1
+starlette==0.36.3
+typing_extensions==4.10.0
 urllib3==2.2.1
+uvicorn==0.29.0
+uvloop==0.19.0
+watchfiles==0.21.0
+websockets==12.0

routes/__init__.py

@@ -0,0 +1 @@
+

routes/commands.py

@@ -0,0 +1,17 @@
+# commands.py: register commands
+
+from telebot import TeleBot
+from telebot import types
+
+import telebot
+
+def register(bot:TeleBot):
+    bot.set_my_commands([
+        telebot.types.BotCommand("help", "Get info"),
+        telebot.types.BotCommand("text", "Random text"),
+        telebot.types.BotCommand("mono", "Monospace random text"),
+        telebot.types.BotCommand("noise", "Random noise"),
+        telebot.types.BotCommand("sine", "Random sine wave audio"),
+        telebot.types.BotCommand("mix", "Random mix"),
+        telebot.types.BotCommand("voice", "Random voice")
+    ])

routes/handlers.py

@@ -0,0 +1,31 @@
+# chat.py: chat routes
+
+from telebot import TeleBot
+
+# handlers
+from app.handlers.info_handler import get_info
+from app.handlers.text_handler import get_random_text
+from app.handlers.text_handler import get_random_text_mono
+from app.handlers.audio_handler import get_random_noise
+from app.handlers.audio_handler import get_random_sine
+from app.handlers.audio_handler import get_random_mix
+from app.handlers.audio_handler import get_random_voice
+from app.handlers.member_handler import get_welcome
+from app.handlers.inline_handler import inline_dispatch
+
+# register handlers in chats
+def register(bot:TeleBot):
+    bot.register_message_handler(get_info, commands=['start', 'help'], pass_bot=True)
+    # random text
+    bot.register_message_handler(get_random_text, commands=['text'], pass_bot=True)
+    bot.register_message_handler(get_random_text_mono, commands=['mono'], pass_bot=True)
+    # random audio
+    bot.register_message_handler(get_random_noise, commands=['noise'], pass_bot=True)
+    bot.register_message_handler(get_random_sine, commands=['sine'], pass_bot=True)
+    bot.register_message_handler(get_random_mix, commands=['mix'], pass_bot=True)
+    # random voice (a mix of noise and sine wave sent as voice)
+    bot.register_message_handler(get_random_voice, commands=['voice'], pass_bot=True)
+    # chat member change
+    bot.register_message_handler(get_welcome, content_types=['new_chat_members'], pass_bot=True)
+    # inline
+    bot.register_inline_handler(inline_dispatch, lambda query: query, pass_bot=True)

server.py

@@ -0,0 +1,55 @@
+# server.py
+
+import fastapi
+import uvicorn
+import telebot
+
+from configs import env
+
+from telebot import TeleBot
+
+TOKEN = env.BOT_TOKEN
+
+HOST = env.SERVER_HOST
+PORT = env.SERVER_PORT
+LISTEN = env.SERVER_LISTEN
+
+SSL_CERT = env.SSL_CERT
+SSL_PRIV = env.SSL_PRIV
+
+URL_BASE = "https://{}:{}".format(HOST, PORT)
+URL_PATH = "/{}/".format(TOKEN)
+
+# when in production
+def run(bot:TeleBot):
+    # create the app
+    app = fastapi.FastAPI(docs=None, redoc_url=None)
+
+    @app.post(f'/{TOKEN}/')
+    def process_webhook(update:dict):
+        if update:
+            update = telebot.types.Update.de_json(update)
+            bot.process_new_updates([update])
+        else:
+            return
+
+    # remove previous webhook(?)
+    bot.remove_webhook()
+
+    # set webhook
+    bot.set_webhook(
+        url=URL_BASE+URL_PATH
+    )
+
+    # run the server
+    uvicorn.run(
+        app,
+        host=HOST,
+        port=PORT,
+        ssl_certfile=SSL_CERT,
+        ssl_keyfile=SSL_PRIV
+    )
+
+# when in dev environment 
+def run_dev(bot:TeleBot):
+    bot.infinity_polling()