ramforth/youtube-chat-terminal
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

Feat: Implement exponential backoff for YouTube API calls to handle quota errors.

Browse Source
This commit is contained in:
ramforth
2025-10-29 13:18:18 +01:00
parent d343375790
commit 00e03209de
1 changed files with 44 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files

20
main.py
View File

@@ -45,6 +45,9 @@ def get_authenticated_service():
API_SERVICE_NAME, API_VERSION, credentials=credentials) API_SERVICE_NAME, API_VERSION, credentials=credentials)
def get_live_chat_id(youtube, video_id): def get_live_chat_id(youtube, video_id):
retries = 0
max_retries = 5
while retries < max_retries:
try: try:
response = youtube.videos().list( response = youtube.videos().list(
part='liveStreamingDetails', part='liveStreamingDetails',
@@ -57,10 +60,20 @@ def get_live_chat_id(youtube, video_id):
return video['liveStreamingDetails']['activeLiveChatId'] return video['liveStreamingDetails']['activeLiveChatId']
return None return None
except googleapiclient.errors.HttpError as e: except googleapiclient.errors.HttpError as e:
if e.resp.status == 403 and "quotaExceeded" in str(e.content):
console.print(f"[red]Quota exceeded. Retrying in {2**retries} seconds...[/red]")
time.sleep(2**retries)
retries += 1
else:
console.print(f"[red]An HTTP error {e.resp.status} occurred: {e.content}[/red]") console.print(f"[red]An HTTP error {e.resp.status} occurred: {e.content}[/red]")
return None return None
console.print(f"[red]Failed to get live chat ID after {max_retries} retries due to quota issues.[/red]")
return None
def fetch_live_chat_messages(youtube, live_chat_id, page_token=None): def fetch_live_chat_messages(youtube, live_chat_id, page_token=None):
retries = 0
max_retries = 5
while retries < max_retries:
try: try:
request = youtube.liveChatMessages().list( request = youtube.liveChatMessages().list(
liveChatId=live_chat_id, liveChatId=live_chat_id,
@@ -70,8 +83,15 @@ def fetch_live_chat_messages(youtube, live_chat_id, page_token=None):
response = request.execute() response = request.execute()
return response return response
except googleapiclient.errors.HttpError as e: except googleapiclient.errors.HttpError as e:
if e.resp.status == 403 and "quotaExceeded" in str(e.content):
console.print(f"[red]Quota exceeded. Retrying in {2**retries} seconds...[/red]")
time.sleep(2**retries)
retries += 1
else:
console.print(f"[red]An HTTP error {e.resp.status} occurred: {e.content}[/red]") console.print(f"[red]An HTTP error {e.resp.status} occurred: {e.content}[/red]")
return None return None
console.print(f"[red]Failed to fetch live chat messages after {max_retries} retries due to quota issues.[/red]")
return None
def main(): def main():
youtube = get_authenticated_service() youtube = get_authenticated_service()