✅ Finalized YouTube upload flow: metadata archiving, title suffixing, and cleanup
- Integrated save_metadata_record() after successful upload - Titles now reflect sequential numbering for same-day sessions - Automatically uploads thumbnail for widescreen videos - Removes session folder after upload if DEBUG == False - Merges session + clip metadata for persistence - Handles fallback logic for missing notes
This commit is contained in:
121
modules/metadata_utils.py
Normal file
121
modules/metadata_utils.py
Normal file
@ -0,0 +1,121 @@
|
||||
"""
|
||||
metadata_utils.py
|
||||
|
||||
Handles metadata extraction from video clip structure and notes.json,
|
||||
and manages persistent storage of finalized metadata records.
|
||||
|
||||
Author: Llama Chile Shop
|
||||
"""
|
||||
|
||||
import json
|
||||
import re
|
||||
from pathlib import Path
|
||||
from modules.config import NAS_MOUNT_ROOT
|
||||
|
||||
# Define where to persist finalized metadata records after upload
|
||||
HISTORY_DIR = Path("Z:/LCS/Logs/processed")
|
||||
|
||||
|
||||
def derive_session_metadata(session_dir: Path) -> dict:
|
||||
"""
|
||||
Derives session-level metadata from a session directory.
|
||||
Includes shared attributes, notes.json contents, and clip metadata for all videos found.
|
||||
|
||||
Args:
|
||||
session_dir (Path): Path to the session folder (e.g., 2025.07.24 or 2025.07.24.2)
|
||||
|
||||
Returns:
|
||||
dict: A dictionary representing session metadata, including notes and per-clip info.
|
||||
"""
|
||||
session_dir = Path(session_dir)
|
||||
session_name = session_dir.name
|
||||
|
||||
# Validate session folder format: YYYY.MM.DD or YYYY.MM.DD.N
|
||||
match = re.match(r"(\d{4})\.(\d{2})\.(\d{2})(?:\.(\d+))?", session_name)
|
||||
if not match:
|
||||
raise ValueError(f"Invalid session folder format: {session_name}")
|
||||
|
||||
year, month, day, session_index = match.groups()
|
||||
session_date = f"{year}-{month}-{day}"
|
||||
session_number = int(session_index) if session_index else 1
|
||||
|
||||
# Attempt to load notes.json from the session root
|
||||
notes_path = session_dir / "notes.json"
|
||||
notes_data = {}
|
||||
if notes_path.exists():
|
||||
try:
|
||||
with open(notes_path, "r", encoding="utf-8") as f:
|
||||
notes_data = json.load(f)
|
||||
except Exception as e:
|
||||
raise RuntimeError(f"Failed to parse notes.json: {e}")
|
||||
|
||||
# Extract shared fields (with fallback defaults)
|
||||
session_meta = {
|
||||
"session_date": session_date,
|
||||
"session_number": session_number,
|
||||
"highlight": notes_data.get("highlight", "Fortnite highlight moment"),
|
||||
"tags": notes_data.get("tags", []),
|
||||
"gag_name": notes_data.get("gag_name", None),
|
||||
"notes": notes_data,
|
||||
"clips": []
|
||||
}
|
||||
|
||||
# Scan for all .mp4 clips within expected subdirectories
|
||||
for subfolder in ["hits", "misses", "montages", "outtakes"]:
|
||||
clip_dir = session_dir / subfolder
|
||||
if not clip_dir.exists():
|
||||
continue
|
||||
|
||||
for clip_path in clip_dir.glob("*.mp4"):
|
||||
stem = clip_path.stem.lower()
|
||||
is_vertical = stem.endswith("-vert") or stem.endswith("-vertical")
|
||||
format = "vertical" if is_vertical else "wide"
|
||||
|
||||
clip_meta = {
|
||||
"path": str(clip_path),
|
||||
"filename": clip_path.name,
|
||||
"stem": clip_path.stem,
|
||||
"format": format,
|
||||
"clip_type": subfolder,
|
||||
"youtube_urls": [],
|
||||
"peertube_urls": []
|
||||
}
|
||||
|
||||
session_meta["clips"].append(clip_meta)
|
||||
|
||||
return session_meta
|
||||
|
||||
|
||||
def save_metadata_record(metadata: dict) -> None:
|
||||
"""
|
||||
Saves a finalized metadata record to disk for future lookup or audit.
|
||||
|
||||
This includes all session-level and clip-level data, plus any added URLs
|
||||
after upload to YouTube or PeerTube.
|
||||
|
||||
Args:
|
||||
metadata (dict): Fully populated metadata record, typically post-upload.
|
||||
|
||||
Raises:
|
||||
RuntimeError: If required fields are missing or write fails.
|
||||
"""
|
||||
try:
|
||||
session_date = metadata.get("session_date")
|
||||
filename = metadata.get("filename") or metadata.get("stem")
|
||||
|
||||
if not session_date or not filename:
|
||||
raise ValueError("Metadata missing required fields: session_date or filename/stem")
|
||||
|
||||
# Use YYYY.MM.DD folder for archival
|
||||
dest_dir = HISTORY_DIR / session_date.replace("-", ".")
|
||||
dest_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
# Save as <stem>.json
|
||||
dest_file = dest_dir / f"{Path(filename).stem}.json"
|
||||
with open(dest_file, "w", encoding="utf-8") as f:
|
||||
json.dump(metadata, f, indent=2)
|
||||
|
||||
print(f"📁 Saved metadata record to: {dest_file}")
|
||||
|
||||
except Exception as e:
|
||||
raise RuntimeError(f"Failed to save metadata record: {e}")
|
||||
@ -3,65 +3,100 @@ yt_poster.py
|
||||
|
||||
Handles video uploads to YouTube using the YouTube Data API.
|
||||
|
||||
This module includes logic for setting titles, descriptions, tags, and
|
||||
privacy status. It integrates with description generation tools and supports
|
||||
automatic metadata based on the video type (e.g., montage).
|
||||
|
||||
Requires authentication via OAuth 2.0 and expects a valid token.pickle file.
|
||||
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
|
||||
"""
|
||||
|
||||
import os
|
||||
import shutil
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
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
|
||||
from modules.thumbnail_utils import generate_thumbnail, generate_thumbnail_prompt
|
||||
from modules.config import DEBUG
|
||||
from dotenv import load_dotenv
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
|
||||
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 dotenv import load_dotenv
|
||||
load_dotenv()
|
||||
|
||||
HISTORY_DIR = Path("Z:/LCS/Logs/processed")
|
||||
|
||||
def upload_video(file_path: Path, is_vertical: bool, stream_date: str, description: str = None, private: bool = DEBUG) -> str:
|
||||
|
||||
def count_existing_uploads(session_date: str) -> int:
|
||||
"""
|
||||
Uploads a video to YouTube, assigns to playlist, sets recording date, and optionally uploads a thumbnail.
|
||||
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
|
||||
|
||||
|
||||
def upload_video(file_path: Path, session_dir: Path, is_vertical: bool) -> str:
|
||||
"""
|
||||
Uploads a single video to YouTube, with title/description/thumbnail handling.
|
||||
|
||||
Args:
|
||||
file_path (Path): Full path to the rendered video file.
|
||||
is_vertical (bool): True if video is vertical.
|
||||
stream_date (str): Stream session date in YYYY.MM.DD[.N] format.
|
||||
description (str): Optional description. Generated if None.
|
||||
private (bool): If True, uploads as private (used for debug mode).
|
||||
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.
|
||||
|
||||
Returns:
|
||||
str: YouTube video URL.
|
||||
str: Public YouTube video URL.
|
||||
"""
|
||||
try:
|
||||
from authorize_youtube import get_authenticated_service
|
||||
youtube = get_authenticated_service()
|
||||
|
||||
title = generate_montage_title(stream_date)
|
||||
if not description:
|
||||
description = generate_montage_description()
|
||||
# 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
|
||||
|
||||
tags = ["Fortnite", "Zero Build", "Solo", "Gramps", "CoolHandGramps"]
|
||||
privacy_status = "private" if private else "public"
|
||||
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": tags,
|
||||
"tags": metadata.get("tags", []),
|
||||
"categoryId": "20", # Gaming
|
||||
},
|
||||
"status": {
|
||||
"privacyStatus": privacy_status,
|
||||
"privacyStatus": "private" if DEBUG else "public",
|
||||
"selfDeclaredMadeForKids": False,
|
||||
}
|
||||
}
|
||||
@ -85,30 +120,19 @@ def upload_video(file_path: Path, is_vertical: bool, stream_date: str, descripti
|
||||
video_url = f"https://youtu.be/{video_id}"
|
||||
print(f"✅ Upload complete: {video_url}")
|
||||
|
||||
# ✅ Generate thumbnail if widescreen
|
||||
# Upload thumbnail if wide
|
||||
if not is_vertical:
|
||||
# Load notes.txt if present
|
||||
notes_file = file_path.with_name("notes.txt")
|
||||
if notes_file.exists():
|
||||
with open(notes_file, "r", encoding="utf-8") as f:
|
||||
notes_text = f.read().strip()
|
||||
else:
|
||||
notes_text = "A funny Fortnite moment with Gramps."
|
||||
|
||||
# Build prompt (future use for AI image generation)
|
||||
thumbnail_prompt = generate_thumbnail_prompt(notes_text)
|
||||
print(f"🧠 Thumbnail prompt: {thumbnail_prompt}")
|
||||
|
||||
# Generate local thumbnail via ffmpeg
|
||||
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 generated and set.")
|
||||
print("✅ Custom thumbnail uploaded.")
|
||||
|
||||
# ✅ Add to playlist
|
||||
# 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(
|
||||
@ -125,11 +149,9 @@ def upload_video(file_path: Path, is_vertical: bool, stream_date: str, descripti
|
||||
).execute()
|
||||
print(f"✅ Added to playlist: {playlist_id}")
|
||||
|
||||
# ✅ Set recording date
|
||||
parts = stream_date.split(".")
|
||||
date_obj = datetime(int(parts[0]), int(parts[1]), int(parts[2]))
|
||||
# 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={
|
||||
@ -141,6 +163,18 @@ def upload_video(file_path: Path, is_vertical: bool, stream_date: str, descripti
|
||||
).execute()
|
||||
print(f"✅ Recording date set: {recording_date}")
|
||||
|
||||
# Save metadata (include YouTube URL)
|
||||
metadata["youtube_urls"] = [video_url]
|
||||
save_metadata_record(metadata)
|
||||
|
||||
# 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}")
|
||||
|
||||
return video_url
|
||||
|
||||
except HttpError as e:
|
||||
@ -148,5 +182,5 @@ def upload_video(file_path: Path, is_vertical: bool, stream_date: str, descripti
|
||||
return ""
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ Unexpected error during upload: {e}")
|
||||
print(f"❌ Unexpected error: {e}")
|
||||
return ""
|
||||
|
||||
Reference in New Issue
Block a user