SumGuyV5/python_scripts
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

file name change

Browse Source
This commit is contained in:
SumGuyV5
2026-07-23 17:34:36 +00:00
parent 7b7116dd16
commit 5e0459c6d7
1 changed files with 55 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files
twitch_video_info.py
+52 -34
View File
@@ -1,10 +1,9 @@
import asyncio
import json
import requests
import httpx # Switched from requests to prevent async loop freezing
import database
from twitchAPI.twitch import Twitch
from twitchAPI.helper import first
# Import the explicit VideoType Enum to prevent the AttributeError
from twitchAPI.type import VideoType
SECRETS = None
@@ -15,9 +14,7 @@ GAME_CACHE = {} # Local cache dictionary to store game_id -> game_name mapping
CHANNEL_NAME = "teampgp"
async def get_twitch():
global SECRETS
global TWITCH
global USER
global SECRETS, TWITCH, USER
if SECRETS is None:
with open('twitch_secrets.json', 'r') as f:
SECRETS = json.load(f)
@@ -44,8 +41,9 @@ async def get_game_name_by_id(game_id: str) -> str:
pass
return "Unknown Game"
async def get_vod_game_name(vod_id: str) -> str:
game_id = 0
async def get_vod_game_name(vod_id: str):
"""Asynchronously query Twitch GQL endpoint for VOD game metadata."""
game_id = "0"
game_name = "Unknown Game"
url = "https://gql.twitch.tv/gql"
@@ -58,7 +56,7 @@ async def get_vod_game_name(vod_id: str) -> str:
"operationName": "VideoMetadata",
"variables": {
"channelLogin": "",
"videoID": vod_id
"videoID": str(vod_id)
},
"extensions": {
"persistedQuery": {
@@ -68,17 +66,21 @@ async def get_vod_game_name(vod_id: str) -> str:
}
}]
response = requests.post(url, headers=headers, json=payload)
# Using httpx async client to prevent blocking the asyncio event loop
async with httpx.AsyncClient() as client:
try:
response = await client.post(url, headers=headers, json=payload)
if response.status_code == 200:
data = response.json()
# Parsing the Game ID out of the response array
video_info = data[0]['data']['video']
video_info = data[0].get('data', {}).get('video')
if video_info and video_info.get('game'):
game_id = video_info['game']['id']
game_id = str(video_info['game']['id'])
game_name = video_info['game']['displayName']
print(f"Game: {game_name} (ID: {game_id})")
print(f"GQL Found: {game_name} (ID: {game_id})")
else:
print("No game information found for this VOD.")
print(f"No game information found in GQL for VOD {vod_id}.")
except Exception as e:
print(f"Error fetching GQL metadata for VOD {vod_id}: {e}")
return game_id, game_name
@@ -91,18 +93,24 @@ async def get_streamer_vods():
all_vods = []
async for v in vod_generator:
# FIXED: Resolving game category using the automatic stream markers
# Resolving game category safely without freezing the event loop
game_id, game_name = await get_vod_game_name(v.id)
# Safely convert game_id to integer if possible, otherwise default to 0
try:
clean_game_id = int(game_id)
except ValueError:
clean_game_id = 0
vod_data = {
"id": v.id,
"title": v.title,
"created_at": str(v.published_at),
"view_count": int(v.view_count),
"view_count": int(v.view_count) if v.view_count else 0,
"duration": v.duration,
"url": v.url,
"thumbnail_url": v.thumbnail_url,
"game_id": int(game_id),
"game_id": clean_game_id,
"game_name": game_name,
"stream_id": str(v.stream_id) if v.stream_id else "0",
"creator_name": CHANNEL_NAME,
@@ -118,24 +126,26 @@ async def get_streamer_clips():
await get_twitch()
print(f"Starting clip extraction for {USER.display_name}...")
clip_generator = TWITCH.get_clips(broadcaster_id=USER.id, first=100)
all_clips = []
async for c in clip_generator:
# Clips DO have game_id attributes natively supported
game_name = await get_game_name_by_id(c.game_id)
try:
clean_game_id = int(c.game_id)
except (ValueError, TypeError):
clean_game_id = 0
clip_data = {
"id": c.id,
"title": c.title,
"created_at": str(c.created_at),
"view_count": int(c.view_count),
"view_count": int(c.view_count) if c.view_count else 0,
"duration": c.duration,
"url": c.url,
"thumbnail_url": c.thumbnail_url,
"game_id": int(c.game_id),
"game_id": clean_game_id,
"game_name": game_name,
"stream_id": "0",
"creator_name": c.creator_name,
@@ -150,22 +160,30 @@ async def get_streamer_clips():
async def main():
print("--- Script Started ---")
# 1. Pull clips (with categories)
#clips_list = await get_streamer_clips()
#print(f"\nSuccessfully received a list of {len(clips_list)} clips in main().")
#if clips_list:
# print(f"Top clip: '{clips_list[0]['title']}' (Game: {clips_list[0]['game_name']})")
db = database.Database()
# 2. Pull VODs (without categories)
vods_list = await get_streamer_vods()
for v in vods_list:
db.insert_video_record(v['id'], v['title'], v['created_at'], v['view_count'], v['duration'], v['url'], v['thumbnail_url'],
v['game_id'], v['game_name'], v['stream_id'], v['creator_name'], v['clip_is'])
#print(f"\nSuccessfully received a list of {len(vods_list)} VODs in main().")
for v in vods_list:
db.insert_video_record(
v['id'], v['title'], v['created_at'], v['view_count'], v['duration'],
v['url'], v['thumbnail_url'], v['game_id'], v['game_name'],
v['stream_id'], v['creator_name'], v['clip_is']
)
if vods_list:
print(f"Recent VOD: '{vods_list[0]['title']}'")
clips_list = await get_streamer_clips()
for v in clips_list:
db.insert_video_record(
v['id'], v['title'], v['created_at'], v['view_count'], v['duration'],
v['url'], v['thumbnail_url'], v['game_id'], v['game_name'],
v['stream_id'], v['creator_name'], v['clip_is']
)
if clips_list:
print(f"Recent VOD: '{clips_list[0]['title']}'")
if __name__ == "__main__":
asyncio.run(main())