Repository navigation
Expand file tree
/
Copy pathdiscordbot.py
More file actions
195 lines (164 loc) · 6.43 KB
/
Copy pathdiscordbot.py
File metadata and controls
195 lines (164 loc) · 6.43 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
import asyncio
import datetime
import logging
from pathlib import Path
from src.console import console
import discord
from discord.ext import commands
async def run(config):
"""
Starts the bot. If you want to create a database connection pool or other session for the bot to use,
create it here and pass it to the bot as a kwarg.
"""
intents = discord.Intents.default()
intents.message_content = True
bot = Bot(
config=config,
description=config['DISCORD']['discord_bot_description'],
intents=intents
)
try:
await bot.start(config['DISCORD']['discord_bot_token'])
except KeyboardInterrupt:
await bot.close()
class Bot(commands.Bot):
def __init__(self, *, config, description, intents):
super().__init__(
command_prefix=self.get_prefix_,
description=description,
intents=intents
)
self.start_time = None
self.app_info = None
self.config = config
async def setup_hook(self):
# Called before the bot connects to Discord
asyncio.create_task(self.track_start())
await self.load_all_extensions()
async def track_start(self):
"""
Waits for the bot to connect to discord and then records the time.
Can be used to work out uptime.
"""
await self.wait_until_ready()
self.start_time = datetime.datetime.utcnow()
async def get_prefix_(self, bot, message):
"""
A coroutine that returns a prefix.
"""
prefix = [self.config['DISCORD']['command_prefix']]
return commands.when_mentioned_or(*prefix)(bot, message)
async def load_all_extensions(self):
"""
Attempts to load all .py files in /cogs/ as cog extensions
"""
await self.wait_until_ready()
await asyncio.sleep(1) # ensure that on_ready has completed and finished printing
cogs = [x.stem for x in Path('cogs').glob('*.py')]
for extension in cogs:
try:
await self.load_extension(f'cogs.{extension}')
print(f'loaded {extension}')
except Exception as e:
error = f'{extension}\n {type(e).__name__} : {e}'
print(f'failed to load extension {error}')
print('-' * 10)
async def on_ready(self):
"""
This event is called every time the bot connects or resumes connection.
"""
print('-' * 10)
self.app_info = await self.application_info()
print(f'Logged in as: {self.user.name}\n'
f'Using discord.py version: {discord.__version__}\n'
f'Owner: {self.app_info.owner}\n')
print('-' * 10)
channel = self.get_channel(int(self.config['DISCORD']['discord_channel_id']))
if channel:
await channel.send(f'{self.user.name} is now online')
async def on_message(self, message):
"""
This event triggers on every message received by the bot. Including one's that it sent itself.
"""
if message.author.bot:
return # ignore all bots
await self.process_commands(message)
async def send_discord_notification(config, bot, message, debug=False, meta=None):
"""
Send a notification message to Discord channel.
Args:
bot: Discord bot instance (can be None)
message: Message string to send
meta: Optional meta dict for debug logging
Returns:
bool: True if message was sent successfully, False otherwise
"""
only_unattended = config.get('DISCORD', {}).get('only_unattended', False)
if only_unattended and meta and not meta.get('unattended', False):
return False
if not bot or not hasattr(bot, 'is_ready') or not bot.is_ready():
if debug:
console.print("[yellow]Discord bot not ready - skipping notifications")
return False
try:
channel_id = int(config['DISCORD']['discord_channel_id'])
channel = bot.get_channel(channel_id)
if channel:
await channel.send(message)
if debug:
console.print(f"[green]Discord notification sent: {message}")
return True
else:
console.print("[yellow]Discord channel not found")
return False
except Exception as e:
console.print(f"[yellow]Discord notification error: {e}")
return False
async def send_upload_status_notification(config, bot, meta):
"""Send Discord notification with upload status including failed trackers."""
only_unattended = config.get('DISCORD', {}).get('only_unattended', False)
if only_unattended and meta and not meta.get('unattended', False):
return False
if not bot or not hasattr(bot, 'is_ready') or not bot.is_ready():
return False
tracker_status = meta.get('tracker_status', {})
if not tracker_status:
return False
# Get list of trackers where upload is True
successful_uploads = [
tracker for tracker, status in tracker_status.items()
if status.get('upload', False)
]
# Get list of failed trackers with reasons
failed_trackers = []
for tracker, status in tracker_status.items():
if not status.get('upload', False):
if status.get('banned', False):
failed_trackers.append(f"{tracker} (banned)")
elif status.get('skipped', False):
failed_trackers.append(f"{tracker} (skipped)")
elif status.get('dupe', False):
failed_trackers.append(f"{tracker} (dupe)")
release_name = meta.get('name', meta.get('title', 'Unknown Release'))
message_parts = []
if successful_uploads:
message_parts.append(f"✅ **Uploaded to:** {', '.join(successful_uploads)} - {release_name}")
if failed_trackers:
message_parts.append(f"❌ **Failed:** {', '.join(failed_trackers)}")
if not message_parts:
return False
message = "\n".join(message_parts)
try:
channel_id = int(config['DISCORD']['discord_channel_id'])
channel = bot.get_channel(channel_id)
if channel:
await channel.send(message)
return True
except Exception as e:
console.print(f"[yellow]Discord notification error: {e}")
return False
if __name__ == '__main__':
# Only used when running discordbot.py directly
from data.config import config
logging.basicConfig(level=logging.INFO)
asyncio.run(run(config))