✅ Update yt_poster.py: [describe your changes briefly]
This commit is contained in:
@ -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)
|
||||
|
||||
Reference in New Issue
Block a user