suti7yk5032/Dislocker
1
0
Fork 0
Code Issues 11 Pull requests Projects 3 Releases Packages Wiki Activity Actions

ボタン操作ができるように、タイムアウト機能の実装、スラッシュコマンド対応への準備

Browse source
This commit is contained in:
suti7yk5032 2024-07-21 16:36:16 +09:00
parent f7b6caca30
commit b908ce17ed
1 changed files with 230 additions and 2 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

232
dislocker.py
View file

@ -1,13 +1,17 @@
import json
import discord
from discord import Interaction, TextStyle, app_commands
from discord.ui import TextInput, View, Modal
import os
import psycopg2
from psycopg2 import sql
import hashlib
import string
import random
from datetime import datetime
from datetime import datetime, timedelta
from openpyxl import Workbook
import threading
import time
class DL():
def __init__(self):
@ -70,7 +74,10 @@ class DL():
print(find_pc_list_table)
if find_pc_list_table[0][0] == False:
cursor.execute("CREATE TABLE pc_list (pc_number INTEGER NOT NULL, using_user_id INTEGER, password_hash VARCHAR(32), PRIMARY KEY (pc_number), FOREIGN KEY (using_user_id) REFERENCES club_member(id))")
self.db.commit
for i in range(10):
print(i)
cursor.execute("INSERT INTO pc_list (pc_number) VALUES (%s)", (i + 1,))
self.db.commit()
cursor.execute("SELECT EXISTS (SELECT FROM information_schema.tables WHERE table_schema = 'public' AND table_name = 'pc_usage_history')")
find_pc_usage_history_table = cursor.fetchall()
@ -91,6 +98,7 @@ class DL():
class Bot(discord.Client):
def password_generate(self, length):
numbers = string.digits # (1)
password = ''.join(random.choice(numbers) for _ in range(length)) # (2)
@ -371,9 +379,36 @@ class Bot(discord.Client):
cursor.close()
return result
async def timeout_notify(self, **kwargs):
try:
pc_number = kwargs["pc_number"]
discord_display_name = kwargs["discord_display_name"]
await self.get_channel(dislocker.server_config["bot"]["log_channel_id"]).send(f':negative_squared_cross_mark: {discord_display_name} さんのPC {pc_number} の使用登録はタイムアウトにより解除されました。')
result = {"result": "ok"}
except Exception as error:
print("自動停止処理中にエラーが発生しました。\nエラー内容")
print(str(error.__class__.__name__))
print(str(error.args))
print(str(error))
result = {"result": "error"}
finally:
return result
async def on_ready(self):
print("DiscordのBotが起動しました。")
async def on_interaction(self, interaction:discord.Interaction):
try:
if interaction.data["component_type"] == 2:
await self.on_button(interaction)
except KeyError:
pass
async def on_message(self, message):
if message.author.bot:
pass
@ -490,6 +525,29 @@ class Bot(discord.Client):
else:
await message.channel.send("# :warning: PCを登録できませんでした。\n構文が間違っています。\n-# /pcregister PC番号")
elif msg_split[0] == "/registerbutton":
pc_button_view = View(timeout=None)
for i in range(1, 11):
pc_register_button = discord.ui.Button(style=discord.ButtonStyle.primary, label=f"{i}", custom_id=f"pcregister_{i}")
pc_button_view.add_item(pc_register_button)
await self.get_channel(dislocker.server_config["bot"]["config_public_channel_id"]).send(f'# :index_pointing_at_the_viewer: 使いたいPCの番号を選んでください！', view=pc_button_view)
elif msg_split[0] == "/stopbutton":
stop_button_view = View(timeout=None)
stop_button = discord.ui.Button(style=discord.ButtonStyle.danger, label="PCの使用を停止", custom_id="stop")
stop_button_view.add_item(stop_button)
await self.get_channel(dislocker.server_config["bot"]["config_public_channel_id"]).send(f'# :index_pointing_at_the_viewer: 使用を停止しますか？', view=stop_button_view)
elif msg_split[0] == "/userbutton":
user_register_button_view = View(timeout=None)
user_register_button = discord.ui.Button(style=discord.ButtonStyle.green, label="ユーザー登録", custom_id="user_register")
user_register_button_view.add_item(user_register_button)
await self.get_channel(dislocker.server_config["bot"]["config_public_channel_id"]).send(f'# :index_pointing_at_the_viewer: ユーザー登録はお済ですか？', view=user_register_button_view)
elif message.channel.id == dislocker.server_config["bot"]["config_public_channel_id"]:
msg_split = message.content.split()
@ -507,6 +565,174 @@ class Bot(discord.Client):
else:
await message.channel.send("# :skull_crossbones: 登録できませんでした。\n\n-# もしかして...\n-# 手動でメンバーを登録したいですか？\n-# もしそうなら、このチャンネルにはその権限がありません。\n-# そのチャンネルに移動してから、もう一度試してみてください!")
async def on_button(self, interaction: Interaction):
custom_id = interaction.data["custom_id"]
custom_id_split = custom_id.split("_")
print(custom_id, custom_id_split[0])
if custom_id_split[0] == "pcregister":
device_register_view = View(timeout=15)
pc_number = custom_id_split[1]
print(custom_id_split)
for i in range(1, 20):
device_register_button = discord.ui.Button(style=discord.ButtonStyle.primary, label=f"{i}", custom_id=f"deviceregister_{str(pc_number)}_{i}")
device_register_view.add_item(device_register_button)
await interaction.response.send_message(f"# :keyboard: デバイス番号を選んでください！\n>>> # PC番号 | {str(pc_number)}", view=device_register_view, ephemeral=True)
elif custom_id_split[0] == "deviceregister":
pc_number = custom_id_split[1]
device_number = custom_id_split[2]
reason_register_view = View(timeout=15)
reason_button = discord.ui.Button(style=discord.ButtonStyle.primary, label="使用目的を入力する", custom_id=f"reasonregister_{str(pc_number)}_{str(device_number)}")
reason_register_view.add_item(reason_button)
await interaction.response.send_message(f"# :regional_indicator_q: 使用目的を書いてください！\n>>> # PC番号 | {str(pc_number)}\n# デバイス番号 | {str(device_number)}", view=reason_register_view, ephemeral=True)
elif custom_id_split[0] == "reasonregister":
pc_number = custom_id_split[1]
device_number = custom_id_split[2]
reason_input_form = Reason(title="Dislocker | 登録", pc_number=str(pc_number), device_number=str(device_number))
await interaction.response.send_modal(reason_input_form)
elif custom_id_split[0] == "stop":
print("STOP running")
pc_stop = self.stop(user_id=interaction.user.id)
print(pc_stop)
stop_view = View(timeout=15)
if pc_stop["result"] == "unused":
await interaction.response.send_message("# :shaking_face: 使用されていないようです...", ephemeral=True)
elif pc_stop["result"] == "ok":
await interaction.response.send_message(f":white_check_mark: PC番号 {pc_stop["pc_number"]} の使用が終了されました。", ephemeral=True)
await self.get_channel(dislocker.server_config["bot"]["log_channel_id"]).send(f':negative_squared_cross_mark: {pc_stop["name"]} さんがPC {pc_stop["pc_number"]} の使用を終了しました。')
else:
await interaction.response.send_message("# :skull_crossbones: 停止できませんでした。\n内部エラーが発生しています。", ephemeral=True)
elif custom_id_split[0] == "user" and custom_id_split[1] == "register":
print("User Register RUnning")
user_register = self.user_register(name=interaction.user.display_name, discord_user_name=interaction.user.name, discord_user_id=interaction.user.id)
if user_register["result"] == "ok":
await interaction.response.send_message(f"# :white_check_mark: ユーザー情報が登録されました。\n>>> ユーザー名:{interaction.user.display_name}", ephemeral=True)
elif user_register["result"] == "already_exists":
await interaction.response.send_message("# :no_entry: 登録できませんでした。\nもう登録されている可能性があります。", ephemeral=True)
else:
await interaction.response.send_message("# :no_entry: 登録できませんでした。\n内部エラーが発生しています。", ephemeral=True)
class Monitor():
def __init__(self) -> None:
pass
def start(self, **kwargs):
self.serach_time = kwargs["search_time"]
self.allowable_time = kwargs["allowable_time"]
serach_thread = threading.Thread(target=self.search)
serach_thread.start()
def search(self):
try:
while True:
cursor = dislocker.db.cursor()
cursor.execute("SELECT * FROM pc_list WHERE password_hash IS NOT NULL")
pc_list = cursor.fetchall()
print(pc_list)
print(len(pc_list))
if pc_list:
if len(pc_list) == 1:
user_id = pc_list[0][1]
cursor.execute("SELECT * FROM pc_usage_history WHERE member_id= %s AND end_use_time IS NULL ORDER BY id DESC LIMIT 1", (user_id,))
pc_usage = cursor.fetchall()
print(pc_usage)
start_time = pc_usage[0][4]
print(start_time)
print(type(start_time))
current_time = datetime.now()
time_difference = current_time - start_time
if time_difference >= timedelta(seconds=self.allowable_time):
cursor.execute("SELECT * FROM club_member WHERE id = %s", (user_id,))
user_info = cursor.fetchall()
stop = bot.stop(user_id=user_info[0][3])
bot.timeout_notify(pc_number=pc_list[0][0], discord_display_name=user_info[0][1])
time.sleep(1)
elif len(pc_list) >= 2:
for i in pc_list:
print(i)
user_id = i[1]
cursor.execute("SELECT * FROM pc_usage_history WHERE member_id= %s AND end_use_time IS NULL ORDER BY id DESC LIMIT 1", (user_id,))
pc_usage = cursor.fetchall()
print(pc_usage)
start_time = pc_usage[0][4]
print(start_time)
print(type(start_time))
current_time = datetime.now()
time_difference = current_time - start_time
if time_difference >= timedelta(seconds=self.allowable_time):
cursor.execute("SELECT * FROM club_member WHERE id = %s", (user_id,))
user_info = cursor.fetchall()
stop = bot.stop(user_id=user_info[0][3])
bot.timeout_notify(pc_number=i[0], discord_display_name=user_info[0][1])
time.sleep(0.1)
else:
result = {"result": "NONE"}
else:
result = {"result": "NONE"}
cursor.close()
print(result["result"])
time.sleep(self.serach_time)
except Exception as error:
print("自動停止処理中にエラーが発生しました。\nエラー内容")
print(str(error.__class__.__name__))
print(str(error.args))
print(str(error))
result = {"result": "error"}
dislocker.db.rollback()
finally:
cursor.close()
print(result["result"])
return result
class Reason(Modal):
def __init__(self, title: str, pc_number: str, device_number: str, timeout=15) -> None:
super().__init__(title=title, timeout=timeout)
print(pc_number)
print(device_number)
self.reason_input_form = TextInput(label="使用目的を入力してください", style=TextStyle.short, custom_id=f"register_{pc_number}_{device_number}")
self.add_item(self.reason_input_form)
async def on_submit(self, interaction: Interaction) -> None:
custom_id = interaction.data["components"][0]["components"][0]["custom_id"]
print(custom_id)
custom_id_split = custom_id.split("_")
pc_number = custom_id_split[1]
device_number = custom_id_split[2]
register = bot.register(user_id=interaction.user.id, name=interaction.user.name, display_name=interaction.user.display_name, pc_number=pc_number, device_number=device_number, detail=self.reason_input_form.value)
print(register["result"])
if register["result"] == "ok":
await interaction.response.send_message(f":white_check_mark: 使用が開始されました。\n>>> # パスワード | {register["password"]}\n## PC番号 | {pc_number}\n## デバイス番号 | {device_number}\n## 使用目的 | {self.reason_input_form.value}", ephemeral=True)
await bot.get_channel(dislocker.server_config["bot"]["log_channel_id"]).send(f':white_check_mark: {register["name"]} さんがPC {pc_number} の使用を開始しました。')
elif register["result"] == "pc_already_in_use_by_you":
await interaction.response.send_message(f"# :exploding_head: あなたはPCをもう使用されているようです。\n使用状態を解除するには 終了ボタン で使用終了をお知らせください。\n>>> # PC番号 | {register["pc_number"]}\n# デバイス番号 | {register["device_number"]}\n# 使用開始時刻 | {register["start_time"]}\n# 使用目的 | {register["detail"]}", ephemeral=True)
elif register["result"] == "pc_already_in_use_by_other":
await interaction.response.send_message(f"# :man_gesturing_no: そのPCは他のメンバーによって使用されています。\n別のPC番号を指定して、再度お試しください。", ephemeral=True)
elif register["result"] == "user_data_not_found":
await interaction.response.send_message("# :dizzy_face: ユーザーとして登録されていないようです。\n最初にサーバーで登録を行ってください。", ephemeral=True)
else:
await interaction.response.send_message("# :skull_crossbones: 登録できませんでした。\n内部エラーが発生しています。", ephemeral=True)
dislocker = DL()
if dislocker.init_result == "ok":
@ -514,6 +740,8 @@ if dislocker.init_result == "ok":
intents = discord.Intents.default()
intents.message_content = True
bot = Bot(intents=intents)
monitor = Monitor()
monitor.start(search_time=10, allowable_time=30)
bot.run(dislocker.server_config['bot']['token'])
else:
pass