TelegramBotAdmin/launcher.py

from __future__ import annotations

import argparse
import asyncio
import os
import runpy
import signal
from pathlib import Path

PROJECT_ROOT = Path(__file__).resolve().parent


def _arguments() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Telegram manager launcher")
    parser.add_argument("--worker", default="")
    parser.add_argument("--port", type=int, default=0)
    return parser.parse_args()


async def _run_supervisor() -> None:
    os.environ["WBB_ADMIN_BOOTSTRAP"] = "1"
    os.environ["WBB_SUPERVISOR_MODE"] = "1"

    import wbb
    from wbb.admin.supervisor import BotSupervisor
    from wbb.web_admin import start_admin_web, stop_admin_web

    supervisor = BotSupervisor(
        wbb.BOT_PROFILES_PATH,
        project_root=PROJECT_ROOT,
        control_port=wbb.ADMIN_WEB_PORT,
    )
    wbb.BOT_SUPERVISOR = supervisor
    await supervisor.start_enabled()
    await start_admin_web()

    stopped = asyncio.Event()
    loop = asyncio.get_running_loop()
    for name in (signal.SIGINT, signal.SIGTERM):
        try:
            loop.add_signal_handler(name, stopped.set)
        except NotImplementedError:
            pass
    try:
        await stopped.wait()
    finally:
        await supervisor.close()
        await stop_admin_web()


def main() -> None:
    args = _arguments()
    if args.worker:
        if args.port:
            os.environ["ADMIN_WEB_PORT"] = str(args.port)
        runpy.run_module("wbb", run_name="__main__")
        return
    asyncio.run(_run_supervisor())


if __name__ == "__main__":
    main()