This repository has been archived on 2024-08-21. You can view files and clone it, but cannot push or open issues or pull requests.
EmojiCaptchaBot/classes/pyrogroup.py

127 lines
3.8 KiB
Python

import logging
from dataclasses import dataclass
from typing import Union
from bson import ObjectId
from libbot import sync
from libbot.pyrogram.classes import PyroClient
from pyrogram.types import User
from modules.database import col_groups
logger = logging.getLogger(__name__)
@dataclass
class PyroGroup:
"""Dataclass of DB entry of a group"""
__slots__ = (
"_id",
"id",
"locale",
"locale_auto",
"ban_failed",
"timeout_join",
"timeout_verify",
)
_id: ObjectId
id: int
locale: Union[str, None]
locale_auto: bool
ban_failed: bool
timeout_join: int
timeout_verify: int
@classmethod
async def find(
cls,
id: Union[int, float],
locale: Union[str, None] = sync.config_get("locale", "defaults", "group"),
locale_auto: bool = sync.config_get("locale_auto", "defaults", "group"),
ban_failed: bool = sync.config_get("ban_failed", "defaults", "group"),
timeout_join: int = sync.config_get("timeout_join", "defaults", "group"),
timeout_verify: int = sync.config_get("timeout_verify", "defaults", "group"),
):
db_entry = await col_groups.find_one(
{
"id": id,
}
)
if db_entry is None:
inserted = await col_groups.insert_one(
{
"id": int(id),
"locale": locale,
"locale_auto": locale_auto,
"ban_failed": ban_failed,
"timeout_join": timeout_join,
"timeout_verify": timeout_verify,
}
)
db_entry = await col_groups.find_one({"_id": inserted.inserted_id})
if db_entry is None:
raise RuntimeError("Could not find inserted group entry.")
return cls(**db_entry)
async def set_locale(self, locale: Union[str, None]) -> None:
logger.debug("Locale of group %s has been set to %s", self.id, locale)
await col_groups.update_one({"_id": self._id}, {"$set": {"locale": locale}})
async def set_locale_auto(self, enabled: bool) -> None:
logger.debug(
"Automatic locale selection of group %s has been set to %s",
self.id,
enabled,
)
await col_groups.update_one(
{"_id": self._id}, {"$set": {"locale_auto": enabled}}
)
async def set_ban_failed(self, enabled: bool) -> None:
logger.debug(
"Banning users that failed the captcha in group %s has been set to %s",
self.id,
enabled,
)
await col_groups.update_one(
{"_id": self._id}, {"$set": {"ban_failed": enabled}}
)
async def set_timeout_join(self, timeout: int) -> None:
logger.debug(
"Join timeout in group %s has been set to %s",
self.id,
timeout,
)
await col_groups.update_one(
{"_id": self._id}, {"$set": {"timeout_join": timeout}}
)
async def set_timeout_verify(self, timeout: int) -> None:
logger.debug(
"Verification timeout in group %s has been set to %s",
self.id,
timeout,
)
await col_groups.update_one(
{"_id": self._id}, {"$set": {"timeout_verify": timeout}}
)
# Group settings
# User locale
def select_locale(
self, app: PyroClient, user: User, ignore_auto: bool = False
) -> str:
if user is None:
return app.default_locale
if not ignore_auto and self.locale_auto is True:
if user.language_code is not None:
return user.language_code
return self.locale if self.locale is not None else app.default_locale
return self.locale if self.locale is not None else app.default_locale