-
Notifications
You must be signed in to change notification settings - Fork 1
/
main.py
289 lines (237 loc) · 12.2 KB
/
main.py
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
import os
import re
import nextcord
import requests
from nextcord.ext import commands, tasks
from googleapiclient.discovery import build
intents = nextcord.Intents.default()
bot = commands.Bot(command_prefix="!", intents=intents)
YOUTUBE_API_KEY = os.getenv('YOUTUBE_API_KEY')
if not YOUTUBE_API_KEY:
print("Error: YOUTUBE_API_KEY is missing!")
youtube = build('youtube', 'v3', developerKey=YOUTUBE_API_KEY)
TWITCH_CLIENT_ID = os.getenv('TWITCH_CLIENT_ID')
TWITCH_CLIENT_SECRET = os.getenv('TWITCH_CLIENT_SECRET')
TWITCH_OAUTH_TOKEN = None
tracked_channels = {'youtube': {}, 'twitch': {}}
notification_channels = {}
tracked_types = {}
last_live_streams = {}
notification_messages = {}
def get_twitch_oauth_token():
global TWITCH_OAUTH_TOKEN
url = "https://id.twitch.tv/oauth2/token"
params = {
'client_id': TWITCH_CLIENT_ID,
'client_secret': TWITCH_CLIENT_SECRET,
'grant_type': 'client_credentials'
}
response = requests.post(url, params=params)
data = response.json()
TWITCH_OAUTH_TOKEN = data['access_token']
def check_twitch_stream(channel_name):
global TWITCH_OAUTH_TOKEN
url = f"https://api.twitch.tv/helix/streams"
headers = {
'Client-ID': TWITCH_CLIENT_ID,
'Authorization': f'Bearer {TWITCH_OAUTH_TOKEN}'
}
params = {'user_login': channel_name}
response = requests.get(url, headers=headers, params=params)
if response.status_code == 401:
get_twitch_oauth_token()
headers['Authorization'] = f'Bearer {TWITCH_OAUTH_TOKEN}'
response = requests.get(url, headers=headers, params=params)
data = response.json()
if 'data' in data and len(data['data']) > 0:
stream_data = data['data'][0]
stream_title = stream_data['title']
stream_thumbnail = stream_data['thumbnail_url'].replace('{width}', '1280').replace('{height}', '720')
stream_url = f"https://www.twitch.tv/{channel_name}"
return (True, stream_title, stream_thumbnail, stream_url)
return (False, None, None, None)
def check_video_uploads(channel_id):
try:
request = youtube.search().list(
part="snippet",
channelId=channel_id,
type="video",
maxResults=1,
order="date"
)
response = request.execute()
if 'items' in response and len(response['items']) > 0:
video = response['items'][0]
video_id = video['id']['videoId']
video_title = video['snippet']['title']
video_thumbnail = video['snippet']['thumbnails']['high']['url']
video_url = f"https://www.youtube.com/watch?v={video_id}"
return (True, video_title, video_thumbnail, video_url, video_id)
except Exception as e:
print(f"Error fetching video upload data: {e}")
return (False, None, None, None, None)
def check_video_details(video_id):
try:
request = youtube.videos().list(
part="contentDetails",
id=video_id
)
response = request.execute()
if 'items' in response and len(response['items']) > 0:
duration = response['items'][0]['contentDetails']['duration']
return duration
except Exception as e:
print(f"Error fetching video details: {e}")
return None
def is_short(duration):
if duration and 'PT' in duration:
minutes = re.search(r'(\d+)M', duration)
seconds = re.search(r'(\d+)S', duration)
if not minutes and seconds and int(seconds.group(1)) <= 60:
return True
return False
@bot.slash_command(name="set_notification_channel", description="Set the channel where notifications will be sent.")
async def set_notification_channel(interaction: nextcord.Interaction, channel: nextcord.TextChannel):
guild_id = interaction.guild.id
notification_channels[guild_id] = channel.id
await interaction.response.send_message(f"Notifications will now be sent to {channel.mention}")
@bot.slash_command(name="add_channel", description="Add a YouTube or Twitch channel to track.")
async def add_channel(interaction: nextcord.Interaction, platform: str, channel_name: str):
await interaction.response.defer()
guild_id = interaction.guild.id
platform = platform.lower()
try:
if platform == 'youtube':
channel_id, channel_title = get_youtube_channel(channel_name)
if not channel_id:
await interaction.followup.send(f"Error: Unable to find YouTube channel '{channel_name}'.")
return
tracked_channels['youtube'].setdefault(guild_id, []).append({'id': channel_id, 'title': channel_title})
await interaction.followup.send(f"Now tracking YouTube channel: {channel_title}")
elif platform == 'twitch':
tracked_channels['twitch'].setdefault(guild_id, []).append(channel_name.lower())
await interaction.followup.send(f"Now tracking Twitch channel: {channel_name}")
else:
await interaction.followup.send("Invalid platform. Please use 'youtube' or 'twitch'.")
except Exception as e:
print(f"Error in add_channel: {e}")
await interaction.followup.send("An error occurred while adding the channel. Please try again.")
@bot.slash_command(name="list_channels", description="List all YouTube and Twitch channels being tracked.")
async def list_channels(interaction: nextcord.Interaction):
guild_id = interaction.guild.id
tracked_youtube = tracked_channels.get('youtube', {}).get(guild_id, [])
tracked_twitch = tracked_channels.get('twitch', {}).get(guild_id, [])
youtube_list = [channel['title'] for channel in tracked_youtube]
if not youtube_list and not tracked_twitch:
await interaction.response.send_message("No channels are currently being tracked.")
return
youtube_channels = "\n".join(youtube_list) if youtube_list else "None"
twitch_channels = "\n".join(tracked_twitch) if tracked_twitch else "None"
await interaction.response.send_message(f"**YouTube Channels:**\n{youtube_channels}\n\n**Twitch Channels:**\n{twitch_channels}")
class RemoveChannelSelect(nextcord.ui.Select):
def __init__(self, guild_id, youtube_channels, twitch_channels):
options = []
for channel in youtube_channels:
options.append(nextcord.SelectOption(label=channel['title'], description="YouTube", value=f"youtube|{channel['id']}"))
for channel_name in twitch_channels:
options.append(nextcord.SelectOption(label=channel_name, description="Twitch", value=f"twitch|{channel_name}"))
super().__init__(placeholder="Select a channel to remove...", options=options)
async def callback(self, interaction: nextcord.Interaction):
guild_id = interaction.guild.id
selected_value = self.values[0]
platform, channel_id_or_name = selected_value.split("|")
if platform == "youtube":
tracked_channels['youtube'][guild_id] = [ch for ch in tracked_channels['youtube'][guild_id] if ch['id'] != channel_id_or_name]
await interaction.followup.send(f"Removed YouTube channel with ID: {channel_id_or_name}")
elif platform == "twitch":
tracked_channels['twitch'][guild_id].remove(channel_id_or_name)
await interaction.followup.send(f"Removed Twitch channel: {channel_id_or_name}")
@bot.slash_command(name="remove_channel", description="Remove a YouTube or Twitch channel from tracking.")
async def remove_channel(interaction: nextcord.Interaction):
guild_id = interaction.guild.id
youtube_channels = tracked_channels.get('youtube', {}).get(guild_id, [])
twitch_channels = tracked_channels.get('twitch', {}).get(guild_id, [])
if not youtube_channels and not twitch_channels:
await interaction.response.send_message("No channels are currently being tracked.")
return
view = nextcord.ui.View()
view.add_item(RemoveChannelSelect(guild_id, youtube_channels, twitch_channels))
await interaction.response.send_message("Select the channel you want to remove:", view=view)
@bot.slash_command(name="set_notification_message", description="Customize the notification message.")
async def set_notification_message(interaction: nextcord.Interaction, message: str):
guild_id = interaction.guild.id
notification_messages[guild_id] = message
await interaction.followup.send("Notification message updated!")
class Poll(nextcord.ui.View):
def __init__(self):
super().__init__()
self.value = None
@nextcord.ui.button(label="Yes", style=nextcord.ButtonStyle.green)
async def yes(self, button: nextcord.ui.Button, interaction: nextcord.Interaction):
await interaction.response.send_message("You voted Yes!")
@nextcord.ui.button(label="No", style=nextcord.ButtonStyle.red)
async def no(self, button: nextcord.ui.Button, interaction: nextcord.Interaction):
await interaction.response.send_message("You voted No!")
@bot.slash_command(name="poll", description="Create a poll for a live stream.")
async def poll(interaction: nextcord.Interaction):
view = Poll()
await interaction.response.send_message("Are you watching the stream?", view=view)
@tasks.loop(minutes=3)
async def check_streams():
print("Checking for new YouTube uploads, live streams, and Twitch streams...")
for guild_id, channels in tracked_channels['youtube'].items():
for channel in channels:
channel_id = channel['id']
is_video, video_title, video_thumbnail, video_url, video_id = check_video_uploads(channel_id)
if not is_video:
last_live_streams[channel_id] = None
continue
if last_live_streams.get(channel_id) == video_id:
continue
guild = bot.get_guild(guild_id)
if guild and is_video:
last_live_streams[channel_id] = video_id
video_duration = check_video_details(video_id)
if is_short(video_duration):
title_prefix = "New Short Uploaded"
embed_color = nextcord.Color.green()
else:
title_prefix = "New Video Uploaded"
embed_color = nextcord.Color.blue()
embed = nextcord.Embed(
title=f"{title_prefix}: {video_title}",
description=f"[Click to watch the video]({video_url})",
color=embed_color
)
embed.set_image(url=video_thumbnail)
channel = bot.get_channel(notification_channels.get(guild_id, guild.text_channels[0].id))
await channel.send(content=notification_messages.get(guild_id, "@everyone"), embed=embed)
for guild_id, channels in tracked_channels['twitch'].items():
for channel_name in channels:
is_live, stream_title, stream_thumbnail, stream_url = check_twitch_stream(channel_name)
if is_live and last_live_streams.get(channel_name) != stream_url:
last_live_streams[channel_name] = stream_url
guild = bot.get_guild(guild_id)
if guild:
embed = nextcord.Embed(
title=f"Twitch Stream Live: {stream_title}",
description=f"[Click to watch the stream]({stream_url})",
color=nextcord.Color.purple()
)
embed.set_image(url=stream_thumbnail)
channel = bot.get_channel(notification_channels.get(guild_id, guild.text_channels[0].id))
await channel.send(content=notification_messages.get(guild_id, "@everyone"), embed=embed)
@bot.slash_command(name="ping", description="Ping the bot to check if it's online.")
async def ping(interaction: nextcord.Interaction):
await interaction.response.send_message("Pong!")
@bot.event
async def on_ready():
print(f'Logged in as {bot.user.name}')
await bot.change_presence(activity=nextcord.Activity(type=nextcord.ActivityType.watching, name="YouTube and Twitch streams"))
get_twitch_oauth_token()
check_streams.start()
DISCORD_BOT_TOKEN = os.getenv('DISCORD_BOT_TOKEN')
if not DISCORD_BOT_TOKEN:
print("Error: DISCORD_BOT_TOKEN is missing!")
else:
bot.run(DISCORD_BOT_TOKEN)