Compare commits

..

No commits in common. "main" and "v1.2" have entirely different histories.
main ... v1.2

51 changed files with 165 additions and 1478 deletions

View File

@ -1,38 +0,0 @@
name: Tests
on:
push:
branches:
- main
tags-ignore:
- v*
pull_request:
jobs:
test:
runs-on: ubuntu-latest
container: catthehacker/ubuntu:act-latest
strategy:
matrix:
python-version: ["3.8", "3.9", "3.10", "3.11"]
steps:
- uses: actions/checkout@v3
- name: Set up Python ${{ matrix.python-version }}
uses: actions/setup-python@v3
with:
python-version: ${{ matrix.python-version }}
env:
AGENT_TOOLSDIRECTORY: /opt/hostedtoolcache
- name: Install dependencies
run: |
python -m pip install --upgrade pip
pip install tox tox-gh-actions build
- name: Test with tox
run: tox
- name: Build
run: python -m build
- uses: actions/upload-artifact@v3
with:
name: Artifacts
path: dist/*

10
.gitignore vendored
View File

@ -161,10 +161,8 @@ cython_debug/
#.idea/ #.idea/
# Project # Project
venv/ venv
venv_linux/ venv_linux
venv_windows/ venv_windows
.vscode/ .vscode
tests/.tmp/

View File

@ -3,15 +3,6 @@
"extends": [ "extends": [
"config:base" "config:base"
], ],
"baseBranches": [
"dev"
],
"pip_requirements": {
"fileMatch": [
"requirements/.*\\.txt$"
],
"enabled": true
},
"packageRules": [ "packageRules": [
{ {
"matchUpdateTypes": [ "matchUpdateTypes": [

View File

@ -1,86 +1,2 @@
<h1 align="center">LibBotUniversal</h1> # LibBotUniversal
<p align="center">
<a href="https://git.end-play.xyz/profitroll/LibBotUniversal/src/branch/master/LICENSE"><img alt="PyPI - License" src="https://img.shields.io/pypi/l/libbot">
<a href="https://git.end-play.xyz/profitroll/LibBotUniversal/releases/latest"><img alt="Gitea Release" src="https://img.shields.io/gitea/v/release/profitroll/LibBotUniversal?gitea_url=https%3A%2F%2Fgit.end-play.xyz"></a>
<a href="https://pypi.org/project/libbot/"><img alt="PyPI - Python Version" src="https://img.shields.io/pypi/pyversions/libbot"></a>
<a href="https://git.end-play.xyz/profitroll/LibBotUniversal"><img alt="Code style: black" src="https://img.shields.io/badge/code%20style-black-000000.svg"></a>
</p>
Handy library for Telegram/Discord bots development.
## Getting started
There are different sub-packages available:
* pyrogram - Telegram bots with Pyrogram's fork "Pyrofork"
* pycord - Discord bots with Pycord
* speed - Performance improvements
* dev - Dependencies for package development purposes
You can freely choose any sub-package you want, as well as add multiple (comma-separated) or none at all.
```shell
# Only general features
pip install libbot
# Only with Pyrogram
pip install libbot[pyrogram]
# With Pycord and Performance improvements
pip install libbot[pycord,speed]
```
## Examples
### Pyrogram
```python
from libbot.pyrogram import PyroClient
def main():
client = PyroClient(scheduler=scheduler)
try:
client.run()
except KeyboardInterrupt:
print("Shutting down...")
finally:
if client.scheduler is not None:
client.scheduler.shutdown()
exit()
if __name__ == "__main__":
main()
```
### Pycord
```python
from discord import Intents
from libbot import sync
from libbot.pycord import PycordBot
async def main():
intents = Intents.default()
bot = PycordBot(intents=intents)
bot.load_extension("cogs")
try:
await bot.start(sync.config_get("bot_token", "bot"))
except KeyboardInterrupt:
logger.warning("Shutting down...")
await bot.close()
if __name__ == "__main__":
loop = asyncio.get_event_loop()
loop.run_until_complete(main())
```
## Config examples
For bot config examples please check the examples directory. Without a valid config file, the bot won't start at all, so you need to make sure the correct config file is used.

View File

@ -1,21 +0,0 @@
{
"help": {
"scopes": [
{
"name": "BotCommandScopeDefault"
},
{
"name": "BotCommandScopeChat",
"chat_id": "owner"
}
]
},
"shutdown": {
"scopes": [
{
"name": "BotCommandScopeChat",
"chat_id": "owner"
}
]
}
}

View File

@ -1,13 +0,0 @@
{
"locale": "en",
"debug": false,
"bot": {
"owners": [
0
],
"debug_guilds": [
0
],
"bot_token": ""
}
}

View File

@ -1,38 +0,0 @@
{
"locale": "en",
"bot": {
"owner": 0,
"api_id": 0,
"api_hash": "",
"bot_token": "",
"workers": 1,
"max_concurrent_transmissions": 1,
"scoped_commands": true
},
"reports": {
"chat_id": "owner"
},
"disabled_plugins": [],
"commands": {
"help": {
"scopes": [
{
"name": "BotCommandScopeDefault"
},
{
"name": "BotCommandScopeChat",
"chat_id": "owner"
}
]
},
"shutdown": {
"scopes": [
{
"name": "BotCommandScopeChat",
"chat_id": "owner"
}
]
}
}
}

View File

@ -1,23 +0,0 @@
{
"metadata": {
"flag": "🇬🇧",
"name": "English",
"codes": [
"en"
]
},
"bot": {
"name": "Your Bot",
"about": "I'm a your bot. Nice to meet you!",
"description": "I'm just your bot. Yet nice to meet you!"
},
"commands": {
"help": "Show help message"
},
"messages": {
"help": "Sample Text"
},
"callbacks": {
"sample": "This button is working!"
}
}

9
libbot/__init__.py Normal file
View File

@ -0,0 +1,9 @@
__name__ = "libbot"
__version__ = "1.2"
__license__ = "GPL3"
__author__ = "Profitroll"
from .__main__ import *
from . import sync
from . import i18n
from . import pyrogram

View File

@ -8,8 +8,7 @@ try:
except ImportError: except ImportError:
from json import dumps, loads from json import dumps, loads
from ._utils import supports_argument from libbot.sync import nested_set
from .sync._nested import nested_delete, nested_set
async def json_read(path: Union[str, Path]) -> Any: async def json_read(path: Union[str, Path]) -> Any:
@ -23,7 +22,6 @@ async def json_read(path: Union[str, Path]) -> Any:
""" """
async with aiofiles.open(str(path), mode="r", encoding="utf-8") as f: async with aiofiles.open(str(path), mode="r", encoding="utf-8") as f:
data = await f.read() data = await f.read()
return loads(data) return loads(data)
@ -37,8 +35,6 @@ async def json_write(data: Any, path: Union[str, Path]) -> None:
async with aiofiles.open(str(path), mode="w", encoding="utf-8") as f: async with aiofiles.open(str(path), mode="w", encoding="utf-8") as f:
await f.write( await f.write(
dumps(data, ensure_ascii=False, escape_forward_slashes=False, indent=4) dumps(data, ensure_ascii=False, escape_forward_slashes=False, indent=4)
if supports_argument(dumps, "escape_forward_slashes")
else dumps(data, ensure_ascii=False, indent=4)
) )
@ -75,10 +71,8 @@ async def config_get(
``` ```
""" """
this_key = await json_read(config_file) this_key = await json_read(config_file)
for dict_key in path: for dict_key in path:
this_key = this_key[dict_key] this_key = this_key[dict_key]
return this_key[key] return this_key[key]
@ -92,39 +86,8 @@ async def config_set(
* value (`Any`): Any JSON serializable data * value (`Any`): Any JSON serializable data
* *path (`str`): Path to the key of the target * *path (`str`): Path to the key of the target
* config_file (`Union[str, Path]`, *optional*): Path-like object or path as a string of a location of the config file. Defaults to `"config.json"` * config_file (`Union[str, Path]`, *optional*): Path-like object or path as a string of a location of the config file. Defaults to `"config.json"`
### Raises:
* `KeyError`: Key is not found under path provided
""" """
await json_write( await json_write(
nested_set(await json_read(config_file), value, *(*path, key)), config_file nested_set(await json_read(config_file), value, *(*path, key)), config_file
) )
return return
async def config_delete(
key: str,
*path: str,
missing_ok: bool = False,
config_file: Union[str, Path] = "config.json",
) -> None:
"""Set config's key by its path
### Args:
* key (`str`): Key to delete
* *path (`str`): Path to the key of the target
* missing_ok (`bool`): Do not raise an exception if the key is missing. Defaults to `False`
* config_file (`Union[str, Path]`, *optional*): Path-like object or path as a string of a location of the config file. Defaults to `"config.json"`
### Raises:
* `KeyError`: Key is not found under path provided and `missing_ok` is `False`
"""
config_data = await json_read(config_file)
try:
nested_delete(config_data, *(*path, key))
except KeyError as exc:
if not missing_ok:
raise exc from exc
await json_write(config_data, config_file)

View File

@ -1,36 +1,35 @@
from os import listdir from os import listdir
from pathlib import Path from pathlib import Path
from typing import Any, Dict, Union from typing import Any, Dict
import libbot from libbot import config_get, json_read, sync
from libbot.i18n.classes.bot_locale import BotLocale
def _( async def _(key: str, *args: str, locale: str = sync.config_get("locale")) -> Any:
key: str,
*args: str,
locale: Union[str, None] = "en",
locales_root: Union[str, Path] = Path("locale"),
) -> Any:
"""Get value of locale string """Get value of locale string
### Args: ### Args:
* key (`str`): The last key of the locale's keys path. * key (`str`): The last key of the locale's keys path
* *args (`list`): Path to key like: `dict[args][key]`. * *args (`list`): Path to key like: `dict[args][key]`
* locale (`Union[str, None]`): Locale to looked up in. Defaults to `"en"`. * locale (`str`): Locale to looked up in. Defaults to config's `"locale"` value
* locales_root (`Union[str, Path]`, *optional*): Folder where locales are located. Defaults to `Path("locale")`.
### Returns: ### Returns:
* `Any`: Value of provided locale key. Is usually `str`, `dict` or `list` * `Any`: Value of provided locale key. Is usually `str`, `dict` or `list`
""" """
if locale is None: if locale is None:
locale = libbot.sync.config_get("locale") locale = sync.config_get("locale")
try: try:
this_dict = libbot.sync.json_read(Path(f"{locales_root}/{locale}.json")) this_dict = await json_read(
Path(f'{await config_get("locale", "locations")}/{locale}.json')
)
except FileNotFoundError: except FileNotFoundError:
try: try:
this_dict = libbot.sync.json_read( this_dict = await json_read(
Path(f'{locales_root}/{libbot.sync.config_get("locale")}.json') Path(
f'{await config_get("locale", "locations")}/{await config_get("locale")}.json'
)
) )
except FileNotFoundError: except FileNotFoundError:
return f'⚠️ Locale in config is invalid: could not get "{key}" in {args} from locale "{locale}"' return f'⚠️ Locale in config is invalid: could not get "{key}" in {args} from locale "{locale}"'
@ -45,27 +44,26 @@ def _(
return f'⚠️ Locale in config is invalid: could not get "{key}" in {args} from locale "{locale}"' return f'⚠️ Locale in config is invalid: could not get "{key}" in {args} from locale "{locale}"'
def in_all_locales( async def in_all_locales(key: str, *args: str) -> list:
key: str, *args: str, locales_root: Union[str, Path] = Path("locale")
) -> list:
"""Get value of the provided key and path in all available locales """Get value of the provided key and path in all available locales
### Args: ### Args:
* key (`str`): The last key of the locale's keys path. * key (`str`): The last key of the locale's keys path.
* *args (`list`): Path to key like: `dict[args][key]`. * *args (`list`): Path to key like: `dict[args][key]`.
* locales_root (`Union[str, Path]`, *optional*): Folder where locales are located. Defaults to `Path("locale")`.
### Returns: ### Returns:
* `list`: List of values in all locales * `list`: List of values in all locales
""" """
output = [] output = []
files_locales = listdir(locales_root) files_locales = listdir(await config_get("locale", "locations"))
valid_locales = [".".join(entry.split(".")[:-1]) for entry in files_locales] valid_locales = [".".join(entry.split(".")[:-1]) for entry in files_locales]
for lc in valid_locales: for lc in valid_locales:
try: try:
this_dict = libbot.sync.json_read(Path(f"{locales_root}/{lc}.json")) this_dict = await json_read(
Path(f'{await config_get("locale", "locations")}/{lc}.json')
)
except FileNotFoundError: except FileNotFoundError:
continue continue
@ -81,27 +79,26 @@ def in_all_locales(
return output return output
def in_every_locale( async def in_every_locale(key: str, *args: str) -> Dict[str, Any]:
key: str, *args: str, locales_root: Union[str, Path] = Path("locale")
) -> Dict[str, Any]:
"""Get value of the provided key and path in every available locale with locale tag """Get value of the provided key and path in every available locale with locale tag
### Args: ### Args:
* key (`str`): The last key of the locale's keys path. * key (`str`): The last key of the locale's keys path.
* *args (`list`): Path to key like: `dict[args][key]`. * *args (`list`): Path to key like: `dict[args][key]`.
* locales_root (`Union[str, Path]`, *optional*): Folder where locales are located. Defaults to `Path("locale")`.
### Returns: ### Returns:
* `Dict[str, Any]`: Locale is a key and it's value from locale file is a value * `Dict[str, Any]`: Locale is a key and it's value from locale file is a value
""" """
output = {} output = {}
files_locales = listdir(locales_root) files_locales = listdir(await config_get("locale", "locations"))
valid_locales = [".".join(entry.split(".")[:-1]) for entry in files_locales] valid_locales = [".".join(entry.split(".")[:-1]) for entry in files_locales]
for lc in valid_locales: for lc in valid_locales:
try: try:
this_dict = libbot.sync.json_read(Path(f"{locales_root}/{lc}.json")) this_dict = await json_read(
Path(f'{await config_get("locale", "locations")}/{lc}.json')
)
except FileNotFoundError: except FileNotFoundError:
continue continue

View File

@ -8,29 +8,25 @@ from libbot import sync
class BotLocale: class BotLocale:
"""Small addon that can be used by bot clients' classes in order to minimize I/O""" """Small addon that can be used by bot clients' classes in order to minimize I/O"""
def __init__( def __init__(self, locales_folder: Union[str, Path, None] = None) -> None:
self, if locales_folder is None:
default_locale: Union[str, None] = "en", locales_folder = Path(sync.config_get("locale", "locations"))
locales_root: Union[str, Path] = Path("locale"), elif isinstance(locales_folder, str):
) -> None: locales_folder = Path(locales_folder)
if isinstance(locales_root, str): elif not isinstance(locales_folder, Path):
locales_root = Path(locales_root) raise TypeError("'locales_folder' must be a valid path or path-like object")
elif not isinstance(locales_root, Path):
raise TypeError("'locales_root' must be a valid path or path-like object")
files_locales: list = listdir(locales_root) files_locales: list = listdir(locales_folder)
valid_locales: list = [ valid_locales: list = [
".".join(entry.split(".")[:-1]) for entry in files_locales ".".join(entry.split(".")[:-1]) for entry in files_locales
] ]
self.default: str = ( self.default: str = sync.config_get("locale")
sync.config_get("locale") if default_locale is None else default_locale
)
self.locales: dict = {} self.locales: dict = {}
for lc in valid_locales: for lc in valid_locales:
self.locales[lc] = sync.json_read(Path(f"{locales_root}/{lc}.json")) self.locales[lc] = sync.json_read(Path(f"{locales_folder}/{lc}.json"))
def _(self, key: str, *args: str, locale: Union[str, None] = None) -> Any: def _(self, key: str, *args: str, locale: Union[str, None] = None) -> Any:
"""Get value of locale string """Get value of locale string

View File

@ -1,37 +1,33 @@
from os import listdir from os import listdir
from pathlib import Path from pathlib import Path
from typing import Any, Dict, Union from typing import Any, Dict
import libbot from libbot import sync
from libbot.i18n import sync from libbot.sync import config_get, json_read
from libbot.i18n.classes.bot_locale import BotLocale
async def _( def _(key: str, *args: str, locale: str = sync.config_get("locale")) -> Any:
key: str,
*args: str,
locale: Union[str, None] = "en",
locales_root: Union[str, Path] = Path("locale"),
) -> Any:
"""Get value of locale string """Get value of locale string
### Args: ### Args:
* key (`str`): The last key of the locale's keys path. * key (`str`): The last key of the locale's keys path
* *args (`list`): Path to key like: `dict[args][key]`. * *args (`list`): Path to key like: `dict[args][key]`
* locale (`Union[str, None]`): Locale to looked up in. Defaults to `"en"`. * locale (`str`): Locale to looked up in. Defaults to config's `"locale"` value
* locales_root (`Union[str, Path]`, *optional*): Folder where locales are located. Defaults to `Path("locale")`.
### Returns: ### Returns:
* `Any`: Value of provided locale key. Is usually `str`, `dict` or `list` * `Any`: Value of provided locale key. Is usually `str`, `dict` or `list`
""" """
locale = libbot.sync.config_get("locale") if locale is None else locale if locale is None:
locale = sync.config_get("locale")
try: try:
this_dict = await libbot.json_read(Path(f"{locales_root}/{locale}.json")) this_dict = json_read(
Path(f'{config_get("locale", "locations")}/{locale}.json')
)
except FileNotFoundError: except FileNotFoundError:
try: try:
this_dict = await libbot.json_read( this_dict = json_read(
Path(f'{locales_root}/{await libbot.config_get("locale")}.json') Path(f'{config_get("locale", "locations")}/{config_get("locale")}.json')
) )
except FileNotFoundError: except FileNotFoundError:
return f'⚠️ Locale in config is invalid: could not get "{key}" in {args} from locale "{locale}"' return f'⚠️ Locale in config is invalid: could not get "{key}" in {args} from locale "{locale}"'
@ -46,27 +42,26 @@ async def _(
return f'⚠️ Locale in config is invalid: could not get "{key}" in {args} from locale "{locale}"' return f'⚠️ Locale in config is invalid: could not get "{key}" in {args} from locale "{locale}"'
async def in_all_locales( def in_all_locales(key: str, *args: str) -> list:
key: str, *args: str, locales_root: Union[str, Path] = Path("locale")
) -> list:
"""Get value of the provided key and path in all available locales """Get value of the provided key and path in all available locales
### Args: ### Args:
* key (`str`): The last key of the locale's keys path. * key (`str`): The last key of the locale's keys path.
* *args (`list`): Path to key like: `dict[args][key]`. * *args (`list`): Path to key like: `dict[args][key]`.
* locales_root (`Union[str, Path]`, *optional*): Folder where locales are located. Defaults to `Path("locale")`.
### Returns: ### Returns:
* `list`: List of values in all locales * `list`: List of values in all locales
""" """
output = [] output = []
files_locales = listdir(locales_root) files_locales = listdir(config_get("locale", "locations"))
valid_locales = [".".join(entry.split(".")[:-1]) for entry in files_locales] valid_locales = [".".join(entry.split(".")[:-1]) for entry in files_locales]
for lc in valid_locales: for lc in valid_locales:
try: try:
this_dict = await libbot.json_read(Path(f"{locales_root}/{lc}.json")) this_dict = json_read(
Path(f'{config_get("locale", "locations")}/{lc}.json')
)
except FileNotFoundError: except FileNotFoundError:
continue continue
@ -82,27 +77,26 @@ async def in_all_locales(
return output return output
async def in_every_locale( def in_every_locale(key: str, *args: str) -> Dict[str, Any]:
key: str, *args: str, locales_root: Union[str, Path] = Path("locale")
) -> Dict[str, Any]:
"""Get value of the provided key and path in every available locale with locale tag """Get value of the provided key and path in every available locale with locale tag
### Args: ### Args:
* key (`str`): The last key of the locale's keys path. * key (`str`): The last key of the locale's keys path.
* *args (`list`): Path to key like: `dict[args][key]`. * *args (`list`): Path to key like: `dict[args][key]`.
* locales_root (`Union[str, Path]`, *optional*): Folder where locales are located. Defaults to `Path("locale")`.
### Returns: ### Returns:
* `Dict[str, Any]`: Locale is a key and it's value from locale file is a value * `Dict[str, Any]`: Locale is a key and it's value from locale file is a value
""" """
output = {} output = {}
files_locales = listdir(locales_root) files_locales = listdir(config_get("locale", "locations"))
valid_locales = [".".join(entry.split(".")[:-1]) for entry in files_locales] valid_locales = [".".join(entry.split(".")[:-1]) for entry in files_locales]
for lc in valid_locales: for lc in valid_locales:
try: try:
this_dict = await libbot.json_read(Path(f"{locales_root}/{lc}.json")) this_dict = json_read(
Path(f'{config_get("locale", "locations")}/{lc}.json')
)
except FileNotFoundError: except FileNotFoundError:
continue continue

View File

@ -1,15 +1,12 @@
import asyncio
import logging import logging
from datetime import datetime, timedelta from datetime import datetime, timedelta
from os import cpu_count, getpid from os import getpid
from pathlib import Path from pathlib import Path
from time import time from time import time
from typing import Any, Dict, List, Union from typing import List, Union
try: try:
import pyrogram import pyrogram
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from apscheduler.schedulers.background import BackgroundScheduler
from pyrogram.client import Client from pyrogram.client import Client
from pyrogram.errors import BadRequest from pyrogram.errors import BadRequest
from pyrogram.handlers.message_handler import MessageHandler from pyrogram.handlers.message_handler import MessageHandler
@ -43,75 +40,29 @@ logger = logging.getLogger(__name__)
class PyroClient(Client): class PyroClient(Client):
def __init__( def __init__(self):
self, with open("config.json", "r", encoding="utf-8") as f:
name: str = "bot_client", self.config: dict = loads(f.read())
owner: Union[int, None] = None,
config: Union[Dict[str, Any], None] = None,
config_path: Union[str, Path] = Path("config.json"),
api_id: Union[int, None] = None,
api_hash: Union[str, None] = None,
bot_token: Union[str, None] = None,
workers: int = min(32, cpu_count() + 4),
locales_root: Union[str, Path, None] = None,
plugins_root: str = "plugins",
plugins_exclude: Union[List[str], None] = None,
sleep_threshold: int = 120,
max_concurrent_transmissions: int = 1,
commands_source: Union[Dict[str, dict], None] = None,
scoped_commands: Union[bool, None] = None,
i18n_bot_info: bool = False,
scheduler: Union[AsyncIOScheduler, BackgroundScheduler, None] = None,
**kwargs,
):
if config is None:
with open(config_path, "r", encoding="utf-8") as f:
self.config: dict = loads(f.read())
else:
self.config = config
super().__init__( super().__init__(
name=name, name="bot_client",
api_id=self.config["bot"]["api_id"] if api_id is None else api_id, api_id=self.config["bot"]["api_id"],
api_hash=self.config["bot"]["api_hash"] if api_hash is None else api_hash, api_hash=self.config["bot"]["api_hash"],
bot_token=self.config["bot"]["bot_token"] bot_token=self.config["bot"]["bot_token"],
if bot_token is None # Workers should be commented when using convopyro, otherwise
else bot_token,
# Workers should be `min(32, cpu_count() + 4)`, otherwise
# handlers land in another event loop and you won't see them # handlers land in another event loop and you won't see them
workers=self.config["bot"]["workers"] workers=self.config["bot"]["workers"],
if "workers" in self.config["bot"] plugins=dict(root="plugins", exclude=self.config["disabled_plugins"]),
else workers, sleep_threshold=120,
plugins=dict(
root=plugins_root,
exclude=self.config["disabled_plugins"]
if plugins_exclude is None
else plugins_exclude,
),
sleep_threshold=sleep_threshold,
max_concurrent_transmissions=self.config["bot"][ max_concurrent_transmissions=self.config["bot"][
"max_concurrent_transmissions" "max_concurrent_transmissions"
] ],
if "max_concurrent_transmissions" in self.config["bot"]
else max_concurrent_transmissions,
**kwargs,
) )
self.owner: int = self.config["bot"]["owner"] if owner is None else owner self.owner: int = self.config["bot"]["owner"]
self.commands: List[PyroCommand] = [] self.commands: List[PyroCommand] = []
self.commands_source: Dict[str, dict] = ( self.scoped_commands: bool = self.config["bot"]["scoped_commands"]
self.config["commands"] if commands_source is None else commands_source
)
self.scoped_commands: bool = (
self.config["bot"]["scoped_commands"]
if scoped_commands is None
else scoped_commands
)
self.start_time: float = 0 self.start_time: float = 0
self.bot_locale: BotLocale = BotLocale( self.bot_locale: BotLocale = BotLocale(Path(self.config["locations"]["locale"]))
default_locale=self.config["locale"],
locales_root=(Path("locale") if locales_root is None else locales_root),
)
self.default_locale: str = self.bot_locale.default self.default_locale: str = self.bot_locale.default
self.locales: dict = self.bot_locale.locales self.locales: dict = self.bot_locale.locales
@ -119,13 +70,7 @@ class PyroClient(Client):
self.in_all_locales = self.bot_locale.in_all_locales self.in_all_locales = self.bot_locale.in_all_locales
self.in_every_locale = self.bot_locale.in_every_locale self.in_every_locale = self.bot_locale.in_every_locale
self.scheduler: Union[AsyncIOScheduler, BackgroundScheduler, None] = scheduler async def start(self, scheduler):
self.scopes_placeholders: Dict[str, int] = {"owner": self.owner}
self.i18n_bot_info: bool = i18n_bot_info
async def start(self, register_commands: bool = True) -> None:
await super().start() await super().start()
self.start_time = time() self.start_time = time()
@ -138,107 +83,39 @@ class PyroClient(Client):
getpid(), getpid(),
) )
if self.i18n_bot_info:
# Register default bot's info
try:
await self.set_bot_info(
name=self._("name", "bot"),
about=self._("about", "bot"),
description=self._("description", "bot"),
lang_code="",
)
logger.info(
"Bot's info for the default locale %s has been updated",
self.default_locale,
)
except KeyError:
logger.warning(
"Default locale %s has incorrect keys or values in bot section",
self.default_locale,
)
# Register bot's info for each available locale
for locale_code in self.locales:
locale = self.locales[locale_code]
if "metadata" not in locale or ("codes" not in locale["metadata"]):
logger.warning(
"Locale %s is missing metadata or metadata.codes key",
locale_code,
)
continue
for code in locale["metadata"]["codes"]:
try:
await self.set_bot_info(
name=locale["bot"]["name"],
about=locale["bot"]["about"],
description=locale["bot"]["description"],
lang_code=code,
)
logger.info(
"Bot's info for the locale %s has been updated",
self.code,
)
except KeyError:
logger.warning(
"Locale %s has incorrect keys or values in bot section",
locale_code,
)
# Send a message to the bot's reports chat about the startup
try: try:
await self.send_message( await self.send_message(
chat_id=self.owner chat_id=self.config["reports"]["chat_id"],
if self.config["reports"]["chat_id"] == "owner"
else self.config["reports"]["chat_id"],
text=f"Bot started PID `{getpid()}`", text=f"Bot started PID `{getpid()}`",
) )
except BadRequest:
logger.warning("Unable to send message to report chat.")
if self.scheduler is None: scheduler.add_job(
return
# Schedule the task to register all commands
if register_commands:
self.scheduler.add_job(
self.register_commands, self.register_commands,
trigger="date", trigger="date",
run_date=datetime.now() + timedelta(seconds=5), run_date=datetime.now() + timedelta(seconds=5),
kwargs={"command_sets": await self.collect_commands()}, kwargs={"command_sets": await self.collect_commands()},
) )
self.scheduler.start() scheduler.start()
async def stop(self, exit_completely: bool = True) -> None:
try:
await self.send_message(
chat_id=self.owner
if self.config["reports"]["chat_id"] == "owner"
else self.config["reports"]["chat_id"],
text=f"Bot stopped with PID `{getpid()}`",
)
await asyncio.sleep(0.5)
except BadRequest: except BadRequest:
logger.warning("Unable to send message to report chat.") logger.warning("Unable to send message to report chat.")
async def stop(self):
try:
await self.send_message(
chat_id=self.config["reports"]["chat_id"],
text=f"Bot stopped with PID `{getpid()}`",
)
except BadRequest:
logger.warning("Unable to send message to report chat.")
await super().stop() await super().stop()
logger.warning("Bot stopped with PID %s.", getpid()) logger.warning("Bot stopped with PID %s.", getpid())
if exit_completely:
try:
exit()
except SystemExit as exc:
raise SystemExit(
"Bot has been shut down, this is not an application error!"
) from exc
async def collect_commands(self) -> Union[List[CommandSet], None]: async def collect_commands(self) -> Union[List[CommandSet], None]:
"""Gather list of the bot's commands """Gather list of the bot's commands
### Returns: ### Returns:
* `List[CommandSet]`: List of the commands' sets. * `List[CommandSet]`: List of the commands' sets
""" """
command_sets = None command_sets = None
@ -249,7 +126,7 @@ class PyroClient(Client):
command_sets = [] command_sets = []
# Iterate through all commands in config # Iterate through all commands in config
for command, contents in self.commands_source.items(): for command, contents in self.config["commands"].items():
# Iterate through all scopes of a command # Iterate through all scopes of a command
for scope in contents["scopes"]: for scope in contents["scopes"]:
if dumps(scope) not in scopes: if dumps(scope) not in scopes:
@ -274,9 +151,8 @@ class PyroClient(Client):
scope_dict = loads(scope) scope_dict = loads(scope)
# Replace "owner" in the bot scope with owner's id # Replace "owner" in the bot scope with owner's id
for placeholder, chat_id in self.scopes_placeholders.items(): if "chat_id" in scope_dict and scope_dict["chat_id"] == "owner":
if "chat_id" in scope_dict and scope_dict["chat_id"] == placeholder: scope_dict["chat_id"] = self.owner
scope_dict["chat_id"] = chat_id
# Create object with the same name and args from the dict # Create object with the same name and args from the dict
try: try:
@ -329,7 +205,7 @@ class PyroClient(Client):
def add_command( def add_command(
self, self,
command: str, command: str,
) -> None: ):
"""Add command to the bot's internal commands list """Add command to the bot's internal commands list
### Args: ### Args:
@ -348,7 +224,7 @@ class PyroClient(Client):
async def register_commands( async def register_commands(
self, command_sets: Union[List[CommandSet], None] = None self, command_sets: Union[List[CommandSet], None] = None
) -> None: ):
"""Register commands stored in bot's 'commands' attribute""" """Register commands stored in bot's 'commands' attribute"""
if command_sets is None: if command_sets is None:
@ -377,9 +253,7 @@ class PyroClient(Client):
language_code=command_set.language_code, language_code=command_set.language_code,
) )
async def remove_commands( async def remove_commands(self, command_sets: Union[List[CommandSet], None] = None):
self, command_sets: Union[List[CommandSet], None] = None
) -> None:
"""Remove commands stored in bot's 'commands' attribute""" """Remove commands stored in bot's 'commands' attribute"""
if command_sets is None: if command_sets is None:

View File

@ -1,9 +1,6 @@
from pathlib import Path from pathlib import Path
from typing import Any, Union from typing import Any, Union
from .._utils import supports_argument
from ._nested import nested_delete, nested_set
try: try:
from ujson import dumps, loads from ujson import dumps, loads
except ImportError: except ImportError:
@ -21,7 +18,6 @@ def json_read(path: Union[str, Path]) -> Any:
""" """
with open(str(path), mode="r", encoding="utf-8") as f: with open(str(path), mode="r", encoding="utf-8") as f:
data = f.read() data = f.read()
return loads(data) return loads(data)
@ -33,11 +29,32 @@ def json_write(data: Any, path: Union[str, Path]) -> None:
* path (`Union[str, Path]`): Path-like object or path as a string of a destination * path (`Union[str, Path]`): Path-like object or path as a string of a destination
""" """
with open(str(path), mode="w", encoding="utf-8") as f: with open(str(path), mode="w", encoding="utf-8") as f:
f.write( f.write(dumps(data, ensure_ascii=False, escape_forward_slashes=False, indent=4))
dumps(data, ensure_ascii=False, escape_forward_slashes=False, indent=4)
if supports_argument(dumps, "escape_forward_slashes")
else dumps(data, ensure_ascii=False, indent=4) def nested_set(target: dict, value: Any, *path: str, create_missing=True) -> dict:
) """Set the key by its path to the value
### Args:
* target (`dict`): Dictionary to perform modifications on
* value (`Any`): Any data
* *path (`str`): Path to the key of the target
* create_missing (`bool`, *optional*): Create keys on the way if they're missing. Defaults to `True`
### Returns:
* `dict`: Changed dictionary
"""
d = target
for key in path[:-1]:
if key in d:
d = d[key]
elif create_missing:
d = d.setdefault(key, {})
else:
return target
if path[-1] in d or create_missing:
d[path[-1]] = value
return target
def config_get( def config_get(
@ -73,10 +90,8 @@ def config_get(
``` ```
""" """
this_key = json_read(config_file) this_key = json_read(config_file)
for dict_key in path: for dict_key in path:
this_key = this_key[dict_key] this_key = this_key[dict_key]
return this_key[key] return this_key[key]
@ -90,37 +105,6 @@ def config_set(
* value (`Any`): Any JSON serializable data * value (`Any`): Any JSON serializable data
* *path (`str`): Path to the key of the target * *path (`str`): Path to the key of the target
* config_file (`Union[str, Path]`, *optional*): Path-like object or path as a string of a location of the config file. Defaults to `"config.json"` * config_file (`Union[str, Path]`, *optional*): Path-like object or path as a string of a location of the config file. Defaults to `"config.json"`
### Raises:
* `KeyError`: Key is not found under path provided
""" """
json_write(nested_set(json_read(config_file), value, *(*path, key)), config_file) json_write(nested_set(json_read(config_file), value, *(*path, key)), config_file)
return return
def config_delete(
key: str,
*path: str,
missing_ok: bool = False,
config_file: Union[str, Path] = "config.json",
) -> None:
"""Set config's key by its path
### Args:
* key (`str`): Key to delete
* *path (`str`): Path to the key of the target
* missing_ok (`bool`): Do not raise an exception if the key is missing. Defaults to `False`
* config_file (`Union[str, Path]`, *optional*): Path-like object or path as a string of a location of the config file. Defaults to `"config.json"`
### Raises:
* `KeyError`: Key is not found under path provided and `missing_ok` is `False`
"""
config_data = json_read(config_file)
try:
nested_delete(config_data, *(*path, key))
except KeyError as exc:
if not missing_ok:
raise exc from exc
json_write(config_data, config_file)

View File

@ -1,19 +1,19 @@
[build-system] [build-system]
requires = ["setuptools>=62.6", "wheel"] requires = ["setuptools>=62.6,<66"]
build-backend = "setuptools.build_meta" build-backend = "setuptools.build_meta"
[project] [project]
name = "libbot" name = "libbot"
dynamic = ["version", "dependencies", "optional-dependencies"] version = "1.2"
authors = [{ name = "Profitroll" }] authors = [{ name = "Profitroll" }]
description = "Universal bot library with functions needed for basic Discord/Telegram bot development." description = "Universal bot library with functions needed for basic Discord/Telegram bot development."
readme = "README.md" readme = "README.md"
requires-python = ">=3.8" requires-python = ">=3.8"
license = { text = "GPLv3" } license = { text = "GPL3" }
classifiers = [ classifiers = [
"Development Status :: 3 - Alpha", "Development Status :: 3 - Alpha",
"Intended Audience :: Developers", "Intended Audience :: Developers",
"License :: OSI Approved :: GNU General Public License v3 (GPLv3)", "License :: OSI Approved :: MIT License",
"Operating System :: OS Independent", "Operating System :: OS Independent",
"Programming Language :: Python :: 3.8", "Programming Language :: Python :: 3.8",
"Programming Language :: Python :: 3.9", "Programming Language :: Python :: 3.9",
@ -22,46 +22,33 @@ classifiers = [
"Topic :: Software Development :: Libraries :: Python Modules", "Topic :: Software Development :: Libraries :: Python Modules",
"Topic :: Utilities", "Topic :: Utilities",
] ]
dependencies = ["aiofiles~=23.1.0"]
[tool.setuptools.dynamic] [project.optional-dependencies]
version = { attr = "libbot.__version__" } pycord = ["py-cord>=2.0.0"]
dependencies = { file = "requirements/_.txt" } pyrogram = ["pyrogram>=2.0.0"]
speed = ["ujson~=5.8.0"]
[tool.setuptools.dynamic.optional-dependencies]
dev = { file = "requirements/dev.txt" }
pycord = { file = "requirements/pycord.txt" }
pyrogram = { file = "requirements/pyrogram.txt" }
speed = { file = "requirements/speed.txt" }
[project.urls] [project.urls]
Source = "https://git.end-play.xyz/profitroll/LibBotUniversal" Source = "https://git.end-play.xyz/profitroll/LibBotUniversal"
Documentation = "https://git.end-play.xyz/profitroll/LibBotUniversal/wiki" Documentation = "https://git.end-play.xyz/profitroll/LibBotUniversal/wiki"
Tracker = "https://git.end-play.xyz/profitroll/LibBotUniversal/issues" Tracker = "https://git.end-play.xyz/profitroll/LibBotUniversal/issues"
[tool.setuptools.packages.find] [tool.setuptools]
where = ["src"] packages = [
"libbot",
"libbot.sync",
"libbot.pyrogram",
"libbot.pyrogram.classes",
"libbot.i18n",
"libbot.i18n.sync",
"libbot.i18n.classes",
]
[tool.setuptools_scm]
[tool.black] [tool.black]
target-version = ['py38', 'py39', 'py310', 'py311'] target-version = ['py38', 'py38', 'py39', 'py310', 'py311']
[tool.isort] [tool.isort]
profile = "black" profile = "black"
[tool.pytest.ini_options]
minversion = "6.0"
python_files = ["test_*.py"]
pythonpath = "."
testpaths = ["tests"]
[tool.mypy]
namespace_packages = true
install_types = true
strict = true
show_error_codes = true
[tool.pylint.main]
extension-pkg-whitelist = ["ujson"]
py-version = 3.8
[tool.coverage.run]
source = ["libbot"]

View File

@ -1 +0,0 @@
aiofiles>=23.0.0

View File

@ -1,11 +0,0 @@
black==24.4.2
build==1.2.1
isort==5.13.2
mypy==1.10.1
pylint==3.2.5
pytest-asyncio==0.23.7
pytest-cov==5.0.0
pytest==8.2.2
tox==4.16.0
types-aiofiles==24.1.0.20240626
types-ujson==5.10.0.20240515

View File

@ -1,2 +0,0 @@
apscheduler~=3.10.4
py-cord~=2.6.0

View File

@ -1,2 +0,0 @@
apscheduler~=3.10.4
pyrofork~=2.3.32

View File

@ -1 +0,0 @@
ujson~=5.10.0

3
setup.cfg Normal file
View File

@ -0,0 +1,3 @@
[metadata]
description_file=README.md
license_files=LICENSE

4
setup.py Normal file
View File

@ -0,0 +1,4 @@
from setuptools import setup
if __name__ == "__main__":
setup()

View File

@ -1,6 +0,0 @@
__version__ = "3.2.3"
__license__ = "GPL3"
__author__ = "Profitroll"
from . import errors, i18n, pycord, pyrogram, sync
from .__main__ import config_delete, config_get, config_set, json_read, json_write

View File

@ -1,22 +0,0 @@
import inspect
from typing import Callable
def supports_argument(func: Callable, arg_name: str) -> bool:
"""Check whether a function has a specific argument
### Args:
* func (`Callable`): Function to be inspected
* arg_name (`str`): Argument to be checked
### Returns:
* `bool`: `True` if argument is supported and `False` if not
"""
if hasattr(func, "__code__"):
return arg_name in inspect.signature(func).parameters
elif hasattr(func, "__doc__"):
if doc := func.__doc__:
first_line = doc.splitlines()[0]
return arg_name in first_line
return False

View File

@ -1 +0,0 @@
from .config import ConfigKeyError, ConfigValueError

View File

@ -1,37 +0,0 @@
from typing import Any, List, Optional, Union
class ConfigKeyError(Exception):
"""Raised when config key is not found.
### Attributes:
* key (`Union[str, List[str]]`): Missing config key.
"""
def __init__(self, key: Union[str, List[str]]) -> None:
self.key: Union[str, List[str]] = key
super().__init__(
f"Config key {'.'.join(key) if isinstance(key, list) else key} is missing. Please set in your config file."
)
def __str__(self):
return f"Config key {'.'.join(self.key) if isinstance(self.key, list) else self.key} is missing. Please set in your config file."
class ConfigValueError(Exception):
"""Raised when config key's value is invalid.
### Attributes:
* key (`Union[str, List[str]]`): Invalid config key.
* value (`Optional[Any]`): Key's correct value.
"""
def __init__(self, key: Union[str, List[str]], value: Optional[Any] = None) -> None:
self.key: Union[str, List[str]] = key
self.value: Optional[Any] = value
super().__init__(
f"Config key {'.'.join(key) if isinstance(key, list) else key} has invalid value. {f'Must be {value}. ' if value else ''}Please set in your config file."
)
def __str__(self):
return f"Config key {'.'.join(self.key) if isinstance(self.key, list) else self.key} has invalid value. {f'Must be {self.value}. ' if self.value else ''}Please set in your config file."

View File

@ -1 +0,0 @@
from .bot import PycordBot

View File

@ -1,61 +0,0 @@
import logging
from pathlib import Path
from typing import Any, Dict, List, Union
try:
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from apscheduler.schedulers.background import BackgroundScheduler
from discord import Bot
except ImportError as exc:
raise ImportError(
"You need to install libbot[pycord] in order to use this class."
) from exc
try:
from ujson import loads
except ImportError:
from json import loads
from libbot.i18n import BotLocale
from libbot.i18n.sync import _
logger = logging.getLogger(__name__)
class PycordBot(Bot):
def __init__(
self,
config: Union[Dict[str, Any], None] = None,
config_path: Union[str, Path] = Path("config.json"),
locales_root: Union[str, Path, None] = None,
scheduler: Union[AsyncIOScheduler, BackgroundScheduler, None] = None,
*args,
**kwargs,
):
if config is None:
with open(config_path, "r", encoding="utf-8") as f:
self.config: dict = loads(f.read())
else:
self.config = config
super().__init__(
debug_guilds=(
self.config["bot"]["debug_guilds"] if self.config["debug"] else None
),
owner_ids=self.config["bot"]["owners"],
*args,
**kwargs,
)
self.bot_locale: BotLocale = BotLocale(
default_locale=self.config["locale"],
locales_root=(Path("locale") if locales_root is None else locales_root),
)
self.default_locale: str = self.bot_locale.default
self.locales: Dict[str, Any] = self.bot_locale.locales
self._ = self.bot_locale._
self.in_all_locales = self.bot_locale.in_all_locales
self.in_every_locale = self.bot_locale.in_every_locale
self.scheduler: Union[AsyncIOScheduler, BackgroundScheduler, None] = scheduler

View File

@ -1 +0,0 @@
from .__main__ import config_delete, config_get, config_set, json_read, json_write

View File

@ -1,64 +0,0 @@
from typing import Any, Dict
def nested_set(
target: dict, value: Any, *path: str, create_missing=True
) -> Dict[str, Any]:
"""Set the key by its path to the value
### Args:
* target (`dict`): Dictionary to perform modifications on
* value (`Any`): Any data
* *path (`str`): Path to the key of the target
* create_missing (`bool`, *optional*): Create keys on the way if they're missing. Defaults to `True`
### Raises:
* `KeyError`: Key is not found under path provided
### Returns:
* `Dict[str, Any]`: Changed dictionary
"""
d = target
for key in path[:-1]:
if key in d:
d = d[key]
elif create_missing:
d = d.setdefault(key, {})
else:
raise KeyError(
f"Key '{key}' is not found under path provided ({path}) and create_missing is False"
)
if path[-1] in d or create_missing:
d[path[-1]] = value
return target
def nested_delete(target: dict, *path: str) -> Dict[str, Any]:
"""Delete the key by its path
### Args:
* target (`dict`): Dictionary to perform modifications on
### Raises:
* `KeyError`: Key is not found under path provided
### Returns:
`Dict[str, Any]`: Changed dictionary
"""
d = target
for key in path[:-1]:
if key in d:
d = d[key]
else:
raise KeyError(f"Key '{key}' is not found under path provided ({path})")
if path[-1] in d:
del d[path[-1]]
else:
raise KeyError(f"Key '{path[-1]}' is not found under path provided ({path})")
return target

View File

@ -1,6 +0,0 @@
{
"locale": "en",
"bot": {
"bot_token": "sample_token"
}
}

View File

@ -1,38 +0,0 @@
from json import dumps, loads
from os import makedirs
from pathlib import Path
from uuid import uuid4
import pytest
from libbot.i18n import BotLocale
@pytest.fixture()
def location_config() -> Path:
makedirs(Path("tests/.tmp"), exist_ok=True)
filename = str(uuid4())
with open(Path("tests/config.json"), "r", encoding="utf-8") as file:
config = loads(file.read())
with open(Path(f"tests/.tmp/{filename}.json"), "w", encoding="utf-8") as file:
file.write(
dumps(
config,
ensure_ascii=False,
indent=4,
)
)
return Path(f"tests/.tmp/{filename}.json")
@pytest.fixture()
def location_locale() -> Path:
return Path("tests/data/locale/")
@pytest.fixture()
def bot_locale(location_locale: Path) -> BotLocale:
return BotLocale(locales_root=location_locale)

View File

@ -1 +0,0 @@
{}

View File

@ -1,3 +0,0 @@
{
"foo": 'bar'
}

View File

@ -1,23 +0,0 @@
{
"metadata": {
"flag": "🇬🇧",
"name": "English",
"codes": [
"en"
]
},
"bot": {
"name": "",
"about": "",
"description": ""
},
"foo": "bar",
"messages": {
"example": "okay"
},
"callbacks": {
"default": {
"nested": "sure"
}
}
}

View File

@ -1,23 +0,0 @@
{
"metadata": {
"flag": "🇺🇦",
"name": "Українська",
"codes": [
"uk"
]
},
"bot": {
"name": "",
"about": "",
"description": ""
},
"foo": "бар",
"messages": {
"example": "окей"
},
"callbacks": {
"default": {
"nested": "авжеж"
}
}
}

View File

@ -1,15 +0,0 @@
{
"foo": "bar",
"abcdefg": [
"higklmnop",
{
"lol": {
"kek": [
1.0000035,
0.2542,
1337
]
}
}
]
}

View File

@ -1,59 +0,0 @@
from pathlib import Path
from typing import Any, List, Union
import pytest
from libbot.i18n import BotLocale
@pytest.mark.parametrize(
"key, args, locale, expected",
[
("foo", [], None, "bar"),
("foo", [], "uk", "бар"),
("example", ["messages"], None, "okay"),
("example", ["messages"], "uk", "окей"),
("nested", ["callbacks", "default"], None, "sure"),
("nested", ["callbacks", "default"], "uk", "авжеж"),
],
)
def test_bot_locale_get(
key: str,
args: List[str],
locale: Union[str, None],
expected: Any,
bot_locale: BotLocale,
):
assert (
bot_locale._(key, *args, locale=locale)
if locale is not None
else bot_locale._(key, *args)
) == expected
@pytest.mark.parametrize(
"key, args, expected",
[
("foo", [], ["bar", "бар"]),
("example", ["messages"], ["okay", "окей"]),
("nested", ["callbacks", "default"], ["sure", "авжеж"]),
],
)
def test_i18n_in_all_locales(
key: str, args: List[str], expected: Any, bot_locale: BotLocale
):
assert (bot_locale.in_all_locales(key, *args)) == expected
@pytest.mark.parametrize(
"key, args, expected",
[
("foo", [], {"en": "bar", "uk": "бар"}),
("example", ["messages"], {"en": "okay", "uk": "окей"}),
("nested", ["callbacks", "default"], {"en": "sure", "uk": "авжеж"}),
],
)
def test_i18n_in_every_locale(
key: str, args: List[str], expected: Any, bot_locale: BotLocale
):
assert (bot_locale.in_every_locale(key, *args)) == expected

View File

@ -1,74 +0,0 @@
from pathlib import Path
from typing import Any, List
import pytest
from libbot import config_delete, config_get, config_set
@pytest.mark.asyncio
@pytest.mark.parametrize(
"args, expected",
[
(["locale"], "en"),
(["bot_token", "bot"], "sample_token"),
],
)
async def test_config_get_valid(args: List[str], expected: str, location_config: Path):
assert await config_get(args[0], *args[1:], config_file=location_config) == expected
@pytest.mark.asyncio
@pytest.mark.parametrize(
"args, expected",
[
(["bot_stonks", "bot"], pytest.raises(KeyError)),
],
)
async def test_config_get_invalid(
args: List[str], expected: Any, location_config: Path
):
with expected:
assert (
await config_get(args[0], *args[1:], config_file=location_config)
== expected
)
@pytest.mark.asyncio
@pytest.mark.parametrize(
"key, path, value",
[
("locale", [], "en"),
("bot_token", ["bot"], "sample_token"),
],
)
async def test_config_set(key: str, path: List[str], value: Any, location_config: Path):
await config_set(key, value, *path, config_file=location_config)
assert await config_get(key, *path, config_file=location_config) == value
@pytest.mark.asyncio
@pytest.mark.parametrize(
"key, path",
[
("bot_token", ["bot"]),
],
)
async def test_config_delete(key: str, path: List[str], location_config: Path):
await config_delete(key, *path, config_file=location_config)
assert key not in (await config_get(*path, config_file=location_config))
@pytest.mark.asyncio
@pytest.mark.parametrize(
"key, path",
[
("bot_lol", ["bot"]),
],
)
async def test_config_delete_missing(key: str, path: List[str], location_config: Path):
assert (
await config_delete(key, *path, missing_ok=True, config_file=location_config)
is None
)

View File

@ -1,66 +0,0 @@
from pathlib import Path
from typing import Any, List
import pytest
from libbot import sync
@pytest.mark.parametrize(
"args, expected",
[
(["locale"], "en"),
(["bot_token", "bot"], "sample_token"),
],
)
def test_config_get_valid(args: List[str], expected: str, location_config: Path):
assert sync.config_get(args[0], *args[1:], config_file=location_config) == expected
@pytest.mark.parametrize(
"args, expected",
[
(["bot_stonks", "bot"], pytest.raises(KeyError)),
],
)
def test_config_get_invalid(args: List[str], expected: Any, location_config: Path):
with expected:
assert (
sync.config_get(args[0], *args[1:], config_file=location_config) == expected
)
@pytest.mark.parametrize(
"key, path, value",
[
("locale", [], "en"),
("bot_token", ["bot"], "sample_token"),
],
)
def test_config_set(key: str, path: List[str], value: Any, location_config: Path):
sync.config_set(key, value, *path, config_file=location_config)
assert sync.config_get(key, *path, config_file=location_config) == value
@pytest.mark.parametrize(
"key, path",
[
("bot_token", ["bot"]),
],
)
def test_config_delete(key: str, path: List[str], location_config: Path):
sync.config_delete(key, *path, config_file=location_config)
assert key not in sync.config_get(*path, config_file=location_config)
@pytest.mark.parametrize(
"key, path",
[
("bot_lol", ["bot"]),
],
)
async def test_config_delete_missing(key: str, path: List[str], location_config: Path):
assert (
sync.config_delete(key, *path, missing_ok=True, config_file=location_config)
is None
)

View File

@ -1,66 +0,0 @@
from pathlib import Path
from typing import Any, List, Union
import pytest
from libbot import i18n
@pytest.mark.asyncio
@pytest.mark.parametrize(
"key, args, locale, expected",
[
("foo", [], None, "bar"),
("foo", [], "uk", "бар"),
("example", ["messages"], None, "okay"),
("example", ["messages"], "uk", "окей"),
("nested", ["callbacks", "default"], None, "sure"),
("nested", ["callbacks", "default"], "uk", "авжеж"),
],
)
async def test_i18n_get(
key: str,
args: List[str],
locale: Union[str, None],
expected: Any,
location_locale: Path,
):
assert (
await i18n._(key, *args, locale=locale, locales_root=location_locale)
if locale is not None
else await i18n._(key, *args, locales_root=location_locale)
) == expected
@pytest.mark.asyncio
@pytest.mark.parametrize(
"key, args, expected",
[
("foo", [], ["bar", "бар"]),
("example", ["messages"], ["okay", "окей"]),
("nested", ["callbacks", "default"], ["sure", "авжеж"]),
],
)
async def test_i18n_in_all_locales(
key: str, args: List[str], expected: Any, location_locale: Path
):
assert (
await i18n.in_all_locales(key, *args, locales_root=location_locale)
) == expected
@pytest.mark.asyncio
@pytest.mark.parametrize(
"key, args, expected",
[
("foo", [], {"en": "bar", "uk": "бар"}),
("example", ["messages"], {"en": "okay", "uk": "окей"}),
("nested", ["callbacks", "default"], {"en": "sure", "uk": "авжеж"}),
],
)
async def test_i18n_in_every_locale(
key: str, args: List[str], expected: Any, location_locale: Path
):
assert (
await i18n.in_every_locale(key, *args, locales_root=location_locale)
) == expected

View File

@ -1,59 +0,0 @@
from pathlib import Path
from typing import Any, List, Union
import pytest
from libbot.i18n import sync
@pytest.mark.parametrize(
"key, args, locale, expected",
[
("foo", [], None, "bar"),
("foo", [], "uk", "бар"),
("example", ["messages"], None, "okay"),
("example", ["messages"], "uk", "окей"),
("nested", ["callbacks", "default"], None, "sure"),
("nested", ["callbacks", "default"], "uk", "авжеж"),
],
)
def test_i18n_get(
key: str,
args: List[str],
locale: Union[str, None],
expected: Any,
location_locale: Path,
):
assert (
sync._(key, *args, locale=locale, locales_root=location_locale)
if locale is not None
else sync._(key, *args, locales_root=location_locale)
) == expected
@pytest.mark.parametrize(
"key, args, expected",
[
("foo", [], ["bar", "бар"]),
("example", ["messages"], ["okay", "окей"]),
("nested", ["callbacks", "default"], ["sure", "авжеж"]),
],
)
def test_i18n_in_all_locales(
key: str, args: List[str], expected: Any, location_locale: Path
):
assert (sync.in_all_locales(key, *args, locales_root=location_locale)) == expected
@pytest.mark.parametrize(
"key, args, expected",
[
("foo", [], {"en": "bar", "uk": "бар"}),
("example", ["messages"], {"en": "okay", "uk": "окей"}),
("nested", ["callbacks", "default"], {"en": "sure", "uk": "авжеж"}),
],
)
def test_i18n_in_every_locale(
key: str, args: List[str], expected: Any, location_locale: Path
):
assert (sync.in_every_locale(key, *args, locales_root=location_locale)) == expected

View File

@ -1,64 +0,0 @@
try:
from ujson import JSONDecodeError, dumps
except ImportError:
from json import dumps, JSONDecodeError
from pathlib import Path
from typing import Any, Union
import pytest
from libbot import json_read, json_write
@pytest.mark.asyncio
@pytest.mark.parametrize(
"path, expected",
[
(
"tests/data/test.json",
{
"foo": "bar",
"abcdefg": ["higklmnop", {"lol": {"kek": [1.0000035, 0.2542, 1337]}}],
},
),
("tests/data/empty.json", {}),
],
)
async def test_json_read_valid(path: Union[str, Path], expected: Any):
assert await json_read(path) == expected
@pytest.mark.asyncio
@pytest.mark.parametrize(
"path, expected",
[
("tests/data/invalid.json", JSONDecodeError),
("tests/data/nonexistent.json", FileNotFoundError),
],
)
async def test_json_read_invalid(path: Union[str, Path], expected: Any):
with pytest.raises(expected):
await json_read(path)
@pytest.mark.asyncio
@pytest.mark.parametrize(
"data, path",
[
(
{
"foo": "bar",
"abcdefg": ["higklmnop", {"lol": {"kek": [1.0000035, 0.2542, 1337]}}],
},
"tests/data/test.json",
),
({}, "tests/data/empty.json"),
],
)
async def test_json_write(data: Any, path: Union[str, Path]):
await json_write(data, path)
assert Path(path).is_file()
with open(path, "r", encoding="utf-8") as f:
assert f.read() == dumps(data, ensure_ascii=False, indent=4)

View File

@ -1,61 +0,0 @@
try:
from ujson import JSONDecodeError, dumps
except ImportError:
from json import dumps, JSONDecodeError
from pathlib import Path
from typing import Any, Union
import pytest
from libbot import sync
@pytest.mark.parametrize(
"path, expected",
[
(
"tests/data/test.json",
{
"foo": "bar",
"abcdefg": ["higklmnop", {"lol": {"kek": [1.0000035, 0.2542, 1337]}}],
},
),
("tests/data/empty.json", {}),
],
)
def test_json_read_valid(path: Union[str, Path], expected: Any):
assert sync.json_read(path) == expected
@pytest.mark.parametrize(
"path, expected",
[
("tests/data/invalid.json", JSONDecodeError),
("tests/data/nonexistent.json", FileNotFoundError),
],
)
def test_json_read_invalid(path: Union[str, Path], expected: Any):
with pytest.raises(expected):
assert sync.json_read(path) == expected
@pytest.mark.parametrize(
"data, path",
[
(
{
"foo": "bar",
"abcdefg": ["higklmnop", {"lol": {"kek": [1.0000035, 0.2542, 1337]}}],
},
"tests/data/test.json",
),
({}, "tests/data/empty.json"),
],
)
def test_json_write(data: Any, path: Union[str, Path]):
sync.json_write(data, path)
assert Path(path).is_file()
with open(path, "r", encoding="utf-8") as f:
assert f.read() == dumps(data, ensure_ascii=False, indent=4)

View File

@ -1,81 +0,0 @@
from typing import Any, Dict, List
import pytest
from libbot.sync._nested import nested_delete, nested_set
@pytest.mark.parametrize(
"target, value, path, create_missing, expected",
[
({"foo": "bar"}, "rab", ["foo"], True, {"foo": "rab"}),
({"foo": "bar"}, {"123": 456}, ["foo"], True, {"foo": {"123": 456}}),
(
{"foo": {"bar": {}}},
True,
["foo", "bar", "test"],
True,
{"foo": {"bar": {"test": True}}},
),
(
{"foo": {"bar": {}}},
True,
["foo", "bar", "test"],
False,
{"foo": {"bar": {}}},
),
],
)
def test_nested_set_valid(
target: Dict[str, Any],
value: Any,
path: List[str],
create_missing: bool,
expected: Any,
):
assert (nested_set(target, value, *path, create_missing=create_missing)) == expected
@pytest.mark.parametrize(
"target, value, path, create_missing, expected",
[
(
{"foo": {"bar": {}}},
True,
["foo", "bar", "test1", "test2"],
False,
KeyError,
),
],
)
def test_nested_set_invalid(
target: Dict[str, Any],
value: Any,
path: List[str],
create_missing: bool,
expected: Any,
):
with pytest.raises(expected):
assert (
nested_set(target, value, *path, create_missing=create_missing)
) == expected
@pytest.mark.parametrize(
"target, path, expected",
[
({"foo": "bar"}, ["foo"], {}),
({"foo": "bar", "bar": "foo"}, ["bar"], {"foo": "bar"}),
(
{"foo": {"bar": {}}},
["foo", "bar"],
{"foo": {}},
),
],
)
def test_nested_delete(
target: Dict[str, Any],
path: List[str],
expected: Any,
):
assert (nested_delete(target, *path)) == expected

View File

@ -1,24 +0,0 @@
from typing import Callable
import pytest
from libbot._utils import supports_argument
def func1(foo: str, bar: str):
pass
def func2(foo: str):
pass
@pytest.mark.parametrize(
"func, arg_name, result",
[
(func1, "foo", True),
(func2, "bar", False),
],
)
def test_supports_argument(func: Callable, arg_name: str, result: bool):
assert supports_argument(func, arg_name) == result

23
tox.ini
View File

@ -1,23 +0,0 @@
[tox]
minversion = 3.8.0
envlist = py38, py39, py310, py311
isolated_build = true
[gh-actions]
python =
3.8: py38
3.9: py39
3.10: py310
3.11: py311
[testenv]
setenv =
PYTHONPATH = {toxinidir}
deps =
-r{toxinidir}/requirements/_.txt
-r{toxinidir}/requirements/dev.txt
-r{toxinidir}/requirements/pycord.txt
-r{toxinidir}/requirements/pyrogram.txt
-r{toxinidir}/requirements/speed.txt
commands =
pytest --basetemp={envtmpdir} --cov=libbot