A friend of mine wants to include a task loop in his bot that let's users know when he goes live on twitch. The best way I can come up with doing this is a task loop that is triggered based off when he joins a specific voice channel within his discord server. How would I go about writing this? I've never dealt with the bot recognizing when someone joins a voice channel and I'm not too versed on task loops. Any examples would be helpful. This task loop is written in a cog as well. Thanks.
#Task loop triggered based off day of week and time of day
scarlet dove
normal shale
-d on_voice_state_update
muted thunderBOT
normal shale
But I mean.. you could simply have a task that just checks twitch API every so often and when the user is live, it sends an alert to your predefined channel. You don't need to trigger it.
But if you wanted to, ^^ is how you would get the voice status of the user to start and stop the task
scarlet dove
@commands.Cog.listener()
async def gone_live(self):
guild = self.bot.get_guild(data["guild_id"])
member = disnake.utils.get(guild.members, id=12345)
if member.voice is not None:
# execute code?
```would it be something like that?Not exactly since gone_live is not an event that currently exists.
scarlet dove
@commands.Cog.listener()
async def on_voice_state_update(self,member,before,after):
guild = self.bot.get_guild(data["guild_id"])
live_member = disnake.utils.get(guild.members, id=12345)
if member == live_member:
if member.voice is not None:
# execute code
I'm not too sure what to do with before and after
better yet
normal shale
a task would look like:
@tasks.loop(minutes=1)
async def go_live_check(self):
'''do checks at Twitch API to see if the user is live'''
async def on_voice_state_update(self,member):
if member.id == 12345:
# execute code
I've looked at the twitch api before and I could never understand it. I'll look at it again
normal shale
Having it trigger off a channel doesn't seem like a good idea, tbh.
scarlet dove
it was the best I could come up with lol I'm looking at https://pytwitchapi.readthedocs.io/en/stable/ right now and I'm just not seeing how or here I would write the code to check for when they go online. It keeps talking about an app_id and an app_secret and other items that I have zero idea how to get, or if mine would be needed, or theirs.
from twitchAPI.twitch import Twitch
from twitchAPI.helper import first
import asyncio
async def twitch_example():
# initialize the twitch instance, this will by default also create a app authentication for you
twitch = await Twitch('app_id', 'app_secret')
# call the API for the data of your twitch user
# this returns a async generator that can be used to iterate over all results
# but we are just interested in the first result
# using the first helper makes this easy.
user = await first(twitch.get_users(logins='your_twitch_user'))
# print the ID of your user or do whatever else you want with it
print(user.id)
# run this example
asyncio.run(twitch_example())
```like they have this example. I understand that I would put in my friends twitch user in the `user = await first(twitch.get_users(logins="your_twitch_user'))` but where would I get the app_id, and the app_secret or who's information would I use there?It would make sense to look at Twitch API docs
https://dev.twitch.tv/docs/api/reference/
Specifically, how authentication works:
https://dev.twitch.tv/docs/authentication/
scarlet dove
so I need to register an app first. Do I need to do that off my twitch account or theirs?
normal shale
Yours
or theirs. Or anyone's.
Doesn't matter.
Just needs to be associated with an account.
scarlet dove
ok so I've gotten the app registered, I got the client id and the client secret
yea I'm so lost. I'm in the get-started part of that link you sent me, and am on the sub-section of Get an OAuth token and I'm just lost
scarlet dove
```python
from twitchAPI.twitch import Twitch
from twitchAPI.helper import first...
using this same example, I put my client id and secret in my config.json file and pulled it into the test.py. so my code looks like
from twitchAPI.twitch import Twitch
from twitchAPI.helper import first
import asyncio
import json
with open('./server/config.json','r',encoding="utf-8-sig") as f:
data = json.load(f)
twitch_client_id = data["twitch_client_id"]
twitch_client_secret = data["twitch_client_secret"]
async def twitch_example():
twitch = await Twitch(twitch_client_id, twitch_client_secret)
user = await first(twitch.get_users(logins='spiderterrestrial'))
print(user.id)
asyncio.run(twitch_example())
```and when I ran the code to see if it would give me their `user.id`, I got this error message https://paste.pythondiscord.com/vudibumaxoam I not supposed to put spiderterrestrial in for your_twitch_user?
normal shale
TwitchAuthorizationException: Authentication failed with code 403
https://id.twitch.tv/oauth2/token?client_id=xxxxx&client_secret=xxxx,%20and&grant_type=client_credentials&scope=
doesn't look right
scarlet dove
I'm looking all over these docs and various pages throughout the dev.twitch.tv and the python one and I have no idea. Is there not an easier way to do this because this is like complicated. I'm not even going to lie. I have zero idea what to do
normal shale
The only time I ever needed to access Twitch API I made my own "library" to handle just what I needed. It's not pretty and I haven't touched it in almost a year, but it worked for me and might help get you what you're looking for. The only thing, it's was for a scheduled webhook so it wasn't designed to be asynchronous, though it could be used as a reference for you to create your own version
https://github.com/dlchamp/TwitchAnnounceWebhook/blob/main/src/twitch/twitch.py
It was specifically designed for getting steamers by ID or channel username and returning their status, etc.
scarlet dove
The only time I ever needed to access Twitch API I made my own "library" to hand...
I'm looking over your code, and it's honestly foreign to me. I'm not going to give up though. What I'm going to do for temporary times is have a command /live <title> <date> <time> <url_link> <details> that he can run for as many live events he wants to log. This will be stored in the json file and removed once the notification is sent. This information, once the command is executed, will not only store the information to the json file, but will also create an event for him automatically within the discord for everyone else to mark interested or not, then when it comes time for the live event, the bot will automatically pull the information from the json file, send the appropriate announcements to the correct channels, and then remove the information from the json file. The only problem I forsee with doing this for me is working with the datetime library and a task loop to check the stored date and time against the current date and time so that when it's the correct date and time, the task loop can do what it needs to do. thoughts?
normal shale
I guess that works, too?
scarlet dove
I'm sorry. I just can't figure out this twitch thing. like it's just so foreign to me, and I don't want to give up on it, but it's not something I'm going to grasp in a days time so I'm trying to think of a temporary alternative
normal shale
It's just a matter of getting the correct authentication information into the request with the correct headers and sending it to the correct endpoint. Once you have the JSON response, it's as simple as parsing a dict like normal.
scarlet dove
right, but I can't seem to get the correct authentication or even figure out how to go about getting it. Basically what I'm saying is I would need someone to get in voice with me, and either let me screen share, or they join a live session in vscode with me and show me how to do this. I'm not that great at self-teaching myself a new concept with an api I've never seen before
normal shale
you have to send the client id and secret to:
https://id.twitch.tv/oauth2/token?client_id=CLIENT_ID&client_secret=CLIENT_SECRET%grant_type=client_credentials
From there you'll get a json rsponse with an access token
That access token is what is used in future requests
scarlet dove
That access token is what is used in future requests
it's not working
even tried to do it via cURL
PS C:\Users\Mekas> curl -X POST 'https://id.twitch.tv/oauth2/token' \ -H 'Content-Type: application/x-www-form-urlencoded' \ -d 'client_id=doqh22tm1dyro8ozhlb5mji6n32jp6&client_secret=ma9hu9ixbo4f0z8ecd63yegyjllp40&grant_type=client_credentials'
Invoke-WebRequest : Cannot bind parameter 'Headers'. Cannot convert the "Content-Type:
application/x-www-form-urlencoded" value of type "System.String" to type "System.Collections.IDictionary".
At line:1 char:55
+ ... token' \ -H 'Content-Type: application/x-www-form-urlencoded' \ -d 'c ...
+ ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+ CategoryInfo : InvalidArgument: (:) [Invoke-WebRequest], ParameterBindingException
+ FullyQualifiedErrorId : CannotConvertArgumentNoMessage,Microsoft.PowerShell.Commands.InvokeWebRequestCommand
PS C:\Users\Mekas>
It should be an encoded url
scarlet dove
ok I got the oauth2 token, and the example code I showed earlier now returns the user id for his twitch username. so now what do I do
but where it says that the access token expires in 4844703, what does that mean?
normal shale
that means that the code will expire in that many seconds.
It would need to be refreshed
scarlet dove
so I need some kind of refresh thing to refresh the token
normal shale
Correct.
I don't have that because "skills issue" and I couldn't be bothered to not just generate a new token every 60 seconds when the script was running.
But it would make more sense to just reuse the available access code until it's expired, then refresh it as needed.
saves an API call until you just have to make it.
scarlet dove
import disnake, json, asyncio
from disnake.ext import commands, tasks
from twitchAPI.twitch import Twitch
from twitchAPI.helper import first
with open('./server/config.json','r',encoding="utf-8-sig") as f:
data = json.load(f)
twitch_id = data["twitch_client_id"]
twitch_secret = data["twitch_client_secret"]
class LiveEvents(commands.Cog):
def __init__(self,bot):
self.bot = bot
self.going_live.start()
@tasks.loop(minutes=1)
async def going_live(self):
self.bot.wait_until_ready()
twitch = await Twitch(twitch_id, twitch_secret)
user_id = (await first(twitch.get_users(logins='spiderterrestrial'))).id
print(user_id)
def setup(bot):
bot.add_cog(LiveEvents(bot))
```so how would I adapt this to capture that "expires_in" so that it will know when to refresh the token? I would like for this to happen automaticallythey give these two examples
normal shale
self.bot.wait_until_ready()
you don't need this unless you need to wait for the bot to connect, but since it's a different API, it isn't needed.
You'd also want to put it in a before.loop function so it's only ran one time on the first iteration of the loop and not every single time the loop runs.
scarlet dove
`self.bot.wait_until_ready()`
you don't need this unless you need to wait for th...
never head of a before.loop function
normal shale
That is for user authentication
You don't need user auth for what you're after.
-d before.loop
muted thunderBOT
scarlet dove
You don't need user auth for what you're after.
to refresh the auth token?
normal shale
the first screenshot.
That is for user authentication "User clicks an Authenticate with Twitch" button
Channel info and stream status doesn't require user auth.
last I look, at least.
the app ID and secret to get the access token, then use the access token and app ID to make the request
scarlet dove
I used the command prompt with the curl command to get the oauth token...
normal shale
Yeah.
scarlet dove
I needed to get the oauth token for some reason. I don't know what that reason is, but I needed it for something. it wouldn't give me the id for spiderterrestrial until I got that oauth token
scarlet dove
so that token expires and will need to be automatically refreshed otherwise it won't work
scarlet dove
ok
normal shale
Looks like the API has changed a bit.
https://api.twitch.tv/helix/users?login=spiderterrestrial works when I submit it with the proper headers with the acces code
{
"data": [
{
"id": "754596439",
"login": "spiderterrestrial",
"display_name": "spiderterrestrial",
"type": "",
"broadcaster_type": "affiliate",
"description": "WELCOME TO THE MOTHERSHIP : I'm ET, and welcome Terran's to my channel. We play some fun games with each other, take tokes in the process, and sometimes some beers! So, grab a pod-chair, and a bevy, and enjoy the vibes",
"profile_image_url": "https://static-cdn.jtvnw.net/jtv_user_pictures/a7eaeee3-c3be-4515-9d36-a3234cc6d6cb-profile_image-300x300.png",
"offline_image_url": "",
"view_count": 0,
"created_at": "2021-12-21T15:49:06Z"
}
]
}
from twitchAPI.twitch import Twitch
from twitchAPI.helper import first
import asyncio
import json
with open('./server/config.json','r',encoding="utf-8-sig") as f:
data = json.load(f)
twitch_client_id = data["twitch_client_id"]
twitch_client_secret = data["twitch_client_secret"]
async def twitch_example():
twitch = await Twitch(twitch_client_id, twitch_client_secret)
user = await first(twitch.get_users(logins='spiderterrestrial'))
print(user.id)
asyncio.run(twitch_example())
```I can run this now, and get his id, but I don't know what other fields I can get access to, which I only need his online/offline status to be able to do what I need to do. When he goes online on twitch, he wants the notifications sent to the appropriate channels within the discord serverand I'm presuming a notification sent to all of his twitch followers as well
import disnake
import json
import asyncio
from disnake.ext import commands, tasks
from twitchAPI.twitch import Twitch
from twitchAPI.helper import first
with open('./server/config.json','r',encoding="utf-8-sig") as f:
data = json.load(f)
twitch_id = data["twitch_client_id"]
twitch_secret = data["twitch_client_secret"]
class LiveEvents(commands.Cog):
def __init__(self,bot):
self.bot = bot
self.going_live.start()
@tasks.loop(minutes=1)
async def going_live(self):
twitch = await Twitch(twitch_id, twitch_secret)
user_id = (await first(twitch.get_users(logins='spiderterrestrial'))).id
print(user_id)
def setup(bot):
bot.add_cog(LiveEvents(bot))
```and that's what this task loop is going to be designed for. It's supposed to know when he's online, send the appropriate notifications. He hasn't told me what he wants to do when he goes offline, so I'm not doing anything with that, but I just need this task loop, or a cog listener to be triggered when he goes online on twitch to send out notificationsso like
@tasks.loop(minutes=1)
async def going_live(self):
user_online_status = # get online status
if user_online_status == True:
await self.send_notifications()
await self.send_twitch_notifications()
async def send_notifications(self):
guild = self.bot.get_guild(id=12345)
live_channel = disnake.utils.get(guild.text_channels, name="going-live")
spiders_id = disnake.utils.get(guild.members, id=12345)
embed = disnake.Embed(
color = disnake.Colour.random(),
title = "Some Title",
description = "Some Description"
).add_field(
name = "You Can View My Stream Here",
value = "https://twitch.tv/spiderterrestrial",
inline = False
).set_thumbnail(url = spiders_id.avatar)
await live_channel.send(embed=embed)
async def send_twitch_notifications(self):
streamers_followers = # obtain list of followers
for follower in streamers_followers:
await follower.send(twitch_default_im_live_notification)
so get online status, send notifications when online
import disnake, json, asyncio
from disnake.ext import commands, tasks
from twitchAPI.twitch import Twitch
from twitchAPI.helper import first
with open('./server/config.json','r',encoding="utf-8-sig") as f:
data = json.load(f)
twitch_id = data["twitch_client_id"]
twitch_secret = data["twitch_client_secret"]
guild_id = data["guild_id"]
class LiveEvents(commands.Cog):
def __init__(self,bot):
self.bot = bot
self.going_live.start()
@tasks.loop(minutes=1)
async def going_live(self):
# Obtain streamer object here and pass off to notification method
twitch = await Twitch(twitch_id, twitch_secret)
user = await first(twitch.get_users(logins='spiderterrestrial'))
await self.send_guild_notification(user)
async def send_guild_notification(self,user):
# obtain necessary user information for embed and list of followers
user_name = user.display_name
user_description = user.description
user_profile_image = user.profile_image_url
user_followers = []
embed = disnake.Embed(
color = disnake.Colour.random(),
title = f"{user_name} Is Going Live!",
description = user_description
).add_field(
name = "You Can View My Stream Here",
value = "https://twitch.tv/spiderterrestrial",
inline = False
).set_thumbnail(url = user_profile_image)
guild = await self.bot.get_guild(guild_id)
live_channel = disnake.utils.get(guild.text_channels, name="going-live")
await live_channel.send(embed=embed)
# get the twitch "i'm going live" notification that twitch sends out when a user goes live and send it to the followers
for follower in user_followers: pass
def setup(bot):
bot.add_cog(LiveEvents(bot))
```this is where I'm at so farso I'll need to incorporate this https://pytwitchapi.readthedocs.io/en/stable/modules/twitchAPI.twitch.html#twitchAPI.twitch.Twitch.get_channel_followers for getting the followers
normal shale
OK
I'm not familiar with that library, unfortunately. So, at this point, you'll need to read the docs.
scarlet dove
ok thank you for your help thus far ❤️
scarlet dove
I'm not familiar with that library, unfortunately. So, at this point, you'll ne...
I have a completely separate question for you using the datetime library if you're alright with that
normal shale
ok?
scarlet dove
from datetime import datetime, timedelta
start_time = "00:00 AM"
end_time = "23:45 PM"
start_time = datetime.strptime(start_time, "%H:%M %p")
end_time = datetime.strptime(end_time, "%H:%M %p")
times = []
while start_time <= end_time:
x = start_time.strftime("%H:%M %p")
times.append(x)
start_time += timedelta(minutes=15)
print(times)
```this code produces
```console
['00:00 AM', '00:15 AM', '00:30 AM', '00:45 AM', '01:00 AM', '01:15 AM', '01:30 AM', '01:45 AM', '02:00 AM', '02:15 AM', '02:30 AM', '02:45 AM', '03:00 AM', '03:15 AM', '03:30 AM', '03:45 AM', '04:00 AM', '04:15 AM', '04:30 AM', '04:45 AM', '05:00 AM', '05:15 AM', '05:30 AM', '05:45 AM', '06:00 AM', '06:15 AM', '06:30 AM', '06:45 AM', '07:00 AM', '07:15 AM', '07:30 AM', '07:45 AM', '08:00 AM', '08:15 AM', '08:30 AM', '08:45 AM', '09:00 AM', '09:15 AM', '09:30 AM', '09:45 AM', '10:00 AM', '10:15 AM', '10:30 AM', '10:45 AM', '11:00 AM', '11:15 AM', '11:30 AM', '11:45 AM', '12:00 PM', '12:15 PM', '12:30 PM', '12:45 PM', '13:00 PM', '13:15 PM', '13:30 PM', '13:45 PM', '14:00 PM', '14:15 PM', '14:30 PM', '14:45 PM', '15:00 PM', '15:15 PM', '15:30 PM', '15:45 PM', '16:00 PM', '16:15 PM', '16:30 PM', '16:45 PM', '17:00 PM', '17:15 PM', '17:30 PM', '17:45 PM', '18:00 PM', '18:15 PM', '18:30 PM', '18:45 PM', '19:00 PM', '19:15 PM', '19:30 PM', '19:45 PM', '20:00 PM', '20:15 PM', '20:30 PM', '20:45 PM', '21:00 PM', '21:15 PM', '21:30 PM', '21:45 PM', '22:00 PM', '22:15 PM', '22:30 PM', '22:45 PM', '23:00 PM', '23:15 PM', '23:30 PM', '23:45 PM']
```when ran. How would I convert these back to 12-hour format?Use %I instead of %H
scarlet dove
you're a life saver. Tyvm ❤️
scarlet dove
Use `%I` instead of `%H`
ok so now I'm having an issue with the time stamps.
code: https://pastebin.com/6r2VV4xK
error: https://pastebin.com/psxzcGmB
/create_live_event <name: str> <description: str> <channel> <start: str> <end: str>
The times that are input are written as like 12:45 PM but it's not wanting to accept them. Am I strf'ing the times wrong?
normal shale
I'd probably user a converter here that will verify that the entered string is the correct format and raise an error otherwise
scarlet dove
ok how would I go about doing that? I've never done that before sorry
if scheduled_start_time not in available_times or scheduled_end_time not in available_times:
return await inter.edit_original_message(
f"{inter.author.mention} Your Start/End Time Must Follow The Format - HH:MM AM/PM and Be In 15 Minute Increments"
)
```because I have this which takes those generated lists of time stamps created with
```python
async def get_times(self):
start_time = datetime.strptime("12:00 AM", "%I:%M %p")
end_time = datetime.strptime("11:45 PM", "%I:%M %p")
times = []
while start_time <= end_time:
x = start_time.strftime("%I:%M %p")
times.append(x)
start_time += timedelta(minutes=15)
return [i for i in times]
```to ensure that the user puts in a correct time with the appropriate format-d guild.create_scheduled_event
ohhh I'm missing the dates
well no because the parameters only call for times, not dates, but in the discord UI for creating a new event, it wants dates too
normal shale
scheduled_start_time = datetime.strptime(scheduled_start_time, "%I:%M %p"),
Let's say you put in 01:00 AM
The output is 1900-01-01 01:00:00
You'd have to do something like.
end_time = datetime.strptime("01:00 AM", "%I:%M %p").time()
current_date = datetime.now().date()
end_datetime = datetime.combine(current_date, end_time)
So that it would be "today" with the provided time.
scarlet dove
well I'm slow and don't know how to implement this. The function doesn't call for dates, and doesn't have a way for me to impliment dates into the newly created event so I'm lost
normal shale
You'd need to know the user's timezone
It does though.
It wants a datetime.datetime object, which includes date and time
It almost seems like.. you should just use Streamcord and let that bot handle go live notifications
scarlet dove
I'm wanting to learn this. This is a separate command from the event. With the task loop event, I'm just needing help getting the users online status and sending the notifications from there. What I'm currently working on now is a command to create a new event within the discord.
normal shale
I see.
then you need the date and time
And you need the user's timezone
or default to UTC and they have to make the conversion to input
otherwise, it will default to your bot's server's local time
Also, just an fyi.
You don't need an autocomplete for the channel.
You should be able to just to channel: disnake.VoiceChannel | disnake.StageChannel or channel: typing.Union[disnake.VoiceChannel, disnake.StageChannel] (if you're not on at least 3.10)
And it will provide a list to the user and provide you objects in the function so you don't have to get the channel
scarlet dove
@commands.slash_command(name="create_live_event",description="Allows a streamer to create a live event",guild_ids=guild_id)
async def create_live_event(self,inter,name: str,description: str,channel: typing.Union[disnake.VoiceChannel, disnake.StageChannel],scheduled_start_time: str,scheduled_end_time: str,scheduled_start_date: str,scheduled_end_date: str,time_zone: str):
await inter.response.defer(ephemeral=True)
start_time = datetime.strptime(scheduled_start_time, "%I:%M %p")
start_date = datetime.fromtimestamp(scheduled_start_date, time_zone)
end_time = datetime.strptime(scheduled_end_time, "%I:%M %p")
end_date = datetime.fromtimestamp(scheduled_end_date, time_zone)
combined_start = datetime.combine(start_date, start_time, time_zone)
combined_end = datetime.combine(end_date, end_time, time_zone)
await inter.guild.create_scheduled_event(
name = name,
description = description,
channel = disnake.utils.get(inter.guild.text_channels, name=channel),
scheduled_start_time = combined_start,
scheduled_end_time = combined_end
)
embed = disnake.Embed(
color = disnake.Colour.random(),
title = name,
description = description
).add_field(
name = "Time Frames",
value = f"Start Time: {scheduled_start_time}\nEnd Time: {scheduled_end_time}\nChannel: {channel}",
inline = False
).set_thumbnail(url = self.bot.user.avatar)
return await inter.edit_original_message(f"{inter.author.mention} Your Event Has Been Created!", embed=embed)
```ok so what about this?Why not just make it a datetime string?
5/25/2023 09:30 PM
instead of 4 args just to get two values
scarlet dove
because I'm not that smart. I'm learning how to do this
@commands.slash_command(name="create_live_event",description="Allows a streamer to create a live event",guild_ids=guild_id)
async def create_live_event(
self,
inter,
name: str,
description: str,
channel,
scheduled_start_time_and_date: str,
scheduled_end_time_and_date: str,
time_zone: str
):
await inter.response.defer(ephemeral=True)
start_time = datetime.strptime(scheduled_start_time_and_date, "%Y-%M-%d %I:%M %p")
end_time = datetime.strptime(scheduled_end_time_and_date, "%Y-%M-%s %I:%M %p")
await inter.guild.create_scheduled_event(
name = name,
description = description,
channel = disnake.utils.get(inter.guild.text_channels, name=channel),
scheduled_start_time = start_time,
scheduled_end_time = end_time
)
embed = disnake.Embed(
color = disnake.Colour.random(),
title = name,
description = description
).add_field(
name = "Time Frames",
value = f"Start Time: {scheduled_start_time}\nEnd Time: {scheduled_end_time}\nChannel: {channel}",
inline = False
).set_thumbnail(url = self.bot.user.avatar)
return await inter.edit_original_message(f"{inter.author.mention} Your Event Has Been Created!", embed=embed)
```what about this?Or just let them create events through the client? It's not really that complicated to need a command to make it not really any more convenient.
channel = disnake.utils.get(inter.guild.text_channels, name=channel), no longer needed since you have the channel object.
Or would if you type hinted it
scarlet dove
ok I'll just delete the file and forget it
normal shale
Alright. Take care
scarlet dove
/closed