現代化 Python 機器人代碼 (2025)
import discord
from discord.ext import commands, tasks
import asyncio
import aiohttp
import json
import os
from datetime import datetime, timedelta
from azure.identity import DefaultAzureCredential
from azure.mgmt.compute import ComputeManagementClient
import logging
# 設定日誌記錄
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class MinecraftBot(commands.Bot):
def __init__(self):
intents = discord.Intents.default()
intents.message_content = True
super().__init__(command_prefix='!', intents=intents)
# Azure 設定
self.subscription_id = os.getenv('AZURE_SUBSCRIPTION_ID')
self.resource_group = os.getenv('AZURE_RESOURCE_GROUP')
self.vm_name = os.getenv('AZURE_VM_NAME')
# 初始化 Azure 客戶端
self.credential = DefaultAzureCredential()
self.compute_client = ComputeManagementClient(
self.credential, self.subscription_id
)
# 狀態追蹤
self.server_status = {
'is_running': False,
'last_checked': None,
'players_online': 0,
'startup_time': None
}
async def on_ready(self):
logger.info(f'{self.user} 已成功連接到 Discord!')
await self.sync_commands()
self.monitor_server.start()
async def sync_commands(self):
"""同步斜線指令"""
try:
synced = await self.tree.sync()
logger.info(f'同步了 {len(synced)} 個指令')
except Exception as e:
logger.error(f'指令同步失敗: {e}')
@tasks.loop(minutes=5)
async def monitor_server(self):
"""定期監控伺服器狀態"""
try:
vm_status = await self.get_vm_status()
await self.update_server_status(vm_status)
# 自動關機邏輯（無玩家 30 分鐘後）
if (self.server_status['is_running'] and
self.server_status['players_online'] == 0 and
self.server_status['last_checked'] and
datetime.now() - self.server_status['last_checked'] > timedelta(minutes=30)):
await self.auto_shutdown()
except Exception as e:
logger.error(f'監控錯誤: {e}')
async def get_vm_status(self):
"""獲取 Azure VM 狀態"""
try:
vm = self.compute_client.virtual_machines.get(
self.resource_group, self.vm_name, expand='instanceView'
)
# 檢查 VM 電源狀態
power_state = None
for status in vm.instance_view.statuses:
if status.code.startswith('PowerState'):
power_state = status.code.split('/')[-1]
break
return {
'power_state': power_state,
'vm_size': vm.hardware_profile.vm_size,
'location': vm.location
}
except Exception as e:
logger.error(f'獲取 VM 狀態失敗: {e}')
return None
async def start_vm(self):
"""啟動 Azure VM"""
try:
logger.info('正在啟動 Azure VM...')
operation = self.compute_client.virtual_machines.begin_start(
self.resource_group, self.vm_name
)
await asyncio.get_event_loop().run_in_executor(None, operation.wait)
logger.info('Azure VM 啟動完成')
return True
except Exception as e:
logger.error(f'VM 啟動失敗: {e}')
return False
async def stop_vm(self):
"""停止 Azure VM"""
try:
logger.info('正在停止 Azure VM...')
operation = self.compute_client.virtual_machines.begin_deallocate(
self.resource_group, self.vm_name
)
await asyncio.get_event_loop().run_in_executor(None, operation.wait)
logger.info('Azure VM 停止完成')
return True
except Exception as e:
logger.error(f'VM 停止失敗: {e}')
return False
async def get_minecraft_status(self, server_ip: str, port: int = 25565):
"""檢查 Minecraft 伺服器狀態"""
try:
async with aiohttp.ClientSession() as session:
url = f'https://api.mcsrvstat.us/2/{server_ip}:{port}'
async with session.get(url) as response:
if response.status == 200:
data = await response.json()
return {
'online': data.get('online', False),
'players': data.get('players', {}),
'version': data.get('version', 'Unknown'),
'motd': data.get('motd', {}).get('clean', [''])[0] if data.get('motd') else 'N/A'
}
except Exception as e:
logger.error(f'獲取 Minecraft 狀態失敗: {e}')
return None
# 斜線指令定義
bot = MinecraftBot()
@bot.tree.command(name="start-server", description="啟動 Minecraft 伺服器")
async def start_server(interaction: discord.Interaction):
await interaction.response.defer()
embed = discord.Embed(
title="🚀 正在啟動 Minecraft 伺服器",
color=discord.Color.orange(),
timestamp=datetime.now()
)
embed.add_field(name="狀態", value="正在啟動 Azure VM...", inline=False)
embed.add_field(name="預計時間", value="2-3 分鐘", inline=True)
embed.set_footer(text="自動化系統")
await interaction.followup.send(embed=embed)
# 啟動 VM
success = await bot.start_vm()
if success:
# 等待 Minecraft 伺服器啟動
await asyncio.sleep(90) # 給予啟動時間
embed = discord.Embed(
title="✅ Minecraft 伺服器已啟動",
color=discord.Color.green(),
timestamp=datetime.now()
)
embed.add_field(name="伺服器 IP", value="mc.lueklake.me", inline=True)
embed.add_field(name="版本", value="1.20.1 (BMC4)", inline=True)
embed.add_field(name="狀態", value="🟢 線上", inline=True)
embed.set_footer(text="準備好開始遊戲！")
await interaction.followup.send(embed=embed)
else:
embed = discord.Embed(
title="❌ 啟動失敗",
description="請聯繫管理員或稍後重試",
color=discord.Color.red(),
timestamp=datetime.now()
)
await interaction.followup.send(embed=embed)
@bot.tree.command(name="stop-server", description="停止 Minecraft 伺服器")
async def stop_server(interaction: discord.Interaction):
await interaction.response.defer()
embed = discord.Embed(
title="🛑 正在停止 Minecraft 伺服器",
description="正在安全關閉伺服器並停止 Azure VM...",
color=discord.Color.orange(),
timestamp=datetime.now()
)
await interaction.followup.send(embed=embed)
success = await bot.stop_vm()
if success:
embed = discord.Embed(
title="✅ 伺服器已安全停止",
description="Azure VM 已停止，節省雲端費用",
color=discord.Color.green(),
timestamp=datetime.now()
)
embed.set_footer(text="使用 /start-server 重新啟動")
await interaction.followup.send(embed=embed)
@bot.tree.command(name="server-status", description="查看伺服器詳細狀態")
async def server_status(interaction: discord.Interaction):
await interaction.response.defer()
# 獲取 VM 狀態
vm_status = await bot.get_vm_status()
mc_status = await bot.get_minecraft_status("mc.lueklake.me")
embed = discord.Embed(
title="📊 Minecraft 伺服器狀態",
color=discord.Color.blue(),
timestamp=datetime.now()
)
if vm_status:
vm_state = "🟢 運行中" if vm_status['power_state'] == 'running' else "🔴 已停止"
embed.add_field(name="Azure VM", value=vm_state, inline=True)
embed.add_field(name="VM 規格", value=vm_status['vm_size'], inline=True)
embed.add_field(name="資料中心", value=vm_status['location'], inline=True)
if mc_status and mc_status['online']:
players = mc_status['players']
player_info = f"{players.get('online', 0)}/{players.get('max', 20)}"
embed.add_field(name="遊戲狀態", value="🟢 線上", inline=True)
embed.add_field(name="玩家數量", value=player_info, inline=True)
embed.add_field(name="版本", value=mc_status['version'], inline=True)
if players.get('list'):
player_names = ', '.join([p['name'] for p in players['list'][:5]])
if len(players['list']) > 5:
player_names += f" (+{len(players['list'])-5} 更多)"
embed.add_field(name="在線玩家", value=player_names, inline=False)
else:
embed.add_field(name="遊戲狀態", value="🔴 離線", inline=True)
embed.set_footer(text="每 5 分鐘自動更新")
await interaction.followup.send(embed=embed)
# 啟動機器人
if __name__ == "__main__":
token = os.getenv('DISCORD_BOT_TOKEN')
if not token:
logger.error('請設定 DISCORD_BOT_TOKEN 環境變數')
exit(1)
bot.run(token)