diff --git a/modules/yt_poster.py b/modules/yt_poster.py index 4f9e3cf..ddf88f5 100644 --- a/modules/yt_poster.py +++ b/modules/yt_poster.py @@ -1,186 +1,115 @@ +#!/usr/bin/env python3 """ yt_poster.py -Handles video uploads to YouTube using the YouTube Data API. +This module handles the upload of videos to YouTube using the YouTube Data API v3. +It supports setting metadata such as title, description, tags, category, and privacy settings. +It also ensures that the game title "Fortnite" is included in the metadata to trigger proper categorization. -This module manages: -- Title and description generation -- Playlist assignment -- Custom thumbnail generation (widescreen only) -- Upload record persistence -- Session cleanup on success - -Author: Llama Chile Shop +Author: gramps@llamachile.shop """ import os -import shutil -from datetime import datetime -from pathlib import Path +import google.auth from googleapiclient.discovery import build -from googleapiclient.errors import HttpError from googleapiclient.http import MediaFileUpload -from modules.title_utils import generate_montage_title -from modules.description_utils import generate_montage_description, generate_clip_description -from modules.thumbnail_utils import generate_thumbnail, generate_thumbnail_prompt -from modules.metadata_utils import derive_session_metadata, save_metadata_record -from modules.config import DEBUG +from modules.config import OPENAI_API_KEY, DEBUG +from modules.archive import save_metadata_record -from dotenv import load_dotenv -load_dotenv() +# Category ID for "Gaming" on YouTube (required for accurate categorization) +CATEGORY_ID = "20" -HISTORY_DIR = Path("Z:/LCS/Logs/processed") +# Default tags to include if none are provided +DEFAULT_TAGS = [ + "Fortnite", "Zero Build", "Gramps", "CoolHandGramps", + "funny", "gaming", "highlights" +] +# Default visibility setting +DEFAULT_PRIVACY = "public" -def count_existing_uploads(session_date: str) -> int: +def ensure_fortnite_tag(metadata): """ - Returns the number of existing metadata records for a given session date. - Used to compute title suffixes (e.g., "Video 2"). - """ - session_folder = HISTORY_DIR / session_date.replace("-", ".") - return len(list(session_folder.glob("*.json"))) if session_folder.exists() else 0 + Ensures that the word 'Fortnite' appears in at least one of the following: + - Title + - Description + - Tags list - -def upload_video(file_path: Path, session_dir: Path, is_vertical: bool) -> str: + This helps YouTube automatically detect the game and associate the video + with Fortnite gameplay. """ - Uploads a single video to YouTube, with title/description/thumbnail handling. + if "fortnite" not in metadata["title"].lower() and \ + "fortnite" not in metadata["description"].lower() and \ + not any("fortnite" in tag.lower() for tag in metadata.get("tags", [])): + metadata.setdefault("tags", []).append("Fortnite") + +def upload_video(youtube, video_path, metadata): + """ + Uploads a video to YouTube with the provided metadata. Args: - file_path (Path): Path to the rendered .mp4 file. - session_dir (Path): Parent directory of the session (used for metadata). - is_vertical (bool): Format flag for Shorts logic. + youtube: Authenticated YouTube API service object. + video_path: Path to the video file to be uploaded. + metadata: Dictionary containing video metadata fields. Returns: - str: Public YouTube video URL. + str: URL of the uploaded YouTube video. """ - try: - from authorize_youtube import get_authenticated_service - youtube = get_authenticated_service() - # Derive session metadata and isolate the clip record - session_meta = derive_session_metadata(session_dir) - session_date = session_meta["session_date"] - stem = file_path.stem + # Ensure the 'Fortnite' keyword is present somewhere in metadata + ensure_fortnite_tag(metadata) - clip_record = next( - (clip for clip in session_meta["clips"] if clip["stem"] == stem), - None - ) - - if clip_record is None: - raise RuntimeError(f"Clip {stem} not found in session metadata.") - - metadata = {**session_meta, **clip_record} - - # Title logic with sequential suffixing - suffix = "" - existing_count = count_existing_uploads(session_date) - if existing_count > 0: - suffix = f"Video {existing_count + 1}" - - title = generate_montage_title(session_date, suffix=suffix) - - # Description - if metadata["highlight"]: - description = generate_clip_description(metadata["highlight"]) - else: - description = generate_montage_description() - - # Upload video - body = { - "snippet": { - "title": title, - "description": description, - "tags": metadata.get("tags", []), - "categoryId": "20", # Gaming - }, - "status": { - "privacyStatus": "private" if DEBUG else "public", - "selfDeclaredMadeForKids": False, - } + # Construct the request body for YouTube API + request_body = { + "snippet": { + "title": metadata["title"], + "description": metadata["description"], + "tags": metadata.get("tags", DEFAULT_TAGS), + "categoryId": CATEGORY_ID # Set to "Gaming" + }, + "status": { + "privacyStatus": metadata.get("privacy", DEFAULT_PRIVACY) } + } - print(f"๐Ÿ“ค Uploading to YouTube: {file_path.name}") - media = MediaFileUpload(str(file_path), chunksize=-1, resumable=True) + # Wrap the video file in a MediaFileUpload object + media = MediaFileUpload(video_path, mimetype="video/*", resumable=True) - request = youtube.videos().insert( - part="snippet,status", - body=body, - media_body=media - ) + print(f"๐Ÿ“ค Uploading {video_path} to YouTube...") - response = None - while response is None: - status, response = request.next_chunk() - if status: - print(f"๐ŸŸก Uploading: {int(status.progress() * 100)}%") + # Execute the video insert request + request = youtube.videos().insert( + part="snippet,status", + body=request_body, + media_body=media + ) - video_id = response["id"] - video_url = f"https://youtu.be/{video_id}" - print(f"โœ… Upload complete: {video_url}") + response = request.execute() + video_id = response["id"] + youtube_url = f"https://www.youtube.com/watch?v={video_id}" - # Upload thumbnail if wide - if not is_vertical: - notes = metadata.get("notes", {}) - prompt = generate_thumbnail_prompt(notes.get("highlight", "Fortnite moment")) - thumbnail_path = generate_thumbnail(file_path, output_path=f"{file_path.stem}_thumb.jpg") - if thumbnail_path: - youtube.thumbnails().set( - videoId=video_id, - media_body=str(thumbnail_path) - ).execute() - print("โœ… Custom thumbnail uploaded.") + print(f"โœ… Uploaded to YouTube: {youtube_url}") - # Add to playlist - playlist_id = os.getenv("YT_PLAYLIST_ID_SHORTS" if is_vertical else "YT_PLAYLIST_ID_CLIPS") - if playlist_id: - youtube.playlistItems().insert( - part="snippet", - body={ - "snippet": { - "playlistId": playlist_id, - "resourceId": { - "kind": "youtube#video", - "videoId": video_id - } - } - } - ).execute() - print(f"โœ… Added to playlist: {playlist_id}") + # Record the YouTube URL in the metadata for archive history + metadata.setdefault("youtube_url", []).append(youtube_url) - # Set recording date - date_obj = datetime.strptime(session_date, "%Y-%m-%d") - recording_date = date_obj.strftime("%Y-%m-%dT00:00:00Z") - youtube.videos().update( - part="recordingDetails", - body={ - "id": video_id, - "recordingDetails": { - "recordingDate": recording_date - } - } - ).execute() - print(f"โœ… Recording date set: {recording_date}") + # Persist the metadata archive only if we're not in DEBUG mode + if not DEBUG: + save_metadata_record(video_path, metadata) - # Save metadata (include YouTube URL) - metadata["youtube_urls"] = [video_url] - save_metadata_record(metadata) + return youtube_url - # Clean up session directory if DEBUG is off - if not DEBUG: - try: - shutil.rmtree(session_dir) - print(f"๐Ÿงน Cleaned up source directory: {session_dir}") - except Exception as e: - print(f"โš ๏ธ Cleanup failed: {e}") +def get_authenticated_service(): + """ + Returns an authenticated YouTube API service using Application Default Credentials. + This requires that `gcloud auth application-default login` has been run successfully, + or that a service account token is available in the environment. - return video_url - - except HttpError as e: - print(f"โŒ YouTube API error: {e}") - return "" - - except Exception as e: - print(f"โŒ Unexpected error: {e}") - return "" + Returns: + googleapiclient.discovery.Resource: The YouTube API client object. + """ + credentials, _ = google.auth.default( + scopes=["https://www.googleapis.com/auth/youtube.upload"] + ) + return build("youtube", "v3", credentials=credentials)