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

update and test twitch clip downloading

Browse Source
This commit is contained in:
SumGuyV5
2026-07-20 12:42:52 -04:00
parent 0505d2a581
commit ff15ded0c9
9 changed files with 553 additions and 85 deletions
Download Patch File Download Diff File Expand all files Collapse all files
twitch_download_vod.py
+93 -42
View File
@@ -74,7 +74,7 @@ def download():
print(f"ID: {record_id} | Date: {record_date} | Game: {game_name} | Title: {title} | was successful")
DB.mark_as_downloaded(record_id)
else:
print(f"ID: {record_id} | Date: {record_date} | Game: {game_name} | Title: {title} | was Failed")
print(f"ID: {record_id} | Download Process failed. {output["stdout"]}. Error: {output["stderr"]}")
elif DB.table == "clips":
for slug, record_date, title, game_name, clip_by, views, downloaded, uploaded_yt, uploaded_shorts_yt in undownloaded:
@@ -98,49 +98,56 @@ def upload():
unuploaded = DB.get_unuploaded()
twitch_datetime = " #Twitch Every Friday and Sunday @7:30 EST https://twitch.tv/teampgp"
file_path = ""
title = ""
description = ""
categoryId = CategoryId.GAMING
privatcyStatus = 'private'
base_tags = ['gaming', 'TeamPGP', 'twitch', 'Level1Techs', 'twitch']
upload_queue = []
if DB.table == "vods":
for record_id, record_date, title, game_name, downloaded, uploaded_yt, chat_upload_yt in unuploaded_vods:
for record_id, record_date, title, game_name, downloaded, uploaded_yt, chats_upload_yt in unuploaded:
print(f"ID: {record_id} | Date: {record_date} | Game: {game_name} | Title: {title}")
file_path = f"save/vods/{record_id}/{title}.mp4"
title = title
description = f"Game: {game_name}, on {record_date}, #VODS #Twitch Every Friday and Sunday @7:30 EST https://twitch.tv/teampgp"
categoryId = CategoryId.GAMING
privatcyStatus = 'private'
tags = ['gaming', 'TeamPGP', f'{game_name}', 'twitch_vods', 'vods', 'Level1Techs', 'twitch']
description = f"Game: {game_name}, on {record_date}, #VODS {twitch_datetime}"
tags = list(base_tags)
tags.extend([f'{game_name}', 'twitch_vods', 'vods'])
tags.extend(top_hashtags(record_id))
output = uploader.upload_video(file_path, title, description, categoryId, privatcyStatus, tags)
upload_queue.append([record_id, file_path, title, categoryId, description, privatcyStatus, tags])
if output is True:
print(f"ID: {record_id} | Was successfully uploaded.")
DB.mark_as_uploaded(record_id)
else:
print(f"ID: {record_id} | Upload Process failed.")
elif DB.table == "clips":
for slug, record_date, title, game_name, clip_by, views, downloaded, uploaded_yt in unuploaded_clips:
for slug, record_date, title, game_name, clip_by, views, downloaded, uploaded_yt in unuploaded:
print(f"Slug: {slug} | Date: {record_date} | Game: {game_name} | By: {clip_by} | Views: {views} | Title: {title}")
file_path = f"save/clips/{slug}/{title}.mp4"
title = title
description = f"Game: {game_name}, Cliped By: {clip_by}, on {record_date}, #Shorts #Clips #Twitch Every Friday and Sunday @7:30 EST https://twitch.tv/teampgp"
categoryId = CategoryId.GAMING
privatcyStatus = 'private'
tags = ['shorts', 'gaming', 'TeamPGP', f'{game_name}', f'{clip_by}', 'twitch_clips', 'clips', 'Level1Techs', 'twitch']
description = f"Game: {game_name}, Cliped By: {clip_by}, on {record_date}, #Shorts #Clips {twitch_datetime}"
tags = list(base_tags)
tags.extend([f'{game_name}', 'twitch_clips', 'clips', f'{clip_by}', 'shorts'])
tags.extend(top_hashtags(slug))
tags.extend(top_hashtags({slug}))
upload_queue.append([slug, file_path, title, categoryId, description, privatcyStatus, tags])
#output = uploader.upload_video(file_path, title, categoryId, description, privatcyStatus, tags)
for db_id, file_path, title, categoryId, description, privatcyStatus, tags in upload_queue:
output = uploader.upload_video(file_path, title, categoryId, description, privatcyStatus, tags)
if output is True:
print(f"Title: {title} | Was successfully uploaded.")
DB.mark_as_uploaded(db_id)
else:
print(f"Title: {title} | Download Process failed.")
if output is True:
print(f"Slug: {slug} | Was successfully uploaded.")
DB.mark_as_uploaded(slug)
else:
print(f"Slug: {slug} | Upload Process failed.")
def create_chats():
pass
def create_chat_vods():
def create_shorts():
pass
def get_vod_ids_simplified(channel_name: str):
"""Queries Twitch's public endpoint directly for trending clips."""
session = requests.Session()
url = "https://gql.twitch.tv/gql"
@@ -150,9 +157,8 @@ def get_vod_ids_simplified(channel_name: str):
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
"Content-Type": "text/plain"
}
# A simplified query string that hardcodes the ARCHIVE type filter into the request structure
query_string = """
vods_query_string = """
query GetChannelVideos($login: String!, $limit: Int!) {
user(login: $login) {
videos(first: $limit, types: [ARCHIVE]) {
@@ -171,12 +177,44 @@ def get_vod_ids_simplified(channel_name: str):
}
"""
clips_query_string = """
query GetChannelClips($login: String!, $limit: Int!) {
user(login: $login) {
clips(first: $limit, criteria: { period: ALL_TIME }) {
edges {
node {
slug
title
createdAt
viewCount
game {
displayName
}
curator {
login
}
}
}
}
}
}
"""
limit = 50
query_string = ""
if DB.table == "vods":
query_string = vods_query_string
limit = 50
elif DB.table == "clips":
query_string = clips_query_string
limit = 40
# A simplified query string that hardcodes the ARCHIVE type filter into the request structure
payload = [{
"operationName": "GetChannelVideos",
"query": query_string,
"variables": {
"login": channel_name.lower(),
"limit": 50
"limit": limit
}
}]
@@ -201,9 +239,13 @@ def get_vod_ids_simplified(channel_name: str):
if not user_data:
print(f"Channel '{channel_name}' not found.")
return []
edges = user_data['videos']['edges']
vod_ids = []
edges = None
if DB.table == "vods":
edges = user_data['videos']['edges']
elif DB.table == "clips":
edges = user_data['clips']['edges']
video_ids = []
print(f"--- Latest VODs for {channel_name} ---")
for edge in edges:
@@ -212,15 +254,24 @@ def get_vod_ids_simplified(channel_name: str):
# Safe extraction in case a VOD has no category set (Just Chatting, Uncategorized, etc.)
game_info = node.get('game')
game_name = game_info.get('displayName') if game_info else "Unknown/No Category"
if DB.table == "vods":
print(f"ID: {node['id']} | Date: {node['publishedAt']} | Game: {game_name} | Title: {node['title']}")
print(f"ID: {node['id']} | Date: {node['publishedAt']} | Game: {game_name} | Title: {node['title']}")
# Pass game_name to your database logic
DB.insert_vods_record(node['id'], node['publishedAt'], node['title'], game_name)
#insert_record(node['id'], node['publishedAt'], node['title'], game_name)
video_ids.append(node['id'])
elif DB.table == "clips":
# Safe extraction in case the curator account was deleted/missing
curator_info = node.get('curator')
clip_by = curator_info.get('login') if curator_info else "Unknown Creator"
print(f"Slug: {node['slug']} | Date: {node['createdAt']} | Game: {game_name} | By: {clip_by} | Views: {node['viewCount']} | Title: {node['title']}")
DB.insert_clips_record(node['slug'], node['createdAt'], node['title'], game_name, clip_by, int(node['viewCount']))
video_ids.append(node['slug'])
# Pass game_name to your database logic
DB.insert_vods_record(node['id'], node['publishedAt'], node['title'], game_name)
#insert_record(node['id'], node['publishedAt'], node['title'], game_name)
vod_ids.append(node['id'])
return vod_ids
return video_ids
except Exception as e:
print(f"An unexpected error occurred: {e}")
@@ -232,4 +283,4 @@ if __name__ == "__main__":
get_vod_ids_simplified(CHANNEL_NAME)
download_vods()
upload_vods()
#close_database()
DB.close_database()