-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathbot.py
More file actions
341 lines (272 loc) Β· 9.64 KB
/
bot.py
File metadata and controls
341 lines (272 loc) Β· 9.64 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
import os
import asyncio
import threading
import signal
import time
import platform
import socket
import getpass
from pathlib import Path
import psutil
import discord
from discord.ext import commands
from dotenv import load_dotenv
load_dotenv()
TOKEN = os.getenv("DISCORD_TOKEN")
if not TOKEN:
raise RuntimeError("DISCORD_TOKEN environment variable not set")
CHANNEL_ID = os.getenv("CHANNEL_ID")
if not CHANNEL_ID:
raise RuntimeError("CHANNEL_ID environment variable not set")
intents = discord.Intents.default()
intents.message_content = True
client = commands.Bot(command_prefix="!", intents=intents)
client.remove_command('help')
@client.event
async def on_ready():
print(f"Bot logged in as {client.user}")
@client.command(name="sysinfo")
async def sysinfo(ctx):
"""
Grabs the PC current system information
"""
info = get_system_info()
message = "**π₯οΈ System Information**\n\n"
for key, value in info.items():
message += f"**{key}:** {value}\n"
await ctx.send(message)
@client.command(name="pwd")
async def pwd(ctx):
"""
Prints currnet working directory
"""
await ctx.send(f"π **Current working directory:**\n```\n{os.getcwd()}\n```")
@client.command(name="ls")
async def ls(ctx):
"""
List the items in the current directory
"""
cwd = Path.cwd()
dirs = []
files = []
for item in cwd.iterdir():
if item.is_dir():
dirs.append(f"[DIR] {item.name}")
else:
files.append(f" {item.name}")
output = "\n".join(sorted(dirs) + sorted(files)) or "(empty)"
await ctx.send(f"π **Listing: {cwd}**\n```\n{output}\n```")
@client.command(name="cd")
async def cd(ctx, *, path: str):
"""
Change directory
e.g !cd /home
use !cd ../ to go back!
"""
new_path = (Path.cwd() / path).resolve()
if not new_path.exists():
await ctx.send(f"β Path does not exist:\n```\n{new_path}\n```")
return
if not new_path.is_dir():
await ctx.send(f"β Not a directory:\n```\n{new_path}\n```")
return
os.chdir(new_path)
await ctx.send(f"π **Directory changed to:**\n```\n{new_path}\n```")
@client.command(name="whoami")
async def whoami(ctx):
"""
Prints the current user login
"""
user = getpass.getuser()
await ctx.send(f"π€ **Current user:**\n```\n{user}\n```")
@client.command(name="search")
async def search(ctx, *, query: str):
"""
Searches for the item
e.g: !search hi.txt
"""
root = Path.cwd()
pattern = query.lower()
max_depth = 5
results = []
try:
fast_search(root, pattern, max_depth, results)
except Exception:
pass
if not results:
await ctx.send(f"π **No results for:** `{query}`")
return
output = "π **Search results:**\n```\n" + "\n".join(results) + "\n```"
if len(output) > 1900:
chunks = [output[i:i+1900] for i in range(0, len(output), 1900)]
for part in chunks:
await ctx.send(part)
else:
await ctx.send(output)
@client.command(name="export_here")
async def export_here(ctx):
"""
Export all the files in the current directory
"""
cwd = Path.cwd()
files = [f for f in cwd.iterdir() if f.is_file()]
if not files:
await ctx.send("π **No files to export in this directory.**")
return
preview = "\n".join(f"- {f.name}" for f in files)
await ctx.send(
f"π¦ **The following files in `{cwd}` will be exported:**\n"
f"```\n{preview}\n```\n"
f"Type **yes** to confirm."
)
def check(msg):
return msg.author == ctx.author and msg.channel == ctx.channel
try:
reply = await client.wait_for("message", timeout=20, check=check)
except asyncio.TimeoutError:
await ctx.send("β **Timed out. Export cancelled.**")
return
if reply.content.lower().strip() != "yes":
await ctx.send("β **Export cancelled.**")
return
zip_path = cwd / "export_here.zip"
try:
import zipfile
with zipfile.ZipFile(zip_path, "w", zipfile.ZIP_DEFLATED) as z:
for f in files:
z.write(f, arcname=f.name)
await ctx.send("π€ **Exporting files...**")
await ctx.send(file=discord.File(zip_path))
except Exception as e:
await ctx.send(f"β Error: `{e}`")
finally:
if zip_path.exists():
zip_path.unlink()
await ctx.send("β
**Export complete.**")
@client.command(name="help")
async def help_command(ctx):
"""Display all available commands"""
help_message = """
```ansi
[1;2m[1;37mAvailable Commands:[0m[0m
[4;2mSystem Information:[0m
[2;31m!sysinfo [0m [2;37m- Display detailed system information (hostname, IP, OS, CPU, RAM, disk)[0m
[2;31m!whoami [0m [2;37m- Show current user login[0m
[4;2mFile Navigation:[0m
[2;31m!pwd[0m [2;37m- Show current working directory[0m
[2;31m!ls[0m [2;37m- List files and directories in the current location[0m
[2;31m!cd[0m [2;32m<path>[0m [2;37m- Change directory (use !cd ../ to go back)[0m
[2;31m!tree[0m [2;32m[depth] [-a][0m [2;37m- Show directory tree structure[0m
[2;37mExamples: !tree, !tree 5, !tree -a, !tree 4 -a[0m
[4;2mFile Operations:[0m
[2;31m!search[0m [2;32m<query>[0m [2;37m- Search for files/folders by name in current directory[0m
[2;37mExample: !search hi.txt[0m
[2;31m!export_here [0m [2;37m- Export all files in the current directory as a zip file[0m
[4;2mOther:[0m
[2;31m!help[0m [2;37m- Show this help message[0m
[1;33mTips:[0m
[2;37m- All commands are case-sensitive and must start with !
- Use !cd ../ to navigate to parent directory
- Use !tree -a to show hidden files[0m
```
"""
await ctx.send(help_message)
def generate_tree(path: Path, prefix: str = "", depth: int = 0, max_depth: int = 3, show_hidden: bool = False):
"""Recursive directory tree generator with clean formatting."""
if depth > max_depth:
return prefix + "βββ ... (depth limit reached)\n"
tree_output = ""
items = [
item for item in path.iterdir()
if show_hidden or not item.name.startswith(".")
]
items = sorted(items, key=lambda x: (not x.is_dir(), x.name.lower()))
count = len(items)
for idx, item in enumerate(items):
connector = "βββ " if idx == count - 1 else "βββ "
tree_output += prefix + connector + item.name + "\n"
if item.is_dir():
extension = " " if idx == count - 1 else "β "
tree_output += generate_tree(
item,
prefix + extension,
depth + 1,
max_depth,
show_hidden
)
return tree_output
@client.command(name="tree")
async def tree(ctx, *args):
"""
Prints a tree of the current directory
!tree β hidden ignored, depth=3
!tree 4 β depth=4
!tree -a β show hidden
!tree 4 -a β depth=4 + hidden
"""
depth = 3
show_hidden = False
for arg in args:
if arg.isdigit():
depth = int(arg)
if arg in ("-a", "--all", "a"):
show_hidden = True
root = Path.cwd()
output = f"π **Tree for: {root}**\n```\n"
output += generate_tree(root, max_depth=depth, show_hidden=show_hidden)
output += "```"
# Discord message size limit
if len(output) > 1800:
chunks = [output[i:i+1800] for i in range(0, len(output), 1800)]
for part in chunks:
await ctx.send(part)
else:
await ctx.send(output)
def fast_search(root: Path, pattern: str, depth: int, results: list):
if depth < 0:
return
for item in root.iterdir():
name = item.name.lower()
if pattern in name:
results.append(str(item))
if item.is_dir():
try:
fast_search(item, pattern, depth - 1, results)
except Exception:
continue
def get_system_info():
info = {}
info["Hostname"] = socket.gethostname()
info["IP Address"] = socket.gethostbyname(info["Hostname"])
info["OS"] = platform.platform()
info["Architecture"] = platform.machine()
info["Processor"] = platform.processor()
info["Physical Cores"] = psutil.cpu_count(logical=False)
info["Total Cores"] = psutil.cpu_count(logical=True)
ram = psutil.virtual_memory()
info["Total RAM"] = f"{ram.total / (1024**3):.2f} GB"
info["Available RAM"] = f"{ram.available / (1024**3):.2f} GB"
disk = psutil.disk_usage('/')
info["Disk Total"] = f"{disk.total / (1024**3):.2f} GB"
info["Disk Free"] = f"{disk.free / (1024**3):.2f} GB"
return info
shutdown_flag = False
def handle_exit(signum, frame):
global shutdown_flag
shutdown_flag = True
print("\nShutting down cleanly...")
signal.signal(signal.SIGINT, handle_exit)
signal.signal(signal.SIGTERM, handle_exit)
def run_bot():
asyncio.run(client.start(TOKEN))
threading.Thread(target=run_bot, daemon=True).start()
while not shutdown_flag:
try:
time.sleep(0.5)
except KeyboardInterrupt:
break
try:
asyncio.run(client.close())
except Exception:
pass
print("Bot stopped.")