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
Raw Permalink Normal View History

2023-08-11 16:04:21 +03:00
import logging
from dataclasses import dataclass
from typing import Union
from bson import ObjectId
2023-08-14 12:56:59 +03:00
from libbot import sync
2023-08-23 11:36:45 +03:00
from libbot.pyrogram.classes import PyroClient
2023-08-11 16:04:21 +03:00
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"""
2023-08-14 13:34:52 +03:00
__slots__ = (
"_id",
"id",
"locale",
"locale_auto",
"ban_failed",
"timeout_join",
"timeout_verify",
)
2023-08-11 16:04:21 +03:00
_id: ObjectId
id: int
locale: Union[str, None]
locale_auto: bool
2023-08-14 13:11:53 +03:00
ban_failed: bool
2023-08-14 13:34:52 +03:00
timeout_join: int
timeout_verify: int
2023-08-11 16:04:21 +03:00
@classmethod
2023-08-17 17:37:42 +03:00
async def find(
2023-08-11 16:04:21 +03:00
cls,
2023-08-23 12:12:26 +03:00
id: Union[int, float],
2023-08-14 12:56:59 +03:00
locale: Union[str, None] = sync.config_get("locale", "defaults", "group"),
locale_auto: bool = sync.config_get("locale_auto", "defaults", "group"),
2023-08-14 13:11:53 +03:00
ban_failed: bool = sync.config_get("ban_failed", "defaults", "group"),
2023-08-14 13:34:52 +03:00
timeout_join: int = sync.config_get("timeout_join", "defaults", "group"),
timeout_verify: int = sync.config_get("timeout_verify", "defaults", "group"),
2023-08-11 16:04:21 +03:00
):
db_entry = await col_groups.find_one(
{
"id": id,
}
)
2023-08-11 16:04:21 +03:00
if db_entry is None:
inserted = await col_groups.insert_one(
2023-08-14 12:56:59 +03:00
{
2023-08-23 12:12:26 +03:00
"id": int(id),
2023-08-14 12:56:59 +03:00
"locale": locale,
"locale_auto": locale_auto,
2023-08-14 13:11:53 +03:00
"ban_failed": ban_failed,
2023-08-14 13:34:52 +03:00
"timeout_join": timeout_join,
"timeout_verify": timeout_verify,
2023-08-14 12:56:59 +03:00
}
2023-08-11 16:04:21 +03:00
)
db_entry = await col_groups.find_one({"_id": inserted.inserted_id})
if db_entry is None:
raise RuntimeError("Could not find inserted group entry.")
2023-08-11 16:04:21 +03:00
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}}
)
2023-08-14 13:11:53 +03:00
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}}
)
2023-08-14 13:34:52 +03:00
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}}
)
2023-08-11 16:04:21 +03:00
# Group settings
# User locale
def select_locale(
self, app: PyroClient, user: User, ignore_auto: bool = False
2023-08-11 16:04:21 +03:00
) -> str:
2023-08-23 11:34:45 +03:00
if user is None:
return app.default_locale
2023-08-11 16:04:21 +03:00
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