YouTubeLoader / bot /__main__.py
understanding's picture
Update bot/__main__.py
278122e verified
raw
history blame
1.86 kB
# PATH: bot/__main__.py
import asyncio
from logging import getLogger
from logging.config import dictConfig
from uvicorn import Config as UvicornConfig, Server as UvicornServer
from bot.config import Server, LOGGER_CONFIG_JSON
from bot.client import create_client
from bot.startup_log import log_startup
from bot.handlers import setup_handlers
from bot.server import create_app
logger = getLogger("bot")
def _validate_required_secrets() -> None:
from bot.config import Telegram, Workers
missing = []
if not Telegram.API_ID: missing.append("API_ID")
if not Telegram.API_HASH: missing.append("API_HASH")
if not Telegram.OWNER_ID: missing.append("OWNER_ID")
if not (Telegram.SESSION_STRING or Telegram.BOT_TOKEN):
missing.append("SESSION_STRING (preferred) or BOT_TOKEN")
if not Workers.WORKER1_URL: missing.append("WORKER1_URL")
if not Workers.WORKER2_URL: missing.append("WORKER2_URL")
if not Workers.BOT_BACKEND_KEY: missing.append("BOT_BACKEND_KEY")
if not Workers.HF_API_KEY: missing.append("HF_API_KEY")
if missing:
raise RuntimeError("Missing required secrets: " + ", ".join(missing))
async def main():
dictConfig(LOGGER_CONFIG_JSON)
_validate_required_secrets()
# web server
quart_app = create_app()
uv_cfg = UvicornConfig(
app=quart_app,
host=Server.BIND_ADDRESS,
port=Server.PORT,
log_config=LOGGER_CONFIG_JSON,
)
uv_server = UvicornServer(uv_cfg)
# telegram
app = create_client()
setup_handlers(app)
await app.start()
asyncio.create_task(log_startup(app))
asyncio.create_task(uv_server.serve())
logger.info("All services started ✅")
try:
while True:
await asyncio.sleep(3600)
finally:
await app.stop()
if __name__ == "__main__":
asyncio.run(main())