Compare commits

...

3 Commits

Author SHA1 Message Date
8c3f6f6a5d Shortened the welcome messages 2024-06-19 09:43:56 +02:00
7bed604b9d
WIP: Part 2 2024-06-02 21:51:07 +02:00
83ef964122
WIP: Part 1 2024-05-01 21:15:35 +02:00
36 changed files with 2666 additions and 979 deletions

159
.gitignore vendored
View File

@ -1,20 +1,157 @@
# ---> JupyterNotebooks
# gitignore template for Jupyter Notebooks
# website: http://jupyter.org/
# ---> Python
# Byte-compiled / optimized / DLL files
__pycache__/
*.py[cod]
*$py.class
# C extensions
*.so
# Distribution / packaging
.Python
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
wheels/
share/python-wheels/
*.egg-info/
.installed.cfg
*.egg
MANIFEST
# PyInstaller
# Usually these files are written by a python script from a template
# before PyInstaller builds the exe, so as to inject date/other infos into it.
*.manifest
*.spec
# Installer logs
pip-log.txt
pip-delete-this-directory.txt
# Unit test / coverage reports
htmlcov/
.tox/
.nox/
.coverage
.coverage.*
.cache
nosetests.xml
coverage.xml
*.cover
*.py,cover
.hypothesis/
.pytest_cache/
cover/
# Translations
*.mo
*.pot
# Django stuff:
*.log
local_settings.py
db.sqlite3
db.sqlite3-journal
# Flask stuff:
instance/
.webassets-cache
# Scrapy stuff:
.scrapy
# Sphinx documentation
docs/_build/
# PyBuilder
.pybuilder/
target/
# Jupyter Notebook
.ipynb_checkpoints
*/.ipynb_checkpoints/*
# IPython
profile_default/
ipython_config.py
# Remove previous ipynb_checkpoints
# git rm -r .ipynb_checkpoints/
# pyenv
# For a library or package, you might want to ignore these files since the code is
# intended to run in multiple environments; otherwise, check them in:
# .python-version
.vscode
config.json
data.json
# pipenv
# According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
# However, in case of collaboration, if having platform-specific dependencies or dependencies
# having no cross-platform support, pipenv may install dependencies that don't work, or not
# install all needed dependencies.
#Pipfile.lock
loop.sh
start.sh
# poetry
# Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control.
# This is especially recommended for binary packages to ensure reproducibility, and is more
# commonly ignored for libraries.
# https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control
#poetry.lock
# PEP 582; used by e.g. github.com/David-OConnor/pyflow
__pypackages__/
# Celery stuff
celerybeat-schedule
celerybeat.pid
# SageMath parsed files
*.sage.py
# Environments
.env
.venv
env/
venv/
ENV/
env.bak/
venv.bak/
# Spyder project settings
.spyderproject
.spyproject
# Rope project settings
.ropeproject
# mkdocs documentation
/site
# mypy
.mypy_cache/
.dmypy.json
dmypy.json
# Pyre type checker
.pyre/
# pytype static type analyzer
.pytype/
# Cython debug symbols
cython_debug/
# PyCharm
# JetBrains specific template is maintainted in a separate JetBrains.gitignore that can
# be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore
# and can be added to the global gitignore or merged into this file. For a more nuclear
# option (not recommended) you can uncomment the following to ignore the entire idea folder.
#.idea/
# Custom
.old/
config.json

View File

@ -2,5 +2,19 @@
"$schema": "https://docs.renovatebot.com/renovate-schema.json",
"extends": [
"config:base"
],
"baseBranches": [
"dev"
],
"packageRules": [
{
"matchUpdateTypes": [
"minor",
"patch",
"pin",
"digest"
],
"automerge": true
}
]
}

View File

@ -1,2 +1,51 @@
# GloryBot
<h1 align="center">GloryBot</h1>
<p align="center">
<a href="https://git.end-play.xyz/VA-11_Hall-A/GloryBot/src/branch/master/LICENSE"><img alt="License: GPL" src="https://img.shields.io/badge/License-GPL-blue"></a>
<a href="https://git.end-play.xyz/VA-11_Hall-A/GloryBot"><img alt="Code style: black" src="https://img.shields.io/badge/code%20style-black-000000.svg"></a>
</p>
Discord bot based on [Pycord](https://github.com/Pycord-Development/pycord) and [LibBotUniversal](https://git.end-play.xyz/profitroll/LibBotUniversal) that provides interesting captcha for your server.
## Requirements
* MongoDB
* Python 3.8+ (3.11 is recommended)
* Git (when installing from source)
## Installation
There are two kinds of installation supported. Stable one is basically a stable release that you can download and use. Upgrades will need some more complicated actions on your side. The source one provides you with easier code management, yet is meant to be used by more advanced users.
### Stable
1. Install all [requirements](#requirements) except for Git
2. Go to [Releases](/releases) and download the stable release of your choice
3. Unpack the archive somewhere you like (`/opt/GloryBot` on Linux or `C:\Program Files\GloryBot` on Windows will do)
4. Open the terminal in project's folder and create a virtual environment: `python -m venv .venv`
5. Activate the virtual environment: `.venv/bin/activate` on Linux and `.venv\Scripts\activate.bat` on Windows
6. Install all requirements: `pip install requirements.txt`
7. Create a config file by copying the `config_example.json` to `config.json`: `cp config_example.json config.json`
8. Configure the bot's `config.json` using your favorite text editor
9. Run the bot: `python main.py`
### From source
1. Install all [requirements](#requirements)
2. Go to the directory of your liking and open a terminal there: For example, `/opt` on Linux or `C:\Program Files` on Windows
3. Clone the repository: `git clone https://git.end-play.xyz/VA-11_Hall-A/GloryBot.git`
4. Change directory to the cloned project: `cd GloryBot`
5. Open the terminal in project's folder and create a virtual environment: `python -m venv .venv`
6. Activate the virtual environment: `.venv/bin/activate` on Linux and `.venv\Scripts\activate.bat` on Windows
7. Install all requirements: `pip install requirements.txt`
8. Create a config file by copying the `config_example.json` to `config.json`: `cp config_example.json config.json`
9. Configure the bot's `config.json` using your favorite text editor
10. Run the bot: `python main.py`
## Upgrading
To-Do
## Creating a service
To-Do

View File

@ -0,0 +1 @@
from .member_status import MemberStatus

View File

@ -0,0 +1,8 @@
from enum import Enum
class MemberStatus(Enum):
UNVERIFIED = 0
VERIFIED = 1
ADDITIONAL = 2
FAILED = 3

View File

@ -0,0 +1,11 @@
from .check import (
CheckAlreadyAssignedError,
CheckChallengeNotFoundError,
CheckNotFoundError,
)
from .guild import (
GuildAlreadyExistsError,
GuildChallengesEmptyError,
GuildNotFoundError,
)
from .member import MemberAlreadyExistsError, MemberNotFoundError

58
classes/errors/check.py Normal file
View File

@ -0,0 +1,58 @@
from bson import ObjectId
class CheckNotFoundError(Exception):
"""Exception raised when check does not exist in a database.
### Attributes:
* guild (`ObjectId`): Member's guild.
* member (`ObjectId`): Member ID.
"""
def __init__(self, guild: ObjectId, member: ObjectId):
self.guild: ObjectId = guild
self.member: ObjectId = member
super().__init__(
f"Could not find a check for member {str(member)} in the guild with id {str(guild)}."
)
def __str__(self):
return f"Could not find a check for member {str(self.member)} in the guild with id {str(self.guild)}."
class CheckAlreadyAssignedError(Exception):
"""Exception raised when a member already has an active check on their record.
### Attributes:
* guild (`ObjectId`): Member's guild.
* member (`ObjectId`): Member ID.
"""
def __init__(self, guild: ObjectId, member: ObjectId):
self.guild: ObjectId = guild
self.member: ObjectId = member
super().__init__(
f"Member {str(member)} in the guild with id {str(guild)} already has an active check on record."
)
def __str__(self):
return f"Member {str(self.member)} in the guild with id {str(self.guild)} already has an active check on record."
class CheckChallengeNotFoundError(Exception):
"""Exception raised when challenge of the check cannot be found in a database.
### Attributes:
* check (`ObjectId`): Check's ID.
* challenge (`ObjectId`): Challenge's ID.
"""
def __init__(self, check: ObjectId, challenge: ObjectId):
self.check: ObjectId = check
self.challenge: ObjectId = challenge
super().__init__(
f"Challenge {str(challenge)} of the check {str(check)} could not be found."
)
def __str__(self):
return f"Challenge {str(self.challenge)} of the check {str(self.check)} could not be found."

48
classes/errors/guild.py Normal file
View File

@ -0,0 +1,48 @@
from typing import Union
from bson import ObjectId
class GuildNotFoundError(Exception):
"""Exception raised when guild does not exist in a database.
### Attributes:
* id (`Union[int, ObjectId]`): Guild ID.
"""
def __init__(self, id: Union[int, ObjectId]):
self.id: Union[int, ObjectId] = id
super().__init__(f"Could not find guild entry for {str(id)}.")
def __str__(self):
return f"Could not find guild entry for {str(self.id)}."
class GuildAlreadyExistsError(Exception):
"""Exception raised when guild already exists in a database.
### Attributes:
* id (`Union[int, ObjectId]`): Guild ID.
"""
def __init__(self, id: Union[int, ObjectId]):
self.id: Union[int, ObjectId] = id
super().__init__(f"There already is a database entry for {str(id)}.")
def __str__(self):
return f"There already is a database entry for {str(self.id)}."
class GuildChallengesEmptyError(Exception):
"""Exception raised when guild has no active challenges.
### Attributes:
* id (`ObjectId`): Guild ID.
"""
def __init__(self, id: ObjectId):
self.id: ObjectId = id
super().__init__(f"Guild {str(id)} has no active challenges.")
def __str__(self):
return f"Guild {str(self.id)} has no active challenges."

41
classes/errors/member.py Normal file
View File

@ -0,0 +1,41 @@
from typing import Union
from bson import ObjectId
class MemberNotFoundError(Exception):
"""Exception raised when member does not exist in a database.
### Attributes:
* id (`Union[int, ObjectId]`): Member ID.
* guild (`ObjectId`): Member's guild.
"""
def __init__(self, id: Union[int, ObjectId], guild: ObjectId):
self.id: Union[int, ObjectId] = id
self.guild: ObjectId = guild
super().__init__(
f"Could not find member entry for {str(id)} in the guild with id {str(guild)}."
)
def __str__(self):
return f"Could not find member entry for {str(self.id)} in the guild with id {str(self.guild)}."
class MemberAlreadyExistsError(Exception):
"""Exception raised when attempted member creation when it already exists in a database.
### Attributes:
* id (`Union[int, ObjectId]`): Member ID.
* guild (`ObjectId`): Member's guild.
"""
def __init__(self, id: Union[int, ObjectId], guild: ObjectId):
self.id: Union[int, ObjectId] = id
self.guild: ObjectId = guild
super().__init__(
f"Member entry for {str(id)} in the guild with id {str(guild)} already exists."
)
def __str__(self):
return f"Member entry for {str(self.id)} in the guild with id {str(self.guild)} already exists."

84
classes/pycord_bot.py Normal file
View File

@ -0,0 +1,84 @@
from typing import Any, Union
from aiohttp import ClientSession
from bson import ObjectId
from libbot.pycord.classes import PycordBot as LibPycordBot
from classes.enums import MemberStatus
from classes.errors import GuildNotFoundError, MemberNotFoundError
from classes.pycord_guild import PycordGuild
from classes.pycord_member import PycordMember
class PycordBot(LibPycordBot):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.client_session = ClientSession()
if self.scheduler is None:
return
async def find_guild(
self,
id: Union[int, ObjectId],
create: bool = False,
) -> PycordGuild:
"""Find guild by their ID.
### Args:
* id (`Union[int, ObjectId]`): Guild's ID
* create (`bool`): Create DB entry if does not exist. Defaults to `False`.
### Raises:
* `GuildNotFoundError`: Raised when guild entry could not be found.
### Returns:
* `PycordGuild`: Guild in database representation.
"""
try:
guild = await PycordGuild.find(id)
except GuildNotFoundError as exc:
if create and isinstance(id, int):
return await PycordGuild.create(id)
else:
raise exc from exc
return guild
async def find_member(
self,
id: Union[int, ObjectId],
guild: ObjectId,
create: bool = False,
) -> PycordMember:
"""Find member by their ID and guild.
### Args:
* id (`Union[int, ObjectId]`): Member's ID
* guild (`ObjectId`): Discord guild's database ID
* create (`bool`): Create DB entry if does not exist. Defaults to `False`.
### Raises:
* `MemberNotFoundError`: Raised when member entry could not be found.
### Returns:
* `PycordMember`: Member in database representation.
"""
try:
member = await PycordMember.find(id, guild)
except MemberNotFoundError as exc:
if create and isinstance(id, int):
return await PycordMember.create(id, guild, MemberStatus.UNVERIFIED)
else:
raise exc from exc
return member
async def close(self, *args: Any, **kwargs: Any) -> None:
await self.client_session.close()
if self.scheduler is not None:
self.scheduler.shutdown()
await super().close(*args, **kwargs)

146
classes/pycord_challenge.py Normal file
View File

@ -0,0 +1,146 @@
import logging
from dataclasses import dataclass
from re import IGNORECASE, Match, compile
from typing import List, Union
from bson import ObjectId, Regex
from modules.database import col_challenges
logger = logging.getLogger(__name__)
@dataclass
class PycordChallenge:
"""Dataclass of DB entry of a captcha challenge"""
__slots__ = (
"_id",
"guild",
"enabled",
"archived",
"challenge",
"answers",
)
_id: ObjectId
guild: ObjectId
enabled: bool
archived: bool
challenge: str
answers: List[Regex]
@classmethod
async def create(
cls,
guild: ObjectId,
challenge: str,
answers: List[Regex],
enabled: bool = True,
archived: bool = False,
):
"""Create new challenge entry in a database.
### Args:
* guild (`ObjectId`): Guild's ID
* challenge (`str`): Challenge text
* answers (`List[Regex]`): List of regex patterns that are answers to the challenge
* enabled (`bool`, *optional*): Whether the challenge is enabled. Defaults to `True`.
* archived (`bool`, *optional*): Whether the challenge is archived. Defaults to `False`.
### Returns:
* `PycordChallenge`: The challenge object
"""
challenge_entry = {
"guild": guild,
"enabled": enabled,
"archived": archived,
"challenge": challenge,
"answers": answers,
}
inserted = await col_challenges.insert_one(challenge_entry)
challenge_entry["_id"] = inserted.inserted_id
return cls(**challenge_entry)
def solve(self, expression: str) -> bool:
"""Check if the expression is an answer to the challenge
### Args:
* expression (`str`): Some expression to be matched against the challenge's answers.
### Returns:
* `bool`: `True` if the answer is correct and `False` if not.
"""
return bool(self.match(expression))
def match(self, expression: str) -> Union[Match, None]:
"""Match the expression against challenge's answers
### Args:
* expression (`str`): Some expression to be matched against the challenge's answers.
### Returns:
* `Union[Match, None]`: Regex `Match` if a match is found, otherwise `None`.
"""
for answer in self.answers:
if re_match := compile(answer.pattern, IGNORECASE).match(
expression.strip()
):
return re_match
return None
async def enable(self, unarchive: bool = True) -> None:
"""Enable and unarchive the challenge
### Args:
* unarchive (`bool`, *optional*): Whether to unarchive the challenge as well. Defaults to `True`.
"""
self.enabled = True
self.archived = unarchive if self.archived else False
await col_challenges.update_one(
{"_id": self._id},
{
"$set": {
"enabled": True,
"archived": unarchive if self.archived else False,
}
},
)
async def disable(self) -> None:
"""Disable the challenge"""
self.enabled = False
await col_challenges.update_one({"_id": self._id}, {"$set": {"enabled": False}})
async def archive(self, disable: bool = True) -> None:
"""Archive and disable the challenge
### Args:
* disable (`bool`, *optional*): Whether to disable the challenge as well. Defaults to `True`.
"""
self.enabled = not disable if self.enabled else False
self.archived = True
await col_challenges.update_one(
{"_id": self._id},
{
"$set": {
"enabled": not disable if self.enabled else False,
"archived": True,
}
},
)
async def unarchive(self) -> None:
"""Unarchive the challenge"""
self.archived = False
await col_challenges.update_one(
{"_id": self._id}, {"$set": {"archived": False}}
)

125
classes/pycord_check.py Normal file
View File

@ -0,0 +1,125 @@
import logging
from dataclasses import dataclass
from datetime import UTC, datetime
from random import choice
from typing import Optional
from bson import ObjectId
from classes.errors import (
CheckAlreadyAssignedError,
CheckChallengeNotFoundError,
CheckNotFoundError,
GuildChallengesEmptyError,
)
from classes.pycord_challenge import PycordChallenge
from modules.database import col_challenges, col_checks
logger = logging.getLogger(__name__)
@dataclass
class PycordCheck:
"""Dataclass of DB entry of a security check"""
__slots__ = (
"_id",
"guild",
"active",
"thread_id",
"member",
"date_created",
"date_modified",
"challenge",
)
_id: ObjectId
guild: ObjectId
active: bool
thread_id: int
member: ObjectId
date_created: datetime
date_modified: datetime
challenge: ObjectId
@classmethod
async def find(
cls,
guild: ObjectId,
member: ObjectId,
):
db_entry = await col_checks.find_one(
{"active": True, "guild": guild, "member": member}
)
if not db_entry:
raise CheckNotFoundError(guild, member)
return cls(**db_entry)
@classmethod
async def create(
cls,
guild: ObjectId,
thread_id: int,
member: ObjectId,
challenge: Optional[ObjectId] = None,
):
# Check whether active check already exists
if await col_checks.find_one(
{"active": True, "guild": guild, "member": member}
):
raise CheckAlreadyAssignedError(guild, member)
# Get all enabled guild challenges
guild_challenges = [
challenge
async for challenge in col_challenges.find(
{"guild": guild, "enabled": True, "archived": False}
)
]
# Check whether guild has challenges at all
if not challenge and not guild_challenges:
raise GuildChallengesEmptyError(guild)
# Create a check dict
check = {
"guild": guild,
"active": True,
"thread_id": thread_id,
"member": member,
"date_created": datetime.now(UTC),
"date_modified": datetime.now(UTC),
"challenge": challenge or choice(guild_challenges)["_id"],
}
# Insert the check into the database
inserted = await col_checks.insert_one(check)
check["_id"] = inserted.inserted_id
return cls(**check)
async def deactivate(self) -> None:
"""Deactivate the check"""
await col_checks.update_one(
{"_id": self._id},
{"$set": {"active": False, "date_modified": datetime.now(UTC)}},
)
async def get_challenge(self) -> PycordChallenge:
"""Get check's bound challenge
### Raises:
* `CheckChallengeNotFoundError`: Challenge could not be found
### Returns:
* `PycordChallenge`: Captcha challenge
"""
challenge = await col_challenges.find_one({"_id": self.challenge})
if not challenge:
raise CheckChallengeNotFoundError(self._id, self.challenge)
return PycordChallenge(**challenge)

93
classes/pycord_guild.py Normal file
View File

@ -0,0 +1,93 @@
import logging
from dataclasses import dataclass
from typing import List, Union
from bson import ObjectId, Regex
from classes.errors import GuildAlreadyExistsError, GuildNotFoundError
from classes.pycord_challenge import PycordChallenge
from classes.pycord_guild_channels import PycordGuildChannels
from classes.pycord_guild_roles import PycordGuildRoles
from modules.database import col_guilds
logger = logging.getLogger(__name__)
@dataclass
class PycordGuild:
"""Dataclass of DB entry of a member"""
__slots__ = ("_id", "id", "channels", "roles")
_id: ObjectId
id: int
channels: PycordGuildChannels
roles: PycordGuildRoles
@classmethod
async def find(cls, id: Union[int, ObjectId]):
"""Find guild in the database.
### Args:
* id (`Union[int, ObjectId]`): Guild's Discord ID
### Raises:
* `GuildNotFoundError`: Raised when guild entry could not be found.
### Returns:
* `PycordGuild`: Guild with its database data.
"""
db_entry = await col_guilds.find_one(
({"id": id} if isinstance(id, int) else {"_id": id})
)
if not db_entry:
raise GuildNotFoundError(id)
db_entry["channels"] = PycordGuildChannels(**db_entry["channels"])
db_entry["roles"] = PycordGuildRoles(**db_entry["roles"])
return cls(**db_entry)
@classmethod
async def create(
cls,
id: int,
channels: PycordGuildChannels = PycordGuildChannels(),
roles: PycordGuildRoles = PycordGuildRoles(),
):
if await col_guilds.find_one({"id": id}):
raise GuildAlreadyExistsError(id)
guild = {"id": id, "channels": channels.dict(), "roles": roles.dict()}
inserted = await col_guilds.insert_one(guild)
guild["_id"] = inserted.inserted_id
return cls(**guild)
def is_valid(self) -> bool:
"""Check if all attributes are valid and return boolean of that.
### Returns:
* `bool`: `True` if all attributes are valid and `False` if not
"""
return bool(self.id and self.channels.is_valid() and self.roles.is_valid())
async def add_challenge(
self, challenge: str, answers: List[Regex], enabled: bool
) -> PycordChallenge:
"""Create new guild's challenge entry in a database.
### Args:
* challenge (`str`): Challenge text
* answers (`List[Regex]`): List of regex patterns that are answers to the challenge
* enabled (`bool`, *optional*): Whether the challenge is enabled. Defaults to `True`.
### Returns:
* `PycordChallenge`: The challenge object
"""
return await PycordChallenge.create(
self._id, challenge, answers, enabled=enabled
)

View File

@ -0,0 +1,31 @@
from dataclasses import asdict, dataclass
from typing import Optional, Union
@dataclass
class PycordGuildChannels:
__slots__ = ("captcha", "logging", "welcome")
captcha: Union[int, None]
logging: Union[int, None]
welcome: Union[int, None]
def __init__(
self,
captcha: Optional[int] = None,
logging: Optional[int] = None,
welcome: Optional[int] = None,
) -> None:
self.captcha = captcha
self.logging = logging
self.welcome = welcome
def dict(self):
return {key: value for key, value in asdict(self).items()}
def is_valid(self) -> bool:
"""Check if all attributes are not `None` and return boolean of that.
### Returns:
* `bool`: `True` if all attributes are there and `False` if not
"""
return bool(self.captcha and self.logging and self.welcome)

View File

@ -0,0 +1,42 @@
from dataclasses import asdict, dataclass
from typing import Optional, Union
@dataclass
class PycordGuildRoles:
__slots__ = (
"moderator",
"verified",
"failed",
"additional",
)
moderator: Union[int, None]
verified: Union[int, None]
failed: Union[int, None]
additional: Union[int, None]
def __init__(
self,
moderator: Optional[int] = None,
verified: Optional[int] = None,
failed: Optional[int] = None,
additional: Optional[int] = None,
) -> None:
self.moderator = moderator
self.verified = verified
self.failed = failed
self.additional = additional
def dict(self):
return {key: value for key, value in asdict(self).items()}
def is_valid(self) -> bool:
"""Check if all attributes are not `None` and return boolean of that.
### Returns:
* `bool`: `True` if all attributes are there and `False` if not
"""
return bool(
self.moderator and self.verified and self.failed and self.additional
)

113
classes/pycord_member.py Normal file
View File

@ -0,0 +1,113 @@
import logging
from dataclasses import dataclass
from typing import Literal, Union
from bson import ObjectId
from discord import Thread
from classes.enums import MemberStatus
from classes.errors import MemberAlreadyExistsError, MemberNotFoundError
from classes.pycord_check import PycordCheck
from modules.database import col_members
logger = logging.getLogger(__name__)
@dataclass
class PycordMember:
"""Dataclass of DB entry of a member"""
__slots__ = ("_id", "id", "guild", "status")
_id: ObjectId
id: int
guild: ObjectId
status: Literal[
MemberStatus.VERIFIED,
MemberStatus.UNVERIFIED,
MemberStatus.FAILED,
MemberStatus.ADDITIONAL,
]
@classmethod
async def find(cls, id: Union[int, ObjectId], guild: ObjectId):
"""Find member in the database.
### Args:
* id (`Union[int, ObjectId]`): Member's Discord ID
* guild (`ObjectId`): Discord guild's database ID
### Raises:
* `MemberNotFoundError`: Raised when member entry could not be found.
### Returns:
* `PycordMember`: Member with its database data.
"""
db_entry = await col_members.find_one(
(
{"id": id, "guild": guild}
if isinstance(id, int)
else {"_id": id, "guild": guild}
)
)
if not db_entry:
raise MemberNotFoundError(id, guild)
return cls(**db_entry)
@classmethod
async def create(
cls,
id: int,
guild: ObjectId,
status: Literal[
MemberStatus.VERIFIED,
MemberStatus.UNVERIFIED,
MemberStatus.FAILED,
MemberStatus.ADDITIONAL,
],
):
# Check whether member already exists
if await col_members.find_one({"id": True, "guild": guild}):
raise MemberAlreadyExistsError(id, guild)
# Create a member dict
check = {
"id": id,
"guild": guild,
"status": status.value,
}
# Insert the member into the database
inserted = await col_members.insert_one(check)
check["_id"] = inserted.inserted_id
return cls(**check)
async def new_check(self, thread_id: int) -> PycordCheck:
return await PycordCheck.create(self.guild, thread_id, self._id)
async def get_check(self) -> PycordCheck:
"""Get an ongoing check
### Returns:
* `PycordCheck`: An ongoing check
### Raises:
* `CheckNotFoundError`: Member does not have an ongoing check
"""
return await PycordCheck.find(self.guild, self._id)
async def set_status(
self,
status: Literal[
MemberStatus.VERIFIED,
MemberStatus.UNVERIFIED,
MemberStatus.FAILED,
MemberStatus.ADDITIONAL,
],
) -> Union[Thread, None]:
await col_members.update_one({"_id": self._id}, {"$set": {status.value}})
self.status = status

164
cogs/cog_member_join.py Normal file
View File

@ -0,0 +1,164 @@
import logging
from typing import Union
from discord import Cog, Member, TextChannel, Thread
from discord.abc import GuildChannel
from classes.errors import (
CheckAlreadyAssignedError,
CheckNotFoundError,
GuildChallengesEmptyError,
GuildNotFoundError,
)
from classes.pycord_bot import PycordBot
from classes.pycord_challenge import PycordChallenge
from classes.pycord_check import PycordCheck
from classes.pycord_guild import PycordGuild
from classes.pycord_member import PycordMember
logger = logging.getLogger(__name__)
class CogMemberJoin(Cog):
def __init__(self, client: PycordBot) -> None:
super().__init__()
self.client: PycordBot = client
@Cog.listener()
async def on_member_join(self, member: Member):
"""Process the event of new or returning member joining the guild.
This handler must take care of the new users, returning verified as well as returning banned users.
Guild must exist and be set up correctly in order to use this handler and process the input.
### Args:
* member (`Member`): Member that has joined the guild
"""
# Check whether the guild exists in the database
# We should not create it here when it does not exist. Config will be empty
try:
pycord_guild: PycordGuild = await self.client.find_guild(member.guild.id)
except GuildNotFoundError:
# Let's notify guild admins about this
logger.error(
"Guild %s is not set up properly: guild is missing from the database.",
member.guild.id,
)
return
# Get the member and create if it does not exist
pycord_member: PycordMember = await self.client.find_member(
member.id, pycord_guild._id, create=True
)
# Check whether guild is set up properly
if not pycord_guild.is_valid() or pycord_guild.channels.captcha is None:
# Let's notify guild admins about this
logger.error(
"Guild %s is not set up properly: check %s in database for details.",
member.guild.id,
pycord_guild._id,
)
return
# Check whether member already has an active check
try:
await pycord_member.get_check()
# Let's notify guild admins and user about this
logger.error(
"Member %s of guild %s could not get a check: user already has an active check.",
member.id,
member.guild.id,
)
return
except CheckNotFoundError:
logger.info(
"Member %s of guild %s does bot have a check yet.",
member.id,
member.guild.id,
)
captcha_channel: Union[GuildChannel, None] = member.guild.get_channel(
pycord_guild.channels.captcha
)
# Check whether the channel exists and is a text channel
if captcha_channel is None or not isinstance(captcha_channel, TextChannel):
logger.error(
"Captcha channel of the guild %s either does not exist or is incorrect."
)
return
# Try creating a thread
try:
captcha_thread: Thread = await captcha_channel.create_thread(
name=f"Verification - @{member.name}",
invitable=False,
reason=f"Verification - @{member.name} ({member.id})",
)
except Exception as exc:
logger.error(
"Could not create captcha thread for the channel %s in guild %s (%s) due to %s",
captcha_channel.id,
pycord_guild.id,
pycord_guild._id,
exc,
)
return
try:
check: PycordCheck = await pycord_member.new_check(captcha_thread.id)
challenge: PycordChallenge = await check.get_challenge()
except GuildChallengesEmptyError:
logger.error(
"Guild %s (%s) has no active challenges, check is aborted.",
pycord_guild.id,
pycord_guild._id,
)
await captcha_thread.delete()
return
except CheckAlreadyAssignedError:
logger.error(
"Member %s of the guild %s (%s) already has an active check, check is aborted.",
member.id,
pycord_guild.id,
pycord_guild._id,
)
await captcha_thread.delete()
return
try:
await captcha_thread.send(
f"Welcome, {member.mention}!\n\nTo keep this Discord server safe, we need to verify that you are a human.\n\nPlease, complete the sentence below:\n`{challenge.challenge} ...`\n\nUse the command `/verify` and the verification answer as its argument to pass the check."
)
except Exception as exc:
logger.error(
"Could not send the challenge to the thread %s in guild %s (%s) due to %s",
captcha_thread.id,
pycord_guild.id,
pycord_guild._id,
exc,
)
return
try:
await captcha_channel.send(
f"Welcome, {member.mention}! Please proceed to {captcha_thread.mention} in order to complete the verification.",
delete_after=180,
)
except Exception as exc:
logger.error(
"Could not send a notification about the challenge to the channel %s in guild %s (%s) due to %s",
captcha_channel.id,
pycord_guild.id,
pycord_guild._id,
exc,
)
return
def setup(client: PycordBot):
client.add_cog(CogMemberJoin(client))

52
cogs/cog_member_remove.py Normal file
View File

@ -0,0 +1,52 @@
import logging
from discord import Cog, Member
from classes.enums import MemberStatus
from classes.errors import GuildNotFoundError, MemberNotFoundError
from classes.pycord_bot import PycordBot
from classes.pycord_guild import PycordGuild
from classes.pycord_member import PycordMember
logger = logging.getLogger(__name__)
class CogMemberRemove(Cog):
def __init__(self, client: PycordBot) -> None:
super().__init__()
self.client = client
@Cog.listener()
async def on_member_remove(self, member: Member):
# Check whether the guild exists in the database
# We should not create it here when it does not exist. Config will be empty
try:
pycord_guild: PycordGuild = await self.client.find_guild(member.guild.id)
except GuildNotFoundError:
# Let's notify guild admins about this
logger.error(
"Guild %s is not set up properly: guild is missing from the database.",
member.guild.id,
)
return
# Get the member
try:
pycord_member: PycordMember = await self.client.find_member(
member.id, pycord_guild._id
)
except MemberNotFoundError:
return
# Verify whether a member has an active check
if check := await pycord_member.get_check():
# Update member's status
await pycord_member.set_status(MemberStatus.FAILED)
# Delete the thread (if possible)
if thread := member.guild.get_thread(check.thread_id):
await thread.delete()
def setup(client: PycordBot):
client.add_cog(CogMemberRemove(client))

152
cogs/cog_verify.py Normal file
View File

@ -0,0 +1,152 @@
import logging
from discord import ApplicationContext, Cog, option
from discord.ext import commands
from classes.enums import MemberStatus
from classes.errors import (
CheckChallengeNotFoundError,
CheckNotFoundError,
GuildNotFoundError,
MemberNotFoundError,
)
from classes.pycord_bot import PycordBot
logger = logging.getLogger(__name__)
class CogVerifySetup(Cog):
def __init__(self, client: PycordBot) -> None:
super().__init__()
self.client = client
@commands.slash_command(
name="verify", description="Submit verification check answer"
)
@option("answer", description="Answer to the challenge")
async def command_verify(self, ctx: ApplicationContext, answer: str):
# Verify that this command has been called in a guild
if ctx.guild is None:
await ctx.respond(
"This command can only be used in a guild.", ephemeral=True
)
return
# Verify that the guild exists and is configured
try:
pycord_guild = await self.client.find_guild(ctx.guild.id)
if not pycord_guild.is_valid():
raise GuildNotFoundError(pycord_guild.id)
except GuildNotFoundError:
await ctx.respond(
"This guild has not been set up for checks.", ephemeral=True
)
return
# Verify that the member is already known
try:
pycord_member = await self.client.find_member(ctx.user.id, pycord_guild._id)
except MemberNotFoundError:
await ctx.respond("There's no check on your record.", ephemeral=True)
return
# Verify that the member has an active check
try:
check = await pycord_member.get_check()
except CheckNotFoundError:
await ctx.respond("You do not have an active check.", ephemeral=True)
return
# Verify that the check has a valid challenge
try:
challenge = await check.get_challenge()
except CheckChallengeNotFoundError:
await ctx.respond(
"Your check does not have a valid channel.", ephemeral=True
)
return
# If the answer provided is incorrect
if not challenge.solve(answer):
logger.info(
"Member %s of guild %s has failed a check %s (challenge %s) with answer '%s'",
pycord_member.id,
pycord_guild.id,
check._id,
challenge._id,
answer,
)
# Update member's status
await pycord_member.set_status(MemberStatus.ADDITIONAL)
await ctx.respond("Your answer to the challenge is incorrect.")
# Assign the member an additional verification role
try:
member = ctx.guild.get_member(ctx.user.id)
additional_role = ctx.guild.get_role(pycord_guild.roles.additional) # type: ignore
await member.add_roles(additional_role)
except Exception as exc:
logger.error(
"Could not give role %s to %s of guild %s after passing the check due to %s",
pycord_guild.roles.verified,
pycord_member.id,
pycord_guild.id,
exc,
)
moderator_role = ctx.guild.get_role(pycord_guild.roles.moderator) # type: ignore
# Notify moderators (if the thread is available)
if thread := ctx.guild.get_thread(check.thread_id):
await thread.send(f"{moderator_role.mention} ACTION REQUIRED")
return
logger.info(
"Member %s of guild %s has passed a check %s (challenge %s) with answer '%s'",
pycord_member.id,
pycord_guild.id,
check._id,
challenge._id,
answer,
)
# Deactivate the check and update member's status
await check.deactivate()
await pycord_member.set_status(MemberStatus.VERIFIED)
await ctx.respond("You have passed the challenge!")
# Assign the member a verified role
try:
member = ctx.guild.get_member(ctx.user.id)
verified_role = ctx.guild.get_role(pycord_guild.roles.verified) # type: ignore
await member.add_roles(verified_role)
except Exception as exc:
logger.error(
"Could not give role %s to %s of guild %s after passing the check due to %s",
pycord_guild.roles.verified,
pycord_member.id,
pycord_guild.id,
exc,
)
# Delete the thread (if possible)
if thread := ctx.guild.get_thread(check.thread_id):
try:
await thread.delete()
except Exception as exc:
logger.error(
"Could not delete thread %s after the check of %s in guild %s due to %s",
thread.id,
pycord_member.id,
pycord_guild.id,
exc,
)
def setup(client: PycordBot):
client.add_cog(CogVerifySetup(client))

View File

@ -1,260 +1,38 @@
{
"token": "",
"locale": "en",
"debug": false,
"guild": 0,
"admins": [],
"channels": {
"nsfw": {
"access": 0,
"ass": 0,
"ass_2d": 0,
"tits": 0,
"tits_2d": 0
},
"verification": {
"captcha": 0,
"captcha_log": 0,
"verification": 0,
"additional_verification": 0
},
"general": {
"chat": 0,
"art": 0
"bot": {
"owners": [
0
],
"debug_guilds": [
0
],
"bot_token": "",
"status": {
"enabled": true,
"activity_type": 0,
"activity_text": "The Game Of Life"
}
},
"roles": {
"nsfw": 0,
"admin": 0,
"verified": 0,
"failed_captcha": 0,
"additional_verification": 0
"database": {
"user": null,
"password": null,
"host": "127.0.0.1",
"port": 27017,
"name": "javelina"
},
"msg": {
"denied": "Команду можуть виконувати лише адміністратори",
"denied_guild": "Команда працює лише на сервері Crue-11 Raise"
},
"auto_nsfw": [
{
"channel": 0,
"booru": "3d",
"tags": "ass",
"limit": 60
},
{
"channel": 0,
"booru": "3d",
"tags": "boobs",
"limit": 60
},
{
"channel": 0,
"booru": "2d",
"tags": "ass",
"limit": 60
},
{
"channel": 0,
"booru": "2d",
"tags": "boobs",
"limit": 60
"modules": {
"captcha": {
"challenges": [
{
"question": "Sample Question",
"answers": [
"sample answer"
]
}
],
"blacklist": []
}
],
"forbidden_replies": [
"русский военный корабль, иди нахуй"
],
"forbidden_answers": [
"славароссии",
"славаросии",
"славаросси",
"славаросії",
"славарф",
"славаросеи",
"пошёлтынахуй",
"пошёлнахуй",
"пошелнахуй",
"идинах",
"пошелтынахуй",
"славапутину",
"славасоветскомусоюзу",
"славассср",
"путинбог",
"путінбог",
"говно",
"хуйня",
"хуй",
"славапутіну"
],
"captchas": [
{
"question": "Як умру",
"answer": [
"то поховайте"
]
},
{
"question": "Хіба ревуть воли",
"answer": [
"як ясла повні"
]
},
{
"question": "Слуга",
"answer": [
"народу"
]
},
{
"question": "Олені, Олені",
"answer": [
"небриті і неголені"
]
},
{
"question": "У всякого своя доля",
"answer": [
"І свій шлях широкий"
]
},
{
"question": "Хто ти, човне? Що шукаєш?",
"answer": [
"Відки і куди пливеш",
"Звідки і куди пливеш",
"Відки й куди пливеш",
"Звідки й куди пливеш"
]
},
{
"question": "Доброго вечора",
"answer": [
"Ми з України",
"Ми з Украіни"
]
},
{
"question": "Душу й тіло ми положим",
"answer": [
"за нашу свободу"
]
},
{
"question": "Згинуть наші вороженьки",
"answer": [
"як роса на сонці"
]
},
{
"question": "Садок вишневий",
"answer": [
"коло хати"
]
},
{
"question": "Ой, хто п'є",
"answer": [
"Тому наливайте"
]
},
{
"question": "Хто не п'є",
"answer": [
"Тому не давайте"
]
},
{
"question": "Ой Житомир",
"answer": [
"це не місто не село",
"то не місто не село"
]
},
{
"question": "Файне місто",
"answer": [
"Тернопіль"
]
},
{
"question": "Хто покаже в чарці дно",
"answer": [
"Тому щастя і добро",
"Тому щастя й добро"
]
},
{
"question": "Батько наш Бандера",
"answer": [
"Україна мати",
"Україна - мати",
"Украіна мати",
"Украіна - мати",
"Україна мати",
"Украіна мати",
"Україна — мати",
"Украіна — мати"
]
},
{
"question": "Ми за Україну",
"answer": [
"підем воювати",
"будем воювати"
]
},
{
"question": "Зродились ми",
"answer": [
"великої години",
"великої годинки"
]
},
{
"question": "Ах, Бандеро",
"answer": [
"Український апостол",
"Украінський апостол",
"Тобі жилось непросто",
"Народний наш герой"
]
},
{
"question": "Той мурує",
"answer": [
"той руйнує"
]
},
{
"question": "Як умру то поховайте",
"answer": [
"мене на могилі"
]
},
{
"question": "Серед степу широкого",
"answer": [
"На Вкраїні милій",
"На Вкраіні милій"
]
},
{
"question": "Щоб лани широкополі",
"answer": [
"І Дніпро, і кручі",
"І Дніпро і кручі"
]
},
{
"question": "Ой у лузі",
"answer": [
"Червона калина"
]
},
{
"question": "Чогось наша славна Україна",
"answer": [
"зажурилася",
"зажурилась"
]
}
]
}
}

View File

@ -1,6 +0,0 @@
@client.slash_command(name="link", description="Connect to your AutoZoom")
async def nazi(ctx: discord.ApplicationContext, code: discord.Option(str, "Code you got in AutoZoom app")):
client.run(getConfig("token"))

View File

@ -1,593 +0,0 @@
import asyncio
from datetime import datetime
from random import choice
import traceback
from typing import List
import discord, ujson
import os
from modules.booruGet import booruGet
from discord import Embed, ButtonStyle, ApplicationContext, Option # type: ignore
#from discord_slash import SlashCommand, SlashContext
intents = discord.Intents().all()
client = discord.Bot(intents=intents)
# Technical ==============================================================================================================
def nowtime():
return datetime.now().strftime("%d.%m.%Y | %H:%M:%S")
def logWrite(message):
print(f"[{nowtime()}] {message}")
def jsonSave(filename, value):
with open(filename, 'w', encoding="utf-8") as f:
f.write(ujson.dumps(value, indent=4, ensure_ascii=False))
def jsonLoad(filename):
with open(filename, 'r', encoding="utf-8") as f:
value = ujson.loads(f.read())
return value
#=========================================================================================================================
# Configuration ==========================================================================================================
def configGet(key: str, *args: str):
this_dict = jsonLoad("config.json")
this_key = this_dict
for dict_key in args:
this_key = this_key[dict_key]
return this_key[key]
def configSet(key, value):
config = jsonLoad("config.json")
config[key] = value
jsonSave("config.json", config)
#=========================================================================================================================
# Discord Utils ==========================================================================================================
def getChan(channels, id):
return discord.utils.get(channels, id=id)
async def rmMsg(channel_id, message_id):
try:
channel = client.get_channel(channel_id)
message = await channel.fetch_message(message_id)
await message.delete()
except:
pass
def makeEmbed(title="", description="", footer="", image=None, color=0xffffff):
embed=Embed(title=title, description=description, color=color)
if footer is not None:
embed.set_footer(text=footer)
if image is not None:
embed.set_image(url=image)
return embed
async def is_correct_answer(answer: str, correct: List[str]) -> bool:
for entry in correct:
if (answer.lower().replace(" ", "").replace(",", "").replace(".", "").replace("", "-").replace("", "-")).startswith(entry.lower().replace(" ", "")):
return True
return False
#=========================================================================================================================
# Booru Function =========================================================================================================
async def booruAuto(channel, booru, tags, limit):
try:
output = await booruGet(booru, tags, limit=limit)
if output["kind"] == "image" and output["error"] == "":
await channel.send(
embed=makeEmbed(image=output["image"], footer=f'Теги: {output["tags"]}', color=0x2400e8),
view=discord.ui.View(discord.ui.Button(style=ButtonStyle.link, label="Джерело Фото", url=output["source"]))
)
elif output["kind"] == "video" and output["thumbnail"] != "":
await channel.send(
embed=makeEmbed(image=output["thumbnail"], footer=f'Теги: {output["tags"]}', color=0x2400e8),
view=discord.ui.View(discord.ui.Button(style=ButtonStyle.link, label="Дивитись Відео", url=output["source"]))
)
except:
# traceback.print_exc()
pass
async def booruCommand(ctx, booru, tags, limit, page, times):
for _ in range(times):
try:
output = await booruGet(booru, tags, limit=limit, page=page)
if output["kind"] == "image" and output["error"] == "":
await ctx.respond(
embed=makeEmbed(image=output["image"], footer=f'Теги: {output["tags"]}', color=0x2400e8),
view=discord.ui.View(discord.ui.Button(style=ButtonStyle.link, label="Джерело Фото", url=output["source"]))
)
elif output["kind"] == "video" and output["thumbnail"] != "":
await ctx.respond(
embed=makeEmbed(image=output["thumbnail"], footer=f'Теги: {output["tags"]}', color=0x2400e8),
view=discord.ui.View(discord.ui.Button(style=ButtonStyle.link, label="Дивитись Відео", url=output["source"]))
)
else:
traceback.print_exc()
if configGet("debug"):
await ctx.respond(embed=makeEmbed(image=output["image"], footer="Exception: "+output["error"], color=0xf74b3a))
else:
await ctx.respond(embed=makeEmbed(image=output["image"], color=0xf74b3a))
except:
# traceback.print_exc()
pass
await asyncio.sleep(.25)
#=========================================================================================================================
# Auto Nudes Sender ======================================================================================================
async def sendNudes():
await client.wait_until_ready()
guild = client.get_guild(configGet("guild"))
while True:
for entry in configGet("auto_nsfw"):
await booruAuto(getChan(guild.channels, entry["channel"]), entry["booru"], entry["tags"], entry["limit"])
await asyncio.sleep(.25)
await asyncio.sleep(60)
#=========================================================================================================================
# @client.slash_command(name="autonsfw", description="2д пікча з бордів за тегом")
# async def autonsfw(
# ctx: ApplicationContext,
# booru: Option(str, "Назва booru сервісу", choices=["realbooru", "yandere", "danbooru", "konachan", "rule34"]), # type: ignore
# tags: Option(str, "Теги (через пробіл)", required=False, default=""), # type: ignore
# limit: Option(int, "Сторінки (Кількість зображень/сторінку) (20 якщо не задано)", required=False, default=20), # type: ignore
# page: Option(int, "Номер сторінки результатів (Рандомна якщо не задано)", required=False, default=None), # type: ignore
# cooldown: Option(int, "Затримка між запитами", required=False, default=1, min_value=1, max_value=5) # type: ignore
# ):
# await booruCommand(ctx, "2d", tags, limit, page, times)
# NSFW Commands ==========================================================================================================
nsfw = client.create_group("nsfw", "Команди з відвертим вмістом")
@nsfw.command(name="2d", description="2д пікча з бордів за тегом")
async def nsfw_2d(
ctx: ApplicationContext,
tags: Option(str, "Теги (через пробіл)", required=False, default=""), # type: ignore
limit: Option(int, "Сторінки (Кількість зображень/сторінку)", required=False, default=20), # type: ignore
page: Option(int, "Номер сторінки результатів", required=False, default=None), # type: ignore
times: Option(int, "Кількість бажаних зображень", required=False, default=1, min_value=1, max_value=5) # type: ignore
):
await booruCommand(ctx, "2d", tags, limit, page, times)
@nsfw.command(name="3d", description="3д пікча з бордів за тегом")
async def nsfw_3d(
ctx: ApplicationContext,
tags: Option(str, "Теги (через пробіл)", required=False, default=""), # type: ignore
limit: Option(int, "Сторінки (Кількість зображень/сторінку)", required=False, default=20), # type: ignore
page: Option(int, "Номер сторінки результатів", required=False, default=None), # type: ignore
times: Option(int, "Кількість бажаних зображень", required=False, default=1, min_value=1, max_value=5) # type: ignore
):
await booruCommand(ctx, "3d", tags, limit, page, times)
@nsfw.command(name="yandere", description="Пікча з yande.re за тегом")
async def nsfw_yandere(
ctx: ApplicationContext,
tags: Option(str, "Теги (через пробіл)", required=False, default=""), # type: ignore
limit: Option(int, "Сторінки (Кількість зображень/сторінку)", required=False, default=20), # type: ignore
page: Option(int, "Номер сторінки результатів", required=False, default=None), # type: ignore
times: Option(int, "Кількість бажаних зображень", required=False, default=1, min_value=1, max_value=5) # type: ignore
):
await booruCommand(ctx, "yandere", tags, limit, page, times)
@nsfw.command(name="konachan", description="Пікча з konachan.com за тегом")
async def nsfw_konachan(
ctx: ApplicationContext,
tags: Option(str, "Теги (через пробіл)", required=False, default=""), # type: ignore
limit: Option(int, "Сторінки (Кількість зображень/сторінку)", required=False, default=20), # type: ignore
page: Option(int, "Номер сторінки результатів", required=False, default=None), # type: ignore
times: Option(int, "Кількість бажаних зображень", required=False, default=1, min_value=1, max_value=5) # type: ignore
):
await booruCommand(ctx, "konachan", tags, limit, page, times)
@nsfw.command(name="danbooru", description="Пікча з danbooru.donmai.us за тегом")
async def nsfw_danbooru(
ctx: ApplicationContext,
tags: Option(str, "Теги (через пробіл)", required=False, default=""), # type: ignore
limit: Option(int, "Сторінки (Кількість зображень/сторінку)", required=False, default=20), # type: ignore
page: Option(int, "Номер сторінки результатів", required=False, default=None), # type: ignore
times: Option(int, "Кількість бажаних зображень", required=False, default=1, min_value=1, max_value=5) # type: ignore
):
await booruCommand(ctx, "danbooru", tags, limit, page, times)
@nsfw.command(name="rule34", description="Пікча rule34.xxx за тегом")
async def nsfw_rule34(
ctx: ApplicationContext,
tags: Option(str, "Теги (через пробіл)", required=False, default=""), # type: ignore
limit: Option(int, "Сторінки (Кількість зображень/сторінку)", required=False, default=20), # type: ignore
page: Option(int, "Номер сторінки результатів", required=False, default=None), # type: ignore
times: Option(int, "Кількість бажаних зображень", required=False, default=1, min_value=1, max_value=5) # type: ignore
):
await booruCommand(ctx, "rule34", tags, limit, page, times)
@nsfw.command(name="realbooru", description="Пікча з realbooru.com за тегом")
async def nsfw_realbooru(
ctx: ApplicationContext,
tags: Option(str, "Теги (через пробіл)", required=False, default=""), # type: ignore
limit: Option(int, "Сторінки (Кількість зображень/сторінку)", required=False, default=20), # type: ignore
page: Option(int, "Номер сторінки результатів", required=False, default=None), # type: ignore
times: Option(int, "Кількість бажаних зображень", required=False, default=1, min_value=1, max_value=5) # type: ignore
):
await booruCommand(ctx, "realbooru", tags, limit, page, times)
#=========================================================================================================================
async def pidozra(ctx, user, reason="Не вказана"):
userdata = jsonLoad("data.json")
if str(user.id) not in userdata["dms"]:
captcha = choice(configGet("captchas"))
userdata["dms"][str(user.id)] = {}
userdata["dms"][str(user.id)]["captcha"] = captcha
else:
captcha = userdata["dms"][str(user.id)]["captcha"]
logWrite(f"User {user.name}#{user.discriminator} manually sent to verification due to '{reason}' (q: '{captcha['question']}' a: {str(captcha['answer'])})")
try:
sent_msg = await user.send(content=f"{user.mention} Вибачте, але вам потрібно пройти невеличкий тест для продовження користування сервером Crue-11 Raise\n\n**{captcha['question']} ...**\n\n```\nПродовжіть речення або надішліть правильну відповідь щоб пройти перевірку\n```")
if not isinstance(ctx, discord.Message):
await ctx.respond(content=f"Юзеру `{user.name}#{user.discriminator}` оголешно про підозру")
except Exception as exp:
if not isinstance(ctx, discord.Message):
await ctx.respond(content=f"Не вдалось оголосити `{user.name}#{user.discriminator}` про підозру:\n`{exp}`")
return
await getChan(user.guild.channels, configGet("captcha_log", "channels", "verification")).send(embed=makeEmbed(title="❓ Відправлено на перевірку", description=f"**Ім'я:** `{user.name}#{user.discriminator}`\n**Причина:** `{reason}`\n**Питання:** `{captcha['question']}`\n**Відповіді:** `{str(captcha['answer'])}`", color=0x6996e4))
await user.remove_roles(getChan(user.guild.roles, configGet("verified", "roles")))
await user.add_roles(getChan(user.guild.roles, configGet("additional_verification", "roles")))
userdata["dms"][str(user.id)]["question_id"] = sent_msg.id
jsonSave("data.json", userdata)
@client.event
async def on_ready():
print(f"Logged in as {client.user}")
await client.change_presence(activity=discord.Activity(type=discord.ActivityType.listening, name="Гімн України"))
await sendNudes()
# Guild Commands =========================================================================================================
@client.slash_command(name="verify", description="Підтвердити що юзер не московит")
async def verify(ctx: ApplicationContext, user: Option(discord.User, "Юзер сервера")): # type: ignore
if ctx.author.id in configGet("admins"):
if ctx.guild is not None:
userdata = jsonLoad("data.json")
logWrite(f"User {user.name}#{user.discriminator} verified by admin {ctx.author.name}#{ctx.author.discriminator}")
await ctx.respond(content=f"{user.mention} Ласкаво просимо!\nУ вас тепер є повний доступ до всього сервера. Приємного спілкування!", delete_after=3)
await user.add_roles(getChan(user.guild.roles, configGet("verified", "roles")))
await getChan(user.guild.channels, configGet("chat", "channels", "general")).send(content=f"У нас поповнення у вигляді {user.mention}. Познайомтесь :)")
await user.remove_roles(getChan(user.guild.roles, configGet("failed_captcha", "roles")))
await getChan(user.guild.channels, configGet("captcha_log", "channels", "verification")).send(embed=makeEmbed(title="✅ Юзера верифіковано вручну", description=f"**Ім'я:** `{user.name}#{user.discriminator}`\n**Адмін**: `{ctx.author.name}#{ctx.author.discriminator}`", color=0xffc300))
try:
del userdata[str(user.id)]
jsonSave("data.json", userdata)
except:
pass
return
else:
await ctx.respond(content=configGet("denied_guild", "msg"))
else:
await ctx.respond(content=configGet("denied", "msg"))
@client.slash_command(name="pidozra", description="Оголосити юзеру про підозру")
async def pidozra_cmd(ctx: ApplicationContext, user: Option(discord.User, "Юзер сервера"), reason: Option(str, "Причина", required=False, default="Не вказана")): # type: ignore
if ctx.author.id in configGet("admins"):
if ctx.guild is not None:
await pidozra(ctx, user, reason=reason)
else:
await ctx.respond(content=configGet("denied_guild", "msg"))
else:
await ctx.respond(content=configGet("denied", "msg"))
@client.slash_command(name="nazi", description="Денацифікувати обраного московита")
async def nazi(ctx: ApplicationContext, user: Option(discord.User, "Обраний московит")): # type: ignore
if ctx.author.id in configGet("admins"):
if ctx.guild is not None:
userdata = jsonLoad("data.json")
logWrite(f"User {user.name}#{user.discriminator} was sent to russian warship by {ctx.author.name}#{ctx.author.discriminator}")
await ctx.respond(content=f"{user.mention} {choice(configGet('forbidden_replies'))}", delete_after=4)
await asyncio.sleep(3)
await user.ban(reason="Московит йобаний", delete_message_days=3)
await getChan(user.guild.channels, configGet("captcha_log", "channels", "verification")).send(embed=makeEmbed(title="☠ Юзера послано нахуй", description=f"**Ім'я:** `{user.name}#{user.discriminator}`\n**Адмін:** `{ctx.author.name}#{ctx.author.discriminator}`", color=0xea3319))
try:
del userdata[str(user.id)]
jsonSave("data.json", userdata)
except:
pass
return
else:
await ctx.respond(content=configGet("denied_guild", "msg"))
else:
await ctx.respond(content=configGet("denied", "msg"))
#=========================================================================================================================
# Utility Commands =======================================================================================================
@client.slash_command(name="debug", description="Переключити режим відлагодження")
async def debug(ctx: ApplicationContext, value: Option(bool, "Стан режиму")): # type: ignore
if ctx.author.id in configGet("admins"):
configSet("debug", value)
logWrite(f"User {ctx.author.name}#{ctx.author.discriminator} set debug mode to {str(value)}")
await ctx.respond(content=f"Режим відлагодження змінено на `{str(value)}`")
else:
await ctx.respond(content=configGet("denied", "msg"))
@client.slash_command(name="reboot", description="Перезапустити бота")
async def reboot(ctx: ApplicationContext):
if ctx.author.id in configGet("admins"):
logWrite(f"User {ctx.author.name}#{ctx.author.discriminator} called reboot")
await ctx.respond(content=f"Вимикаюсь з номером процесу `{os.getpid()}`")
os.system(f"kill -9 {os.getpid()}")
else:
await ctx.respond(content=configGet("denied", "msg"))
#=========================================================================================================================
# Discord Events =========================================================================================================
@client.event
async def on_member_join(member):
userdata = jsonLoad("data.json")
if str(member.id) not in userdata:
captcha = choice(configGet("captchas"))
userdata[str(member.id)] = {}
userdata[str(member.id)]["captcha"] = captcha
else:
captcha = userdata[str(member.id)]["captcha"]
logWrite(f"User {member.name}#{member.discriminator} joined with (q: '{captcha['question']}' a: {str(captcha['answer'])})")
await getChan(member.guild.channels, configGet("captcha_log", "channels", "verification")).send(embed=makeEmbed(title="🆕 Юзер зайшов", description=f"**Ім'я:** `{member.name}#{member.discriminator}`\n**Питання:** `{captcha['question']}`\n**Відповіді:** `{str(captcha['answer'])}`", color=0x6996e4))
await asyncio.sleep(3)
sent_msg = await getChan(member.guild.channels, configGet("captcha", "channels", "verification")).send(content=f"{member.mention} Ласкаво просимо!\n\n**{captcha['question']} ...**\n\n```\nПродовжіть речення або надішліть правильну відповідь щоб пройти перевірку\n```", delete_after=1800)
userdata[str(member.id)]["question_id"] = sent_msg.id
jsonSave("data.json", userdata)
await asyncio.sleep(300)
userdata = jsonLoad("data.json")
if str(member.id) in userdata:
try:
logWrite(f"User {member.name}#{member.discriminator} ignored verification (q: '{userdata[str(member.id)]['captcha']['question']}' a: {str(userdata[str(member.id)]['captcha']['answer'])})")
await getChan(member.guild.channels, configGet("captcha_log", "channels", "verification")).send(embed=makeEmbed(title="🔇 Юзер проігнорував тест", description=f"**Ім'я:** `{member.name}#{member.discriminator}`\n**Питання:** `{userdata[str(member.id)]['captcha']['question']}`\n**Відповіді:** `{str(userdata[str(member.id)]['captcha']['answer'])}`", color=0xffc300))
await getChan(member.guild.channels, configGet("captcha", "channels", "verification")).send(content=f"{member.mention} Тест проігноровано, до побачення", delete_after=4)
await asyncio.sleep(3)
await member.kick(reason="Ігнорування вступного тесту")
del userdata[str(member.id)]
jsonSave("data.json", userdata)
except:
traceback.print_exc()
@client.event
async def on_member_remove(member):
logWrite(f"User {member.name}#{member.discriminator} left")
userdata = jsonLoad("data.json")
if str(member.id) in userdata:
try:
await getChan(member.guild.channels, configGet("captcha_log", "channels", "verification")).send(embed=makeEmbed(title="🚪 Юзер вийшов", description=f"**Ім'я:** `{member.name}#{member.discriminator}`\n**Питання:** `{userdata[str(member.id)]['question']}`\n**Відповіді:** `{str(userdata[str(member.id)]['answer'])}`", color=0x6996e4))
except:
await getChan(member.guild.channels, configGet("captcha_log", "channels", "verification")).send(embed=makeEmbed(title="🚪 Юзер вийшов", description=f"**Ім'я:** `{member.name}#{member.discriminator}`", color=0x6996e4))
if "question_id" in userdata[str(member.id)]:
await rmMsg(configGet("captcha", "channels", "verification"), userdata[str(member.id)]["question_id"])
del userdata[str(member.id)]["question_id"]
if userdata[str(member.id)] == {}:
del userdata[str(member.id)]
jsonSave("data.json", userdata)
@client.event
async def on_raw_reaction_add(payload):
if payload.channel_id == configGet("access", "channels", "nsfw"):
if str(payload.emoji) == "":
logWrite(f"User {payload.member.name}#{payload.member.discriminator} gets NSFW role")
await payload.member.add_roles(getChan(payload.member.guild.roles, configGet("nsfw", "roles")))
@client.event
async def on_raw_reaction_remove(payload):
user = discord.utils.get(client.get_all_members(), id=payload.user_id)
if payload.channel_id == configGet("access", "channels", "nsfw"):
if str(payload.emoji) == "":
logWrite(f"User {user.name}#{user.discriminator} lost NSFW role")
await user.remove_roles(getChan(user.guild.roles, configGet("nsfw", "roles")))
@client.event
async def on_message(message):
userdata = jsonLoad("data.json")
if message.channel.id == configGet("art", "channels", "general"):
await message.add_reaction('🔺')
await message.add_reaction('🔻')
thread = await message.create_thread(name=f"{message.author.name} ({datetime.now().strftime('%d.%m.%Y')})", auto_archive_duration=10080)
await thread.send(content="**Гілку для коментарів створено!**\nНадсилайте коментарі до цієї роботи/фотографії користуючись гілкою.")
elif message.channel.id == configGet("captcha", "channels", "verification"):
if message.author != client.user:
if await is_correct_answer(message.content, userdata[str(message.author.id)]["captcha"]["answer"]):
logWrite(f"User {message.author.name}#{message.author.discriminator} verified")
await message.reply(content=f"{message.author.mention} Ласкаво просимо!\nУ вас тепер є повний доступ до всього сервера. Приємного спілкування!", delete_after=3)
await message.delete()
await rmMsg(configGet("captcha", "channels", "verification"), userdata[str(message.author.id)]["question_id"])
await message.author.add_roles(getChan(message.guild.roles, configGet("verified", "roles")))
await getChan(message.author.guild.channels, configGet("captcha_log", "channels", "verification")).send(embed=makeEmbed(title="✅ Юзера верифіковано", description=f"**Ім'я:** `{message.author.name}#{message.author.discriminator}`\n**Питання:** `{userdata[str(message.author.id)]['captcha']['question']}`\n**Відповіді:** `{str(userdata[str(message.author.id)]['captcha']['answer'])}`", color=0xffc300))
await getChan(message.author.guild.channels, configGet("chat", "channels", "general")).send(content=f"У нас поповнення у вигляді {message.author.mention}. Познайомтесь :)")
del userdata[str(message.author.id)]["captcha"]
del userdata[str(message.author.id)]
jsonSave("data.json", userdata)
return
if message.content.lower().replace(" ", "") in configGet("forbidden_answers"):
logWrite(f"User {message.author.name}#{message.author.discriminator} was piece of shit and replied '{message.content}'")
await message.reply(content=f"{message.author.mention} {choice(configGet('forbidden_replies'))}", delete_after=4)
await message.delete()
await rmMsg(configGet("captcha", "channels", "verification"), userdata[str(message.author.id)]["question_id"])
await asyncio.sleep(3)
await message.author.ban(reason="Московит йобаний", delete_message_days=1)
await getChan(message.author.guild.channels, configGet("captcha_log", "channels", "verification")).send(embed=makeEmbed(title="☠ Юзер руснявий виблядок", description=f"**Ім'я:** `{message.author.name}#{message.author.discriminator}`\n**Питання:** `{userdata[str(message.author.id)]['captcha']['question']}`\n**Відповів:** `{message.content}`\n**Правильні відповіді:** `{str(userdata[str(message.author.id)]['captcha']['answer'])}`", color=0xea3319))
del userdata[str(message.author.id)]["captcha"]
del userdata[str(message.author.id)]
jsonSave("data.json", userdata)
return
else:
logWrite(f"User {message.author.name}#{message.author.discriminator} failed verification (q: '{userdata[str(message.author.id)]['captcha']['question']}' a: {str(userdata[str(message.author.id)]['captcha']['answer'])} replied: '{message.content}')")
await message.reply(content=f"{message.author.mention} відповідь неправильна!\nВідправляйтесь у канал {getChan(message.guild.channels, configGet('verification', 'channels', 'verification')).mention}", delete_after=3)
await getChan(message.guild.channels, configGet("verification", "channels", "verification")).send(content=f"Користувач {message.author.mention} потребує перевірки {getChan(message.guild.roles, configGet('admin', 'roles')).mention}\n\nПитання: `{userdata[str(message.author.id)]['captcha']['question']}`\nВідповідь: `{str(userdata[str(message.author.id)]['captcha']['answer'])}`\n\nВідповів: `{message.content}`", delete_after=300)
await message.delete()
await rmMsg(configGet("captcha", "channels", "verification"), userdata[str(message.author.id)]["question_id"])
await asyncio.sleep(1)
await message.author.add_roles(getChan(message.guild.roles, configGet("failed_captcha", "roles")))
await getChan(message.author.guild.channels, configGet("captcha_log", "channels", "verification")).send(embed=makeEmbed(title="☠ Юзер не пройшов", description=f"**Ім'я:** `{message.author.name}#{message.author.discriminator}`\n**Питання:** `{userdata[str(message.author.id)]['captcha']['question']}`\n**Відповів:** `{message.content}`\n**Правильні відповіді:** `{str(userdata[str(message.author.id)]['captcha']['answer'])}`", color=0xea3319))
del userdata[str(message.author.id)]["captcha"]
del userdata[str(message.author.id)]
jsonSave("data.json", userdata)
return
elif isinstance(message.channel, discord.DMChannel) and str(message.author.id) in userdata["dms"]:
guild = client.get_guild(configGet("guild"))
member = guild.get_member(message.author.id)
for answer in userdata["dms"][str(message.author.id)]["captcha"]["answer"]:
if answer.lower().replace(" ", "") in message.content.lower().replace(" ", "").replace(",", "").replace(".", ""):
logWrite(f"User {message.author.name}#{message.author.discriminator} additionally verified")
await message.reply(content=f"{message.author.mention} Ласкаво просимо!\nУ вас тепер є повний доступ до всього сервера. Приємного спілкування!")
#await rmMsg(message.channel.id, userdata["dms"][str(message.author.id)]["question_id"])
await member.add_roles(guild.get_role(configGet("verified", "roles")))
await member.remove_roles(guild.get_role(configGet("additional_verification", "roles")))
await getChan(guild.channels, configGet("captcha_log", "channels", "verification")).send(embed=makeEmbed(title="✅ Юзера додатково верифіковано", description=f"**Ім'я:** `{message.author.name}#{message.author.discriminator}`\n**Питання:** `{userdata['dms'][str(message.author.id)]['captcha']['question']}`\n**Відповіді:** `{str(userdata['dms'][str(message.author.id)]['captcha']['answer'])}`", color=0xffc300))
del userdata["dms"][str(message.author.id)]["captcha"]
del userdata["dms"][str(message.author.id)]
jsonSave("data.json", userdata)
return
if message.content.lower().replace(" ", "") in configGet("forbidden_answers"):
logWrite(f"User {message.author.name}#{message.author.discriminator} was piece of shit and replied '{message.content}'")
await message.reply(content=f"{choice(configGet('forbidden_replies'))}")
await rmMsg(message.channel.id, userdata["dms"][str(message.author.id)]["question_id"])
await member.ban(reason="Московит йобаний", delete_message_days=7)
await getChan(guild.channels, configGet("captcha_log", "channels", "verification")).send(embed=makeEmbed(title="☠ Юзер руснявий виблядок", description=f"**Ім'я:** `{message.author.name}#{message.author.discriminator}`\n**Питання:** `{userdata['dms'][str(message.author.id)]['captcha']['question']}`\n**Відповів:** `{message.content}`\n**Правильні відповіді:** `{str(userdata['dms'][str(message.author.id)]['captcha']['answer'])}`", color=0xea3319))
del userdata["dms"][str(message.author.id)]["captcha"]
del userdata["dms"][str(message.author.id)]
jsonSave("data.json", userdata)
return
else:
logWrite(f"User {message.author.name}#{message.author.discriminator} failed verification (q: '{userdata['dms'][str(message.author.id)]['captcha']['question']}' a: {str(userdata['dms'][str(message.author.id)]['captcha']['answer'])} replied: '{message.content}')")
await message.reply(content=f"{message.author.mention} відповідь неправильна!\nВідправляйтесь у канал {getChan(guild.channels, configGet('verification', 'channels', 'verification')).mention}")
await getChan(guild.channels, configGet("verification", "channels", "verification")).send(content=f"Користувач {message.author.mention} потребує перевірки {getChan(guild.roles, configGet('admin_id')).mention}", delete_after=300)
await rmMsg(message.channel.id, userdata["dms"][str(message.author.id)]["question_id"])
await asyncio.sleep(2)
await member.add_roles(guild.get_role(configGet("failed_captcha", "roles")))
await member.remove_roles(guild.get_role(configGet("additional_verification", "roles")))
await getChan(guild.channels, configGet("captcha_log", "channels", "verification")).send(embed=makeEmbed(title="☠ Юзер не пройшов", description=f"**Ім'я:** `{message.author.name}#{message.author.discriminator}`\n**Питання:** `{userdata['dms'][str(message.author.id)]['captcha']['question']}`\n**Відповів:** `{message.content}`\n**Правильні відповіді:** `{str(userdata['dms'][str(message.author.id)]['captcha']['answer'])}`", color=0xea3319))
del userdata["dms"][str(message.author.id)]["captcha"]
del userdata["dms"][str(message.author.id)]
jsonSave("data.json", userdata)
return
if "🇷🇺" in message.content:
await pidozra(message, message.author, "Прапор московитів")
await message.delete()
return
#=========================================================================================================================
client.run(configGet("token"))

21
locale/de.json Normal file
View File

@ -0,0 +1,21 @@
{
"messages": {
"welcome": {
"morning": [
"{0} Guten Morgen und willkommen! {1}"
],
"midday": [
"{0} Guten Tag und willkommen! {1}"
],
"evening": [
"{0} Guten Abend und willkommen! {1}"
],
"night": [
"{0} Gute Nacht und willkommen! {1}"
],
"unknown": [
"{0} Hallo und willkommen! {1}"
]
}
}
}

21
locale/en.json Normal file
View File

@ -0,0 +1,21 @@
{
"messages": {
"welcome": {
"morning": [
"{0} Good morning and welcome! {1}"
],
"midday": [
"{0} Good day and welcome! {1}"
],
"evening": [
"{0} Good evening and welcome! {1}"
],
"night": [
"{0} Good night and welcome! {1}"
],
"unknown": [
"{0} Hello and welcome! {1}"
]
}
}
}

35
locale/uk.json Normal file
View File

@ -0,0 +1,35 @@
{
"messages": {
"welcome": {
"morning": [
"{0} Добрий ранок та ласкаво просимо! {1}",
"{0} Доброго ранку та ласкаво просимо! {1}",
"{0} Вітаннячко! Ласкаво просимо! {1}",
"{0} Доброго ранку! Ласкаво просимо! {1}"
],
"midday": [
"{0} Добрий день! Ласкаво просимо! {1}",
"{0} Добридень! Ласкаво просимо! {1}",
"{0} День добрий! Ласкаво просимо! {1}",
"{0} Мої вітання! Ласкаво просимо! {1}",
"{0} Здоровенькі були! Ласкаво просимо! {1}",
"{0} Раді вітати вас! Ласкаво просимо! {1}",
"{0} Доброго здоров’ячка! Ласкаво просимо! {1}"
],
"evening": [
"{0} Добрий вечір! Ласкаво просимо! {1}",
"{0} Доброго вечора! Ласкаво просимо! {1}",
"{0} Добривечір! Ласкаво просимо! {1}",
"{0} Доброго вечора та ласкаво просимо! {1}",
"{0} Добрий вечір та ласкаво просимо! {1}"
],
"night": [
"{0} Доброї ночі! Ласкаво просимо! {1}",
"{0} Здоровенькі були! Ласкаво просимо! {1}"
],
"unknown": [
"{0} Вітаннячко! Ласкаво просимо! {1}"
]
}
}
}

39
main.py Normal file
View File

@ -0,0 +1,39 @@
import asyncio
import logging
from os import getpid
from discord import Intents
from libbot import sync
from classes.pycord_bot import PycordBot
from modules.scheduler import scheduler
logging.basicConfig(
level=logging.DEBUG if sync.config_get("debug") else logging.INFO,
format="%(name)s.%(funcName)s | %(levelname)s | %(message)s",
datefmt="[%X]",
)
logger = logging.getLogger(__name__)
async def main():
# Define the bot's intents
intents = Intents.default()
intents.members = True
# Create a bot connection object
bot = PycordBot(scheduler=scheduler, intents=intents)
bot.load_extension("cogs")
try:
await bot.start(sync.config_get("bot_token", "bot"))
except KeyboardInterrupt:
logger.warning("Forcefully shutting down with PID %s...", getpid())
await bot.close()
if __name__ == "__main__":
loop = asyncio.get_event_loop()
loop.run_until_complete(main())

3
migrate.py Normal file
View File

@ -0,0 +1,3 @@
from modules.migrator import migrate_database
migrate_database()

0
migrations/.gitkeep Normal file
View File

View File

@ -1,111 +0,0 @@
import json
import traceback
import requests
import webbrowser
import random
import asyncio
import xml.etree.ElementTree as ET
async def booruGet(booru, tags, page=None, limit=30):
try:
booru_list = ["realbooru", "yandere", "danbooru", "konachan", "rule34"]
real_list = ["realbooru"]
anime_list = ["yandere", "danbooru", "konachan"]
thumbnail = ""
if page is None:
page = random.randint(0, 30)
if booru == "2d":
booru = random.choice(anime_list)
elif booru == "3d":
booru = random.choice(real_list)
elif booru == "random" or booru not in booru_list:
booru = random.choice(booru_list)
if booru == "realbooru":
url = f"https://realbooru.com/index.php?page=dapi&s=post&q=index&limit={limit}&pid={page}&tags={tags}"
base_url = "https://realbooru.com/images/"
thumbnail_url = "https://realbooru.com/thumbnails/"
post = random.choice(ET.fromstring(requests.get(url).text))
file_format = post.get("file_url").split(".")[-1]
thumb_format = post.get("preview_url").split(".")[-1]
md5 = post.get("md5")
output = f"{md5[:2]}/{md5[2:4]}/{md5}.{file_format}"
output_thumb = f"{md5[:2]}/{md5[2:4]}/thumbnail_{md5}.{thumb_format}"
image = source = base_url+output
thumbnail = thumbnail_url+output_thumb
tags = post.get("tags")
elif booru == "yandere":
url = f"https://yande.re/post.json?limit={limit}&tags={tags}&page={page}"
output = random.choice(json.loads(requests.get(url).text))
image = output["sample_url"]
source = output["file_url"]
tags = output["tags"]
elif booru == "danbooru":
url = f"https://danbooru.donmai.us/posts.json?&page={page}&tags={tags}&limit={limit}"
output = random.choice(json.loads(requests.get(url).text))
image = output["large_file_url"]
source = output["file_url"]
tags = output["tag_string"]
elif booru == "konachan":
url = f"https://konachan.com/post.json?&page={page}&tags={tags}&limit={limit}"
output = random.choice(json.loads(requests.get(url).text))
image = output["sample_url"]
source = output["file_url"]
tags = output["tags"]
elif booru == "rule34":
url = f"https://api.rule34.xxx/index.php?page=dapi&s=post&q=index&limit={limit}&pid={page}&tags={tags}&json=1"
print(url)
print(requests.get(url).json)
output = random.choice(json.loads(requests.get(url).json))
image = output["sample_url"]
source = output["file_url"]
tags = output["tags"]
if (".jpg" in image) or (".png" in image) or (".webp" in image) or (".jpeg" in image) or (".gif" in image):
kind = "image"
else:
kind = "video"
return {"image": image, "thumbnail": thumbnail, "source": source, "kind": kind, "tags": tags.strip(), "error": ""}
except IndexError:
return {"image": "http://res.end-play.xyz/discord/invalid.jpg", "thumbnail": "", "source": "N/A", "kind": "image", "tags": "", "error": "Found page with such tags is empty.\nDecrease images limit or page number and try again."}
except Exception as exp:
return {"image": "http://res.end-play.xyz/discord/invalid.jpg", "thumbnail": "", "source": "N/A", "kind": "image", "tags": "", "error": str(traceback.format_exc())}
if __name__ == "__main__":
booru = input("Input booru: ")
tags = input("Input tags: ")
url = asyncio.run(booruGet(booru, tags))
print(url)
webbrowser.open(url["image"])

29
modules/database.py Normal file
View File

@ -0,0 +1,29 @@
"""Module that provides all database collections"""
from typing import Any, Mapping
from async_pymongo import AsyncClient, AsyncCollection, AsyncDatabase
from libbot.sync import config_get
db_config: Mapping[str, Any] = config_get("database")
if db_config["user"] is not None and db_config["password"] is not None:
con_string = "mongodb://{0}:{1}@{2}:{3}/{4}".format(
db_config["user"],
db_config["password"],
db_config["host"],
db_config["port"],
db_config["name"],
)
else:
con_string = "mongodb://{0}:{1}/{2}".format(
db_config["host"], db_config["port"], db_config["name"]
)
db_client = AsyncClient(con_string)
db: AsyncDatabase = db_client.get_database(name=db_config["name"])
col_guilds: AsyncCollection = db.get_collection("guilds")
col_checks: AsyncCollection = db.get_collection("checks")
col_members: AsyncCollection = db.get_collection("members")
col_challenges: AsyncCollection = db.get_collection("challenges")

22
modules/migrator.py Normal file
View File

@ -0,0 +1,22 @@
from typing import Any, Mapping
from libbot.sync import config_get
from mongodb_migrations.cli import MigrationManager
from mongodb_migrations.config import Configuration
def migrate_database() -> None:
"""Apply migrations from folder `migrations/` to the database"""
db_config: Mapping[str, Any] = config_get("database")
manager_config = Configuration(
{
"mongo_host": db_config["host"],
"mongo_port": db_config["port"],
"mongo_database": db_config["name"],
"mongo_username": db_config["user"],
"mongo_password": db_config["password"],
}
)
manager = MigrationManager(manager_config)
manager.run()

3
modules/scheduler.py Normal file
View File

@ -0,0 +1,3 @@
from apscheduler.schedulers.asyncio import AsyncIOScheduler
scheduler = AsyncIOScheduler()

View File

@ -1,3 +1,6 @@
py-cord[speed]==2.5.0
requests==2.31.0
ujson==5.9.0
apscheduler~=3.10.4
async_pymongo==0.1.5
mongodb-migrations==1.3.1
pytz~=2024.1
--extra-index-url https://git.end-play.xyz/api/packages/profitroll/pypi/simple
libbot[speed,pycord]==3.2.2

View File

@ -1 +0,0 @@
[]

1075
validation/GloryBot.dmm Normal file

File diff suppressed because it is too large Load Diff

0
validation/members.json Normal file
View File