Compare commits

3 Commits
v0.1.0 ... main

2 changed files with 59 additions and 25 deletions

View File

@@ -1,5 +1,17 @@
# YouTube Live Chat in Terminal # YouTube Live Chat in Terminal
---
> **ARCHIVED PROJECT**
>
> This project is considered archived and is no longer in active development.
>
> **Reason:** It relies on the YouTube Data API v3, which has significant quota limitations (10,000 points/day). Continuous polling for live chat messages quickly exhausts this quota, making this approach unsustainable for long-term or frequent use.
>
> **Recommendation:** For a more sustainable and quota-friendly solution, please refer to the **`youtube-chat-webhook-v2`** project. That project explores alternative methods for fetching YouTube live chat data that do not rely on the official API and its restrictive quotas.
---
## Project Description ## Project Description
This project aims to provide a customizable way to view YouTube Live Chat directly in your terminal. It is implemented in `main.py`. This project aims to provide a customizable way to view YouTube Live Chat directly in your terminal. It is implemented in `main.py`.

72
main.py
View File

@@ -45,33 +45,53 @@ 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):
try: retries = 0
response = youtube.videos().list( max_retries = 5
part='liveStreamingDetails', while retries < max_retries:
id=video_id try:
).execute() response = youtube.videos().list(
part='liveStreamingDetails',
id=video_id
).execute()
if 'items' in response and response['items']: if 'items' in response and response['items']:
video = response['items'][0] video = response['items'][0]
if 'liveStreamingDetails' in video and 'activeLiveChatId' in video['liveStreamingDetails']: if 'liveStreamingDetails' in video and 'activeLiveChatId' in video['liveStreamingDetails']:
return video['liveStreamingDetails']['activeLiveChatId'] return video['liveStreamingDetails']['activeLiveChatId']
return None return None
except googleapiclient.errors.HttpError as e: except googleapiclient.errors.HttpError as e:
console.print(f"[red]An HTTP error {e.resp.status} occurred: {e.content}[/red]") if e.resp.status == 403 and "quotaExceeded" in str(e.content):
return None 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]")
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):
try: retries = 0
request = youtube.liveChatMessages().list( max_retries = 5
liveChatId=live_chat_id, while retries < max_retries:
part='snippet,authorDetails', try:
pageToken=page_token request = youtube.liveChatMessages().list(
) liveChatId=live_chat_id,
response = request.execute() part='snippet,authorDetails',
return response pageToken=page_token
except googleapiclient.errors.HttpError as e: )
console.print(f"[red]An HTTP error {e.resp.status} occurred: {e.content}[/red]") response = request.execute()
return None return response
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]")
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()
@@ -162,7 +182,9 @@ def main():
next_page_token = response.get('nextPageToken') next_page_token = response.get('nextPageToken')
polling_interval_millis = response['pollingIntervalMillis'] polling_interval_millis = response['pollingIntervalMillis']
time.sleep(polling_interval_millis / 1000.0) # Ensure a minimum sleep duration to conserve API quota
sleep_time = max(5, polling_interval_millis / 1000.0)
time.sleep(sleep_time)
else: else:
console.print("[yellow]No new messages or an error occurred. Retrying...[/yellow]") console.print("[yellow]No new messages or an error occurred. Retrying...[/yellow]")
time.sleep(5) # Wait 5 seconds before retrying on error time.sleep(5) # Wait 5 seconds before retrying on error