pawabot ¤

def get_parser() -> argparse.ArgumentParser:
    """Return the CLI argument parser.

    Returns:
        An argparse parser.
    """
    parser = argparse.ArgumentParser(prog="pawabot")
    subparsers = parser.add_subparsers(dest="subcommand", title="Commands", metavar="", prog="pawabot")
    subcommand_help = "Show this help message and exit."

    global_options = parser.add_argument_group(title="Global options")
    global_options.add_argument(
        "-L",
        "--log-level",
        dest="log_level",
        default="INFO",
        help="Log level to use",
        choices=("TRACE", "DEBUG", "INFO", "SUCCESS", "WARNING", "ERROR", "CRITICAL"),
        type=str.upper,
    )
    # global_options.add_argument(
    #     "-P", "--log-path", dest="log_path", default=None, help="Log path to use. Can be a directory or a file."
    # )

    def create_subparser(command: str, text: str, **kwargs: object) -> argparse.ArgumentParser:
        sub = subparsers.add_parser(command, add_help=False, help=text, description=text, **kwargs)  # ty:ignore[invalid-argument-type]
        sub.add_argument("-h", "--help", action="help", help=subcommand_help)
        return sub

    create_subparser("run", "Run the bot.")

    create_admin = create_subparser("create-admin", "Create an administrator in the database.")
    create_admin.add_argument("-i", "--uid", dest="uid", help="Telegram user id.")
    create_admin.add_argument("-u", "--username", dest="username", help="Telegram user name.")

    create_user = create_subparser("create-user", "Create a user in the database.")
    create_user.add_argument("-i", "--uid", dest="uid", help="Telegram user id.")
    create_user.add_argument("-u", "--username", dest="username", help="Telegram user name.")
    create_user.add_argument("-a", "--admin", action="store_true", dest="admin", help="Give admin access.")

    create_subparser("list-users", "List registered users.")

    # delete_users = subparser("delete-users", "Delete users by ID.")
    # delete_users.add_argument("uids", nargs="+", dest="uids", help="IDs of the users to delete.")

    # TODO: list-privileges
    # TODO: grant
    # TODO: revoke

    return parser

def main(cmd_args: list[str] | None = None) -> int:
    """Run the main program.

    This function is executed when you type `pawabot` or `python -m pawabot`.

    Arguments:
        cmd_args: Arguments passed from the command line.

    Returns:
        An exit code.
    """
    parser = get_parser()
    args = parser.parse_args(args=cmd_args)

    def log_level_to_name(level: int) -> str | None:
        for log_name in ["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"]:
            if level == getattr(logging, log_name):
                return log_name
        return None

    class InterceptHandler(logging.Handler):
        def emit(self, record: logging.LogRecord) -> None:
            # Retrieve context where the logging call occurred, this happens to be in the 6th frame upward
            logger_opt = logger.opt(depth=6, exception=record.exc_info)
            log_name = log_level_to_name(record.levelno)
            if log_name is not None:
                logger_opt.log(log_name, record.getMessage())

    log_level = args.log_level
    logger.configure(
        handlers=[
            {
                "sink": sys.stdout,
                "format": "{time:YYYY-MM-DD HH:mm:ss.SSS} | <lvl>{level:<8}</lvl> | {message}",
                "level": log_level,
            },
        ],
    )
    logging.basicConfig(handlers=[InterceptHandler()], level=0)

    _init(db_path="sqlite:///" + str(_DATA_DIR / "db.sqlite3"))

    if args.subcommand is None:
        print(parser.format_help(), file=sys.stderr)
        return 1

    # with open("owner_id.txt") as stream:
    #     OWNER_ID = stream.read().rstrip("\n")

    if args.subcommand == "create-admin":
        if args.uid is None or args.username is None:
            print("Error: --uid and --username are required", file=sys.stderr)
            return 1
        _User.create(uid=args.uid, username=args.username, is_admin=True)
        return 0
    if args.subcommand == "create-user":
        if args.uid is None or args.username is None:
            print("Error: --uid and --username are required", file=sys.stderr)
            return 1
        _User.create(uid=args.uid, username=args.username, is_admin=args.admin)
        return 0
    if args.subcommand == "list-users":
        print(f"{'ID':>10}  {'USERNAME':<20}  ADMIN")
        print("---------------------------------------")
        for user in _User.all():
            print(f"{user.uid:>10}  {user.username:<20}  {user.is_admin}")
            # TODO: also show privileges
        return 0
    # elif args.subcommand == "delete-users":
    #     for uid in args.uids:
    #         User
    if args.subcommand == "run":
        if "BOT_TOKEN" in os.environ:
            bot_token = os.environ.get("BOT_TOKEN")
        else:
            with (Path.home() / ".config" / "pawabot" / "bot_token.txt").open() as stream:
                bot_token = stream.read().rstrip("\n")

        if bot_token is None:
            print("Error: BOT_TOKEN environment variable is not set", file=sys.stderr)
            return 1

        app = ApplicationBuilder().token(bot_token).build()

        app.add_handler(CommandHandler("start", callbacks._start))
        app.add_handler(CommandHandler("help", callbacks._help))
        app.add_handler(CommandHandler("myID", callbacks._my_id))
        app.add_handler(CommandHandler("myPrivileges", callbacks._my_privileges))
        app.add_handler(CommandHandler("requestAccess", callbacks._request_access))
        app.add_handler(CommandHandler("grant", callbacks._grant))
        app.add_handler(CommandHandler("revoke", callbacks._revoke))

        # Interactive movie search: choose a movie, then a torrent.
        search_conversation = ConversationHandler(
            entry_points=[CommandHandler("search", callbacks._search)],
            allow_reentry=True,
            states={
                callbacks._SELECTING_RESULT: [
                    MessageHandler(filters.TEXT & ~filters.COMMAND, callbacks._select_search_result),
                ],
                callbacks._SELECTING_TORRENT: [
                    MessageHandler(filters.TEXT & ~filters.COMMAND, callbacks._select_torrent),
                ],
            },
            fallbacks=[CommandHandler("cancel", callbacks._cancel)],
        )
        app.add_handler(search_conversation)

        app.add_handler(MessageHandler(filters.Regex(callbacks._MAGNET_RE), callbacks._parse_magnet))

        app.add_handler(CommandHandler("test", callbacks._test))

        app.add_handler(MessageHandler(filters.COMMAND, callbacks._unknown_command))
        app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, callbacks._unknown))

        logger.info("Starting Bot")
        app.run_polling()

        return 0

    print(parser.format_help(), file=sys.stderr)
    return 1