26 Commits
Author SHA1 Message Date
k.eaven 27967b8f7b Merge pull request 'Chore: Minor docstring updates for release prep' (#3) from development into main
CI/CD Pipeline / Unit Tests (Python 3.12) (push) Successful in 37s
CI/CD Pipeline / Unit Tests (Python 3.13) (push) Successful in 46s
CI/CD Pipeline / Code Quality & Linting (push) Successful in 41s
CI/CD Pipeline / Integration Tests (push) Successful in 23s
CI/CD Pipeline / Security Scanning (push) Successful in 38s
CI/CD Pipeline / Generate Test Report (push) Successful in 3s
CI/CD Pipeline / Build Docker Image (push) Successful in 33s
CI/CD Pipeline / CI/CD Pipeline Status (push) Successful in 2s
CI/CD Pipeline / Unit Tests (Python 3.11) (push) Successful in 30s
Reviewed-on: k.eaven/pterodisbot#3
2026-10-24 23:11:00 +00:00
k.eaven 2323ea5eab Chore: Minor docstring updates for release prep
CI/CD Pipeline / Unit Tests (Python 3.12) (push) Successful in 33s
CI/CD Pipeline / Unit Tests (Python 3.13) (push) Successful in 32s
CI/CD Pipeline / Code Quality & Linting (push) Successful in 26s
CI/CD Pipeline / Security Scanning (push) Successful in 24s
CI/CD Pipeline / Unit Tests (Python 3.11) (push) Successful in 30s
CI/CD Pipeline / Integration Tests (push) Successful in 18s
CI/CD Pipeline / Generate Test Report (push) Successful in 11s
CI/CD Pipeline / Build Docker Image (push) Successful in 55s
CI/CD Pipeline / CI/CD Pipeline Status (push) Successful in 2s
CI/CD Pipeline / Unit Tests (Python 3.11) (pull_request) Successful in 46s
CI/CD Pipeline / Unit Tests (Python 3.12) (pull_request) Successful in 47s
CI/CD Pipeline / Unit Tests (Python 3.13) (pull_request) Successful in 48s
CI/CD Pipeline / Code Quality & Linting (pull_request) Successful in 29s
CI/CD Pipeline / Security Scanning (pull_request) Successful in 33s
CI/CD Pipeline / Build Docker Image (pull_request) Skipped
CI/CD Pipeline / CI/CD Pipeline Status (pull_request) Successful in 2s
CI/CD Pipeline / Integration Tests (pull_request) Successful in 19s
CI/CD Pipeline / Generate Test Report (pull_request) Successful in 3s
2026-10-24 23:02:16 +00:00
k.eaven 92104fb4bf Merge pull request 'Update: Major code refactor for modular extensibility' (#2) from development into main
CI/CD Pipeline / Unit Tests (Python 3.11) (push) Successful in 53s
CI/CD Pipeline / Unit Tests (Python 3.12) (push) Successful in 35s
CI/CD Pipeline / Unit Tests (Python 3.13) (push) Successful in 53s
CI/CD Pipeline / Code Quality & Linting (push) Successful in 28s
CI/CD Pipeline / Integration Tests (push) Successful in 20s
CI/CD Pipeline / Security Scanning (push) Successful in 29s
CI/CD Pipeline / Generate Test Report (push) Successful in 3s
CI/CD Pipeline / Build Docker Image (push) Successful in 25s
CI/CD Pipeline / CI/CD Pipeline Status (push) Successful in 1s
Reviewed-on: k.eaven/pterodisbot#2
2026-10-24 15:33:41 +00:00
k.eaven 49848286d7 Redirect program entrypoint via shell script in image
CI/CD Pipeline / Unit Tests (Python 3.12) (push) Successful in 38s
CI/CD Pipeline / Unit Tests (Python 3.13) (push) Successful in 47s
CI/CD Pipeline / Code Quality & Linting (push) Successful in 35s
CI/CD Pipeline / Security Scanning (push) Successful in 32s
CI/CD Pipeline / Unit Tests (Python 3.11) (push) Successful in 23s
CI/CD Pipeline / Integration Tests (push) Successful in 27s
CI/CD Pipeline / Build Docker Image (push) Successful in 55s
CI/CD Pipeline / CI/CD Pipeline Status (push) Successful in 2s
CI/CD Pipeline / Unit Tests (Python 3.11) (pull_request) Successful in 31s
CI/CD Pipeline / Unit Tests (Python 3.13) (pull_request) Successful in 33s
CI/CD Pipeline / Unit Tests (Python 3.12) (pull_request) Successful in 33s
CI/CD Pipeline / Integration Tests (pull_request) Successful in 26s
CI/CD Pipeline / Code Quality & Linting (pull_request) Successful in 33s
CI/CD Pipeline / Security Scanning (pull_request) Successful in 33s
CI/CD Pipeline / Build Docker Image (pull_request) Skipped
CI/CD Pipeline / Generate Test Report (pull_request) Successful in 3s
CI/CD Pipeline / CI/CD Pipeline Status (pull_request) Successful in 2s
CI/CD Pipeline / Generate Test Report (push) Successful in 4s
2026-10-24 14:52:21 +00:00
k.eaven 434eb7eea6 Fix: Update Dockerfile ENTRYPOINT to use entrypoint.sh for config generation 2026-10-24 14:34:42 +00:00
k.eaven 8105f11c32 Fix: Update entrypoint.sh to use main.py instead of pterodisbot.py
CI/CD Pipeline / Unit Tests (Python 3.12) (push) Successful in 39s
CI/CD Pipeline / Unit Tests (Python 3.11) (push) Successful in 39s
CI/CD Pipeline / Unit Tests (Python 3.13) (push) Successful in 47s
CI/CD Pipeline / Code Quality & Linting (push) Successful in 22s
CI/CD Pipeline / Security Scanning (push) Successful in 28s
CI/CD Pipeline / Integration Tests (push) Successful in 32s
CI/CD Pipeline / Generate Test Report (push) Successful in 3s
CI/CD Pipeline / Build Docker Image (push) Successful in 51s
CI/CD Pipeline / CI/CD Pipeline Status (push) Successful in 1s
2026-10-24 14:28:35 +00:00
k.eaven 451e202832 Fix: Auto-generate config.ini from environment variables with priority logic
CI/CD Pipeline / Unit Tests (Python 3.11) (push) Successful in 26s
CI/CD Pipeline / Unit Tests (Python 3.12) (push) Successful in 33s
CI/CD Pipeline / Unit Tests (Python 3.13) (push) Successful in 33s
CI/CD Pipeline / Code Quality & Linting (push) Successful in 33s
CI/CD Pipeline / Security Scanning (push) Successful in 30s
CI/CD Pipeline / Integration Tests (push) Successful in 31s
CI/CD Pipeline / Generate Test Report (push) Successful in 4s
CI/CD Pipeline / Build Docker Image (push) Successful in 44s
CI/CD Pipeline / CI/CD Pipeline Status (push) Successful in 1s
2026-10-24 14:18:56 +00:00
k.eaven 316416bcbe Add pydantic to requirements.txt for Docker container
CI/CD Pipeline / Unit Tests (Python 3.11) (push) Successful in 31s
CI/CD Pipeline / Unit Tests (Python 3.12) (push) Successful in 34s
CI/CD Pipeline / Unit Tests (Python 3.13) (push) Successful in 34s
CI/CD Pipeline / Code Quality & Linting (push) Successful in 30s
CI/CD Pipeline / Security Scanning (push) Successful in 31s
CI/CD Pipeline / Integration Tests (push) Successful in 30s
CI/CD Pipeline / Generate Test Report (push) Successful in 3s
CI/CD Pipeline / Build Docker Image (push) Successful in 7m24s
CI/CD Pipeline / CI/CD Pipeline Status (push) Successful in 2s
2026-10-24 13:51:23 +00:00
k.eaven 0adc7f44e8 Update CI/CD workflow for modular code structure: update test targets, linting targets, coverage modules, and add Pelican Panel test keys
CI/CD Pipeline / Unit Tests (Python 3.11) (push) Successful in 40s
CI/CD Pipeline / Unit Tests (Python 3.12) (push) Successful in 40s
CI/CD Pipeline / Unit Tests (Python 3.13) (push) Successful in 43s
CI/CD Pipeline / Code Quality & Linting (push) Successful in 25s
CI/CD Pipeline / Security Scanning (push) Successful in 38s
CI/CD Pipeline / Integration Tests (push) Successful in 32s
CI/CD Pipeline / Generate Test Report (push) Successful in 4s
CI/CD Pipeline / Build Docker Image (push) Successful in 27s
CI/CD Pipeline / CI/CD Pipeline Status (push) Successful in 1s
2026-10-24 13:08:52 +00:00
k.eaven 346e6d9604 Update: Major code refactor for modular extensibility
CI/CD Pipeline / Unit Tests (Python 3.11) (push) Successful in 29s
CI/CD Pipeline / Unit Tests (Python 3.12) (push) Successful in 30s
CI/CD Pipeline / Code Quality & Linting (push) Successful in 23s
CI/CD Pipeline / Unit Tests (Python 3.13) (push) Successful in 27s
CI/CD Pipeline / Integration Tests (push) Successful in 21s
CI/CD Pipeline / Security Scanning (push) Successful in 2m16s
CI/CD Pipeline / Generate Test Report (push) Successful in 4s
CI/CD Pipeline / Build Docker Image (push) Successful in 44s
CI/CD Pipeline / CI/CD Pipeline Status (push) Successful in 2s
2026-10-24 10:58:01 +00:00
k.eaven ac59b97500 Merge pull request 'Add Pelican Panel API key support' (#1) from development into main
CI/CD Pipeline / Unit Tests (Python 3.11) (push) Successful in 53s
CI/CD Pipeline / Unit Tests (Python 3.12) (push) Successful in 55s
CI/CD Pipeline / Unit Tests (Python 3.13) (push) Successful in 56s
CI/CD Pipeline / Code Quality & Linting (push) Successful in 32s
CI/CD Pipeline / Integration Tests (push) Successful in 31s
CI/CD Pipeline / Security Scanning (push) Successful in 2m17s
CI/CD Pipeline / Generate Test Report (push) Successful in 3s
CI/CD Pipeline / Build Docker Image (push) Successful in 1m2s
CI/CD Pipeline / CI/CD Pipeline Status (push) Successful in 2s
Reviewed-on: k.eaven/pterodisbot#1
2026-10-24 08:01:48 +00:00
k.eaven f1f39f35ca Merge branch 'main' into development
CI/CD Pipeline / Unit Tests (Python 3.11) (push) Successful in 31s
CI/CD Pipeline / Unit Tests (Python 3.12) (push) Successful in 33s
CI/CD Pipeline / Unit Tests (Python 3.13) (push) Successful in 27s
CI/CD Pipeline / Code Quality & Linting (push) Successful in 40s
CI/CD Pipeline / Unit Tests (Python 3.11) (pull_request) Successful in 26s
CI/CD Pipeline / Unit Tests (Python 3.12) (pull_request) Successful in 26s
CI/CD Pipeline / Unit Tests (Python 3.13) (pull_request) Successful in 43s
CI/CD Pipeline / Security Scanning (push) Successful in 2m23s
CI/CD Pipeline / Integration Tests (pull_request) Successful in 25s
CI/CD Pipeline / Security Scanning (pull_request) Successful in 2m34s
CI/CD Pipeline / Build Docker Image (pull_request) Skipped
CI/CD Pipeline / Generate Test Report (push) Successful in 9s
CI/CD Pipeline / Build Docker Image (push) Successful in 1m11s
CI/CD Pipeline / Generate Test Report (pull_request) Successful in 4s
CI/CD Pipeline / CI/CD Pipeline Status (push) Successful in 9s
CI/CD Pipeline / CI/CD Pipeline Status (pull_request) Successful in 9s
CI/CD Pipeline / Code Quality & Linting (pull_request) Successful in 25s
CI/CD Pipeline / Integration Tests (push) Successful in 24s
2026-10-24 07:46:55 +00:00
k.eaven d891b21ffd Resolve CI/CD branch conflicts
CI/CD Pipeline / Unit Tests (Python 3.11) (push) Successful in 42s
CI/CD Pipeline / Unit Tests (Python 3.13) (push) Successful in 46s
CI/CD Pipeline / Unit Tests (Python 3.12) (push) Successful in 46s
CI/CD Pipeline / Code Quality & Linting (push) Successful in 30s
CI/CD Pipeline / Unit Tests (Python 3.11) (pull_request) Successful in 45s
CI/CD Pipeline / Unit Tests (Python 3.12) (pull_request) Successful in 33s
CI/CD Pipeline / Unit Tests (Python 3.13) (pull_request) Successful in 36s
CI/CD Pipeline / Code Quality & Linting (pull_request) Successful in 33s
CI/CD Pipeline / Security Scanning (push) Successful in 2m28s
CI/CD Pipeline / Security Scanning (pull_request) Successful in 2m24s
CI/CD Pipeline / Build Docker Image (pull_request) Skipped
CI/CD Pipeline / Integration Tests (push) Successful in 25s
CI/CD Pipeline / Integration Tests (pull_request) Successful in 21s
CI/CD Pipeline / Build Docker Image (push) Successful in 39s
CI/CD Pipeline / Generate Test Report (push) Successful in 9s
CI/CD Pipeline / Generate Test Report (pull_request) Successful in 12s
CI/CD Pipeline / CI/CD Pipeline Status (pull_request) Successful in 14s
CI/CD Pipeline / CI/CD Pipeline Status (push) Successful in 6s
2026-10-24 07:46:22 +00:00
k.eaven d830033b8f Update README
CI/CD Pipeline / Unit Tests (Python 3.12) (push) Successful in 27s
CI/CD Pipeline / Integration Tests (push) Successful in 17s
CI/CD Pipeline / Security Scanning (push) Successful in 20s
CI/CD Pipeline / Code Quality & Linting (push) Successful in 47s
CI/CD Pipeline / Generate Test Report (push) Successful in 3s
CI/CD Pipeline / Build Docker Image (push) Successful in 25s
CI/CD Pipeline / CI/CD Pipeline Status (push) Successful in 1s
CI/CD Pipeline / Unit Tests (Python 3.11) (pull_request) Successful in 31s
CI/CD Pipeline / Security Scanning (pull_request) Successful in 20s
CI/CD Pipeline / Unit Tests (Python 3.13) (pull_request) Successful in 3m30s
CI/CD Pipeline / Unit Tests (Python 3.12) (pull_request) Successful in 3m31s
CI/CD Pipeline / Integration Tests (pull_request) Successful in 21s
CI/CD Pipeline / Generate Test Report (pull_request) Successful in 3s
CI/CD Pipeline / CI/CD Pipeline Status (pull_request) Successful in 2s
CI/CD Pipeline / Unit Tests (Python 3.11) (push) Successful in 25s
CI/CD Pipeline / Unit Tests (Python 3.13) (push) Successful in 28s
CI/CD Pipeline / Code Quality & Linting (pull_request) Successful in 44s
CI/CD Pipeline / Build Docker Image (pull_request) Skipped
2026-10-24 07:29:19 +00:00
k.eaven ec73053ac9 Update CI/CD to run a matrix of unit tests for compatible Python versions
CI/CD Pipeline / Unit Tests (Python 3.12) (push) Successful in 21s
CI/CD Pipeline / Code Quality & Linting (push) Successful in 46s
CI/CD Pipeline / Unit Tests (Python 3.11) (push) Successful in 3m40s
CI/CD Pipeline / Security Scanning (push) Successful in 19s
CI/CD Pipeline / Unit Tests (Python 3.13) (push) Successful in 4m17s
CI/CD Pipeline / Integration Tests (push) Successful in 18s
CI/CD Pipeline / Generate Test Report (push) Successful in 3s
CI/CD Pipeline / Build Docker Image (push) Successful in 34s
CI/CD Pipeline / CI/CD Pipeline Status (push) Successful in 2s
2026-10-24 06:56:43 +00:00
k.eaven 448d1d1448 Add Pelican Panel API key support: accept pacc_ (client) and papp_ (application) prefixes alongside Pterodactyl's ptlc_ and ptla_
CI/CD Pipeline / Code Quality & Linting (push) Successful in 58s
CI/CD Pipeline / Security Scanning (push) Successful in 1m13s
CI/CD Pipeline / Unit Tests (Python 3.12) (push) Successful in 4m16s
CI/CD Pipeline / Build Docker Image (push) Successful in 54s
CI/CD Pipeline / Integration Tests (push) Successful in 3m18s
CI/CD Pipeline / Generate Test Report (push) Successful in 3s
CI/CD Pipeline / CI/CD Pipeline Status (push) Successful in 2s
2026-10-24 06:51:57 +00:00
k.eaven 338ebe0836 Update CI/CD unit test matrix
CI/CD Pipeline / Unit Tests (Python 3.12) (push) Successful in 38s
CI/CD Pipeline / Unit Tests (Python 3.11) (push) Successful in 39s
CI/CD Pipeline / Unit Tests (Python 3.13) (push) Successful in 41s
CI/CD Pipeline / Code Quality & Linting (push) Successful in 25s
CI/CD Pipeline / Integration Tests (push) Successful in 28s
CI/CD Pipeline / Security Scanning (push) Successful in 2m20s
CI/CD Pipeline / Generate Test Report (push) Successful in 4s
CI/CD Pipeline / Build Docker Image (push) Successful in 11m39s
CI/CD Pipeline / CI/CD Pipeline Status (push) Successful in 1s
2026-10-23 22:05:56 +00:00
k.eaven 64e8acf697 Update CI/CD runner workflow
CI/CD Pipeline / Unit Tests (Python 3.11) (push) Successful in 1m45s
CI/CD Pipeline / Unit Tests (Python 3.12) (push) Successful in 1m36s
CI/CD Pipeline / Code Quality & Linting (push) Successful in 26s
CI/CD Pipeline / Security Scanning (push) Successful in 2m17s
CI/CD Pipeline / Unit Tests (Python 3.10) (push) Failing after 11s
CI/CD Pipeline / Integration Tests (push) Skipped
CI/CD Pipeline / Build Docker Image (push) Skipped
CI/CD Pipeline / Generate Test Report (push) Successful in 2s
CI/CD Pipeline / CI/CD Pipeline Status (push) Successful in 2s
2026-10-23 21:59:40 +00:00
k.eaven 1f0a63067b Update all dependencies to latest stable versions: Python 3.12, discord.py 2.7.1, aiohttp 3.14.5, matplotlib 3.11.2, and all test dependencies
CI/CD Pipeline / Code Quality & Linting (push) Successful in 46s
CI/CD Pipeline / Unit Tests (Python 3.12) (push) Successful in 23s
CI/CD Pipeline / Integration Tests (push) Successful in 30s
CI/CD Pipeline / Generate Test Report (push) Successful in 14s
CI/CD Pipeline / Build Docker Image (push) Successful in 1m11s
CI/CD Pipeline / CI/CD Pipeline Status (push) Successful in 2s
CI/CD Pipeline / Security Scanning (push) Successful in 28s
2026-10-23 20:52:18 +00:00
k.eaven 872b016707 Fix Gitea CI/CD checkout: use fetch-depth: 50 with git protocol config to resolve 'bad pack header' error
CI/CD Pipeline / Unit Tests (Python 3.10) (push) Successful in 6m40s
CI/CD Pipeline / Unit Tests (Python 3.11) (push) Successful in 6m38s
CI/CD Pipeline / Unit Tests (Python 3.9) (push) Successful in 6m40s
CI/CD Pipeline / Code Quality & Linting (push) Successful in 54s
CI/CD Pipeline / Security Scanning (push) Successful in 24s
CI/CD Pipeline / Integration Tests (push) Successful in 3m30s
CI/CD Pipeline / Generate Test Report (push) Successful in 3s
CI/CD Pipeline / Build Docker Image (push) Successful in 16m2s
CI/CD Pipeline / CI/CD Pipeline Status (push) Successful in 3s
2026-10-21 08:32:00 +00:00
k.eaven 8e7a68d6a2 Fix Gitea CI/CD checkout: add fetch-depth: 0 to resolve 'bad pack header' error 2026-10-21 08:02:59 +00:00
k.eaven 9d092514d9 Add: Kubernetes guide to README 2026-10-21 07:05:21 +00:00
k.eaven 4bd84e2c7d Improved README 2025-11-06 16:14:51 +00:00
k.eaven 7fd2e5e89b Update README.md
Simplifying main README in preparation for full wiki documentation of project
2025-11-06 13:26:04 +00:00
k.eaven 8fa5f18d83 README back in project root directory 2025-11-04 13:07:32 +00:00
k.eaven 21208897b0 Add: README directory 2025-11-04 11:54:52 +00:00
20 changed files with 3968 additions and 641 deletions

No files matched your search

+104 -48
View File
@@ -31,30 +31,36 @@ jobs:
strategy:
fail-fast: false
matrix:
python-version: ['3.9', '3.10', '3.11']
python-version: ['3.11', '3.12', '3.13']
steps:
- name: Checkout code
uses: actions/checkout@v3
with:
fetch-depth: 50
- name: Set up Python ${{ matrix.python-version }}
uses: actions/setup-python@v4
- name: Configure Git
run: |
git config --global protocol.version 2
git config --global http.postBuffer 524288000
git config --global http.lowSpeedLimit 0
git config --global http.lowSpeedTime 999999
- name: Set up uv & Python ${{ matrix.python-version }}
uses: astral-sh/setup-uv@v5
with:
python-version: ${{ matrix.python-version }}
- name: Cache pip dependencies
uses: actions/cache@v3
with:
path: ~/.cache/pip
key: ${{ runner.os }}-py${{ matrix.python-version }}-pip-${{ hashFiles('requirements.txt', 'requirements-test.txt') }}
restore-keys: |
${{ runner.os }}-py${{ matrix.python-version }}-pip-
activate-environment: true
enable-cache: true
cache-dependency-glob: |
requirements.txt
requirements-test.txt
- name: Install dependencies
run: |
python -m pip install --upgrade pip setuptools wheel
pip install -r requirements.txt
pip install -r requirements-test.txt
uv pip install --upgrade setuptools wheel
uv pip install -r requirements.txt
uv pip install -r requirements-test.txt
- name: Create test configuration
run: |
@@ -64,6 +70,9 @@ jobs:
PanelURL = https://panel.example.com
ClientAPIKey = ptlc_test_client_key_123456789
ApplicationAPIKey = ptla_test_app_key_987654321
# Pelican Panel test keys
# ClientAPIKey = pacc_test_client_key_123456789
# ApplicationAPIKey = papp_test_app_key_987654321
[Discord]
Token = test_discord_token_placeholder
@@ -72,11 +81,16 @@ jobs:
- name: Run unit tests with coverage
run: |
pytest test_pterodisbot.py \
python -m pytest test_pterodisbot.py \
-v \
--tb=short \
--cov=pterodisbot \
--cov=server_metrics_graphs \
--cov=config \
--cov=api \
--cov=embeds \
--cov=views \
--cov=bot \
--cov=metrics \
--cov=models \
--cov-report=xml \
--cov-report=term \
--cov-report=html \
@@ -99,20 +113,31 @@ jobs:
steps:
- name: Checkout code
uses: actions/checkout@v3
- name: Set up Python
uses: actions/setup-python@v4
with:
python-version: '3.11'
fetch-depth: 50
- name: Configure Git
run: |
git config --global protocol.version 2
git config --global http.postBuffer 524288000
git config --global http.lowSpeedLimit 0
git config --global http.lowSpeedTime 999999
- name: Set up uv & Python
uses: astral-sh/setup-uv@v5
with:
python-version: '3.12'
activate-environment: true
enable-cache: true
- name: Install linting tools
run: |
python -m pip install --upgrade pip
pip install flake8 pylint black isort mypy
uv pip install --upgrade setuptools wheel
uv pip install flake8 pylint black isort mypy
- name: Run flake8
run: |
flake8 pterodisbot.py server_metrics_graphs.py \
flake8 config.py api.py embeds.py views.py bot.py metrics.py models.py \
--max-line-length=140 \
--ignore=E501,W503,E203 \
--exclude=venv,__pycache__,build,dist \
@@ -122,7 +147,7 @@ jobs:
- name: Run pylint
run: |
pylint pterodisbot.py server_metrics_graphs.py \
pylint config.py api.py embeds.py views.py bot.py metrics.py models.py \
--disable=C0111,C0103,R0913,R0914,R0915,W0718 \
--max-line-length=140 \
--output-format=text \
@@ -131,17 +156,17 @@ jobs:
- name: Check code formatting with black
run: |
black --check --line-length=140 --diff pterodisbot.py server_metrics_graphs.py | tee black-report.txt
black --check --line-length=140 --diff config.py api.py embeds.py views.py bot.py metrics.py models.py | tee black-report.txt
continue-on-error: true
- name: Check import ordering
run: |
isort --check-only --profile black --line-length=140 pterodisbot.py server_metrics_graphs.py
isort --check-only --profile black --line-length=140 config.py api.py embeds.py views.py bot.py metrics.py models.py
continue-on-error: true
- name: Type checking with mypy
run: |
mypy pterodisbot.py server_metrics_graphs.py --ignore-missing-imports > mypy-report.txt || true
mypy config.py api.py embeds.py views.py bot.py metrics.py models.py --ignore-missing-imports > mypy-report.txt || true
continue-on-error: true
- name: Upload linting reports
@@ -162,20 +187,31 @@ jobs:
steps:
- name: Checkout code
uses: actions/checkout@v3
- name: Set up Python
uses: actions/setup-python@v4
with:
python-version: '3.11'
fetch-depth: 50
- name: Configure Git
run: |
git config --global protocol.version 2
git config --global http.postBuffer 524288000
git config --global http.lowSpeedLimit 0
git config --global http.lowSpeedTime 999999
- name: Set up uv & Python
uses: astral-sh/setup-uv@v5
with:
python-version: '3.12'
activate-environment: true
enable-cache: true
- name: Install security tools
run: |
python -m pip install --upgrade pip
pip install bandit safety pip-audit
uv pip install --upgrade setuptools wheel
uv pip install bandit safety pip-audit
- name: Run bandit security scan
run: |
bandit -r . \
bandit -r config.py api.py embeds.py views.py bot.py metrics.py models.py \
-f json \
-o bandit-report.json \
-ll \
@@ -184,7 +220,7 @@ jobs:
- name: Run safety dependency check
run: |
pip install -r requirements.txt
uv pip install -r requirements.txt
safety check --json --output safety-report.json || true
continue-on-error: true
@@ -211,23 +247,31 @@ jobs:
steps:
- name: Checkout code
uses: actions/checkout@v3
- name: Set up Python
uses: actions/setup-python@v4
with:
python-version: '3.11'
fetch-depth: 50
- name: Cache dependencies
uses: actions/cache@v3
- name: Configure Git
run: |
git config --global protocol.version 2
git config --global http.postBuffer 524288000
git config --global http.lowSpeedLimit 0
git config --global http.lowSpeedTime 999999
- name: Set up uv & Python
uses: astral-sh/setup-uv@v5
with:
path: ~/.cache/pip
key: ${{ runner.os }}-pip-integration-${{ hashFiles('requirements.txt') }}
python-version: '3.12'
activate-environment: true
enable-cache: true
cache-dependency-glob: |
requirements.txt
requirements-test.txt
- name: Install dependencies
run: |
python -m pip install --upgrade pip
pip install -r requirements.txt
pip install -r requirements-test.txt
uv pip install --upgrade setuptools wheel
uv pip install -r requirements.txt
uv pip install -r requirements-test.txt
- name: Create test configuration
run: |
@@ -237,6 +281,9 @@ jobs:
PanelURL = https://panel.example.com
ClientAPIKey = ptlc_test_client_key_123456789
ApplicationAPIKey = ptla_test_app_key_987654321
# Pelican Panel test keys
# ClientAPIKey = pacc_test_client_key_123456789
# ApplicationAPIKey = papp_test_app_key_987654321
[Discord]
Token = test_discord_token_placeholder
@@ -245,7 +292,7 @@ jobs:
- name: Run integration tests
run: |
pytest test_pterodisbot.py::TestIntegration \
python -m pytest test_pterodisbot.py::TestIntegration \
-v \
--tb=short \
--timeout=60
@@ -266,6 +313,15 @@ jobs:
steps:
- name: Checkout code
uses: actions/checkout@v3
with:
fetch-depth: 50
- name: Configure Git
run: |
git config --global protocol.version 2
git config --global http.postBuffer 524288000
git config --global http.lowSpeedLimit 0
git config --global http.lowSpeedTime 999999
- name: Set up QEMU
uses: docker/setup-qemu-action@v2
+524 -126
View File
@@ -1,4 +1,4 @@
# Pterodactyl Discord Bot
# Pterodisbot the Pterodactyl Discord Bot
![Pterodactyl X Discord](md_images/Pterodisbot_README_Header.png)
@@ -6,10 +6,11 @@ A feature-rich Discord bot for monitoring and managing Pterodactyl game servers
## Table of Contents
- [Features](#features)
- [Architecture Overview](#architecture-overview)
- [Installation](#installation)
- [Discord Bot Setup](#discord-bot-setup)
- [Traditional Installation](#traditional-installation)
- [Docker Installation](#docker-installation)
- [Kubernetes Installation](#kubernetes-installation)
- [Configuration](#configuration)
- [Environment Variables](#environment-variables)
- [Configuration File](#configuration-file)
@@ -20,58 +21,160 @@ A feature-rich Discord bot for monitoring and managing Pterodactyl game servers
- [Permissions](#permissions)
- [Logging System](#logging-system)
- [Troubleshooting](#troubleshooting)
- [Development](#development)
- [Contributing](#contributing)
- [License](#license)
## Features
### Core Functionality
- **Real-time Server Monitoring**: Auto-updating embeds showing server status (online/offline) and detailed resource usage
- **Power Management**: Start, stop, and restart servers directly from Discord with button controls
- **Connection Information**: One-click display of server addresses and ports
- **Multi-channel Support**: Status embeds can be placed in any channel with proper permissions
### 🎮 Server Management
- **Interactive Server Selection**: Choose servers from an intuitive dropdown menu showing all available game servers
- **Real-time Power Controls**: Start, stop, and restart servers with color-coded button controls
- 🟢 Green Start button
- 🔴 Red Stop button
- 🔵 Blue Restart button
- **Connection Information**: Instantly view server IP addresses and ports with the "Show Address" button
- **Multi-channel Embeds**: Deploy server status displays across any channels where the bot has permissions
### Intelligent Updates
- **State Change Detection**: Embeds update immediately when server power state changes
- **Resource Thresholds**: CPU usage changes >50% trigger updates
- **Optimized API Calls**: Minimizes requests to the Discord API
### 📊 Server Monitoring
- **Live Status Tracking**: Auto-updating embeds display current server state (online/offline/starting)
- **Resource Usage Metrics**: Detailed monitoring of:
- CPU usage with dynamic scaling for multi-core servers
- Memory consumption in real-time
- Disk space utilization
- Network activity (incoming/outgoing traffic)
- Server uptime counter with human-readable formatting
- **Visual Metrics Graphs**:
- Combined CPU and memory usage graphs over the last minute
- 6 data points captured at 10-second intervals
- Discord dark theme-optimized styling
- Automatic graph generation and embedding for running servers
- Dynamic CPU scaling (100% increments) for accurate multi-vCPU visualization
- Trend indicators (increasing/decreasing/stable) for quick status assessment
### Access Control
- **Guild Restriction**: Operates only in specified Discord server
- **Role-based Permissions**: "Game Server User" role required for interactions
- **Admin Commands**: Special commands restricted to server administrators
### 🔄 Intelligent Update System
- **Smart Update Triggers**: Embeds refresh automatically when:
- Server power state changes (offline ↔ starting ↔ running)
- CPU usage shifts by more than 50%
- Server is detected for the first time
- Running servers reach 10-minute intervals (for uptime accuracy)
- **Rate Limit Protection**: Built-in delays prevent Discord API throttling
- **Optimized API Calls**: Minimizes unnecessary requests through change detection
- **Background Task Management**: Asynchronous update loop runs every 10 seconds
### Reliability Features
- **Persistent Storage**: Remembers embed locations between bot restarts
- **Error Handling**: Comprehensive error recovery and logging
- **Rate Limit Protection**: Built-in delays between API calls
### 🔒 Security & Access Control
- **Single-Guild Restriction**: Bot operates exclusively in your designated Discord server
- **Role-based Permissions**: "Game Server User" role required for all server interactions
- **Administrator Commands**: Special management commands restricted to server administrators
- **Interaction Validation**: Every button press and command verified for proper authorization
## Architecture Overview
### 💾 Data Persistence
- **Embed Location Storage**: Automatically remembers where embeds are posted (survives bot restarts)
- **JSON-based Storage**: Human-readable embed tracking in `embed_locations.json`
- **Metrics History**: Maintains 1 minute of server metrics data (6 data points per server)
- **Automatic Cleanup**: Removes tracking for deleted embeds and inactive servers
The bot is built with several key components:
### 🛡️ Reliability Features
- **Comprehensive Error Handling**: Graceful recovery from API failures and network issues
- **Rotating Log Files**: Automatic log rotation (5MB max, 3 backup files)
- **Debug-level Logging**: Detailed operation logs for troubleshooting
- **Graceful Shutdown**: Proper cleanup on SIGINT/SIGTERM signals (Docker-friendly)
- **Configuration Validation**: Startup verification of all required settings
- **API Key Validation**: Checks for correct Pterodactyl API key formats
1. **PterodactylAPI Class**: Handles all communication with the Pterodactyl Panel API
2. **ServerStatusView Class**: Manages the interactive Discord UI with control buttons
3. **PterodactylBot Class**: Main bot class that orchestrates all operations
4. **Background Task**: Regularly checks server status and updates embeds when needed
5. **Configuration System**: Validates and manages all settings from config.ini
### 🎨 User Interface
- **Color-coded Status**: Visual indicators for server states
- 🔵 Blue for online servers
- 🔴 Red for offline servers
- **Ephemeral Interactions**: Private responses visible only to command users
- **Real-time Feedback**: Instant confirmation messages for all actions
- **Persistent Controls**: Server control buttons remain functional indefinitely
- **Responsive Design**: Clean, Discord-optimized embed layouts
The system uses aiohttp for asynchronous HTTP requests and discord.py for Discord interactions.
### 🔧 Administrator Tools
- **Bulk Embed Refresh**: Force-refresh all server status embeds with one command
- **Complete Purge System**: Remove all bot embeds from the server with detailed statistics
- **Progress Tracking**: Real-time progress updates during multi-embed operations
- **Comprehensive Logging**: All administrative actions logged with user attribution
## Installation
### Discord Bot Setup
Before installing the bot software, you need to create a Discord application and bot account.
#### 1. Create Discord Application
1. Navigate to the [Discord Developer Portal](https://discord.com/developers/applications)
2. Click the **"New Application"** button in the top-right corner
3. Enter a name for your application (e.g., "Pterodactyl Bot")
4. Accept the Terms of Service and click **"Create"**
#### 2. Configure Bot Settings
1. In the left sidebar, click on **"Bot"**
2. Click **"Add Bot"** and confirm by clicking **"Yes, do it!"**
3. Under the bot's username, you'll see a **Token** section:
- Click **"Reset Token"** and then **"Copy"** to get your bot token
- ⚠️ **Keep this token secret!** Never share it publicly or commit it to version control
4. Scroll down to **"Privileged Gateway Intents"** and enable:
- ✅ **Message Content Intent** (required for the bot to read message content)
5. Click **"Save Changes"** at the bottom
#### 3. Set Bot Permissions
1. In the left sidebar, click on **"OAuth2"** → **"URL Generator"**
2. Under **"Scopes"**, select:
- ✅ `bot`
- ✅ `applications.commands`
3. Under **"Bot Permissions"**, select:
- ✅ **View Channels** (allows bot to see channels)
- ✅ **Send Messages** (allows bot to post embeds)
- ✅ **Embed Links** (required for rich embeds)
- ✅ **Attach Files** (required for metrics graphs)
- ✅ **Read Message History** (allows bot to edit existing messages)
- ✅ **Use Slash Commands** (required for `/server_status` commands)
#### 4. Invite Bot to Your Server
1. Copy the **Generated URL** at the bottom of the OAuth2 URL Generator page
2. Paste this URL into your browser and press Enter
3. Select the Discord server where you want to add the bot
4. Click **"Authorize"** and complete the CAPTCHA if prompted
5. The bot should now appear in your server (offline until you start the application)
#### 5. Get Your Guild ID
1. In Discord, enable **Developer Mode**:
- Click the ⚙️ gear icon (User Settings)
- Navigate to **"Advanced"** under "App Settings"
- Toggle on **"Developer Mode"**
2. Right-click on your server icon in the left sidebar
3. Click **"Copy Server ID"** at the bottom of the menu
4. Save this ID - you'll need it for the bot configuration
#### 6. Create Server Role
1. In your Discord server, click the server name at the top to open the dropdown menu
2. Select **"Server Settings"** → **"Roles"**
3. Click **"Create Role"**
4. Name the role exactly: `Game Server User`
5. The role doesn't need any special permissions - it's only used for bot access control
6. Click **"Save Changes"**
7. Assign this role to any users who should be able to control game servers
### Prerequisites
- Python 3.8 or higher **or** Docker
- Discord bot token with proper permissions
- Discord bot token (from setup above)
- Discord server (guild) ID (from setup above)
- Pterodactyl panel with API access
- Server with both Client and Application API keys
- Pterodactyl Client API key (starts with `ptlc_`)
- Pterodactyl Application API key (starts with `ptla_`)
### Traditional Installation
1. Clone the repository:
```bash
git clone https://git.serendipity.systems/k.eaven/pterodactyl-discord-bot.git
git clone https://git.serendipity.systems/k.eaven/pterodisbot.git
cd pterodactyl-discord-bot
```
@@ -91,7 +194,7 @@ The system uses aiohttp for asynchronous HTTP requests and discord.py for Discor
1. Clone the repository:
```bash
git clone https://git.serendipity.systems/k.eaven/pterodactyl-discord-bot.git
git clone https://git.serendipity.systems/k.eaven/pterodisbot.git
cd pterodactyl-discord-bot
```
@@ -121,13 +224,13 @@ The system uses aiohttp for asynchronous HTTP requests and discord.py for Discor
pterodisbot:
build: .
environment:
# Remmove brackets and spaces in environment variables
# Remove brackets and spaces in environment variables
# Ideally, use .env file to assign values to these instead
- DISCORD_TOKEN={Your Discord bot token}
- ALLOWED_GUILD_ID={The Discord server ID where the bot should operate}
- PANEL_URL={Your Pterodactyl panel URL (must include http:// or https://)}
- CLIENT_API_KEY={Pterodactyl client API key (starts with `ptlc_`)}
- APPLICATION_API_KEY={Pterodactyl application API key (starts with `ptla_`)}
- DISCORD_TOKEN={Your Discord bot token from Developer Portal}
- ALLOWED_GUILD_ID={Your Discord server ID from step 5 above}
- PANEL_URL={Your Pterodactyl panel URL with https://}
- CLIENT_API_KEY={Pterodactyl client API key starting with ptlc_}
- APPLICATION_API_KEY={Pterodactyl application API key starting with ptla_}
volumes:
- bot_logs:/app/logs
- bot_embed:/app/embed
@@ -142,13 +245,199 @@ The system uses aiohttp for asynchronous HTTP requests and discord.py for Discor
docker-compose up -d
```
### Kubernetes Installation
1. Clone the repository:
```bash
git clone https://git.serendipity.systems/k.eaven/pterodisbot.git
cd pterodactyl-discord-bot
```
2. Build and push the Docker image to your container registry:
```bash
docker build -t your-registry/pterodisbot:latest .
docker push your-registry/pterodisbot:latest
```
3. Create a Kubernetes namespace (optional but recommended):
```bash
kubectl create namespace pterodisbot
```
4. Create a Kubernetes Secret with your configuration:
```bash
kubectl create secret generic pterodisbot-config \
--from-literal=DISCORD_TOKEN='your-discord-bot-token' \
--from-literal=ALLOWED_GUILD_ID='your-discord-guild-id' \
--from-literal=PANEL_URL='https://your.pterodactyl.panel' \
--from-literal=CLIENT_API_KEY='ptlc_your-client-api-key' \
--from-literal=APPLICATION_API_KEY='ptla_your-application-api-key' \
-n pterodisbot
```
5. Create a PersistentVolumeClaim for logs and embed data:
```yaml
# pvc.yaml
apiVersion: v1
kind: PersistentVolumeClaim
metadata:
name: pterodisbot-data
namespace: pterodisbot
spec:
accessModes:
- ReadWriteOnce
resources:
requests:
storage: 1Gi
```
```bash
kubectl apply -f pvc.yaml
```
6. Create the Deployment:
```yaml
# deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
name: pterodisbot
namespace: pterodisbot
labels:
app: pterodisbot
spec:
replicas: 1
selector:
matchLabels:
app: pterodisbot
template:
metadata:
labels:
app: pterodisbot
spec:
containers:
- name: pterodisbot
image: your-registry/pterodisbot:latest
imagePullPolicy: Always
envFrom:
- secretRef:
name: pterodisbot-config
volumeMounts:
- name: pterodisbot-data
mountPath: /app/logs
subPath: logs
- name: pterodisbot-data
mountPath: /app/embed
subPath: embed
resources:
requests:
memory: "256Mi"
cpu: "100m"
limits:
memory: "512Mi"
cpu: "500m"
livenessProbe:
exec:
command: ["python", "-c", "import sys; sys.exit(0)"]
initialDelaySeconds: 30
periodSeconds: 30
readinessProbe:
exec:
command: ["python", "-c", "import sys; sys.exit(0)"]
initialDelaySeconds: 10
periodSeconds: 10
volumes:
- name: pterodisbot-data
persistentVolumeClaim:
claimName: pterodisbot-data
```
```bash
kubectl apply -f deployment.yaml
```
7. (Optional) Create a Service if you need to expose metrics or health checks:
```yaml
# service.yaml
apiVersion: v1
kind: Service
metadata:
name: pterodisbot
namespace: pterodisbot
labels:
app: pterodisbot
spec:
selector:
app: pterodisbot
ports:
- name: http
port: 8080
targetPort: 8080
type: ClusterIP
```
```bash
kubectl apply -f service.yaml
```
8. (Optional) Create a HorizontalPodAutoscaler for automatic scaling:
```yaml
# hpa.yaml
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
name: pterodisbot
namespace: pterodisbot
spec:
scaleTargetRef:
apiVersion: apps/v1
kind: Deployment
name: pterodisbot
minReplicas: 1
maxReplicas: 3
metrics:
- type: Resource
resource:
name: cpu
target:
type: Utilization
averageUtilization: 70
```
```bash
kubectl apply -f hpa.yaml
```
9. Verify the deployment:
```bash
kubectl get pods -n pterodisbot
kubectl logs -n pterodisbot -l app=pterodisbot -f
```
10. For production, consider using a Helm chart or Kustomize for better configuration management. Create a `kustomization.yaml`:
```yaml
# kustomization.yaml
apiVersion: kustomize.config.k8s.io/v1beta1
kind: Kustomization
namespace: pterodisbot
resources:
- pvc.yaml
- deployment.yaml
- service.yaml
- hpa.yaml
images:
- name: pterodisbot
newTag: latest
commonLabels:
app: pterodisbot
```
```bash
kubectl apply -k .
```
## Configuration
### Environment Variables
When using Docker or generating the config automatically, these environment variables are required:
- `DISCORD_TOKEN`: Your Discord bot token
- `DISCORD_TOKEN`: Your Discord bot token (from Discord Developer Portal)
- `ALLOWED_GUILD_ID`: The Discord server ID where the bot should operate
- `PANEL_URL`: Your Pterodactyl panel URL (must include http:// or https://)
- `CLIENT_API_KEY`: Pterodactyl client API key (starts with `ptlc_`)
@@ -178,142 +467,251 @@ python generate_config.py
### First Run Setup
1. Invite your bot to your Discord server with these permissions:
- View Channels
- Send Messages
- Embed Links
- Manage Messages (for embed updates)
- Use Slash Commands
1. Ensure your bot is invited to your Discord server with proper permissions (see [Discord Bot Setup](#discord-bot-setup))
2. Create a "Game Server User" role for users who should control servers
2. Verify the "Game Server User" role exists in your server
3. Use the `/server_status` command to create your first status embed
3. Start the bot using your preferred installation method
4. The bot should appear online in your server
5. Use the `/server_status` command to create your first status embed
### Bot Commands
#### `/server_status`
Display an interactive context menu containing all available servers which can be selected to create an embed for a given server.
Display an interactive dropdown menu containing all available servers from your Pterodactyl panel. Select a server to create a permanent status embed in the current channel with real-time monitoring and control buttons.
**Features:**
- Shows total server count and online/offline statistics
- Lists all servers with descriptions
- Creates persistent, auto-updating embeds
- Automatically removes old embeds when recreating for the same server
#### `/refresh_embeds` (Admin only)
Force a refresh of all server status embeds. This is useful if embeds become out of sync.
Force a complete refresh of all server status embeds. This command:
- Recreates all tracked embeds with fresh data
- Deletes old embed messages to prevent duplication
- Updates embed tracking storage
- Provides statistics on deleted and created embeds
**Use cases:**
- Embeds appear out of sync with actual server states
- After bot configuration changes
- Following Discord API issues
#### `/purge_embeds` (Admin only)
Delete all embeds from the Discord guild.
Permanently delete all server status embeds from the Discord server. This command:
- Removes all tracked embed messages
- Clears the embed tracking file
- Provides real-time progress updates
- Shows detailed statistics (deleted, missing, errors)
**Use cases:**
- Starting fresh with new embed layout
- Removing bot from server
- Cleaning up after testing
### Embed System
Status embeds include:
- Server name and current state (color-coded)
- Resource usage (when online):
- CPU percentage
- Memory usage
- Disk usage
- Network activity
- Interactive buttons:
- Start (green)
- Stop (red)
- Restart (blue)
- Show Address (grey)
Status embeds dynamically display:
Embed locations are stored in `embed_locations.json` and persist between bot restarts.
**Header Information:**
- Server name with current state (color-coded)
- Server description
- Last updated timestamp
**Status Fields:**
- 🆔 Server ID (Pterodactyl identifier)
- ℹ️ Status (Active or Suspended)
- ⏱️ Uptime (for running servers)
**Resource Usage (when online):**
- CPU percentage with dynamic scaling
- Memory usage (current vs. allocated)
- Disk usage (current vs. allocated)
- Network traffic (incoming/outgoing)
**Metrics Visualization (when running):**
- Combined CPU and memory graph
- Last 1 minute of data (6 data points)
- Trend indicators
- Data point counter
**Interactive Buttons:**
- 🟢 **Start** - Powers on the server
- 🔴 **Stop** - Gracefully shuts down the server
- 🔵 **Restart** - Reboots the server
- ⚪ **Show Address** - Displays connection information (IP:Port)
**Automatic Updates:**
Embeds refresh when server state changes, including:
- Power state transitions
- Significant resource usage changes
- Periodic updates for uptime accuracy
Embed locations are stored in `embed/embed_locations.json` and persist between bot restarts.
## Permissions
The bot implements two levels of access control:
1. **Guild Restriction**: Only works in the Discord server specified in `AllowedGuildID`
2. **Role Requirement**: Users must have the "Game Server User" role to interact with server controls
### 1. Guild Restriction
The bot only responds to commands and interactions from the Discord server specified in `AllowedGuildID`. This prevents unauthorized use if the bot is accidentally added to other servers.
Administrators can use the `/refresh_embeds` and `/purge_embeds` command.
### 2. Role-Based Access
Users must have the **"Game Server User"** role (exact name match) to:
- Use server control buttons (Start/Stop/Restart)
- View server addresses
- Interact with any bot features
### 3. Administrator Access
Server administrators (users with Discord's Administrator permission) can additionally:
- Use `/refresh_embeds` command
- Use `/purge_embeds` command
- No special role required - based on Discord permissions
**Note:** The `/server_status` command can be used by anyone, but only users with the "Game Server User" role can interact with the resulting embeds.
## Logging System
The bot maintains comprehensive logs in two locations:
1. **File Logs**: Rotating log files (5MB max, 3 backups) in `pterodactyl_bot.log`
2. **Console Output**: Real-time logging to stdout
### 1. File Logs
- **Location**: `logs/pterodisbot.log`
- **Rotation**: Automatic rotation when file reaches 5MB
- **Backups**: Keeps 3 previous log files (`pterodisbot.log.1`, `.2`, `.3`)
- **Format**: `YYYY-MM-DD HH:MM:SS - LEVEL - MESSAGE`
Log levels:
- **DEBUG**: Detailed operational information (API calls, state changes)
- **INFO**: General operational messages (bot startup, command usage)
- **WARNING**: Potential issues (missing embeds, API warnings)
- **ERROR**: Operational failures (API errors, permission issues)
- **CRITICAL**: Critical failures (configuration errors, startup failures)
### 2. Console Output
- **Real-time**: Logs appear in stdout/stderr
- **Format**: Same as file logs
- **Level**: INFO and above (less verbose than file logs)
### Log Levels
- **DEBUG**: Detailed operational information (API calls, state changes, cache operations)
- **INFO**: General operational messages (bot startup, command usage, successful operations)
- **WARNING**: Potential issues (missing embeds, API warnings, permission issues)
- **ERROR**: Operational failures (API errors, Discord API issues, failed operations)
- **CRITICAL**: Critical failures (configuration errors, startup failures, unrecoverable errors)
### Example Log Entries
```
2025-09-29 03:51:13,743 - INFO - Initialized logging system with file and console output
2025-09-29 03:51:16,828 - INFO - Starting full refresh of all server embeds
2025-09-29 03:51:16,829 - INFO - Bot connected as Petrodisbot (ID: 1419694924458360964)
2025-09-29 03:52:10,481 - INFO - Start button pressed for Minecraft Server by username
2025-09-29 03:52:14,406 - DEBUG - Power state changed for Minecraft Server: offline -> starting
```
## Troubleshooting
### Common Issues
**Embeds not updating:**
- Verify the bot has proper channel permissions
- Check API keys have correct permissions
- Use `/refresh_embeds` to reset all embeds
#### Embeds not updating
**Symptoms**: Embeds show outdated information or don't change when server state changes
**Buttons not working:**
- Confirm users have the "Game Server User" role
- Ensure the bot has these permissions:
**Solutions:**
- Verify the bot has proper channel permissions (View Channel, Send Messages, Embed Links)
- Check that both API keys are valid and have correct permissions
- Review logs for API errors or rate limiting
- Use `/refresh_embeds` to force-recreate all embeds
- Ensure the bot hasn't been rate-limited by Discord (check logs for HTTP 429 errors)
#### Buttons not working
**Symptoms**: Clicking Start/Stop/Restart buttons shows permission errors
**Solutions:**
- Confirm users have the "Game Server User" role (exact name match)
- Verify the role is properly assigned to users
- Check bot permissions:
- View Channel
- Send Messages
- Embed Links
- Use Slash Commands
- Ensure the interaction is happening in the correct Discord server (guild)
**API errors:**
- Double-check your Pterodactyl panel URL
#### Commands not appearing
**Symptoms**: Slash commands don't show up when typing `/`
**Solutions:**
- Wait 5-10 minutes after bot startup (command sync takes time)
- Verify bot has "Use Slash Commands" permission
- Check that the bot is in the correct guild (server)
- Review logs for command sync errors
- Try kicking and re-inviting the bot with proper OAuth2 URL
#### API errors
**Symptoms**: Bot logs show Pterodactyl API failures
**Solutions:**
- Double-check your Pterodactyl panel URL (must include `https://`)
- Verify API keys are correct and not expired
- Confirm the server exists in Pterodactyl
- Confirm the server exists in Pterodactyl panel
- Check API key permissions:
- Client key needs: server resource access, power control
- Application key needs: server list access, server details
- Test API keys directly using Pterodactyl API documentation
#### Graphs not appearing
**Symptoms**: Embeds show "Usage Trends" section but no graph image
**Solutions:**
- Wait at least 20 seconds after server starts (needs 2 data points minimum)
- Check that the server is in "running" state
- Verify matplotlib is installed correctly (`pip install matplotlib`)
- Review logs for graph generation errors
- Ensure `/tmp/matplotlib` directory has write permissions (Docker)
#### Bot crashes on startup
**Symptoms**: Bot exits immediately with configuration errors
**Solutions:**
- Verify all required fields in `config.ini`:
- No empty values
- Correct format for all URLs (must include protocol)
- API keys have correct prefixes (`ptlc_` and `ptla_`)
- Guild ID is a valid integer
- Check file permissions on `config.ini`
- Review startup logs for specific validation errors
- Test configuration validation: `python -c "from pterodisbot import validate_config; validate_config()"`
### Checking Logs
For real-time log monitoring:
```bash
tail -f pterodactyl_bot.log
# Traditional installation
tail -f logs/pterodisbot.log
# Docker installation
docker logs -f pterodisbot
# Docker Compose
docker-compose logs -f pterodisbot
```
## Development
### Requirements
- Python 3.8+
- Poetry (recommended) or pip
### Setup
1. Clone the repository
2. Install dependencies:
```bash
poetry install
```
or
```bash
pip install -r requirements.txt -r requirements-dev.txt
```
3. Create a `.env` file with your development environment variables
4. Run in development mode:
```bash
python pterodisbot.py
```
### Testing
The project includes unit tests (to be expanded). Run with:
For searching logs:
```bash
pytest
# Find all ERROR level messages
grep "ERROR" logs/pterodisbot.log
# Find logs for specific server
grep "Server_Name" logs/pterodisbot.log
# View last 100 lines
tail -n 100 logs/pterodisbot.log
```
## Contributing
### Getting Help
Contributions are welcome! Please follow these guidelines:
If issues persist after trying these solutions:
1. Fork the repository
2. Create a feature branch (`git checkout -b feature/your-feature`)
3. Commit your changes (`git commit -am 'Add some feature'`)
4. Push to the branch (`git push origin feature/your-feature`)
5. Open a pull request
### Coding Standards
- Follow PEP 8 style guidelines
- Include type hints for all function signatures
- Document public methods with docstrings
- Write tests for new functionality
1. Enable DEBUG logging by reviewing recent log files
2. Check the [Git Issues](https://git.serendipity.systems/k.eaven/pterodactyl-discord-bot/issues) for similar problems
3. Verify your setup matches all requirements in [Prerequisites](#prerequisites)
4. Test each component individually (Discord bot, Pterodactyl API, permissions)
## License
+44
View File
@@ -0,0 +1,44 @@
"""
Pterodisbot
A feature-rich Discord bot for monitoring and managing Pterodactyl/Pelican Panel
game servers through Discord, providing real-time status monitoring and control
capabilities with an intuitive interface.
Modules:
- config: Configuration management and validation
- api: Panel API client (Pterodactyl/Pelican)
- models: Pydantic data models
- metrics: Server metrics tracking and graph generation
- embeds: Discord embed management and persistence
- views: Discord UI components (buttons, dropdowns)
- bot: Main bot class and command definitions
- exceptions: Custom exception classes
"""
__version__ = "1.2.0"
__author__ = "Eaven Kimura"
from config import Config, load_config, generate_config_from_env
from exceptions import (
PterodisBotError,
ConfigValidationError,
APIError,
ServerNotFoundError,
PermissionError,
EmbedNotFoundError,
MetricsError,
)
__all__ = [
"Config",
"load_config",
"generate_config_from_env",
"PterodisBotError",
"ConfigValidationError",
"APIError",
"ServerNotFoundError",
"PermissionError",
"EmbedNotFoundError",
"MetricsError",
]
+341
View File
@@ -0,0 +1,341 @@
"""
Pterodactyl/Pelican Panel API Client Module
Provides a clean interface for interacting with Pterodactyl and Pelican Panel APIs.
Supports both client API (server operations) and application API (admin operations).
"""
import asyncio
import logging
from typing import Any, Dict, List, Optional
import aiohttp
from config import Config
from exceptions import APIError, ServerNotFoundError
logger = logging.getLogger("pterodisbot")
class PanelAPI:
"""
Unified API client for Pterodactyl and Pelican Panel.
Handles all HTTP requests to the panel API with proper authentication,
error handling, and rate limiting protection.
Features:
- Automatic session management
- Dual API key support (client + application)
- Concurrent request protection via asyncio.Lock
- Comprehensive error handling and logging
- Response parsing with structured error information
Args:
config: Validated Config instance containing panel credentials
"""
def __init__(self, config: Config) -> None:
"""
Initialize the Panel API client.
Args:
config: Validated Config instance with panel URL and API keys
"""
self.config = config
self.panel_url = config.pterodactyl_url.rstrip("/")
self.client_api_key = config.client_api_key
self.application_api_key = config.application_api_key
self.session: Optional[aiohttp.ClientSession] = None
self._lock = asyncio.Lock()
self._is_initialized = False
logger.info(f"Initialized {config.get_panel_display_name()} API client")
async def initialize(self) -> None:
"""Initialize the aiohttp client session."""
if self.session is None or self.session.closed:
self.session = aiohttp.ClientSession()
logger.debug("Created new aiohttp ClientSession")
self._is_initialized = True
async def close(self) -> None:
"""Cleanly close the aiohttp session."""
if self.session and not self.session.closed:
await self.session.close()
logger.debug("Closed aiohttp ClientSession")
self._is_initialized = False
async def _request(
self,
method: str,
endpoint: str,
data: Optional[Dict[str, Any]] = None,
use_application_key: bool = False,
) -> Dict[str, Any]:
"""
Make an authenticated request to the panel API.
Args:
method: HTTP method (GET, POST, PUT, DELETE)
endpoint: API endpoint (e.g., 'application/servers')
data: Optional JSON payload for POST/PUT requests
use_application_key: Whether to use application API key (admin endpoints)
Returns:
Dictionary containing API response data
Raises:
APIError: For network errors, HTTP errors, or parsing failures
"""
if not self._is_initialized or self.session is None or self.session.closed:
await self.initialize()
url = f"{self.panel_url}/api/{endpoint}"
api_key_type = "Application" if use_application_key else "Client"
logger.debug(f"Preparing {method} request to {endpoint} using {api_key_type} API key")
api_key = self.application_api_key if use_application_key else self.client_api_key
headers = {
"Authorization": f"Bearer {api_key}",
"Accept": "application/json",
"Content-Type": "application/json",
}
try:
async with self._lock:
logger.debug(f"Acquired lock for API request to {endpoint}")
async with self.session.request(
method, url, headers=headers, json=data
) as response:
if response.status == 204:
logger.debug(f"Received 204 No Content from {endpoint}")
return {"status": "success"}
response_data = await response.json()
logger.debug(
f"Received response from {endpoint} with status {response.status}"
)
if response.status >= 400:
error_msg = self._extract_error_message(response_data)
logger.error(
f"API request to {endpoint} failed with status "
f"{response.status}: {error_msg}"
)
raise APIError(
message=error_msg,
endpoint=endpoint,
status_code=response.status,
)
return response_data
except aiohttp.ClientError as e:
logger.error(f"Network error during API request to {endpoint}: {str(e)}")
raise APIError(
message=f"Network error: {str(e)}",
endpoint=endpoint,
)
except Exception as e:
logger.error(f"Unexpected error during API request to {endpoint}: {str(e)}")
raise APIError(
message=f"Unexpected error: {str(e)}",
endpoint=endpoint,
)
def _extract_error_message(self, response_data: Dict[str, Any]) -> str:
"""
Extract human-readable error message from API response.
Args:
response_data: Parsed JSON response from API
Returns:
Formatted error message string
"""
if "errors" in response_data and isinstance(response_data["errors"], list):
errors = response_data["errors"]
if errors and isinstance(errors[0], dict):
return errors[0].get("detail", "Unknown API error")
return str(errors[0]) if errors else "Unknown API error"
return response_data.get("message", "Unknown error")
# ============================================================
# Server Management Methods (Client API)
# ============================================================
async def get_server_resources(self, server_id: str) -> Dict[str, Any]:
"""
Get resource usage for a specific server.
Args:
server_id: The panel server identifier
Returns:
Dictionary containing server resource usage and state
Raises:
ServerNotFoundError: If server doesn't exist
APIError: For other API errors
"""
logger.debug(f"Fetching resource usage for server {server_id}")
try:
response = await self._request(
"GET", f"client/servers/{server_id}/resources"
)
return response
except APIError as e:
if e.status_code == 404:
raise ServerNotFoundError(f"Server {server_id} not found")
raise
async def send_power_action(self, server_id: str, action: str) -> Dict[str, Any]:
"""
Send a power action to a server.
Args:
server_id: The panel server identifier
action: Power action (start, stop, restart)
Returns:
Dictionary containing API response status
Raises:
ValueError: If action is invalid
APIError: For API errors
"""
valid_actions = ["start", "stop", "restart"]
if action not in valid_actions:
raise ValueError(
f"Invalid power action: {action}. Must be one of: {valid_actions}"
)
logger.info(f"Sending {action} command to server {server_id}")
result = await self._request(
"POST", f"client/servers/{server_id}/power", {"signal": action}
)
if result.get("status") == "success":
logger.info(f"Successfully executed {action} on server {server_id}")
else:
logger.error(f"Failed to execute {action} on server {server_id}")
return result
# ============================================================
# Server Information Methods (Application API)
# ============================================================
async def get_servers(self) -> List[Dict[str, Any]]:
"""
Get a list of all servers from the panel.
Uses application API key as this is an admin endpoint.
Returns:
List of server dictionaries with full attributes
"""
logger.info("Fetching list of all servers from panel")
response = await self._request("GET", "application/servers", use_application_key=True)
servers = response.get("data", [])
logger.info(f"Retrieved {len(servers)} servers from panel")
return servers
async def get_server_details(self, server_id: str) -> Dict[str, Any]:
"""
Get detailed server information including allocations.
Uses application API key as this is an admin endpoint.
Args:
server_id: The panel server identifier
Returns:
Dictionary containing detailed server information
"""
logger.debug(f"Fetching detailed information for server {server_id}")
return await self._request(
"GET", f"application/servers/{server_id}", use_application_key=True
)
async def get_server_allocations(self, server_id: str) -> Dict[str, Any]:
"""
Get allocation information for a server (IP addresses and ports).
Uses application API key as this is an admin endpoint.
Args:
server_id: The panel server identifier
Returns:
Dictionary containing server allocation information
"""
logger.debug(f"Fetching allocation information for server {server_id}")
return await self._request(
"GET",
f"application/servers/{server_id}/allocations",
use_application_key=True,
)
async def get_client_server_details(self, server_id: str) -> Dict[str, Any]:
"""
Get server details using client API (for connection info).
Args:
server_id: The panel server identifier
Returns:
Dictionary containing server details with allocations
"""
logger.debug(f"Fetching client server details for {server_id}")
return await self._request("GET", f"client/servers/{server_id}")
# ============================================================
# Convenience Methods
# ============================================================
async def get_server_default_allocation(
self, server_id: str
) -> tuple[str, str]:
"""
Get the default allocation IP and port for a server.
Args:
server_id: The panel server identifier
Returns:
Tuple of (ip_alias, port)
Raises:
ServerNotFoundError: If server or allocations not found
APIError: For API errors
"""
details = await self.get_client_server_details(server_id)
if details.get("status") == "error":
raise ServerNotFoundError(
details.get("message", "Failed to get server details")
)
attributes = details.get("attributes", {})
relationships = attributes.get("relationships", {})
allocations = relationships.get("allocations", {}).get("data", [])
if not allocations:
raise ServerNotFoundError("No allocations found for this server")
# Find default allocation or fallback to first
default_allocation = next(
(
alloc
for alloc in allocations
if alloc.get("attributes", {}).get("is_default", False)
),
allocations[0],
)
allocation_attrs = default_allocation.get("attributes", {})
ip_alias = allocation_attrs.get("ip_alias", "Unknown")
port = str(allocation_attrs.get("port", "Unknown"))
return ip_alias, port
+707
View File
@@ -0,0 +1,707 @@
"""
Main Bot Module
Core bot class and command definitions.
Manages the bot lifecycle, background tasks, and Discord interactions.
"""
import asyncio
import logging
import signal
import sys
import types
from datetime import datetime
from typing import TYPE_CHECKING, Optional, Tuple
import discord
from discord.ext import commands, tasks
from api import PanelAPI
from config import Config
from embeds import EmbedManager
from exceptions import ConfigValidationError
from metrics import ServerMetricsManager
from models import ServerData, ServerResources
if TYPE_CHECKING:
from views import ServerStatusView
logger = logging.getLogger("pterodisbot")
# Singleton instance holder
_bot_instance: Optional["PterodactylBot"] = None
def get_bot_instance() -> Optional["PterodactylBot"]:
"""Get the singleton bot instance."""
return _bot_instance
class PterodactylBot(commands.Bot):
"""
Main bot class for Pterodactyl/Pelican Panel server management.
Handles Discord interactions, embed management, and background tasks.
Manages server status embeds and user commands.
Features:
- Background status update task with smart update detection
- Persistent embed tracking with JSON storage
- Real-time server monitoring with metrics graphs
- Role-based access control
- Single-guild restriction
- Graceful shutdown handling
"""
def __init__(self, config: Config, *args, **kwargs) -> None:
"""
Initialize the Pterodactyl bot instance.
Args:
config: Validated application configuration
*args: Additional arguments for commands.Bot
**kwargs: Additional keyword arguments for commands.Bot
"""
global _bot_instance
_bot_instance = self
super().__init__(*args, **kwargs)
self.config = config
self.api = PanelAPI(config)
self.metrics_manager = ServerMetricsManager()
self.embed_manager = EmbedManager(config, self.api, self.metrics_manager)
# Server cache from panel
self.server_cache: dict = {}
# Concurrency control
self.update_lock = asyncio.Lock()
# State tracking for smart updates
# Format: {server_id: (state, cpu_usage, last_force_update)}
self.previous_states: dict = {}
logger.info("Initialized PterodactylBot instance with state tracking")
@classmethod
def get_instance(cls) -> Optional["PterodactylBot"]:
"""Get the singleton bot instance."""
return _bot_instance
async def setup_hook(self) -> None:
"""
Bot setup routine called when the bot is starting.
Initializes API client, loads saved data, and starts background tasks.
"""
logger.info("Running bot setup hook")
# Initialize API client
await self.api.initialize()
logger.info(f"Initialized {self.config.get_panel_display_name()} API client")
# Load saved embed locations
await self.embed_manager.load_locations()
# Start background update task
self.update_status.start()
logger.info("Started background status update task")
@tasks.loop(seconds=10) # UPDATE_INTERVAL = 10
async def update_status(self) -> None:
"""
Background task to update server status embeds when:
1. Server power state changes (started/stopped/restarted)
2. Significant CPU usage change (>50% difference)
3. First time seeing the server
4. Server has been running for 10 minutes (force update for uptime)
This minimizes API calls to Discord and updates while maintaining
real-time awareness of important server changes.
"""
logger.info("Starting optimized server status update cycle")
async with self.update_lock:
try:
# Fetch current server list from panel
servers = await self.api.get_servers()
if not servers:
logger.warning("No servers found in panel during update")
return
# Update our local cache with fresh server data
self.server_cache = {s["attributes"]["identifier"]: s for s in servers}
logger.debug(f"Updated server cache with {len(servers)} servers")
# Clean up metrics for servers that no longer exist
active_server_ids = list(self.server_cache.keys())
self.metrics_manager.cleanup_old_servers(active_server_ids)
# Update statistics
update_count = 0
error_count = 0
missing_count = 0
skipped_count = 0
current_time = datetime.now().timestamp()
# Process each server we're tracking embeds for
for server_id, location in list(self.embed_manager.embed_locations.items()):
# Skip if server no longer exists in panel
if server_id not in self.server_cache:
logger.warning(f"Server {server_id} not found in cache, skipping update")
continue
server_data = self.server_cache[server_id]
server_name = server_data["attributes"]["name"]
try:
logger.debug(f"Checking status for server {server_name} ({server_id})")
# Get current server resource usage
resources = await self.api.get_server_resources(server_id)
current_state = resources["attributes"]["current_state"]
cpu_usage = round(
resources["attributes"]["resources"].get("cpu_absolute", 0), 2
)
# Collect metrics data for running servers
if current_state == "running":
memory_usage = round(
resources["attributes"]["resources"].get("memory_bytes", 0)
/ (1024**2),
2,
)
self.metrics_manager.add_server_data(
server_id, server_name, cpu_usage, memory_usage
)
logger.debug(
f"Added metrics data for {server_name}: "
f"CPU={cpu_usage}%, Memory={memory_usage}MB"
)
# Retrieve previous recorded state
prev_state, prev_cpu, last_force_update = self.previous_states.get(
server_id, (None, 0, None)
)
# DECISION LOGIC: Should we update the embed?
needs_update = False
# 1. Check if power state changed (most important)
if current_state != prev_state:
logger.debug(
f"Power state changed for {server_name}: "
f"{prev_state} -> {current_state}"
)
needs_update = True
# 2. Check for significant CPU change (only if server is running)
elif current_state == "running" and abs(cpu_usage - prev_cpu) > 50:
logger.debug(
f"Significant CPU change for {server_name}: "
f"{prev_cpu}% -> {cpu_usage}%"
)
needs_update = True
# 3. First time we're seeing this server (initial update)
elif prev_state is None:
logger.debug(f"First check for {server_name}, performing initial update")
needs_update = True
# 4. Force update every 10 minutes for running servers (for uptime counter)
elif current_state == "running" and (
last_force_update is None
or current_time - last_force_update >= 600
): # 10 minutes = 600 seconds
logger.debug(f"Executing 10-minute force update for running server {server_name}")
needs_update = True
last_force_update = current_time
# PERFORM UPDATE IF NEEDED
if needs_update:
# Convert to models for embed creation
server_obj = ServerData.model_validate(server_data)
resources_obj = ServerResources.model_validate(resources)
# Update the embed
success = await self.embed_manager.update_single_embed(
server_id=server_id,
location=location,
server_data=server_obj,
resources=resources_obj,
current_state=current_state,
cpu_usage=cpu_usage,
)
if success:
update_count += 1
# Update state tracking
new_last_force_update = (
last_force_update
if needs_update
and current_state == "running"
and current_time - (last_force_update or 0) >= 600
else last_force_update
)
self.previous_states[server_id] = (
current_state,
cpu_usage,
new_last_force_update,
)
else:
# No significant changes, but update tracking with current state
self.previous_states[server_id] = (
current_state,
cpu_usage,
last_force_update,
)
skipped_count += 1
logger.debug(f"No changes detected for {server_name}, skipping update")
except discord.NotFound:
# Embed message was deleted - clean up tracking
logger.warning(f"Embed for server {server_id} not found, removing from tracking")
await self.embed_manager.untrack_embed(server_id)
missing_count += 1
except Exception as e:
logger.error(f"Failed to update status for server {server_id}: {str(e)}")
error_count += 1
# Small delay between servers to avoid rate limits
await asyncio.sleep(0.5)
# Log summary of this update cycle
logger.info(
f"Update cycle complete: "
f"{update_count} updated, "
f"{skipped_count} skipped, "
f"{missing_count} missing, "
f"{error_count} errors"
)
except Exception as e:
logger.error(f"Error in update_status task: {str(e)}")
await asyncio.sleep(5)
@update_status.before_loop
async def before_update_status(self) -> None:
"""Wait for bot to be ready before starting update task."""
logger.debug("Waiting for bot readiness before starting update task")
await self.wait_until_ready()
await self.embed_manager.refresh_all_embeds()
@update_status.after_loop
async def after_update_status(self) -> None:
"""Handle update task stopping."""
if self.update_status.is_being_cancelled():
logger.info("Server status update task was cancelled")
elif self.update_status.failed():
logger.error("Server status update task failed")
async def close(self) -> None:
"""Cleanup when bot is shutting down."""
await self.embed_manager.save_locations()
if self.api:
await self.api.close()
await super().close()
async def check_allowed_guild(interaction: discord.Interaction, config: Config) -> bool:
"""
Verify that an interaction is coming from the allowed guild.
Args:
interaction: Discord interaction object
config: Application configuration
Returns:
bool: True if interaction is allowed, False otherwise
"""
if interaction.guild_id != config.allowed_guild_id:
logger.warning(
f"Command attempted from unauthorized guild {interaction.guild_id} "
f"by {interaction.user.name}"
)
await interaction.response.send_message(
"This bot is only available in a specific server.", ephemeral=True
)
return False
return True
async def setup_commands(bot: PterodactylBot) -> None:
"""
Set up all slash commands and event handlers for the bot.
Args:
bot: The bot instance
"""
config = bot.config
# ==============================================
# EVENT HANDLERS
# ==============================================
@bot.event
async def on_interaction(interaction: discord.Interaction) -> None:
"""Global interaction handler to check guild before processing any interaction."""
if interaction.guild_id != config.allowed_guild_id:
logger.debug(f"Ignoring interaction from unauthorized guild {interaction.guild_id}")
await interaction.response.send_message(
"This bot is only available in a specific server.", ephemeral=True
)
return
@bot.event
async def on_ready() -> None:
"""Called when the bot successfully connects to Discord."""
logger.info(f"Bot connected as {bot.user.name} (ID: {bot.user.id})")
try:
# Sync commands only to the allowed guild
guild = discord.Object(id=bot.config.allowed_guild_id)
bot.tree.copy_global_to(guild=guild)
synced = await bot.tree.sync(guild=guild)
logger.info(
f"Successfully synced {len(synced)} command(s) to guild {bot.config.allowed_guild_id}: "
f"{[cmd.name for cmd in synced]}"
)
except Exception as e:
logger.error(f"Command sync failed: {str(e)}")
# ==============================================
# SLASH COMMANDS
# ==============================================
@bot.tree.command(
name="server_status", description="Get a list of available game servers to control"
)
async def server_status(interaction: discord.Interaction) -> None:
"""
Slash command to display server status dashboard with interactive dropdown selection.
Workflow:
1. Validates guild permissions and defers ephemeral response
2. Refreshes server cache from Panel API
3. Calculates online/offline statistics by checking each server's state
4. Creates statistics embed with visual server status breakdown
5. Generates dropdown menu with all servers (name + description)
6. Handles user selection via ephemeral dropdown interaction
7. Creates permanent status embed in channel upon selection
8. Manages embed tracking and cleanup of previous embeds
"""
if not await check_allowed_guild(interaction, config):
return
logger.info(f"Server status command invoked by {interaction.user.name}")
await interaction.response.defer(ephemeral=True)
try:
# Refresh server cache with current data from panel
servers = await bot.api.get_servers()
if not servers:
logger.warning("No servers found in panel")
await interaction.followup.send(
"No servers found in the panel.", ephemeral=True
)
return
bot.server_cache = {s["attributes"]["identifier"]: s for s in servers}
logger.debug(f"Refreshed server cache with {len(servers)} servers")
# Count online/offline servers
online_count = 0
offline_count = 0
for server_id, server_data in bot.server_cache.items():
resources = await bot.api.get_server_resources(server_id)
current_state = resources.get("attributes", {}).get("current_state", "offline")
if current_state == "running":
online_count += 1
else:
offline_count += 1
# Create statistics embed
stats_embed = discord.Embed(
title="🏗️ Server Status Dashboard",
description="Select a server from the dropdown below to view its detailed status and controls.",
color=discord.Color.blue(),
timestamp=datetime.now(),
)
stats_embed.add_field(
name="📊 Server Statistics",
value=f"**Total Servers:** {len(servers)}\n"
f"✅ **Online:** {online_count}\n"
f"❌ **Offline:** {offline_count}",
inline=False,
)
stats_embed.add_field(
name="ℹ️ How to Use",
value="Use the dropdown menu below to select a server. "
"The server's status embed will be posted in this channel.",
inline=False,
)
stats_embed.set_footer(text="Server status will update automatically")
# Create dropdown menu options
from views import ServerDropdown, ServerDropdownView
server_options = []
for server_id, server_data in bot.server_cache.items():
server_name = server_data["attributes"]["name"]
server_description = server_data["attributes"].get("description", "No description")
# Truncate description if too long for dropdown constraints
if len(server_description) > 50:
server_description = server_description[:47] + "..."
server_options.append(
discord.SelectOption(
label=server_name, value=server_id, description=server_description
)
)
# Send the initial dashboard embed with dropdown (ephemeral)
await interaction.followup.send(
embed=stats_embed,
view=ServerDropdownView(server_options),
ephemeral=True,
)
logger.info(f"Sent server status dashboard to {interaction.user.name} with {len(server_options)} servers")
except Exception as e:
logger.error(f"Server status command failed: {str(e)}")
await interaction.followup.send(
f"❌ Failed to load server status: {str(e)}", ephemeral=True
)
@bot.tree.command(
name="refresh_embeds", description="Refresh all server status embeds (admin only)"
)
async def refresh_embeds(interaction: discord.Interaction) -> None:
"""Slash command to refresh all server embeds."""
if not await check_allowed_guild(interaction, config):
return
logger.info(f"Refresh embeds command invoked by {interaction.user.name}")
await interaction.response.defer(ephemeral=True)
# Require administrator permissions
if not interaction.user.guild_permissions.administrator:
logger.warning(f"Unauthorized refresh attempt by {interaction.user.name}")
await interaction.followup.send(
"You need administrator permissions to refresh all embeds.", ephemeral=True
)
return
try:
logger.info("Starting full embed refresh per admin request")
deleted, created = await bot.embed_manager.refresh_all_embeds()
await interaction.followup.send(
f"Refreshed all embeds. Deleted {deleted} old embeds, created {created} new ones.",
ephemeral=True,
)
logger.info(f"Embed refresh completed: {deleted} deleted, {created} created")
except Exception as e:
logger.error(f"Embed refresh failed: {str(e)}")
await interaction.followup.send(f"Failed to refresh embeds: {str(e)}", ephemeral=True)
@bot.tree.command(
name="purge_embeds",
description="Permanently delete all server status embeds (admin only)",
)
async def purge_embeds(interaction: discord.Interaction) -> None:
"""
Slash command to permanently purge all server status embeds from Discord channels.
Workflow:
1. Validates administrator permissions
2. Checks if any embeds are currently tracked
3. Sends initial progress embed
4. Processes each embed sequentially with error handling
5. Updates progress embed in real-time
6. Saves cleared tracking data to disk
7. Sends final results with comprehensive statistics
"""
if not await check_allowed_guild(interaction, config):
return
logger.info(f"Purge embeds command invoked by {interaction.user.name}")
await interaction.response.defer(ephemeral=True)
# Require administrator permissions
if not interaction.user.guild_permissions.administrator:
logger.warning(f"Unauthorized purge attempt by {interaction.user.name}")
await interaction.followup.send(
"You need administrator permissions to purge all embeds.", ephemeral=True
)
return
try:
logger.info("Starting embed purge per admin request")
deleted_count = 0
not_found_count = 0
error_count = 0
total_embeds = len(bot.embed_manager.embed_locations)
if total_embeds == 0:
await interaction.followup.send(
"No embeds are currently being tracked. Nothing to purge.", ephemeral=True
)
return
# Create progress embed
progress_embed = discord.Embed(
title="🔄 Purging Server Embeds",
description=f"Processing {total_embeds} embeds...",
color=discord.Color.orange(),
)
progress_embed.add_field(name="Deleted", value="0", inline=True)
progress_embed.add_field(name="Not Found", value="0", inline=True)
progress_embed.add_field(name="Errors", value="0", inline=True)
progress_embed.set_footer(text="This may take a while...")
progress_message = await interaction.followup.send(embed=progress_embed, ephemeral=True)
# Process each tracked embed
for server_id, location in list(bot.embed_manager.embed_locations.items()):
try:
channel = bot.get_channel(int(location["channel_id"]))
if channel:
try:
message = await channel.fetch_message(int(location["message_id"]))
await message.delete()
deleted_count += 1
logger.debug(f"Successfully purged embed for server {server_id}")
except discord.NotFound:
not_found_count += 1
logger.debug(f"Embed for server {server_id} already deleted")
except discord.Forbidden:
error_count += 1
logger.error(f"Permission denied when deleting embed for server {server_id}")
except Exception as e:
error_count += 1
logger.error(f"Error deleting embed for server {server_id}: {str(e)}")
else:
not_found_count += 1
logger.warning(f"Channel not found for server {server_id}")
# Remove from tracking immediately
await bot.embed_manager.untrack_embed(server_id)
# Update progress every 5 embeds or for the last one
processed = deleted_count + not_found_count + error_count
if processed % 5 == 0 or processed == total_embeds:
progress_embed.description = f"Processed {processed}/{total_embeds} embeds"
progress_embed.set_field_at(0, name="Deleted", value=str(deleted_count), inline=True)
progress_embed.set_field_at(1, name="Not Found", value=str(not_found_count), inline=True)
progress_embed.set_field_at(2, name="Errors", value=str(error_count), inline=True)
await progress_message.edit(embed=progress_embed)
# Small delay to avoid rate limits
await asyncio.sleep(0.3)
except Exception as e:
error_count += 1
logger.error(f"Unexpected error processing server {server_id}: {str(e)}")
# Create results embed
result_embed = discord.Embed(
title="✅ Embed Purge Complete",
color=discord.Color.green(),
timestamp=datetime.now(),
)
result_embed.add_field(name="Total Tracked", value=str(total_embeds), inline=True)
result_embed.add_field(name="✅ Successfully Deleted", value=str(deleted_count), inline=True)
result_embed.add_field(name="❌ Already Missing", value=str(not_found_count), inline=True)
result_embed.add_field(name="⚠️ Errors", value=str(error_count), inline=True)
result_embed.add_field(
name="📊 Success Rate",
value=f"{((deleted_count + not_found_count) / total_embeds * 100):.1f}%",
inline=True,
)
result_embed.set_footer(text="Embed tracking file has been cleared")
await progress_message.edit(embed=result_embed)
logger.info(f"Embed purge completed: {deleted_count} deleted, {not_found_count} not found, {error_count} errors")
except Exception as e:
logger.error(f"Embed purge failed: {str(e)}")
await interaction.followup.send(f"❌ Failed to purge embeds: {str(e)}", ephemeral=True)
# ==============================================
# SYSTEM SIGNAL HANDLERS
# ==============================================
def handle_sigint(signum: int, frame: types.FrameType) -> None:
"""
Handle SIGINT signals (Ctrl+C) by initiating graceful shutdown.
Args:
signum: The signal number (signal.SIGINT)
frame: Current stack frame (unused but required by signal handler signature)
"""
logger.info("Received SIGINT (Ctrl+C), initiating graceful shutdown...")
raise KeyboardInterrupt
def handle_sigterm(signum: int, frame: types.FrameType) -> None:
"""
Handle SIGTERM signals (container stop) by initiating graceful shutdown.
Args:
signum: The signal number (signal.SIGTERM)
frame: Current stack frame (unused but required by signal handler signature)
"""
logger.info("Received SIGTERM (container stop), initiating graceful shutdown...")
raise KeyboardInterrupt
# ==============================================
# BOT STARTUP
# ==============================================
async def run_bot(config: Config) -> None:
"""
Main async entry point for the bot application.
Args:
config: Validated application configuration
"""
logger.info("Starting bot initialization")
# Create bot instance
intents = discord.Intents.default()
intents.message_content = True
bot = PterodactylBot(config, command_prefix="!", intents=intents)
# Set up commands and events
await setup_commands(bot)
# Register signal handlers
signal.signal(signal.SIGINT, handle_sigint)
signal.signal(signal.SIGTERM, handle_sigterm)
logger.info("System signal handlers registered")
try:
await bot.start(config.discord_token)
except KeyboardInterrupt:
logger.info("Bot shutting down")
except Exception as e:
logger.error(f"Bot crashed: {str(e)}")
sys.exit(1)
finally:
logger.info("Bot shutdown initiated - performing cleanup")
await bot.close()
logger.info("Bot shutdown complete")
+207
View File
@@ -0,0 +1,207 @@
"""
Configuration Management Module
Handles loading, validation, and access to application configuration.
Supports both config.ini files and environment variable overrides.
"""
import configparser
import logging
import os
from pathlib import Path
from typing import Optional
from exceptions import ConfigValidationError
logger = logging.getLogger("pterodisbot")
class Config:
"""
Application configuration manager.
Loads configuration from config.ini file, validates all required values,
and provides typed access to configuration parameters.
Supports both Pterodactyl (ptlc_/ptla_) and Pelican Panel (pacc_/papp_)
API key formats.
Attributes:
pterodactyl_url: URL of the panel (must include http:// or https://)
client_api_key: Client API key for server operations
application_api_key: Application API key for admin operations
discord_token: Discord bot token
allowed_guild_id: Discord server ID where bot operates
required_role: Role name required for server control permissions
update_interval: Background task update interval in seconds
embed_storage_path: Path to embed locations JSON file
"""
def __init__(self, config_path: str = "config.ini") -> None:
"""
Initialize configuration from file.
Args:
config_path: Path to config.ini file
Raises:
ConfigValidationError: If required values are missing or invalid
"""
self._config = configparser.ConfigParser()
self._config_path = Path(config_path)
if not self._config_path.exists():
raise ConfigValidationError(f"Configuration file not found: {config_path}")
self._config.read(self._config_path)
self._validate()
# Validated configuration values
self.pterodactyl_url: str = self._config.get("Pterodactyl", "PanelURL").rstrip("/")
self.client_api_key: str = self._config.get("Pterodactyl", "ClientAPIKey")
self.application_api_key: str = self._config.get("Pterodactyl", "ApplicationAPIKey")
self.discord_token: str = self._config.get("Discord", "Token")
self.allowed_guild_id: int = self._config.getint("Discord", "AllowedGuildID")
self.required_role: str = "Game Server User"
self.update_interval: int = 10
self.embed_storage_path: Path = Path("./embed/embed_locations.json")
logger.info("Configuration loaded and validated successfully")
def _validate(self) -> None:
"""
Validate all required configuration values.
Checks:
- Required sections exist
- All required keys are present
- AllowedGuildID is a positive integer
- API keys have correct prefixes (ptlc_/pacc_ and ptla_/papp_)
- PanelURL has valid protocol
Raises:
ConfigValidationError: If any validation fails
"""
errors = []
# Validate Pterodactyl section
if not self._config.has_section("Pterodactyl"):
errors.append("Missing [Pterodactyl] section in config.ini")
else:
required_ptero = ["PanelURL", "ClientAPIKey", "ApplicationAPIKey"]
for key in required_ptero:
if not self._config.get("Pterodactyl", key, fallback=None):
errors.append(f"Missing required Pterodactyl config value: {key}")
# Validate Discord section
if not self._config.has_section("Discord"):
errors.append("Missing [Discord] section in config.ini")
else:
required_discord = ["Token", "AllowedGuildID"]
for key in required_discord:
if not self._config.get("Discord", key, fallback=None):
errors.append(f"Missing required Discord config value: {key}")
# Validate AllowedGuildID is a valid integer
try:
guild_id = self._config.getint("Discord", "AllowedGuildID", fallback=0)
if guild_id <= 0:
errors.append("AllowedGuildID must be a positive integer")
except ValueError:
errors.append("AllowedGuildID must be a valid integer")
# Validate API keys have correct prefixes (Pterodactyl and Pelican Panel)
client_key = self._config.get("Pterodactyl", "ClientAPIKey", fallback="")
if client_key and not (client_key.startswith("ptlc_") or client_key.startswith("pacc_")):
errors.append("ClientAPIKey should start with 'ptlc_' (Pterodactyl) or 'pacc_' (Pelican Panel)")
app_key = self._config.get("Pterodactyl", "ApplicationAPIKey", fallback="")
if app_key and not (app_key.startswith("ptla_") or app_key.startswith("papp_")):
errors.append("ApplicationAPIKey should start with 'ptla_' (Pterodactyl) or 'papp_' (Pelican Panel)")
# Validate PanelURL is a valid URL
panel_url = self._config.get("Pterodactyl", "PanelURL", fallback="")
if panel_url and not (panel_url.startswith("http://") or panel_url.startswith("https://")):
errors.append("PanelURL must start with http:// or https://")
if errors:
error_msg = "Configuration validation failed:\n- " + "\n- ".join(errors)
logger.error(error_msg)
raise ConfigValidationError(error_msg)
logger.info("Configuration validation passed")
@property
def is_pelican_panel(self) -> bool:
"""Check if using Pelican Panel API keys (pacc_/papp_ prefixes)."""
return (
self.client_api_key.startswith("pacc_")
or self.application_api_key.startswith("papp_")
)
@property
def panel_type(self) -> str:
"""Return the detected panel type."""
return "Pelican Panel" if self.is_pelican_panel else "Pterodactyl"
def get_panel_display_name(self) -> str:
"""Get display name for the panel type."""
return "Pelican Panel" if self.is_pelican_panel else "Pterodactyl"
@classmethod
def get_instance(cls) -> Optional["Config"]:
"""Get the singleton config instance."""
return cls._instance
# Singleton instance for access from views
Config._instance = None
def load_config(config_path: str = "config.ini") -> Config:
"""
Load and validate configuration from file.
Args:
config_path: Path to config.ini file
Returns:
Config: Validated configuration instance
Raises:
ConfigValidationError: If configuration is invalid
"""
config = Config(config_path)
Config._instance = config
return config
def generate_config_from_env(output_path: str = "config.ini") -> None:
"""
Generate config.ini from environment variables.
Useful for Docker/container deployments where config comes from env vars.
Required environment variables:
DISCORD_TOKEN, ALLOWED_GUILD_ID, PANEL_URL,
CLIENT_API_KEY, APPLICATION_API_KEY
Args:
output_path: Path to write generated config.ini
"""
config = configparser.ConfigParser()
config["Pterodactyl"] = {
"PanelURL": os.environ.get("PANEL_URL", ""),
"ClientAPIKey": os.environ.get("CLIENT_API_KEY", ""),
"ApplicationAPIKey": os.environ.get("APPLICATION_API_KEY", ""),
}
config["Discord"] = {
"Token": os.environ.get("DISCORD_TOKEN", ""),
"AllowedGuildID": os.environ.get("ALLOWED_GUILD_ID", ""),
}
with open(output_path, "w") as f:
config.write(f)
logger.info(f"Generated config.ini from environment variables at {output_path}")
+2 -2
View File
@@ -1,5 +1,5 @@
# Builder stage using slim Python image
FROM python:3.11-slim as builder
FROM python:3.12-slim as builder
# Install system dependencies for building
RUN apt-get update && apt-get install -y \
@@ -19,7 +19,7 @@ RUN --mount=type=cache,target=/root/.cache/pip \
pip install --no-cache-dir -r requirements.txt
# Final stage - using slim
FROM python:3.11-slim
FROM python:3.12-slim
# Install minimal runtime dependencies
RUN apt-get update && apt-get install -y --no-install-recommends \
+424
View File
@@ -0,0 +1,424 @@
"""
Embed Management Module
Handles creation, tracking, and persistence of Discord server status embeds.
Manages embed locations storage and provides utilities for embed lifecycle.
"""
import json
import logging
from datetime import datetime
from pathlib import Path
from typing import Any, Dict, List, Optional, Tuple
import discord
from api import PanelAPI
from config import Config
from exceptions import EmbedNotFoundError
from metrics import ServerMetricsManager
from models import ServerData, ServerResources
logger = logging.getLogger("pterodisbot")
class EmbedManager:
"""
Manages server status embeds lifecycle.
Handles:
- Embed creation with real-time server data
- Persistent storage of embed locations (channel_id, message_id)
- Embed refresh and recreation
- Cleanup of deleted/missing embeds
- Rate limit protection during bulk operations
Attributes:
config: Application configuration
api: Panel API client
metrics_manager: Metrics manager for graphs
storage_path: Path to embed locations JSON file
"""
def __init__(
self,
config: Config,
api: PanelAPI,
metrics_manager: ServerMetricsManager,
) -> None:
"""
Initialize the embed manager.
Args:
config: Validated application configuration
api: Initialized PanelAPI instance
metrics_manager: ServerMetricsManager instance
"""
self.config = config
self.api = api
self.metrics_manager = metrics_manager
self.storage_path = config.embed_storage_path
self.embed_locations: Dict[str, Dict[str, int]] = {}
self.previous_states: Dict[str, Tuple[str, float, Optional[float]]] = {}
# Ensure storage directory exists
self.storage_path.parent.mkdir(parents=True, exist_ok=True)
logger.info("Initialized EmbedManager")
async def load_locations(self) -> None:
"""Load saved embed locations from JSON storage file."""
logger.debug("Loading embed locations from storage")
if not self.storage_path.exists():
logger.info("No existing embed locations file found")
return
try:
with open(self.storage_path, "r") as f:
self.embed_locations = json.load(f)
logger.info(f"Loaded {len(self.embed_locations)} embed locations from storage")
except Exception as e:
logger.error(f"Failed to load embed locations: {str(e)}")
async def save_locations(self) -> None:
"""Save current embed locations to JSON storage file."""
logger.debug("Saving embed locations to storage")
try:
with open(self.storage_path, "w") as f:
json.dump(self.embed_locations, f, indent=2)
logger.debug("Successfully saved embed locations")
except Exception as e:
logger.error(f"Failed to save embed locations: {str(e)}")
async def track_embed(self, server_id: str, message: discord.Message) -> None:
"""
Track a newly created embed in storage.
Args:
server_id: The server's panel identifier
message: Discord message containing the embed
"""
logger.debug(f"Tracking new embed for server {server_id} in channel {message.channel.id}")
self.embed_locations[server_id] = {
"channel_id": str(message.channel.id),
"message_id": str(message.id),
}
await self.save_locations()
async def untrack_embed(self, server_id: str) -> None:
"""
Remove a server's embed from tracking.
Args:
server_id: The server's panel identifier
"""
self.embed_locations.pop(server_id, None)
self.previous_states.pop(server_id, None)
await self.save_locations()
async def create_server_embed(
self, server_data: ServerData, resources: ServerResources
) -> Tuple[discord.Embed, discord.ui.View]:
"""
Create a status embed and interactive view for a server.
Args:
server_data: Server information from panel
resources: Current resource usage data
Returns:
Tuple of (embed, view) ready for display
"""
attributes = server_data.attributes
identifier = attributes.identifier
name = attributes.name
description = attributes.description or "No description available"
logger.debug(f"Building status embed for server {name} ({identifier})")
# Parse resource data
current_state = resources.attributes.current_state.title()
is_suspended = attributes.suspended
# Create embed with appropriate color based on status
embed = discord.Embed(
title=f"{name} - {current_state}",
description=description,
color=(
discord.Color.blue()
if current_state.lower() == "running"
else discord.Color.red()
),
timestamp=datetime.now(),
)
embed.add_field(name="🆔 Server ID", value=f"`{identifier}`", inline=True)
if is_suspended:
embed.add_field(name="ℹ️ Status", value="⛔ `Suspended`", inline=True)
else:
embed.add_field(name="ℹ️ Status", value="✅ `Active`", inline=True)
# Add resource usage if server is running
if current_state.lower() != "offline":
embed = self._add_resource_fields(embed, resources, attributes)
embed = self._add_graph_fields(embed, identifier)
embed.set_footer(text="Last updated")
# Create interactive view with control buttons
from views import ServerStatusView
view = ServerStatusView(
server_id=identifier,
server_name=name,
api=self.api,
server_data=server_data.model_dump(),
)
logger.debug(f"Successfully built status components for {name}")
return embed, view
def _add_resource_fields(
self, embed: discord.Embed, resources: ServerResources, attributes: Any
) -> discord.Embed:
"""Add resource usage fields to embed."""
resource_attrs = resources.attributes.resources
limits = attributes.limits
# Current usage
cpu_usage = round(resource_attrs.cpu_absolute, 2)
memory_usage = round(resource_attrs.memory_bytes / (1024**2), 2)
disk_usage = round(resource_attrs.disk_bytes / (1024**2), 2)
network_rx = round(resource_attrs.network_rx_bytes / (1024**2), 2)
network_tx = round(resource_attrs.network_tx_bytes / (1024**2), 2)
# Maximum allocated resources
cpu_limit = limits.cpu
memory_limit = limits.memory
disk_limit = limits.disk
# Format limit values - display ∞ for unlimited (0 limit)
def format_limit(value: int, unit: str = "") -> str:
if value == 0:
return f"{'∞':>8} {unit}"
return f"{value:>8} {unit}"
# Format uptime
uptime_ms = resource_attrs.uptime
if uptime_ms > 0:
uptime_seconds = uptime_ms // 1000
if uptime_seconds < 60:
uptime_text = f"`{uptime_seconds}s`"
elif uptime_seconds < 3600:
uptime_text = f"`{uptime_seconds // 60}m {uptime_seconds % 60}s`"
elif uptime_seconds < 86400:
hours = uptime_seconds // 3600
minutes = (uptime_seconds % 3600) // 60
uptime_text = f"`{hours}h {minutes}m`"
else:
days = uptime_seconds // 86400
hours = (uptime_seconds % 86400) // 3600
uptime_text = f"`{days}d {hours}h`"
else:
uptime_text = "`Just started`"
embed.add_field(name="⏱️ Uptime", value=uptime_text, inline=True)
# Resource usage text box
usage_text = (
f"```properties\n"
f"CPU : [{cpu_usage:>8} / {format_limit(cpu_limit, '%')}\n"
f"Memory : [{memory_usage:>8} / {format_limit(memory_limit, 'MiB')}\n"
f"Disk : [{disk_usage:>8} / {format_limit(disk_limit, 'MiB')}\n"
f"```"
)
embed.add_field(name="📊 Resource Usage", value=usage_text, inline=False)
embed.add_field(name="Network In", value=f"📥 `{network_rx} MiB`", inline=True)
embed.add_field(name="Network Out", value=f"📤 `{network_tx} MiB`", inline=True)
return embed
def _add_graph_fields(
self, embed: discord.Embed, server_id: str
) -> discord.Embed:
"""Add metrics graph fields to embed if data available."""
server_graphs = self.metrics_manager.get_server_graphs(server_id)
if server_graphs and server_graphs.has_sufficient_data:
summary = server_graphs.get_data_summary()
graph_description = (
f">>> `Data points: {summary['point_count']}/6`\n"
f"`CPU trend: {summary['cpu_trend']} • Memory trend: {summary['memory_trend']}`"
)
embed.add_field(
name="📈 Usage Trends (Last Minute)",
value=graph_description,
inline=False,
)
embed.set_image(url=f"attachment://metrics_graph_{server_id}.png")
return embed
async def refresh_all_embeds(self) -> Tuple[int, int]:
"""
Perform a complete refresh of all server status embeds.
Creates new embeds and deletes old ones to prevent duplication.
Returns:
Tuple of (deleted_count, created_count)
"""
logger.info("Starting full refresh of all server embeds")
from bot import PterodactylBot # Avoid circular import
bot = PterodactylBot.get_instance()
deleted_count = 0
created_count = 0
skipped_count = 0
try:
# Get current server list
servers = await self.api.get_servers()
if not servers:
logger.warning("No servers found in panel during refresh")
return 0, 0
server_cache = {s["attributes"]["identifier"]: s for s in servers}
# Create new embeds
new_embeds = {}
for server_id, server_data in server_cache.items():
if server_id not in self.embed_locations:
skipped_count += 1
continue
channel_id = self.embed_locations[server_id]["channel_id"]
channel = bot.get_channel(int(channel_id))
if not channel:
logger.warning(f"Channel {channel_id} not found for server {server_id}")
continue
try:
resources = await self.api.get_server_resources(server_id)
server_obj = ServerData.model_validate(server_data)
resources_obj = ServerResources.model_validate(resources)
embed, view = await self.create_server_embed(server_obj, resources_obj)
message = await channel.send(embed=embed, view=view)
new_embeds[server_id] = {
"channel_id": str(channel.id),
"message_id": str(message.id),
}
created_count += 1
logger.info(f"Created new embed for server {server_data['attributes']['name']}")
await asyncio.sleep(1) # Rate limit protection
except Exception as e:
logger.error(f"Failed to create new embed for server {server_id}: {str(e)}")
if not new_embeds:
logger.warning("No new embeds created during refresh")
return 0, 0
# Delete old embeds
for server_id, location in list(self.embed_locations.items()):
try:
channel = bot.get_channel(int(location["channel_id"]))
if channel:
try:
message = await channel.fetch_message(int(location["message_id"]))
await message.delete()
deleted_count += 1
logger.debug(f"Deleted old embed for server {server_id}")
await asyncio.sleep(0.5)
except discord.NotFound:
logger.debug(f"Old embed for server {server_id} already deleted")
except Exception as e:
logger.error(f"Failed to delete old embed for server {server_id}: {str(e)}")
except Exception as e:
logger.error(f"Error processing old embed for server {server_id}: {str(e)}")
logger.info(f"Deleted {deleted_count} old embeds, created {created_count} new embeds")
# Update storage
self.embed_locations = new_embeds
await self.save_locations()
return deleted_count, created_count
except Exception as e:
logger.error(f"Critical error during embed refresh: {str(e)}")
raise
async def update_single_embed(
self,
server_id: str,
location: Dict[str, str],
server_data: ServerData,
resources: ServerResources,
current_state: str,
cpu_usage: float,
) -> bool:
"""
Update a single server's status embed.
Args:
server_id: Server identifier
location: Embed location dict with channel_id and message_id
server_data: Current server data
resources: Current resource data
current_state: Current power state
cpu_usage: Current CPU usage
Returns:
True if embed was updated, False otherwise
"""
from bot import PterodactylBot
bot = PterodactylBot.get_instance()
try:
embed, view = await self.create_server_embed(server_data, resources)
channel = bot.get_channel(int(location["channel_id"]))
if not channel:
logger.warning(f"Channel {location['channel_id']} not found for server {server_id}")
return False
message = await channel.fetch_message(int(location["message_id"]))
# Handle graph attachments
files = []
server_graphs = self.metrics_manager.get_server_graphs(server_id)
if current_state == "running" and server_graphs and server_graphs.has_sufficient_data:
combined_graph = server_graphs.generate_combined_graph()
if combined_graph:
files.append(
discord.File(
combined_graph,
filename=f"metrics_graph_{server_id}.png",
)
)
# Empty files list removes existing attachments
await message.edit(embed=embed, view=view, attachments=files)
logger.debug(f"Updated status for {server_data.attributes.name}")
return True
except discord.NotFound:
logger.warning(f"Embed for server {server_id} not found, removing from tracking")
self.embed_locations.pop(server_id, None)
self.previous_states.pop(server_id, None)
await self.save_locations()
return False
except Exception as e:
logger.error(f"Failed to update status for server {server_id}: {str(e)}")
return False
# Import at bottom to avoid circular imports
import asyncio
Regular → Executable
+1 -1
View File
@@ -5,4 +5,4 @@ if ! python generate_config.py; then
exit 1
fi
exec python pterodisbot.py
exec python main.py
+84
View File
@@ -0,0 +1,84 @@
"""
Custom Exception Classes
Defines application-specific exceptions for better error handling
and clearer error messages throughout the codebase.
"""
class PterodisBotError(Exception):
"""Base exception class for all PterodisBot errors."""
pass
class ConfigValidationError(PterodisBotError):
"""
Raised when configuration validation fails.
Contains detailed error messages for all validation failures.
"""
pass
class APIError(PterodisBotError):
"""
Raised when an API request fails.
Attributes:
endpoint: The API endpoint that failed
status_code: HTTP status code if available
message: Error message from the API
"""
def __init__(
self,
message: str,
endpoint: str = "",
status_code: int = 0,
) -> None:
super().__init__(message)
self.endpoint = endpoint
self.status_code = status_code
self.message = message
class ServerNotFoundError(PterodisBotError):
"""
Raised when a requested server is not found in the panel.
"""
pass
class PermissionError(PterodisBotError):
"""
Raised when a user lacks required permissions for an action.
"""
pass
class EmbedNotFoundError(PterodisBotError):
"""
Raised when a tracked embed message is not found in Discord.
"""
pass
class MetricsError(PterodisBotError):
"""
Raised when metrics graph generation fails.
"""
pass
class ConfigurationError(PterodisBotError):
"""
Raised when there's a configuration issue not caught during startup validation.
"""
pass
+70 -20
View File
@@ -3,6 +3,7 @@ import configparser
from pathlib import Path
from dotenv import load_dotenv
def validate_env_vars():
"""Validate that required environment variables are set and non-empty."""
required_vars = [
@@ -17,33 +18,82 @@ def validate_env_vars():
if missing_vars:
raise ValueError(f"Missing required environment variables: {', '.join(missing_vars)}")
def get_env_config():
"""Extract configuration from environment variables."""
return {
'Discord': {
'Token': os.getenv('DISCORD_TOKEN'),
'AllowedGuildID': os.getenv('ALLOWED_GUILD_ID')
},
'Pterodactyl': {
'PanelURL': os.getenv('PANEL_URL'),
'ClientAPIKey': os.getenv('CLIENT_API_KEY'),
'ApplicationAPIKey': os.getenv('APPLICATION_API_KEY')
}
}
def configs_match(existing_config, env_config):
"""Compare existing config with environment variables."""
for section, values in env_config.items():
if not existing_config.has_section(section):
return False
for key, env_value in values.items():
if env_value is None:
# Environment variable not set, don't require match
continue
existing_value = existing_config.get(section, key, fallback=None)
if existing_value != env_value:
return False
return True
def generate_config():
"""Generate config.ini from environment variables."""
# Validate environment variables first
validate_env_vars()
"""
Generate config.ini from environment variables if:
1. No config file exists, OR
2. Config file exists but values don't match environment variables
config = configparser.ConfigParser()
# Discord section - ensure all values are strings
config['Discord'] = {
'Token': str(os.getenv('DISCORD_TOKEN')),
'AllowedGuildID': str(os.getenv('ALLOWED_GUILD_ID'))
}
# Pterodactyl section - ensure all values are strings
config['Pterodactyl'] = {
'PanelURL': str(os.getenv('PANEL_URL')),
'ClientAPIKey': str(os.getenv('CLIENT_API_KEY')),
'ApplicationAPIKey': str(os.getenv('APPLICATION_API_KEY'))
}
# Write to file
Environment variables always take priority.
"""
config_path = Path(os.getenv('CONFIG_PATH', 'config.ini'))
# Check if config file exists
if not config_path.exists():
print(f"No config file found at {config_path}, generating from environment variables...")
validate_env_vars()
_write_config(config_path, get_env_config())
return
# Config file exists, read it and compare with environment
existing_config = configparser.ConfigParser()
existing_config.read(config_path)
env_config = get_env_config()
if not configs_match(existing_config, env_config):
print(f"Config file at {config_path} differs from environment variables, regenerating...")
# Validate that required env vars are present for regeneration
validate_env_vars()
_write_config(config_path, env_config)
return
print(f"Config file at {config_path} matches environment variables, using existing config.")
def _write_config(config_path, config_dict):
"""Write config dictionary to file."""
config = configparser.ConfigParser()
for section, values in config_dict.items():
config[section] = {k: str(v) for k, v in values.items() if v is not None}
config_path.parent.mkdir(exist_ok=True, parents=True)
with open(config_path, 'w') as configfile:
config.write(configfile)
print(f"Config file generated at {config_path}")
print(f"Config file written to {config_path}")
if __name__ == "__main__":
load_dotenv()
+66
View File
@@ -0,0 +1,66 @@
"""
Logging Configuration Module
Sets up application-wide logging with:
- Rotating file handler (5MB max, 3 backups) at DEBUG level
- Console handler at INFO level
- Consistent formatting for both handlers
"""
import logging
import os
from logging.handlers import RotatingFileHandler
from pathlib import Path
def setup_logging(log_dir: str = "logs", log_file: str = "pterodisbot.log") -> logging.Logger:
"""
Configure application-wide logging.
Args:
log_dir: Directory for log files (relative to this file)
log_file: Name of the log file
Returns:
Configured logger instance
"""
# Create logs directory if it doesn't exist
logs_path = Path(__file__).parent / log_dir
logs_path.mkdir(parents=True, exist_ok=True)
# Get or create the application logger
logger = logging.getLogger("pterodisbot")
logger.setLevel(logging.DEBUG)
# Avoid adding duplicate handlers if called multiple times
if logger.handlers:
return logger
# File handler for logs (rotates when reaching 5MB, keeps 3 backups)
file_handler = RotatingFileHandler(
filename=logs_path / log_file,
maxBytes=5 * 1024 * 1024, # 5 MiB max log file size
backupCount=3, # Rotate through 3 files
encoding="utf-8",
)
file_handler.setLevel(logging.DEBUG)
file_handler.setFormatter(
logging.Formatter("%(asctime)s - %(levelname)s - %(message)s")
)
logger.addHandler(file_handler)
# Console handler for real-time output (INFO level and above)
console_handler = logging.StreamHandler()
console_handler.setLevel(logging.INFO)
console_handler.setFormatter(
logging.Formatter("%(asctime)s - %(levelname)s - %(message)s")
)
logger.addHandler(console_handler)
logger.info("Initialized logging system with file and console output")
return logger
def get_logger() -> logging.Logger:
"""Get the configured logger instance."""
return logging.getLogger("pterodisbot")
+54
View File
@@ -0,0 +1,54 @@
#!/usr/bin/env python3
"""
Main entry point for the bot.
Loads configuration and starts the modular bot application.
"""
import asyncio
import logging
import sys
from bot import run_bot
from config import load_config, ConfigValidationError
from exceptions import PterodisBotError
from logging_config import setup_logging
# Configure application logging
setup_logging()
logger = logging.getLogger("pterodisbot")
def main() -> int:
"""
Synchronous entry point for the bot application.
Returns:
Exit code (0 for success, 1 for error)
"""
try:
# Load and validate configuration
logger.info("Loading configuration...")
config = load_config("config.ini")
logger.info(f"Configuration loaded - Panel type: {config.get_panel_display_name()}")
# Run the bot
asyncio.run(run_bot(config))
return 0
except ConfigValidationError as e:
logger.critical(f"Configuration error: {str(e)}")
return 1
except PterodisBotError as e:
logger.critical(f"Application error: {str(e)}")
return 1
except KeyboardInterrupt:
pass # Already logged by signal handler
except Exception as e:
logger.critical(f"Unexpected error during startup: {str(e)}", exc_info=True)
return 1
if __name__ == "__main__":
sys.exit(main())
+538
View File
@@ -0,0 +1,538 @@
"""
Server Metrics Graphs Module
Provides graphing capabilities for server CPU and memory usage.
Generates line graphs as PNG images for embedding in Discord messages.
"""
import io
import logging
import math
from collections import deque
from datetime import datetime
from typing import Any, Dict, Optional
import matplotlib
import matplotlib.dates as mdates
import matplotlib.pyplot as plt
matplotlib.use("Agg") # Use non-interactive backend for server environments
logger = logging.getLogger("pterodisbot")
class ServerMetricsGraphs:
"""
Manages CPU and memory usage graphs for individual servers.
Features:
- Stores last 6 data points (1 minute of history at 10-second intervals)
- Generates PNG images of line graphs for Discord embedding
- Automatic data rotation (FIFO queue with max 6 points)
- Separate tracking for CPU percentage and memory MB usage
- Dynamic CPU scaling in 100% increments for multi-vCPU servers
- Clean graph styling optimized for Discord dark theme
"""
def __init__(self, server_id: str, server_name: str) -> None:
"""
Initialize metrics tracking for a server.
Args:
server_id: Panel server identifier
server_name: Human-readable server name
"""
self.server_id = server_id
self.server_name = server_name
# Use deque with maxlen=6 for automatic FIFO rotation
# Each entry is a tuple: (timestamp, cpu_percent, memory_mb)
self.data_points = deque(maxlen=6)
# Track if we have enough data for meaningful graphs (at least 2 points)
self.has_sufficient_data = False
logger.debug(f"Initialized metrics tracking for server {server_name} ({server_id})")
def add_data_point(
self, cpu_percent: float, memory_mb: float, timestamp: Optional[datetime] = None
) -> None:
"""
Add a new data point to the metrics history.
Args:
cpu_percent: Current CPU usage percentage
memory_mb: Current memory usage in megabytes
timestamp: Optional timestamp, defaults to current time
"""
if timestamp is None:
timestamp = datetime.now()
# Add new data point (automatically rotates old data due to maxlen=6)
self.data_points.append((timestamp, cpu_percent, memory_mb))
# Update sufficient data flag
self.has_sufficient_data = len(self.data_points) >= 2
logger.debug(
f"Added metrics data point for {self.server_name}: "
f"CPU={cpu_percent}%, Memory={memory_mb}MB"
)
def _calculate_cpu_scale_limit(self, max_cpu_value: float) -> int:
"""
Calculate appropriate CPU scale limit in 100% increments.
Args:
max_cpu_value: Maximum CPU value in the dataset
Returns:
Scale limit rounded up to nearest 100% increment
"""
if max_cpu_value <= 100:
return 100
# Round up to nearest 100% increment
# e.g., 150% -> 200%, 250% -> 300%, 350% -> 400%
return math.ceil(max_cpu_value / 100) * 100
def generate_cpu_graph(self) -> Optional[io.BytesIO]:
"""
Generate a CPU usage line graph as a PNG image.
Returns:
BytesIO object containing PNG image data, or None if insufficient data
"""
if not self.has_sufficient_data:
logger.debug(f"Insufficient data for CPU graph generation: {self.server_name}")
return None
try:
# Extract timestamps and CPU data
timestamps = [point[0] for point in self.data_points]
cpu_values = [point[1] for point in self.data_points]
# Calculate dynamic CPU scale limit
max_cpu = max(cpu_values)
cpu_scale_limit = self._calculate_cpu_scale_limit(max_cpu)
# Create figure with dark theme styling
plt.style.use("dark_background")
fig, ax = plt.subplots(figsize=(8, 4), dpi=100)
fig.patch.set_facecolor("#2f3136") # Discord dark theme background
ax.set_facecolor("#36393f") # Slightly lighter for graph area
# Plot CPU line with gradient fill
ax.fill_between(timestamps, cpu_values, alpha=0.3, color="#7289da")
# Customize axes with dynamic scaling
ax.set_ylabel("CPU Usage (%)", color="#ffffff", fontsize=10)
ax.set_ylim(0, cpu_scale_limit)
# Add horizontal grid lines at 100% increments for better readability
for i in range(100, cpu_scale_limit + 1, 100):
ax.axhline(y=i, color="#ffffff", alpha=0.2, linestyle="--", linewidth=0.8)
# Format time axis
ax.xaxis.set_major_formatter(mdates.DateFormatter("%H:%M:%S"))
ax.xaxis.set_major_locator(mdates.SecondLocator(interval=20))
plt.setp(
ax.xaxis.get_majorticklabels(),
rotation=45,
ha="right",
color="#ffffff",
fontsize=8,
)
# Style the graph
ax.tick_params(colors="#ffffff", labelsize=8)
ax.grid(True, alpha=0.3, color="#ffffff")
ax.spines["bottom"].set_color("#ffffff")
ax.spines["left"].set_color("#ffffff")
ax.spines["top"].set_visible(False)
ax.spines["right"].set_visible(False)
# Add title with scale info for multi-vCPU servers
title = f"{self.server_name} - CPU Usage"
if cpu_scale_limit > 100:
estimated_vcpus = cpu_scale_limit // 100
title += f" (~{estimated_vcpus} vCPU cores)"
ax.set_title(title, color="#ffffff", fontsize=12, pad=20)
# Tight layout to prevent label cutoff
plt.tight_layout()
# Save to BytesIO
img_buffer = io.BytesIO()
plt.savefig(
img_buffer,
format="png",
facecolor="#2f3136",
edgecolor="none",
bbox_inches="tight",
dpi=100,
)
img_buffer.seek(0)
# Clean up matplotlib resources
plt.close(fig)
logger.debug(f"Generated CPU graph for {self.server_name} (scale: 0-{cpu_scale_limit}%)")
return img_buffer
except Exception as e:
logger.error(f"Failed to generate CPU graph for {self.server_name}: {str(e)}")
plt.close("all") # Clean up any remaining figures
return None
def generate_memory_graph(self) -> Optional[io.BytesIO]:
"""
Generate a memory usage line graph as a PNG image.
Returns:
BytesIO object containing PNG image data, or None if insufficient data
"""
if not self.has_sufficient_data:
logger.debug(f"Insufficient data for memory graph generation: {self.server_name}")
return None
try:
# Extract timestamps and memory data
timestamps = [point[0] for point in self.data_points]
memory_values = [point[2] for point in self.data_points]
# Create figure with dark theme styling
plt.style.use("dark_background")
fig, ax = plt.subplots(figsize=(8, 4), dpi=100)
fig.patch.set_facecolor("#2f3136") # Discord dark theme background
ax.set_facecolor("#36393f") # Slightly lighter for graph area
# Plot memory line with gradient fill
ax.fill_between(timestamps, memory_values, alpha=0.3, color="#43b581")
# Customize axes
ax.set_ylabel("Memory Usage (MB)", color="#ffffff", fontsize=10)
ax.set_ylim(0, max(memory_values) * 1.1) # Dynamic scaling with 10% padding
# Format time axis
ax.xaxis.set_major_formatter(mdates.DateFormatter("%H:%M:%S"))
ax.xaxis.set_major_locator(mdates.SecondLocator(interval=20))
plt.setp(
ax.xaxis.get_majorticklabels(),
rotation=45,
ha="right",
color="#ffffff",
fontsize=8,
)
# Style the graph
ax.tick_params(colors="#ffffff", labelsize=8)
ax.grid(True, alpha=0.3, color="#ffffff")
ax.spines["bottom"].set_color("#ffffff")
ax.spines["left"].set_color("#ffffff")
ax.spines["top"].set_visible(False)
ax.spines["right"].set_visible(False)
# Add title
ax.set_title(
f"{self.server_name} - Memory Usage",
color="#ffffff",
fontsize=12,
pad=20,
)
# Tight layout to prevent label cutoff
plt.tight_layout()
# Save to BytesIO
img_buffer = io.BytesIO()
plt.savefig(
img_buffer,
format="png",
facecolor="#2f3136",
edgecolor="none",
bbox_inches="tight",
dpi=100,
)
img_buffer.seek(0)
# Clean up matplotlib resources
plt.close(fig)
logger.debug(f"Generated memory graph for {self.server_name}")
return img_buffer
except Exception as e:
logger.error(f"Failed to generate memory graph for {self.server_name}: {str(e)}")
plt.close("all") # Clean up any remaining figures
return None
def generate_combined_graph(self) -> Optional[io.BytesIO]:
"""
Generate a combined CPU and memory usage graph as a PNG image.
Returns:
BytesIO object containing PNG image data, or None if insufficient data
"""
if not self.has_sufficient_data:
logger.debug(f"Insufficient data for combined graph generation: {self.server_name}")
return None
try:
# Extract data
timestamps = [point[0] for point in self.data_points]
cpu_values = [point[1] for point in self.data_points]
memory_values = [point[2] for point in self.data_points]
# Calculate dynamic CPU scale limit
max_cpu = max(cpu_values)
cpu_scale_limit = self._calculate_cpu_scale_limit(max_cpu)
# Create figure with two subplots
plt.style.use("dark_background")
fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(8, 6), dpi=100, sharex=True)
fig.patch.set_facecolor("#2f3136")
# CPU subplot
ax1.set_facecolor("#36393f")
ax1.plot(
timestamps,
cpu_values,
color="#7289da",
linewidth=2.5,
marker="o",
markersize=4,
)
ax1.fill_between(timestamps, cpu_values, alpha=0.3, color="#7289da")
ax1.set_ylabel("CPU Usage (%)", color="#ffffff", fontsize=10)
ax1.set_ylim(0, cpu_scale_limit)
ax1.tick_params(colors="#ffffff", labelsize=8)
ax1.grid(True, alpha=0.3, color="#ffffff")
# Add horizontal grid lines at 100% increments for CPU subplot
for i in range(100, cpu_scale_limit + 1, 100):
ax1.axhline(y=i, color="#ffffff", alpha=0.2, linestyle="--", linewidth=0.8)
# Title with vCPU info if applicable
title = f"{self.server_name} - Resource Usage"
if cpu_scale_limit > 100:
estimated_vcpus = cpu_scale_limit // 100
title += f" (~{estimated_vcpus} vCPU cores)"
ax1.set_title(title, color="#ffffff", fontsize=12)
# Memory subplot
ax2.set_facecolor("#36393f")
ax2.plot(
timestamps,
memory_values,
color="#43b581",
linewidth=2.5,
marker="o",
markersize=4,
)
ax2.fill_between(timestamps, memory_values, alpha=0.3, color="#43b581")
ax2.set_ylabel("Memory (MB)", color="#ffffff", fontsize=10)
ax2.set_ylim(0, max(memory_values) * 1.1)
ax2.tick_params(colors="#ffffff", labelsize=8)
ax2.grid(True, alpha=0.3, color="#ffffff")
# Format time axis (only on bottom subplot)
ax2.xaxis.set_major_formatter(mdates.DateFormatter("%H:%M:%S"))
ax2.xaxis.set_major_locator(mdates.SecondLocator(interval=20))
plt.setp(
ax2.xaxis.get_majorticklabels(),
rotation=45,
ha="right",
color="#ffffff",
fontsize=8,
)
# Style both subplots
for ax in [ax1, ax2]:
ax.spines["bottom"].set_color("#ffffff")
ax.spines["left"].set_color("#ffffff")
ax.spines["top"].set_visible(False)
ax.spines["right"].set_visible(False)
plt.tight_layout()
# Save to BytesIO
img_buffer = io.BytesIO()
plt.savefig(
img_buffer,
format="png",
facecolor="#2f3136",
edgecolor="none",
bbox_inches="tight",
dpi=100,
)
img_buffer.seek(0)
plt.close(fig)
logger.debug(f"Generated combined graph for {self.server_name} (CPU scale: 0-{cpu_scale_limit}%)")
return img_buffer
except Exception as e:
logger.error(f"Failed to generate combined graph for {self.server_name}: {str(e)}")
plt.close("all")
return None
def get_data_summary(self) -> Dict[str, Any]:
"""
Get summary statistics for the current data points.
Returns:
Dictionary containing data point count, latest values, and trends
"""
if not self.data_points:
return {
"point_count": 0,
"has_data": False,
"latest_cpu": 0,
"latest_memory": 0,
}
# Get latest values
latest_point = self.data_points[-1]
latest_cpu = latest_point[1]
latest_memory = latest_point[2]
# Calculate CPU scale info
max_cpu = max(point[1] for point in self.data_points)
cpu_scale_limit = self._calculate_cpu_scale_limit(max_cpu)
estimated_vcpus = cpu_scale_limit // 100
# Calculate trends if we have multiple points
cpu_trend = "stable"
memory_trend = "stable"
if len(self.data_points) >= 2:
first_point = self.data_points[0]
cpu_change = latest_cpu - first_point[1]
memory_change = latest_memory - first_point[2]
# Determine trends (>5% change considered significant)
if abs(cpu_change) > 5:
cpu_trend = "increasing" if cpu_change > 0 else "decreasing"
if abs(memory_change) > 50: # 50MB change threshold
memory_trend = "increasing" if memory_change > 0 else "decreasing"
return {
"point_count": len(self.data_points),
"has_data": self.has_sufficient_data,
"latest_cpu": latest_cpu,
"latest_memory": latest_memory,
"cpu_trend": cpu_trend,
"memory_trend": memory_trend,
"cpu_scale_limit": cpu_scale_limit,
"estimated_vcpus": estimated_vcpus,
"time_span_minutes": len(self.data_points) * 10 / 60, # Convert to minutes
}
class ServerMetricsManager:
"""
Global manager for all server metrics graphs.
Handles:
- Creation and cleanup of ServerMetricsGraphs instances
- Bulk operations across all tracked servers
- Memory management for graph storage
"""
def __init__(self) -> None:
"""Initialize the metrics manager."""
self.server_graphs: Dict[str, ServerMetricsGraphs] = {}
logger.info("Initialized ServerMetricsManager")
def get_or_create_server_graphs(self, server_id: str, server_name: str) -> ServerMetricsGraphs:
"""
Get existing ServerMetricsGraphs instance or create a new one.
Args:
server_id: Panel server identifier
server_name: Human-readable server name
Returns:
ServerMetricsGraphs instance for the specified server
"""
if server_id not in self.server_graphs:
self.server_graphs[server_id] = ServerMetricsGraphs(server_id, server_name)
logger.debug(f"Created new metrics graphs for server {server_name}")
return self.server_graphs[server_id]
def add_server_data(
self, server_id: str, server_name: str, cpu_percent: float, memory_mb: float
) -> None:
"""
Add data point to a server's metrics tracking.
Args:
server_id: Panel server identifier
server_name: Human-readable server name
cpu_percent: Current CPU usage percentage
memory_mb: Current memory usage in megabytes
"""
graphs = self.get_or_create_server_graphs(server_id, server_name)
graphs.add_data_point(cpu_percent, memory_mb)
def remove_server(self, server_id: str) -> None:
"""
Remove a server from metrics tracking.
Args:
server_id: Panel server identifier to remove
"""
if server_id in self.server_graphs:
del self.server_graphs[server_id]
logger.debug(f"Removed metrics tracking for server {server_id}")
def get_server_graphs(self, server_id: str) -> Optional[ServerMetricsGraphs]:
"""
Get ServerMetricsGraphs instance for a specific server.
Args:
server_id: Panel server identifier
Returns:
ServerMetricsGraphs instance or None if not found
"""
return self.server_graphs.get(server_id)
def cleanup_old_servers(self, active_server_ids: list) -> None:
"""
Remove tracking for servers that no longer exist.
Args:
active_server_ids: List of currently active server IDs
"""
servers_to_remove = []
for server_id in self.server_graphs:
if server_id not in active_server_ids:
servers_to_remove.append(server_id)
for server_id in servers_to_remove:
self.remove_server(server_id)
if servers_to_remove:
logger.info(f"Cleaned up metrics for {len(servers_to_remove)} inactive servers")
def get_summary(self) -> Dict[str, Any]:
"""
Get summary of all tracked servers.
Returns:
Dictionary with tracking statistics
"""
return {
"total_servers": len(self.server_graphs),
"servers_with_data": sum(
1 for graphs in self.server_graphs.values() if graphs.has_sufficient_data
),
"total_data_points": sum(
len(graphs.data_points) for graphs in self.server_graphs.values()
),
}
+112
View File
@@ -0,0 +1,112 @@
"""
Data Models Module
Pydantic models for type-safe data handling throughout the application.
Provides validation and serialization for API responses and internal data.
"""
from typing import Any, Dict, List, Optional
from pydantic import BaseModel, Field
class ResourceAttributes(BaseModel):
"""Current resource usage attributes from panel API."""
current_state: str
resources: "ResourceUsage"
class ResourceUsage(BaseModel):
"""Individual resource metrics."""
cpu_absolute: float = 0.0
memory_bytes: int = 0
disk_bytes: int = 0
network_rx_bytes: int = 0
network_tx_bytes: int = 0
uptime: int = 0
class ServerAttributes(BaseModel):
"""Server attributes from panel API."""
identifier: str
name: str
description: Optional[str] = None
suspended: bool = False
limits: "ServerLimits" = Field(default_factory=lambda: ServerLimits())
class ServerLimits(BaseModel):
"""Server resource limits."""
cpu: int = 0
memory: int = 0
disk: int = 0
class ServerRelationships(BaseModel):
"""Server relationships data."""
allocations: Optional["AllocationsData"] = None
class AllocationsData(BaseModel):
"""Allocations data container."""
data: List["Allocation"] = Field(default_factory=list)
class Allocation(BaseModel):
"""Individual allocation (IP/port)."""
attributes: "AllocationAttributes"
class AllocationAttributes(BaseModel):
"""Allocation attributes."""
ip_alias: str = "Unknown"
port: int = 0
is_default: bool = False
class ServerData(BaseModel):
"""Complete server data from panel API."""
attributes: ServerAttributes
relationships: Optional[ServerRelationships] = None
class ServerResources(BaseModel):
"""Server resources response from panel API."""
attributes: ResourceAttributes
class APIResponse(BaseModel):
"""Generic API response wrapper."""
status: str = "success"
data: Any = None
message: Optional[str] = None
errors: Optional[List[Dict[str, Any]]] = None
class ServerListResponse(BaseModel):
"""Server list response from panel API."""
data: List[Dict[str, Any]] = Field(default_factory=list)
meta: Optional[Dict[str, Any]] = None
class EmbedLocation(BaseModel):
"""Embed location tracking data."""
channel_id: str
message_id: str
# Rebuild models to resolve forward references
ResourceAttributes.model_rebuild()
ServerAttributes.model_rebuild()
+4 -4
View File
@@ -119,12 +119,12 @@ def validate_config():
# Validate API keys have correct prefixes
client_key = config.get("Pterodactyl", "ClientAPIKey", fallback="")
if client_key and not client_key.startswith("ptlc_"):
errors.append("ClientAPIKey should start with 'ptlc_'")
if client_key and not (client_key.startswith("ptlc_") or client_key.startswith("pacc_")):
errors.append("ClientAPIKey should start with 'ptlc_' or 'pacc_'")
app_key = config.get("Pterodactyl", "ApplicationAPIKey", fallback="")
if app_key and not app_key.startswith("ptla_"):
errors.append("ApplicationAPIKey should start with 'ptla_'")
if app_key and not (app_key.startswith("ptla_") or app_key.startswith("papp_")):
errors.append("ApplicationAPIKey should start with 'ptla_' or 'papp_'")
# Validate PanelURL is a valid URL
panel_url = config.get("Pterodactyl", "PanelURL", fallback="")
+14 -14
View File
@@ -1,26 +1,26 @@
# Testing Dependencies for Pterodactyl Discord Bot
# Core testing framework
pytest>=7.4.0
pytest-asyncio>=0.21.0
pytest-cov>=4.1.0
pytest-mock>=3.11.1
pytest-timeout>=2.1.0
pytest>=9.1.1
pytest-asyncio>=1.4.0
pytest-cov>=7.1.0
pytest-mock>=3.17.0
pytest-timeout>=2.4.0
# Code quality and linting
flake8>=6.0.0
pylint>=2.17.0
black>=23.7.0
isort>=5.12.0
flake8>=7.4.1
pylint>=4.1.3
black>=26.10.1
isort>=9.0.2
# Security scanning
bandit>=1.7.5
safety>=2.3.5
bandit>=1.9.4
safety>=3.8.1
# Mocking and fixtures
pytest-fixtures>=0.1.0
freezegun>=1.2.2
freezegun>=1.5.5
# Coverage reporting
coverage>=7.2.7
coverage-badge>=1.1.0
coverage>=7.17.0
coverage-badge>=1.1.2
+6 -5
View File
@@ -1,5 +1,6 @@
discord.py>=2.3.0
aiohttp>=3.8.0
configparser>=5.3.0
python-dotenv
matplotlib
discord.py>=2.7.1
aiohttp>=3.14.5
configparser>=7.3.0
python-dotenv>=1.2.4
matplotlib>=3.11.2
pydantic>=2.14.1
+351 -421
View File
File diff suppressed because it is too large. Load diff
+315
View File
@@ -0,0 +1,315 @@
"""
Discord Views Module
Contains all Discord UI views and interactive components.
Handles server control buttons, dropdowns, and permission checks.
"""
import logging
from typing import TYPE_CHECKING
import discord
from discord.ext import commands
from config import Config
from exceptions import PermissionError
if TYPE_CHECKING:
from api import PanelAPI
logger = logging.getLogger("pterodisbot")
class ServerStatusView(discord.ui.View):
"""
Interactive view containing server control buttons.
Provides persistent controls for server management with role-based access.
Buttons: Start (green), Stop (red), Restart (blurple), Show Address (grey).
Features:
- Guild restriction (only works in configured server)
- Role-based access control (requires "Game Server User" role)
- Persistent views (timeout=None for infinite persistence)
- Comprehensive error handling with user feedback
"""
def __init__(
self,
server_id: str,
server_name: str,
api: "PanelAPI",
server_data: dict,
) -> None:
"""
Initialize the server status view with control buttons.
Args:
server_id: The server's panel identifier
server_name: Human-readable server name
api: Panel API client instance
server_data: Full server data from panel
"""
super().__init__(timeout=None) # Persistent view
self.server_id = server_id
self.server_name = server_name
self.api = api
self.server_data = server_data
logger.debug(f"Created ServerStatusView for {server_name} ({server_id})")
async def interaction_check(self, interaction: discord.Interaction) -> bool:
"""
Verify the interacting user has the required role and is in the allowed guild.
Args:
interaction: Discord interaction object
Returns:
bool: True if authorized, False otherwise
"""
config = Config.get_instance()
# First check if interaction is from the allowed guild
if interaction.guild_id != config.allowed_guild_id:
logger.warning(f"Unauthorized interaction attempt from guild {interaction.guild_id}")
await interaction.response.send_message(
"This bot is only available in a specific server.", ephemeral=True
)
return False
# Then check for required role
logger.debug(f"Checking permissions for {interaction.user.name} on server {self.server_name}")
has_role = any(role.name == config.required_role for role in interaction.user.roles)
if not has_role:
logger.warning(
f"Permission denied for {interaction.user.name} - missing '{config.required_role}' role"
)
await interaction.response.send_message(
f"You don't have permission to control servers. You need the '{config.required_role}' role.",
ephemeral=True,
)
return False
logger.debug(f"Permission granted for {interaction.user.name}")
return True
async def on_error(
self, interaction: discord.Interaction, error: Exception, item: discord.ui.Item
) -> None:
"""
Handle errors in button interactions.
Args:
interaction: Discord interaction object
error: Exception that occurred
item: The UI item that triggered the error
"""
logger.error(f"View error in {self.server_name} by {interaction.user.name}: {str(error)}")
if not interaction.response.is_done():
await interaction.response.send_message(
"An error occurred while processing your request.", ephemeral=True
)
else:
await interaction.followup.send(
"An error occurred while processing your request.", ephemeral=True
)
@discord.ui.button(
label="Start", style=discord.ButtonStyle.green, custom_id="start_button"
)
async def start_button(
self, interaction: discord.Interaction, button: discord.ui.Button
) -> None:
"""Send a start command to the server."""
logger.info(f"Start button pressed for {self.server_name} by {interaction.user.name}")
await interaction.response.defer(ephemeral=True)
result = await self.api.send_power_action(self.server_id, "start")
if result.get("status") == "success":
message = f"Server '{self.server_name}' is starting..."
logger.info(f"Successfully started server {self.server_name}")
else:
message = f"Failed to start server: {result.get('message', 'Unknown error')}"
logger.error(f"Failed to start server {self.server_name}: {message}")
await interaction.followup.send(message, ephemeral=True)
@discord.ui.button(
label="Stop", style=discord.ButtonStyle.red, custom_id="stop_button"
)
async def stop_button(
self, interaction: discord.Interaction, button: discord.ui.Button
) -> None:
"""Send a stop command to the server."""
logger.info(f"Stop button pressed for {self.server_name} by {interaction.user.name}")
await interaction.response.defer(ephemeral=True)
result = await self.api.send_power_action(self.server_id, "stop")
if result.get("status") == "success":
message = f"Server '{self.server_name}' is stopping..."
logger.info(f"Successfully stopped server {self.server_name}")
else:
message = f"Failed to stop server: {result.get('message', 'Unknown error')}"
logger.error(f"Failed to stop server {self.server_name}: {message}")
await interaction.followup.send(message, ephemeral=True)
@discord.ui.button(
label="Restart", style=discord.ButtonStyle.blurple, custom_id="restart_button"
)
async def restart_button(
self, interaction: discord.Interaction, button: discord.ui.Button
) -> None:
"""Send a restart command to the server."""
logger.info(f"Restart button pressed for {self.server_name} by {interaction.user.name}")
await interaction.response.defer(ephemeral=True)
result = await self.api.send_power_action(self.server_id, "restart")
if result.get("status") == "success":
message = f"Server '{self.server_name}' is restarting..."
logger.info(f"Successfully restarted server {self.server_name}")
else:
message = f"Failed to restart server: {result.get('message', 'Unknown error')}"
logger.error(f"Failed to restart server {self.server_name}: {message}")
await interaction.followup.send(message, ephemeral=True)
@discord.ui.button(
label="Show Address",
style=discord.ButtonStyle.grey,
custom_id="show_address_button",
)
async def show_address_button(
self, interaction: discord.Interaction, button: discord.ui.Button
) -> None:
"""Show server's default allocation IP and port."""
logger.info(f"Show Address button pressed for {self.server_name} by {interaction.user.name}")
try:
await interaction.response.defer(ephemeral=True)
logger.debug(f"Fetching server details for {self.server_id}")
ip_alias, port = await self.api.get_server_default_allocation(self.server_id)
# Create and send embed
embed = discord.Embed(
title=f"{self.server_name} Connection Info",
color=discord.Color.blue(),
description=f"Server ID: `{self.server_id}`",
)
embed.add_field(name="Address", value=f"`{ip_alias}`", inline=True)
embed.add_field(name="Port", value=f"`{port}`", inline=True)
await interaction.followup.send(embed=embed, ephemeral=True)
logger.info(f"Displayed connection info for {self.server_name}")
except Exception as e:
logger.error(f"Failed to show address for {self.server_name}: {str(e)}")
await interaction.followup.send(
"⚠️ Failed to get connection info. The server may not have any ports allocated.",
ephemeral=True,
)
class ServerDropdown(discord.ui.Select):
"""Dropdown menu for server selection in status dashboard."""
def __init__(self, server_options: list) -> None:
"""
Initialize the server dropdown.
Args:
server_options: List of discord.SelectOption objects
"""
super().__init__(
placeholder="Select a server to display...",
options=server_options,
min_values=1,
max_values=1,
)
async def callback(self, interaction: discord.Interaction) -> None:
"""
Handle server selection from dropdown menu.
Creates permanent status embed in the channel for the selected server.
"""
from bot import PterodactylBot
await interaction.response.defer(ephemeral=True)
bot = PterodactylBot.get_instance()
selected_server_id = self.values[0]
server_data = bot.server_cache.get(selected_server_id)
if not server_data:
await interaction.followup.send(
"❌ Selected server no longer available. Please try again.", ephemeral=True
)
return
server_name = server_data["attributes"]["name"]
logger.info(f"User {interaction.user.name} selected server: {server_name}")
try:
# Get current server status for embed creation
resources = await bot.api.get_server_resources(selected_server_id)
# Delete old embed if it exists to prevent duplication
if selected_server_id in bot.embed_manager.embed_locations:
logger.debug(f"Found existing embed for {selected_server_id}, attempting to delete")
try:
old_location = bot.embed_manager.embed_locations[selected_server_id]
old_channel = bot.get_channel(int(old_location["channel_id"]))
if old_channel:
try:
old_message = await old_channel.fetch_message(
int(old_location["message_id"])
)
await old_message.delete()
logger.debug(f"Deleted old embed for {selected_server_id}")
except discord.NotFound:
logger.debug(f"Old embed for {selected_server_id} already deleted")
except Exception as e:
logger.error(f"Failed to delete old embed: {str(e)}")
# Create and send new permanent status embed in channel
from models import ServerData, ServerResources
server_obj = ServerData.model_validate(server_data)
resources_obj = ServerResources.model_validate(resources)
embed, view = await bot.embed_manager.create_server_embed(server_obj, resources_obj)
message = await interaction.channel.send(embed=embed, view=view)
await bot.embed_manager.track_embed(selected_server_id, message)
await interaction.followup.send(
f"✅ **{server_name}** status has been posted in {interaction.channel.mention}",
ephemeral=True,
)
logger.info(f"Successfully posted status for {server_name}")
except Exception as e:
logger.error(f"Failed to create status embed for {server_name}: {str(e)}")
await interaction.followup.send(
f"❌ Failed to create status embed for **{server_name}**: {str(e)}", ephemeral=True
)
class ServerDropdownView(discord.ui.View):
"""View containing the server dropdown with timeout for auto-cleanup."""
def __init__(self, server_options: list, timeout: int = 180) -> None:
"""
Initialize dropdown view with timeout.
Args:
server_options: List of discord.SelectOption objects
timeout: Timeout in seconds (default 3 minutes for ephemeral cleanup)
"""
super().__init__(timeout=timeout)
self.server_options = server_options
self.add_item(ServerDropdown(server_options))
async def on_timeout(self) -> None:
"""Clean up when dropdown times out (ephemeral auto-removal)."""
logger.debug("Server dropdown timed out and was automatically cleaned up")