diff --git a/modules/metadata_utils.py b/modules/metadata_utils.py new file mode 100644 index 0000000..ceb1e31 --- /dev/null +++ b/modules/metadata_utils.py @@ -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 .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}") diff --git a/modules/yt_poster.py b/modules/yt_poster.py index 1259262..4f9e3cf 100644 --- a/modules/yt_poster.py +++ b/modules/yt_poster.py @@ -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 ""