226 lines
7.6 KiB
Python
226 lines
7.6 KiB
Python
"""
|
||
YouTube Description Updater
|
||
Requires: pip install google-api-python-client google-auth-oauthlib google-auth
|
||
|
||
1. Enable YouTube Data API v3 in Google Cloud Console
|
||
2. Create OAuth 2.0 credentials (Desktop App)
|
||
3. Download as client_secret.json
|
||
"""
|
||
|
||
import os
|
||
import sys
|
||
from google.auth.transport.requests import Request
|
||
from google.oauth2.credentials import Credentials
|
||
from google_auth_oauthlib.flow import InstalledAppFlow
|
||
from googleapiclient.discovery import build
|
||
from googleapiclient.errors import HttpError
|
||
|
||
# Finds the absolute path to 'project_directory' and adds it to Python's search path
|
||
PROJECT_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "../"))
|
||
sys.path.append(PROJECT_ROOT)
|
||
|
||
from package.config import DB as DB
|
||
from package.youtube.config import CREDENTIALS as CREDENTIALS
|
||
from package.database import YoutubeInfo as YoutubeInfo
|
||
|
||
"""
|
||
SCOPES = ["https://www.googleapis.com/auth/youtube.force-ssl"]
|
||
TOKEN_FILE = "token.json"
|
||
CLIENT_SECRET_FILE = "client_secret.json"
|
||
|
||
def get_authenticated_service():
|
||
""Authenticates and returns a YouTube service object.""
|
||
creds = None
|
||
if os.path.exists(TOKEN_FILE):
|
||
creds = Credentials.from_authorized_user_file(TOKEN_FILE, SCOPES)
|
||
|
||
if not creds or not creds.valid:
|
||
if creds and creds.expired and creds.refresh_token:
|
||
creds.refresh(Request())
|
||
else:
|
||
if not os.path.exists(CLIENT_SECRET_FILE):
|
||
raise FileNotFoundError(
|
||
f"{CLIENT_SECRET_FILE} not found. Download it from Google Cloud Console."
|
||
)
|
||
flow = InstalledAppFlow.from_client_secrets_file(CLIENT_SECRET_FILE, SCOPES)
|
||
creds = flow.run_local_server(port=0)
|
||
|
||
with open(TOKEN_FILE, "w") as token:
|
||
token.write(creds.to_json())
|
||
|
||
return build("youtube", "v3", credentials=creds)
|
||
"""
|
||
|
||
def get_video_details(youtube, video_id: str):
|
||
"""Finds the video and returns its current snippet."""
|
||
try:
|
||
response = youtube.videos().list(
|
||
part="snippet,status",
|
||
id=video_id
|
||
).execute()
|
||
|
||
if not response["items"]:
|
||
print(f"❌ Video not found: {video_id}")
|
||
return None
|
||
|
||
return response["items"][0]
|
||
except HttpError as e:
|
||
print(f"❌ API Error while fetching video: {e}")
|
||
return None
|
||
|
||
def update_video_description(video_id: str, new_description: str, append: bool = False):
|
||
"""
|
||
Takes a YouTube video ID, finds the video, and changes its description.
|
||
|
||
Args:
|
||
video_id (str): The YouTube video ID (e.g., 'dQw4w9WgXcQ')
|
||
new_description (str): The new description to set
|
||
append (bool): If True, appends new_description to existing description instead of replacing
|
||
|
||
Returns:
|
||
dict | None: Updated video snippet or None on failure
|
||
"""
|
||
# Load credentials
|
||
if not CREDENTIALS:
|
||
print("❌ Error: YouTube API credentials not configured.")
|
||
return None
|
||
|
||
# Create YouTube API client
|
||
youtube = build('youtube', 'v3', credentials=CREDENTIALS)
|
||
|
||
video = get_video_details(youtube, video_id)
|
||
if not video:
|
||
return None
|
||
|
||
snippet = video["snippet"]
|
||
current_title = snippet["title"]
|
||
current_category = snippet.get("categoryId", "22")
|
||
|
||
print(f"Found video: '{current_title}' [{video_id}]")
|
||
print(f"Current description length: {len(snippet.get('description', ''))}")
|
||
current_description = snippet.get("description", "") or ""
|
||
|
||
# Decide final description
|
||
if append:
|
||
if new_description.strip().lower() in current_description.lower():
|
||
print("ℹ️ Description already contains that string, skipping")
|
||
return snippet
|
||
final_description = (snippet.get("description", "") + "\n\n" + new_description).strip()
|
||
else:
|
||
final_description = new_description
|
||
|
||
# YouTube requires you to send title + categoryId + description when updating snippet
|
||
body = {
|
||
"id": video_id,
|
||
"snippet": {
|
||
"title": current_title,
|
||
"description": final_description,
|
||
"categoryId": current_category,
|
||
# Keep existing tags, etc if you want
|
||
"tags": snippet.get("tags", []),
|
||
"defaultLanguage": snippet.get("defaultLanguage"),
|
||
}
|
||
}
|
||
|
||
try:
|
||
update_response = youtube.videos().update(
|
||
part="snippet",
|
||
body=body
|
||
).execute()
|
||
|
||
print(f"✅ Description updated for {video_id}")
|
||
print(f"New length: {len(final_description)}")
|
||
return update_response["snippet"]
|
||
|
||
except HttpError as e:
|
||
print(f"❌ Failed to update video {video_id}: {e}")
|
||
if "quotaExceeded" in str(e):
|
||
print("You hit the YouTube API quota.")
|
||
return None
|
||
|
||
def append_tag_if_not_exists(video_id: str, new_tag: str):
|
||
"""
|
||
Appends a new tag to a YouTube video if it doesn't already exist.
|
||
Case-insensitive check.
|
||
"""
|
||
if not new_tag or not new_tag.strip():
|
||
print("❌ Tag is empty")
|
||
return None
|
||
|
||
new_tag_clean = new_tag.strip()
|
||
|
||
# Load credentials
|
||
if not CREDENTIALS:
|
||
print("❌ Error: YouTube API credentials not configured.")
|
||
return None
|
||
|
||
# Create YouTube API client
|
||
youtube = build('youtube', 'v3', credentials=CREDENTIALS)
|
||
|
||
video = get_video_details(youtube, video_id)
|
||
if not video:
|
||
return None
|
||
|
||
snippet = video["snippet"]
|
||
existing_tags = snippet.get("tags", []) or []
|
||
existing_lower = {t.lower().strip() for t in existing_tags}
|
||
|
||
if new_tag_clean.lower() in existing_lower:
|
||
print(f"ℹ️ Tag '{new_tag_clean}' already exists on {video_id}, skipping")
|
||
return existing_tags
|
||
|
||
updated_tags = existing_tags + [new_tag_clean]
|
||
|
||
total_chars = sum(len(t) for t in updated_tags)
|
||
if total_chars > 450:
|
||
print(f"⚠️ Tags approaching 500 char limit ({total_chars})")
|
||
if len(updated_tags) > 15:
|
||
print(f"⚠️ {len(updated_tags)} tags - YouTube ignores after ~15-20")
|
||
|
||
print(f"Current tags: {existing_tags}")
|
||
print(f"Adding: '{new_tag_clean}'")
|
||
|
||
body = {
|
||
"id": video_id,
|
||
"snippet": {
|
||
"title": snippet["title"],
|
||
"description": snippet.get("description", ""),
|
||
"categoryId": snippet.get("categoryId", "22"),
|
||
"tags": updated_tags,
|
||
}
|
||
}
|
||
try:
|
||
resp = youtube.videos().update(part="snippet", body=body).execute()
|
||
print(f"✅ Tag '{new_tag_clean}' added to {video_id}")
|
||
return resp["snippet"].get("tags", [])
|
||
except HttpError as e:
|
||
print(f"❌ Failed to add tag: {e}")
|
||
return None
|
||
|
||
def main():
|
||
videos = DB.get_youtube_info_all()
|
||
for video in videos:
|
||
if video.yt_id:
|
||
update_video_description(video.yt_id, f"({video.id})", True)
|
||
append_tag_if_not_exists(video.yt_id, video.id)
|
||
if video.yt_short_id:
|
||
update_video_description(video.yt_short_id, f"({video.id})", True)
|
||
append_tag_if_not_exists(video.yt_short_id, video.id)
|
||
|
||
|
||
# --- Example Usage ---
|
||
if __name__ == "__main__":
|
||
main()
|
||
|
||
# Example 1: Replace description
|
||
# update_video_description("YOUR_VIDEO_ID_HERE", "This is my new description!\nGame: Valorant")
|
||
|
||
# Example 2: Append to description
|
||
# update_video_description("YOUR_VIDEO_ID_HERE", "\n\nCheck out https://twitch.tv/teampgp", append=True)
|
||
|
||
# For integration with your existing code:
|
||
#vid = input("Enter YouTube Video ID: ").strip()
|
||
#if vid:
|
||
# new_desc = input("Enter new description: ").strip()
|
||
# update_video_description(vid, new_desc)
|