cogs refactoring, removing useless bloat, main.py refactor
This commit is contained in:
33
.gitignore
vendored
Normal file
33
.gitignore
vendored
Normal file
@@ -0,0 +1,33 @@
|
||||
.env
|
||||
bot.log
|
||||
__pycache__/
|
||||
*.pyc
|
||||
|
||||
*.py[cod]
|
||||
*$py.class
|
||||
*.so
|
||||
.Python
|
||||
build/
|
||||
develop-eggs/
|
||||
dist/
|
||||
downloads/
|
||||
eggs/
|
||||
.eggs/
|
||||
lib/
|
||||
lib64/
|
||||
parts/
|
||||
sdist/
|
||||
var/
|
||||
wheels/
|
||||
*.egg-info/
|
||||
.installed.cfg
|
||||
*.egg
|
||||
|
||||
.vscode/
|
||||
.idea/
|
||||
*.swp
|
||||
*.swo
|
||||
*.db
|
||||
.venv
|
||||
|
||||
*.log
|
||||
55
README.md
Normal file
55
README.md
Normal file
@@ -0,0 +1,55 @@
|
||||
# muzovkantV2
|
||||
### shitty (half-)vibecoded discord bot
|
||||
## overall
|
||||
you can use this as a base for your discord bot by writing your own [cogs](/cogs/), or as is.
|
||||
|
||||
## installation
|
||||
1. clone the repo
|
||||
```bash
|
||||
git clone https://github.com/rejnronuzz/muzovkantv2.git
|
||||
cd muzovkantv2
|
||||
```
|
||||
2. install the requirements. you might need to enter a venv on some systems.
|
||||
```bash
|
||||
pip install -r requirements.txt
|
||||
```
|
||||
3. insert your discord bot token and [thecatapi](https://thecatapi.com) token into the .env file.
|
||||
*it should look something like this:*
|
||||
```
|
||||
DISCORD_TOKEN=abc123abc
|
||||
CAT_API_KEY=live_abc123abc
|
||||
```
|
||||
you can then edit the [config.py](/config.py) to your liking.
|
||||
|
||||
## usage
|
||||
the bot is running when the main.py script is running.
|
||||
|
||||
### systemd service
|
||||
you can configure a systemd service for this bot.
|
||||
[systemd_service.sh](systemd_service.sh) is ONLY for systemd systems.
|
||||
|
||||
(*tested on Ubuntu 24.04*)
|
||||
|
||||
simply do:
|
||||
```
|
||||
chmod +x systemd_service.sh
|
||||
./systemd_service.sh
|
||||
```
|
||||
the systemd service will now start and auto start on reboot.
|
||||
|
||||
### updating with systemd service
|
||||
keep in mind that after updating with the systemd service enabled, you will need to restart the service.
|
||||
so the update workflow looks something like this:
|
||||
```
|
||||
git pull origin main
|
||||
sudo systemctl restart muzovkantv2
|
||||
```
|
||||
|
||||
## contacting
|
||||
if you have any feedback, contact me on github issues or/and discord. if you want to add any functions to this bot, make a PR.
|
||||
when submitting a bug report, make sure to attach the bot.log files. they generate automatically in root directory.
|
||||
|
||||
*discord: rejnronuz*
|
||||
|
||||
|
||||

|
||||
10
bot_config.json
Normal file
10
bot_config.json
Normal file
@@ -0,0 +1,10 @@
|
||||
{
|
||||
"settings": {
|
||||
"command_prefix": "!",
|
||||
"sync_commands": true
|
||||
},
|
||||
"cogs": [
|
||||
"cogs.uptime.uptime",
|
||||
"cogs.help.help"
|
||||
]
|
||||
}
|
||||
0
cogs/__init__.py
Normal file
0
cogs/__init__.py
Normal file
15
cogs/help/config.json
Normal file
15
cogs/help/config.json
Normal file
@@ -0,0 +1,15 @@
|
||||
{
|
||||
"description": "хмм даже не знаюююююю",
|
||||
"usages": ["help"],
|
||||
"hidden": false,
|
||||
"messages": {
|
||||
"upper_desc": "команды:",
|
||||
"footer": "что то поломалось - пинг меня\nесли имеются предложения по улучшению, в лс или пинг",
|
||||
"pfx_message": "префикс:"
|
||||
},
|
||||
"settings": {
|
||||
"command_pfx": "!",
|
||||
"cogs_dir": "..",
|
||||
"column_width": 20
|
||||
}
|
||||
}
|
||||
107
cogs/help/help.py
Normal file
107
cogs/help/help.py
Normal file
@@ -0,0 +1,107 @@
|
||||
import logging
|
||||
from discord.ext import commands
|
||||
import os
|
||||
import json
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
CONFIG_FILE = 'config.json'
|
||||
REQUIRED_MESSAGES = ['upper_desc', 'footer', 'pfx_message']
|
||||
|
||||
class HelpCog(commands.Cog):
|
||||
def __init__(self, bot: commands.Bot):
|
||||
self.bot = bot
|
||||
self.help_config = self._load_and_validate_config()
|
||||
self.cogs_dir = os.path.join(
|
||||
os.path.dirname(__file__),
|
||||
self.help_config["settings"]["cogs_dir"]
|
||||
)
|
||||
|
||||
def _load_and_validate_config(self) -> dict:
|
||||
path = os.path.join(os.path.dirname(__file__), CONFIG_FILE)
|
||||
|
||||
if not os.path.exists(path):
|
||||
raise FileNotFoundError(f"Help cog config not found at {path}")
|
||||
|
||||
with open(path, 'r', encoding='utf-8') as f:
|
||||
config = json.load(f)
|
||||
|
||||
messages = config.get("messages", {})
|
||||
missing_msgs = [key for key in REQUIRED_MESSAGES if key not in messages]
|
||||
if missing_msgs:
|
||||
raise KeyError(f"Help config 'messages' is missing required keys: {missing_msgs}")
|
||||
|
||||
settings = config.get("settings", {})
|
||||
settings.setdefault("cogs_dir", "..")
|
||||
settings.setdefault("column_width", 20)
|
||||
settings.setdefault("command_pfx", "!")
|
||||
|
||||
config["messages"] = messages
|
||||
config["settings"] = settings
|
||||
|
||||
logger.info("Help cog config loaded successfully")
|
||||
return config
|
||||
|
||||
def _build_help_text(self, prefix: str) -> list[str]:
|
||||
settings = self.help_config["settings"]
|
||||
column_width = settings["column_width"]
|
||||
lines = []
|
||||
|
||||
if not os.path.exists(self.cogs_dir):
|
||||
logger.error(f"Cogs directory not found at: {self.cogs_dir}")
|
||||
return lines
|
||||
|
||||
for item in sorted(os.listdir(self.cogs_dir)):
|
||||
cog_dir = os.path.join(self.cogs_dir, item)
|
||||
if not os.path.isdir(cog_dir):
|
||||
continue
|
||||
|
||||
config_path = os.path.join(cog_dir, CONFIG_FILE)
|
||||
if not os.path.exists(config_path):
|
||||
continue
|
||||
|
||||
try:
|
||||
with open(config_path, 'r', encoding='utf-8') as f:
|
||||
cog_config = json.load(f)
|
||||
except (json.JSONDecodeError, OSError) as e:
|
||||
logger.warning(f"Failed to load config for cog '{item}': {e}")
|
||||
continue
|
||||
|
||||
description = cog_config.get('description', '')
|
||||
usages = cog_config.get('usages', [])
|
||||
hidden = cog_config.get('hidden', False)
|
||||
|
||||
if not usages or hidden:
|
||||
continue
|
||||
|
||||
for usage in usages:
|
||||
usage_text = usage if usage.startswith(prefix) else f"{prefix}{usage}"
|
||||
lines.append(f"{usage_text:<{column_width}} : {description}")
|
||||
|
||||
return lines
|
||||
|
||||
@commands.command(name="help")
|
||||
async def help(self, ctx: commands.Context):
|
||||
cfg_msgs = self.help_config["messages"]
|
||||
cfg_sets = self.help_config["settings"]
|
||||
|
||||
prefix = ctx.prefix or cfg_sets['command_pfx']
|
||||
lines = self._build_help_text(prefix)
|
||||
|
||||
if not lines:
|
||||
await ctx.send("No commands available.")
|
||||
return
|
||||
|
||||
help_text = (
|
||||
f"{cfg_msgs['upper_desc']}\n"
|
||||
f"```\n"
|
||||
f"{chr(10).join(lines)}\n"
|
||||
f"```\n"
|
||||
f"{cfg_msgs['pfx_message']} `{cfg_sets['command_pfx']}`\n"
|
||||
f"{cfg_msgs['footer']}"
|
||||
)
|
||||
|
||||
await ctx.send(help_text)
|
||||
|
||||
async def setup(bot: commands.Bot):
|
||||
await bot.add_cog(HelpCog(bot))
|
||||
23
cogs/uptime/config.json
Normal file
23
cogs/uptime/config.json
Normal file
@@ -0,0 +1,23 @@
|
||||
{
|
||||
"description": "аптайм бота",
|
||||
"usages": ["!uptime"],
|
||||
"first_seen": null,
|
||||
"total_downtime": 0.0,
|
||||
"last_start": null,
|
||||
|
||||
"labels": {
|
||||
"days": ["день", "дня", "дней"],
|
||||
"hours": ["час", "часа", "часов"],
|
||||
"minutes": ["минуту", "минуты", "минут"],
|
||||
"seconds": ["секунду", "секунды","секунд"]
|
||||
},
|
||||
|
||||
"embed": {
|
||||
"session_field": "текущая сессия",
|
||||
"availability_field":"тотал аптайм"
|
||||
},
|
||||
"settings": {
|
||||
"max_units": 4,
|
||||
"show_percent": true
|
||||
}
|
||||
}
|
||||
150
cogs/uptime/uptime.py
Normal file
150
cogs/uptime/uptime.py
Normal file
@@ -0,0 +1,150 @@
|
||||
import discord
|
||||
from discord.ext import commands
|
||||
from datetime import datetime, timezone
|
||||
import json
|
||||
import os
|
||||
|
||||
CONFIG_FILE = "config.json"
|
||||
STATE_FILE = "uptime_state.json"
|
||||
|
||||
REQUIRED_KEYS = ["labels", "embed", "settings"]
|
||||
REQUIRED_LABEL_KEYS = ["days", "hours", "minutes", "seconds"]
|
||||
REQUIRED_EMBED_KEYS = ["session_field", "availability_field"]
|
||||
|
||||
def pluralize(n: int, forms: tuple[str, str, str]) -> str:
|
||||
if 11 <= n % 100 <= 14:
|
||||
return f"{n} {forms[2]}"
|
||||
r = n % 10
|
||||
if r == 1: return f"{n} {forms[0]}"
|
||||
if 2 <= r <= 4: return f"{n} {forms[1]}"
|
||||
return f"{n} {forms[2]}"
|
||||
|
||||
def format_delta(total_seconds: int, labels: dict, max_units: int) -> str:
|
||||
minutes, secs = divmod(total_seconds, 60)
|
||||
hours, minutes = divmod(minutes, 60)
|
||||
days, hours = divmod(hours, 24)
|
||||
|
||||
candidates = [
|
||||
(days, labels["days"]),
|
||||
(hours, labels["hours"]),
|
||||
(minutes, labels["minutes"]),
|
||||
(secs, labels["seconds"]),
|
||||
]
|
||||
|
||||
parts = [pluralize(v, tuple(f)) for v, f in candidates if v][:max_units]
|
||||
return " ".join(parts) if parts else pluralize(0, tuple(labels["seconds"]))
|
||||
|
||||
def _now_ts() -> float:
|
||||
return datetime.now(timezone.utc).timestamp()
|
||||
|
||||
class UptimeSimple(commands.Cog):
|
||||
def __init__(self, bot: commands.Bot):
|
||||
self.bot = bot
|
||||
self.session_start = _now_ts()
|
||||
self.config = self._load_and_validate_config()
|
||||
self.state = self._load_state()
|
||||
|
||||
if self.state["last_start"] is not None:
|
||||
gap = self.session_start - self.state["last_start"]
|
||||
if gap > 5:
|
||||
self.state["total_downtime"] += gap
|
||||
|
||||
if self.state["first_seen"] is None:
|
||||
self.state["first_seen"] = self.session_start
|
||||
|
||||
self.state["last_start"] = self.session_start
|
||||
self._save_state()
|
||||
|
||||
def _parse_color(self, color_val) -> int:
|
||||
if isinstance(color_val, str):
|
||||
return int(color_val.lstrip("#").replace("0x", ""), 16)
|
||||
return int(color_val)
|
||||
|
||||
def _load_and_validate_config(self) -> dict:
|
||||
path = os.path.join(os.path.dirname(__file__), CONFIG_FILE)
|
||||
|
||||
if not os.path.exists(path):
|
||||
raise FileNotFoundError(f"Uptime cog config not found at {path}")
|
||||
|
||||
with open(path, "r", encoding="utf-8") as f:
|
||||
config = json.load(f)
|
||||
|
||||
missing = [k for k in REQUIRED_KEYS if k not in config]
|
||||
if missing:
|
||||
raise KeyError(f"Uptime config is missing required keys: {missing}")
|
||||
|
||||
missing_labels = [k for k in REQUIRED_LABEL_KEYS if k not in config["labels"]]
|
||||
if missing_labels:
|
||||
raise KeyError(f"Uptime config 'labels' is missing keys: {missing_labels}")
|
||||
|
||||
missing_embed = [k for k in REQUIRED_EMBED_KEYS if k not in config["embed"]]
|
||||
if missing_embed:
|
||||
raise KeyError(f"Uptime config 'embed' is missing keys: {missing_embed}")
|
||||
|
||||
settings = config.get("settings", {})
|
||||
settings.setdefault("max_units", 4)
|
||||
settings.setdefault("show_percent", True)
|
||||
|
||||
color_val = config["embed"].get("color", 0x2ecc71)
|
||||
config["embed"]["parsed_color"] = discord.Color(self._parse_color(color_val))
|
||||
|
||||
return config
|
||||
|
||||
def _state_path(self) -> str:
|
||||
return os.path.join(os.path.dirname(__file__), STATE_FILE)
|
||||
|
||||
def _load_state(self) -> dict:
|
||||
path = self._state_path()
|
||||
if os.path.exists(path):
|
||||
with open(path, "r", encoding="utf-8") as f:
|
||||
return json.load(f)
|
||||
return {
|
||||
"first_seen": None,
|
||||
"total_downtime": 0.0,
|
||||
"last_start": None,
|
||||
}
|
||||
|
||||
def _save_state(self) -> None:
|
||||
with open(self._state_path(), "w", encoding="utf-8") as f:
|
||||
json.dump(self.state, f, indent=2)
|
||||
|
||||
def availability_percent(self) -> float:
|
||||
now = _now_ts()
|
||||
total_span = now - self.state["first_seen"]
|
||||
if total_span <= 0:
|
||||
return 100.0
|
||||
return max(0.0, (total_span - self.state["total_downtime"]) / total_span * 100)
|
||||
|
||||
@staticmethod
|
||||
def _pct_bar(pct: float, width: int = 10) -> str:
|
||||
filled = round(pct / 100 * width)
|
||||
return "█" * filled + "░" * (width - filled)
|
||||
|
||||
@commands.command(name="uptime")
|
||||
async def uptime(self, ctx: commands.Context):
|
||||
session_seconds = int(_now_ts() - self.session_start)
|
||||
settings = self.config["settings"]
|
||||
|
||||
embed = discord.Embed(color=self.config["embed"]["parsed_color"])
|
||||
embed.add_field(
|
||||
name = self.config["embed"]["session_field"],
|
||||
value = format_delta(session_seconds, self.config["labels"], settings["max_units"]),
|
||||
inline = False,
|
||||
)
|
||||
|
||||
if settings["show_percent"]:
|
||||
pct = self.availability_percent()
|
||||
embed.add_field(
|
||||
name = self.config["embed"]["availability_field"],
|
||||
value = f"{self._pct_bar(pct)} {pct:.2f}%",
|
||||
inline = False,
|
||||
)
|
||||
|
||||
await ctx.send(embed=embed)
|
||||
|
||||
async def cog_unload(self) -> None:
|
||||
self.state["last_start"] = None
|
||||
self._save_state()
|
||||
|
||||
async def setup(bot: commands.Bot):
|
||||
await bot.add_cog(UptimeSimple(bot))
|
||||
5
cogs/uptime/uptime_state.json
Normal file
5
cogs/uptime/uptime_state.json
Normal file
@@ -0,0 +1,5 @@
|
||||
{
|
||||
"first_seen": 1775231576.45808,
|
||||
"total_downtime": 1324.356654882431,
|
||||
"last_start": 1775232900.814735
|
||||
}
|
||||
101
main.py
Normal file
101
main.py
Normal file
@@ -0,0 +1,101 @@
|
||||
import discord
|
||||
from discord.ext import commands
|
||||
import asyncio
|
||||
import logging
|
||||
import os
|
||||
import json
|
||||
from dotenv import load_dotenv
|
||||
|
||||
CONFIG_FILE = 'bot_config.json'
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
def setup_logging() -> None:
|
||||
logging.basicConfig(
|
||||
level=logging.INFO,
|
||||
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
|
||||
handlers=[
|
||||
logging.FileHandler('bot.log', encoding='utf-8'),
|
||||
logging.StreamHandler()
|
||||
]
|
||||
)
|
||||
|
||||
def load_config() -> dict:
|
||||
if not os.path.exists(CONFIG_FILE):
|
||||
raise FileNotFoundError(f"Main bot config not found at {CONFIG_FILE}")
|
||||
|
||||
with open(CONFIG_FILE, 'r', encoding='utf-8') as f:
|
||||
config = json.load(f)
|
||||
|
||||
# Apply fallbacks
|
||||
settings = config.get("settings", {})
|
||||
settings.setdefault("command_prefix", "!")
|
||||
settings.setdefault("sync_commands", True)
|
||||
config["settings"] = settings
|
||||
config.setdefault("cogs", [])
|
||||
|
||||
return config
|
||||
|
||||
class Bot(commands.Bot):
|
||||
def __init__(self, config: dict):
|
||||
self.config = config
|
||||
|
||||
intents = discord.Intents.default()
|
||||
intents.message_content = True
|
||||
intents.reactions = True
|
||||
intents.members = True
|
||||
intents.guilds = True
|
||||
intents.messages = True
|
||||
|
||||
super().__init__(
|
||||
command_prefix=self.config["settings"]["command_prefix"],
|
||||
intents=intents,
|
||||
help_command=None,
|
||||
)
|
||||
|
||||
async def setup_hook(self) -> None:
|
||||
for cog in self.config["cogs"]:
|
||||
try:
|
||||
await self.load_extension(cog)
|
||||
logger.info("Loaded cog: %s", cog)
|
||||
except Exception as e:
|
||||
logger.error("Failed to load cog %s: %s", cog, e, exc_info=True)
|
||||
|
||||
if self.config["settings"]["sync_commands"]:
|
||||
try:
|
||||
await self.tree.sync()
|
||||
logger.info("Application commands synced successfully")
|
||||
except Exception as e:
|
||||
logger.error("Failed to sync application commands: %s", e)
|
||||
|
||||
async def on_ready(self) -> None:
|
||||
if not hasattr(self, '_ready_fired'):
|
||||
self._ready_fired = True
|
||||
logger.info("Bot ready: %s (id: %s)", self.user, self.user.id)
|
||||
|
||||
async def main() -> None:
|
||||
setup_logging()
|
||||
load_dotenv()
|
||||
|
||||
token = os.getenv('DISCORD_TOKEN')
|
||||
if not token:
|
||||
logger.critical("DISCORD_TOKEN environment variable is missing; Shutting down")
|
||||
return
|
||||
|
||||
try:
|
||||
config = load_config()
|
||||
except Exception as e:
|
||||
logger.critical("Failed to load configuration: %s", e)
|
||||
return
|
||||
|
||||
bot = Bot(config)
|
||||
|
||||
try:
|
||||
await bot.start(token)
|
||||
except discord.LoginFailure:
|
||||
logger.critical("Improper DISCORD_TOKEN passed; Shutting down")
|
||||
|
||||
if __name__ == "__main__":
|
||||
try:
|
||||
asyncio.run(main())
|
||||
except KeyboardInterrupt:
|
||||
logger.info("Bot shutdown via keyboard interrupt")
|
||||
BIN
readme/pic.jpg
Normal file
BIN
readme/pic.jpg
Normal file
Binary file not shown.
|
After Width: | Height: | Size: 86 KiB |
4
requirements.txt
Normal file
4
requirements.txt
Normal file
@@ -0,0 +1,4 @@
|
||||
python-dotenv>=1.0.0
|
||||
discord.py>=2.3.0
|
||||
aiofiles
|
||||
aiosqlite
|
||||
34
systemd_service.sh
Normal file
34
systemd_service.sh
Normal file
@@ -0,0 +1,34 @@
|
||||
#!/bin/bash
|
||||
|
||||
BOT_DIR="$(cd "$(dirname "$0")" && pwd)"
|
||||
BOT_USER="$(whoami)"
|
||||
VENV_PYTHON="$BOT_DIR/venv/bin/python"
|
||||
SERVICE_NAME="muzovkantv2"
|
||||
|
||||
echo "creating systemd service for user '$BOT_USER' in '$BOT_DIR'..."
|
||||
|
||||
sudo tee /etc/systemd/system/$SERVICE_NAME.service > /dev/null <<EOF
|
||||
[Unit]
|
||||
Description=muzovkantv2 discord bot
|
||||
After=network.target
|
||||
|
||||
[Service]
|
||||
Type=simple
|
||||
User=$BOT_USER
|
||||
WorkingDirectory=$BOT_DIR
|
||||
ExecStart=$VENV_PYTHON $BOT_DIR/main.py
|
||||
Restart=on-failure
|
||||
RestartSec=5
|
||||
StandardOutput=journal
|
||||
StandardError=journal
|
||||
|
||||
[Install]
|
||||
WantedBy=multi-user.target
|
||||
EOF
|
||||
|
||||
sudo systemctl daemon-reload
|
||||
sudo systemctl enable $SERVICE_NAME
|
||||
sudo systemctl start $SERVICE_NAME
|
||||
|
||||
echo "done. status:"
|
||||
sudo systemctl status $SERVICE_NAME
|
||||
0
utils/__init__.py
Normal file
0
utils/__init__.py
Normal file
Reference in New Issue
Block a user