advtech/YouTubeChannelPoints
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

FIX: Buffed up the code to make sure it handles dual streams better

Browse Source
This commit is contained in:
Dan 2024-09-02 20:02:48 -04:00
parent a413e70928
commit 0a5a94b03d
1 changed files with 27 additions and 25 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

36
main.py
View File

@ -13,7 +13,7 @@ DATABASE_FILE = 'channel_points.db'
# YouTube Channel ID or Handle # YouTube Channel ID or Handle
CHANNEL_HANDLE = 'UCsVJcf4KbO8Vz308EKpSYxw' CHANNEL_HANDLE = 'UCsVJcf4KbO8Vz308EKpSYxw'
STREAM_KEYWORD = "Live" # Keyword to identify the correct live stream
def get_authenticated_service(): def get_authenticated_service():
flow = InstalledAppFlow.from_client_secrets_file( flow = InstalledAppFlow.from_client_secrets_file(
@ -91,16 +91,19 @@ def get_channel_uploads_playlist_id(youtube, channel_id):
return None return None
def get_latest_video_id_from_playlist(youtube, playlist_id): def find_correct_live_video(youtube, channel_id, keyword):
request = youtube.playlistItems().list( request = youtube.search().list(
part="snippet", part="snippet",
playlistId=playlist_id, channelId=channel_id,
maxResults=1 eventType="live",
type="video"
) )
response = request.execute() response = request.execute()
items = response.get('items', []) items = response.get('items', [])
if items: for item in items:
return items[0]['snippet']['resourceId']['videoId'] title = item['snippet']['title']
if keyword.lower() in title.lower():
return item['id']['videoId']
return None return None
@ -136,7 +139,9 @@ def get_live_chat_id(youtube, video_id):
response = request.execute() response = request.execute()
items = response.get('items', []) items = response.get('items', [])
if items: if items:
return items[0]['liveStreamingDetails'].get('activeLiveChatId') live_chat_id = items[0]['liveStreamingDetails'].get('activeLiveChatId')
print(f"Live Chat ID: {live_chat_id}")
return live_chat_id
return None return None
@ -221,12 +226,14 @@ def monitor_chat(youtube, live_chat_id):
conn.close() conn.close()
next_page_token = response.get('nextPageToken') next_page_token = response.get('nextPageToken')
print(f"Next page token: {next_page_token}")
else: else:
print("No new messages detected; continuing to poll...") print("No new messages detected; continuing to poll...")
except Exception as e: except Exception as e:
print(f"Error while monitoring chat: {e}") print(f"Error while monitoring chat: {e}")
time.sleep(30) # Wait before retrying in case of an error
time.sleep(10) # Adjust this delay as needed time.sleep(10) # Adjust this delay as needed
@ -263,15 +270,9 @@ def main():
print("Channel ID not found!") print("Channel ID not found!")
return return
playlist_id = get_channel_uploads_playlist_id(youtube, channel_id) video_id = find_correct_live_video(youtube, channel_id, STREAM_KEYWORD)
if not playlist_id:
print("Uploads playlist not found!")
return
while True:
video_id = get_latest_video_id_from_playlist(youtube, playlist_id)
if video_id and is_video_live(youtube, video_id): if video_id and is_video_live(youtube, video_id):
print("Channel is live!") print("Correct live stream found and is live!")
live_chat_id = get_live_chat_id(youtube, video_id) live_chat_id = get_live_chat_id(youtube, video_id)
if live_chat_id: if live_chat_id:
print("Monitoring chat...") print("Monitoring chat...")
@ -279,7 +280,8 @@ def main():
else: else:
print("No live chat ID available.") print("No live chat ID available.")
else: else:
print("Channel is not live.") print("Could not find the correct live stream or it is not live.")
time.sleep(300) # Check every 5 minutes time.sleep(300) # Check every 5 minutes