Hi
i have this code
async def on_ready_2(ready_event: EventData):
headers = {
'Authorization': f'Bearer #',
'Client-Id': '#'
}
async def fetch_chatters_cursor(after=None):
url = f'https://api.twitch.tv/helix/chat/chatters?broadcaster_id={br_id}&moderator_id={mod_id}&first={first}'
if after:
url += f'&after={after}'
response = requests.get(url, headers=headers)
if response.status_code != 200:
print(f'Error: {response.status_code} - {response.text}')
return None, None
data = response.json()
cursor = data.get('pagination', {}).get('cursor')
users = [chatter['user_id'] for chatter in data.get('data', [])]
return users, cursor
while True:
stream_found = False
async for streamonline in twitch.get_streams(user_id=br_id):
chater = await twitch.get_chatters(broadcaster_id=br_id, moderator_id=mod_id, first=first)
us_name = [chatter.user_id for chatter in chater.data]
logger.success(f"First chatters list: {len(us_name)} users")
main_list = list(us_name)
cursor = None
while True:
users, cursor = await fetch_chatters_cursor(cursor)
if users is None:
break
logger.success(f"Next chatters list: {len(users)} users")
main_list.extend(users)
if cursor is None:
break
main_list = list(set(main_list))
logger.success(f"Main chatters list: {len(main_list)} users")
for user_id in main_list:
add_name(user_id)
stream_found = True
await asyncio.sleep(60)
if not stream_found:
logger.info("Stream offline")
await asyncio.sleep(300)
Over time, the token in the headers stops working, returning “Invalid token,” but the bot itself continues to operate normally using this token. What should be done?
async def run_bot():
init_db()
target_scope = [AuthScope.CHAT_READ, AuthScope.CHAT_EDIT, AuthScope.CHANNEL_BOT, AuthScope.USER_BOT,
AuthScope.USER_WRITE_CHAT, AuthScope.MODERATOR_READ_CHATTERS, AuthScope.USER_READ_FOLLOWS]
token = "#"
refresh_token = "#"
await twitch.set_user_authentication(token, target_scope, refresh_token)
chat = await Chat(twitch)
chat.register_event(ChatEvent.READY, on_ready)
chat.register_event(ChatEvent.READY, on_ready_2)
chat.start()
try:
await asyncio.Event().wait()
except (KeyboardInterrupt, asyncio.CancelledError):
logger.info("Get turn off signal")
finally:
chat.stop()
await twitch.close()
logger.info("Bot off")
This is the code of the bot in operation.
Also, I think due to too frequent attempts to create a new token to avoid this error, when creating a new token through
def get_tokens(client_id, client_secret, code, redirect_uri):
url = "https://id.twitch.tv/oauth2/token"
params = {
"client_id": client_id,
"client_secret": client_secret,
"code": code,
"grant_type": "authorization_code",
"redirect_uri": redirect_uri
}
response = requests.post(url, params=params)
return response.json()
tokens_info = get_tokens("#", "#", "#", "http://localhost:17563")
print(tokens_info)
I started receiving the error “invalid authorization code” even though I was constantly generating a new one.