559 lines
20 KiB
Python
559 lines
20 KiB
Python
import discord
|
||
from discord.ext import commands, tasks
|
||
from discord import app_commands
|
||
from discord.ui import Button, View
|
||
import aiohttp
|
||
import asyncio
|
||
import platform
|
||
import psutil
|
||
import os
|
||
from datetime import datetime
|
||
import logging
|
||
|
||
# ロギングの設定
|
||
logging.basicConfig(
|
||
level=logging.INFO,
|
||
format='%(asctime)s - %(levelname)s - %(message)s',
|
||
handlers=[
|
||
logging.FileHandler("bot.log"),
|
||
logging.StreamHandler()
|
||
]
|
||
)
|
||
|
||
# BOTの設定
|
||
TOKEN = 'あなたのDiscord Bot Token' # BOTトークン
|
||
CHANNEL_ID = 000000000000000000 # チャンネルID
|
||
CHECK_INTERVAL = 30 # 間隔（秒）
|
||
LOG_FILE = 'ip_changes.log' # IPアドレス変更ログファイル
|
||
|
||
last_ip = None
|
||
status_message = None
|
||
initialization_complete = False
|
||
|
||
intents = discord.Intents.all()
|
||
intents.message_content = True
|
||
intents.reactions = True
|
||
|
||
bot = commands.Bot(command_prefix='!', intents=intents)
|
||
|
||
# グローバルIPを取得
|
||
async def get_global_ip():
|
||
try:
|
||
async with aiohttp.ClientSession() as session:
|
||
async with session.get('https://api.ipify.org/?format=json', timeout=10) as response:
|
||
if response.status == 200:
|
||
data = await response.json()
|
||
return data['ip']
|
||
else:
|
||
logging.error(f"IPの取得に失敗しました: ステータスコード {response.status}")
|
||
return None
|
||
except Exception as e:
|
||
logging.error(f"IPの取得中にエラーが発生しました: {e}")
|
||
return None
|
||
|
||
# ログにIPアドレス変更を記録
|
||
def log_ip_change(old_ip, new_ip):
|
||
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||
log_entry = f"[{timestamp}] IP変更: {old_ip} -> {new_ip}\n"
|
||
|
||
try:
|
||
with open(LOG_FILE, 'a', encoding='utf-8') as log_file:
|
||
log_file.write(log_entry)
|
||
except Exception as e:
|
||
logging.error(f"ログの書き込み中にエラーが発生しました: {e}")
|
||
|
||
# サーバーのステータスを取得
|
||
def get_server_status():
|
||
try:
|
||
# CPU使用率
|
||
cpu_percent = psutil.cpu_percent(interval=1)
|
||
|
||
# メモリ使用率
|
||
memory = psutil.virtual_memory()
|
||
memory_percent = memory.percent
|
||
memory_used = round(memory.used / (1024 * 1024 * 1024), 2) # GB
|
||
memory_total = round(memory.total / (1024 * 1024 * 1024), 2) # GB
|
||
|
||
# ディスク使用率
|
||
disk = psutil.disk_usage('/')
|
||
disk_percent = disk.percent
|
||
disk_used = round(disk.used / (1024 * 1024 * 1024), 2) # GB
|
||
disk_total = round(disk.total / (1024 * 1024 * 1024), 2) # GB
|
||
|
||
# システム稼働時間
|
||
boot_time = datetime.fromtimestamp(psutil.boot_time())
|
||
uptime = datetime.now() - boot_time
|
||
days = uptime.days
|
||
hours, remainder = divmod(uptime.seconds, 3600)
|
||
minutes, seconds = divmod(remainder, 60)
|
||
uptime_str = f"{days}日 {hours}時間 {minutes}分 {seconds}秒"
|
||
|
||
# OSバージョン
|
||
os_info = platform.platform()
|
||
|
||
return {
|
||
"cpu_percent": cpu_percent,
|
||
"memory_percent": memory_percent,
|
||
"memory_used": memory_used,
|
||
"memory_total": memory_total,
|
||
"disk_percent": disk_percent,
|
||
"disk_used": disk_used,
|
||
"disk_total": disk_total,
|
||
"uptime": uptime_str,
|
||
"os_info": os_info
|
||
}
|
||
except Exception as e:
|
||
logging.error(f"サーバーステータスの取得中にエラーが発生しました: {e}")
|
||
return None
|
||
|
||
# 埋め込みメッセージ
|
||
def create_ip_embed(ip, old_ip=None, color=discord.Color.green()):
|
||
if not ip:
|
||
return discord.Embed(
|
||
title="⚠️ エラー",
|
||
description="グローバルIPアドレスの取得に失敗しました",
|
||
color=discord.Color.red()
|
||
)
|
||
|
||
embed = discord.Embed(
|
||
title="🌐 グローバルIPアドレス",
|
||
description=f"**`{ip}`**",
|
||
color=color,
|
||
timestamp=datetime.now()
|
||
)
|
||
|
||
if old_ip and old_ip != ip:
|
||
embed.add_field(name="以前のIPアドレス", value=f"`{old_ip}`", inline=False)
|
||
embed.add_field(name="状態", value="⚠️ IPアドレスが変更されました", inline=False)
|
||
embed.color = discord.Color.orange()
|
||
|
||
embed.set_footer(text="最終更新")
|
||
|
||
return embed
|
||
|
||
# サーバーステータスの埋め込みメッセージ
|
||
def create_status_embed(status_data):
|
||
if not status_data:
|
||
return discord.Embed(
|
||
title="⚠️ エラー",
|
||
description="サーバーステータスの取得に失敗しました",
|
||
color=discord.Color.red()
|
||
)
|
||
|
||
embed = discord.Embed(
|
||
title="🖥️ サーバーステータス",
|
||
color=discord.Color.blue(),
|
||
timestamp=datetime.now()
|
||
)
|
||
|
||
# CPU使用率のプログレスバー
|
||
cpu_bar = create_progress_bar(status_data["cpu_percent"])
|
||
embed.add_field(name="CPU使用率", value=f"{cpu_bar} {status_data['cpu_percent']}%", inline=False)
|
||
|
||
# メモリ使用率のプログレスバー
|
||
memory_bar = create_progress_bar(status_data["memory_percent"])
|
||
embed.add_field(
|
||
name="メモリ使用率",
|
||
value=f"{memory_bar} {status_data['memory_percent']}%\n`{status_data['memory_used']}GB / {status_data['memory_total']}GB`",
|
||
inline=False
|
||
)
|
||
|
||
# ディスク使用率のプログレスバー
|
||
disk_bar = create_progress_bar(status_data["disk_percent"])
|
||
embed.add_field(
|
||
name="ディスク使用率",
|
||
value=f"{disk_bar} {status_data['disk_percent']}%\n`{status_data['disk_used']}GB / {status_data['disk_total']}GB`",
|
||
inline=False
|
||
)
|
||
|
||
# システム情報
|
||
embed.add_field(name="システム稼働時間", value=f"`{status_data['uptime']}`", inline=False)
|
||
embed.add_field(name="OS情報", value=f"`{status_data['os_info']}`", inline=False)
|
||
|
||
embed.set_footer(text="最終更新")
|
||
|
||
return embed
|
||
|
||
# プログレスバーを作成
|
||
def create_progress_bar(percent, bar_length=15):
|
||
filled_length = int(bar_length * percent / 100)
|
||
empty_length = bar_length - filled_length
|
||
|
||
if percent < 60:
|
||
color = "🟢" # 緑
|
||
elif percent < 80:
|
||
color = "🟡" # 黄
|
||
else:
|
||
color = "🔴" # 赤
|
||
|
||
bar = "▰" * filled_length + "▱" * empty_length
|
||
return f"[{bar}] {color}"
|
||
|
||
class IPControlView(View):
|
||
def __init__(self):
|
||
super().__init__(timeout=None)
|
||
|
||
@discord.ui.button(label="今すぐ更新", style=discord.ButtonStyle.primary, custom_id="refresh_ip")
|
||
async def refresh_button(self, interaction: discord.Interaction, button: discord.ui.Button):
|
||
global last_ip
|
||
|
||
await interaction.response.defer(ephemeral=True)
|
||
|
||
current_ip = await get_global_ip()
|
||
|
||
if current_ip is None:
|
||
await interaction.followup.send("IPアドレスの取得に失敗しました", ephemeral=True)
|
||
return
|
||
|
||
if last_ip and last_ip != current_ip:
|
||
log_ip_change(last_ip, current_ip)
|
||
|
||
old_ip = last_ip
|
||
last_ip = current_ip
|
||
|
||
embed = create_ip_embed(current_ip, old_ip)
|
||
await interaction.message.edit(embed=embed, view=self)
|
||
|
||
await interaction.followup.send("IPアドレス情報を更新しました", ephemeral=True)
|
||
|
||
@discord.ui.button(label="サーバーステータス", style=discord.ButtonStyle.success, custom_id="server_status")
|
||
async def status_button(self, interaction: discord.Interaction, button: discord.ui.Button):
|
||
await interaction.response.defer(ephemeral=True)
|
||
|
||
status_data = get_server_status()
|
||
|
||
if status_data is None:
|
||
await interaction.followup.send("サーバーステータスの取得に失敗しました", ephemeral=True)
|
||
return
|
||
|
||
embed = create_status_embed(status_data)
|
||
|
||
view = CloseView()
|
||
|
||
await interaction.followup.send(embed=embed, view=view, ephemeral=False)
|
||
|
||
@discord.ui.button(label="PROXMOX", style=discord.ButtonStyle.secondary, custom_id="proxmox_link")
|
||
async def proxmox_button(self, interaction: discord.Interaction, button: discord.ui.Button):
|
||
global last_ip
|
||
|
||
await interaction.response.defer(ephemeral=True)
|
||
|
||
if not last_ip:
|
||
await interaction.followup.send("IPアドレスが取得されていません", ephemeral=True)
|
||
return
|
||
|
||
proxmox_url = f"https://{last_ip}:8006"
|
||
|
||
await interaction.followup.send(
|
||
f"Proxmox管理画面にアクセス\n"
|
||
f"[Proxmox管理画面]({proxmox_url})\n\n"
|
||
f"または直接アクセス: `{proxmox_url}`",
|
||
ephemeral=True
|
||
)
|
||
|
||
class CloseView(View):
|
||
def __init__(self):
|
||
super().__init__(timeout=None)
|
||
|
||
@discord.ui.button(label="閉じる", style=discord.ButtonStyle.danger, custom_id="close_status")
|
||
async def close_button(self, interaction: discord.Interaction, button: discord.ui.Button):
|
||
await interaction.response.defer(ephemeral=True)
|
||
await interaction.message.delete()
|
||
await interaction.followup.send("ステータス表示を終了しました", ephemeral=True)
|
||
|
||
async def clear_channel_messages():
|
||
try:
|
||
channel = bot.get_channel(CHANNEL_ID)
|
||
if channel is None:
|
||
logging.error(f"チャンネルが見つかりません: {CHANNEL_ID}")
|
||
return False
|
||
|
||
deleted = 0
|
||
logging.info("チャンネル内のメッセージを削除中")
|
||
|
||
try:
|
||
logging.info("最近のメッセージを一削除中")
|
||
deleted_messages = await channel.purge(limit=100, bulk=True)
|
||
deleted += len(deleted_messages)
|
||
logging.info(f"{len(deleted_messages)}件のメッセージの削除完了")
|
||
|
||
remaining = []
|
||
async for message in channel.history(limit=10):
|
||
remaining.append(message)
|
||
|
||
if remaining:
|
||
logging.info(f"残り{len(remaining)}件の古いメッセージを削除中")
|
||
for message in remaining:
|
||
await message.delete()
|
||
deleted += 1
|
||
logging.info(f"メッセージを削除（合計: {deleted}）")
|
||
await asyncio.sleep(0.5) # レート制限回避
|
||
|
||
except discord.Forbidden:
|
||
logging.error("Discordのメッセージ削除権限がありません")
|
||
return False
|
||
except discord.HTTPException as e:
|
||
logging.error(f"一括削除中にHTTPエラーが発生: {e}")
|
||
try:
|
||
async for message in channel.history(limit=100):
|
||
await message.delete()
|
||
deleted += 1
|
||
logging.info(f"個別削除：メッセージを削除（合計: {deleted}）")
|
||
await asyncio.sleep(0.5)
|
||
except Exception as e2:
|
||
logging.error(f"個別削除中にエラーが発生: {e2}")
|
||
|
||
logging.info(f"{deleted}件のメッセージを削除")
|
||
return True
|
||
|
||
except Exception as e:
|
||
logging.error(f"チャンネル内メッセージ削除中にエラーが発生: {e}")
|
||
return False
|
||
|
||
@bot.event
|
||
async def on_ready():
|
||
global initialization_complete
|
||
|
||
logging.info(f'{bot.user}')
|
||
logging.info(f'BOT ID: {bot.user.id}')
|
||
logging.info(f'送信先チャンネルID: {CHANNEL_ID}')
|
||
logging.info(f'間隔: {CHECK_INTERVAL}秒 ({CHECK_INTERVAL/60}分)')
|
||
logging.info('------')
|
||
logging.info('グローバルIP検出・変更通知BOT')
|
||
logging.info('© 2025 uttili_external_system')
|
||
logging.info('https://uttili.com')
|
||
logging.info('------')
|
||
|
||
if not initialization_complete:
|
||
logging.info("チャンネル内メッセージを削除中")
|
||
await clear_channel_messages()
|
||
|
||
logging.info("メッセージを送信中")
|
||
await send_initial_message()
|
||
|
||
if not check_ip_change.is_running():
|
||
check_ip_change.start()
|
||
logging.info("IP検出開始")
|
||
|
||
initialization_complete = True
|
||
logging.info("初期化が完了")
|
||
else:
|
||
if not check_ip_change.is_running():
|
||
check_ip_change.start()
|
||
logging.info("IP検出再開")
|
||
|
||
@tasks.loop(seconds=CHECK_INTERVAL)
|
||
async def check_ip_change():
|
||
global last_ip, status_message
|
||
|
||
try:
|
||
current_ip = await get_global_ip()
|
||
|
||
if current_ip is None:
|
||
logging.warning("IPアドレスの取得に失敗")
|
||
return
|
||
|
||
if last_ip is None:
|
||
last_ip = current_ip
|
||
logging.info(f"初期IPアドレスを確定: {current_ip}")
|
||
|
||
if status_message is None:
|
||
await send_initial_message()
|
||
else:
|
||
await update_ip_message(current_ip)
|
||
|
||
elif last_ip != current_ip:
|
||
old_ip = last_ip
|
||
last_ip = current_ip
|
||
|
||
log_ip_change(old_ip, current_ip)
|
||
|
||
await update_ip_message(current_ip, old_ip)
|
||
|
||
logging.info(f"IPアドレスの変更検出: {old_ip} -> {current_ip}")
|
||
else:
|
||
await update_ip_message(current_ip)
|
||
logging.debug(f"IPアドレスに変更なし: {current_ip}")
|
||
|
||
except Exception as e:
|
||
logging.error(f"IP検出中に予期しないエラーが発生: {e}")
|
||
|
||
@check_ip_change.before_loop
|
||
async def before_check_ip_change():
|
||
await bot.wait_until_ready()
|
||
logging.info("BOT起動")
|
||
|
||
@check_ip_change.error
|
||
async def check_ip_change_error(error):
|
||
logging.error(f"IP検出中にエラーが発生: {error}")
|
||
if not check_ip_change.is_running():
|
||
check_ip_change.restart()
|
||
logging.info("IP検出再開")
|
||
|
||
async def send_initial_message():
|
||
global last_ip, status_message
|
||
|
||
try:
|
||
if status_message is not None:
|
||
logging.info("既にメッセージが存在するためスキップ")
|
||
return
|
||
|
||
channel = bot.get_channel(CHANNEL_ID)
|
||
if channel is None:
|
||
logging.error(f"チャンネル不明: {CHANNEL_ID}")
|
||
return
|
||
|
||
current_ip = await get_global_ip()
|
||
if current_ip is None:
|
||
logging.error("IPアドレスの取得に失敗しました")
|
||
return
|
||
|
||
last_ip = current_ip
|
||
|
||
embed = create_ip_embed(current_ip)
|
||
|
||
view = IPControlView()
|
||
|
||
status_message = await channel.send(embed=embed, view=view)
|
||
|
||
logging.info(f"メッセージを送信 Message ID: {status_message.id}")
|
||
|
||
except Exception as e:
|
||
logging.error(f"メッセージ送信中にエラー: {e}")
|
||
|
||
async def update_ip_message(current_ip, old_ip=None):
|
||
global status_message
|
||
|
||
try:
|
||
if status_message is None:
|
||
await send_initial_message()
|
||
return
|
||
|
||
channel = bot.get_channel(CHANNEL_ID)
|
||
if channel is None:
|
||
logging.error(f"チャンネル不明: {CHANNEL_ID}")
|
||
return
|
||
|
||
try:
|
||
message = await channel.fetch_message(status_message.id)
|
||
except discord.NotFound:
|
||
logging.warning("ステータスメッセージ不明。新規作成開始。")
|
||
status_message = None # status_messageリセット
|
||
await send_initial_message()
|
||
return
|
||
except Exception as e:
|
||
logging.error(f"メッセージ取得中にエラーが発生: {e}")
|
||
status_message = None
|
||
await send_initial_message()
|
||
return
|
||
|
||
color = discord.Color.orange() if old_ip and old_ip != current_ip else discord.Color.green()
|
||
|
||
embed = create_ip_embed(current_ip, old_ip, color)
|
||
|
||
view = IPControlView()
|
||
|
||
await message.edit(embed=embed, view=view)
|
||
|
||
if old_ip and old_ip != current_ip:
|
||
logging.info(f"IPメッセージを更新: {old_ip} -> {current_ip}")
|
||
else:
|
||
logging.debug(f"IPメッセージのタイムスタンプを更新: {current_ip}")
|
||
|
||
except discord.NotFound:
|
||
logging.warning("メッセージ不明。新規作成開始。")
|
||
status_message = None
|
||
await send_initial_message()
|
||
except discord.Forbidden:
|
||
logging.error("メッセージの編集権限がないため動作不可能")
|
||
except discord.HTTPException as e:
|
||
logging.error(f"メッセージ更新中にHTTPエラーが発生: {e}")
|
||
await asyncio.sleep(5)
|
||
try:
|
||
if status_message:
|
||
channel = bot.get_channel(CHANNEL_ID)
|
||
message = await channel.fetch_message(status_message.id)
|
||
embed = create_ip_embed(current_ip, old_ip, discord.Color.green())
|
||
view = IPControlView()
|
||
await message.edit(embed=embed, view=view)
|
||
except Exception:
|
||
status_message = None
|
||
await send_initial_message()
|
||
except Exception as e:
|
||
logging.error(f"メッセージの更新中に予期しないエラーが発生: {e}")
|
||
status_message = None
|
||
await send_initial_message()
|
||
|
||
@bot.command(name="ipreset")
|
||
async def ipreset(ctx):
|
||
global status_message, initialization_complete
|
||
|
||
if not ctx.author.guild_permissions.administrator:
|
||
await ctx.send("❌ このコマンドは管理者権限が必要です", delete_after=5)
|
||
return
|
||
|
||
await ctx.message.delete()
|
||
|
||
await clear_channel_messages()
|
||
status_message = None
|
||
initialization_complete = False
|
||
|
||
initialization_complete = True
|
||
await send_initial_message()
|
||
|
||
if not check_ip_change.is_running():
|
||
check_ip_change.start()
|
||
logging.info("IP検出を開始")
|
||
|
||
await ctx.send("✅ チャンネル内を削除し、ステータスを表示しました", delete_after=5)
|
||
|
||
@bot.command(name="ipclear")
|
||
async def ipclear(ctx):
|
||
global status_message
|
||
|
||
if not ctx.author.guild_permissions.administrator:
|
||
await ctx.send("❌ このコマンドは管理者権限が必要です", delete_after=5)
|
||
return
|
||
|
||
await ctx.message.delete()
|
||
|
||
success = await clear_channel_messages()
|
||
|
||
if success:
|
||
status_message = None
|
||
|
||
await send_initial_message()
|
||
|
||
await ctx.send("✅ チャンネルをクリアしました", delete_after=5)
|
||
else:
|
||
await ctx.send("❌ チャンネルのクリアに失敗しました", delete_after=5)
|
||
|
||
@bot.command(name="ipstatus")
|
||
async def ipstatus(ctx):
|
||
await ctx.message.delete()
|
||
|
||
task_status = "実行中" if check_ip_change.is_running() else "停止中"
|
||
|
||
ip_status = f"`{last_ip}`" if last_ip else "未取得"
|
||
|
||
status_msg = await ctx.send(
|
||
f"**IPチェック状態**\n"
|
||
f"タスク: {task_status}\n"
|
||
f"現在のIP: {ip_status}\n"
|
||
f"更新間隔: {CHECK_INTERVAL}秒",
|
||
delete_after=10
|
||
)
|
||
|
||
if __name__ == "__main__":
|
||
logging.info("BOTを起動中")
|
||
logging.info("グローバルIPの監視開始")
|
||
|
||
try:
|
||
import psutil
|
||
except ImportError:
|
||
logging.warning("psutilモジュールがインストールされていない")
|
||
logging.warning("サーバーステータス機能を使用するには、以下のコマンドでインストール:")
|
||
logging.warning("pip install psutil")
|
||
logging.warning("続行しますが、サーバーステータス機能は機能しない")
|
||
|
||
bot.run(TOKEN) |