YouTube Video Editor & Upload Tools

video_editor

Description:

This video_editor function processes a list of video files, resizing and cropping them to fit YouTube Shorts dimensions (1080x1920). It scales the videos to the correct size, ensures they are properly centered, and concatenates them into a final video that fits the required format.

Parameters:

• video_files: A list of paths to the video files you want to edit.

• output_filename: The path where the final concatenated video will be saved.

Return:

The path to the output video file after the editing is complete.

video_editor

import sys
from moviepy.editor import VideoFileClip, concatenate_videoclips, CompositeVideoClip, vfx

def video_editor(video_files: list[str], output_filename: str):
    """
    This function takes a list of video files, resizes and crops them to fit YouTube Shorts dimensions, and concatenates them into a single video file.

    :param video_files: A list of paths to video files.
    :param output_filename: The path to the output video file.
    :return: The path to the output video file.
    """

    clips = []
    for video in video_files:
        try:
            # Load the video file
            clip = VideoFileClip(video)

            # Set target dimensions for YouTube Shorts
            target_width = 1080
            target_height = 1920

            # Determine the scaling factor to cover the target dimensions
            clip_width, clip_height = clip.size
            width_ratio = target_width / clip_width
            height_ratio = target_height / clip_height
            scaling_factor = max(width_ratio, height_ratio)

            # Resize the clip to cover the target area
            clip_resized = clip.resize(height=int(clip.h * scaling_factor))

            # Center crop the clip to target dimensions
            clip_cropped = clip_resized.crop(
                x_center=clip_resized.w / 2,
                y_center=clip_resized.h / 2,
                width=target_width,
                height=target_height
            )

            # Optionally limit the clip duration to 60 seconds
            # clip_cropped = clip_cropped.subclip(0, min(clip_cropped.duration, 60))

            clips.append(clip_cropped)
        except Exception as e:
            print(f"Error processing {video}: {e}")
            continue

    if not clips:
        print("No valid video clips were processed.")
        sys.exit(1)

    # Concatenate clips
    final_clip = concatenate_videoclips(clips, method='compose')

    # Write the final video file
    final_clip.write_videofile(
        output_filename,
        fps=30,
        codec='libx264',
        audio_codec='aac',
        preset='medium',
        threads=4
    )

    return output_filename

upload_video

Description: Uploads a video file to YouTube using the YouTube Data API. The function sets the video title, description, category, tags (keywords), and privacy settings. It uses a resumable upload process, which is efficient for handling large files and interruptions.

Parameters:

• file (str): Path to the video file to be uploaded.

• title (str): The title of the video to be uploaded.

• description (str): A description of the video.

• category (str): The category ID of the video. Default is "22" (People & Blogs).

• keywords (Optional[str]): Comma-separated list of video keywords (tags). Optional.

• privacyStatus (str): The privacy setting of the video (public, private, or unlisted). Default is "unlisted."

• client_secrets_file (str): Path to the file containing client OAuth secrets. Default is "client_secret.json."

• credentials_file (str): Path to the file where OAuth credentials are saved. Default is "token.pickle."

Returns: None. The function does not return a value but uploads the video to YouTube, providing status updates during the process.

upload_video

def upload_video(
    file: str,
    title: str,
    description: str,
    category: str = "22",
    keywords: Optional[str] = "",
    privacyStatus: str = "unlisted",
    client_secrets_file: str = "client_secret.json",
    credentials_file: str = "token.pickle"
) -> None:
    """
    Uploads a video file to YouTube.

    Args:
        file (str): Path to the video file to upload.
        title (str): Video title.
        description (str): Video description.
        keywords (str): Comma-separated list of video keywords.
    """
    if not os.path.exists(file):
        raise FileNotFoundError(f"Please specify a valid file. '{file}' does not exist.")

    youtube = get_authenticated_service(client_secrets_file, credentials_file)

    tags = None
    if keywords:
        tags = keywords.split(",")

    body = dict(
        snippet=dict(
            title=title,
            description=description,
            tags=tags,
            categoryId=category
        ),
        status=dict(
            privacyStatus=privacyStatus
        )
    )

    # Call the API's videos.insert method to create and upload the video.
    insert_request = youtube.videos().insert(
        part=",".join(body.keys()),
        body=body,
        media_body=MediaFileUpload(file, chunksize=-1, resumable=True)
    )

    resumable_upload(insert_request)

get_authenticated_service

Description: Handles the OAuth 2.0 authentication process for accessing the YouTube Data API. If valid credentials are available, they are loaded from a file; otherwise, the user is prompted to log in and authorize the app. The credentials are saved for future use to avoid repeated login prompts.

Parameters:

• client_secrets_file (str): Path to the client secret file (client_secret.json) for OAuth 2.0 authentication.

• credentials_file (str): Path to the file where the user's access and refresh tokens are stored (token.pickle).

Returns: An authenticated YouTube service object that can be used to make API calls.

get_authenticated_service

def get_authenticated_service(client_secrets_file: str, credentials_file: str):
    credentials = None
    # The file token.pickle stores the user's access and refresh tokens.
    if os.path.exists(credentials_file):
        with open(credentials_file, 'rb') as token:
            credentials = pickle.load(token)
    # If there are no valid credentials, let the user log in.
    if not credentials or not credentials.valid:
        if credentials and credentials.expired and credentials.refresh_token:
            credentials.refresh(Request())
        else:
            if not os.path.exists(client_secrets_file):
                raise FileNotFoundError(f"Missing client_secrets.json file: {client_secrets_file}")
            flow = InstalledAppFlow.from_client_secrets_file(
                client_secrets_file, scopes=[YOUTUBE_UPLOAD_SCOPE])
            
            credentials = flow.run_local_server(
                    host='localhost',
                    port=8088,
                    authorization_prompt_message='Please visit this URL: {url}',
                    success_message='The auth flow is complete; you may close this window.',
                    open_browser=True)
        # Save the credentials for the next run.
        with open(credentials_file, 'wb') as token:
            pickle.dump(credentials, token)
    return build(YOUTUBE_API_SERVICE_NAME, YOUTUBE_API_VERSION, credentials=credentials)

resumable_upload

Description: Uploads large video files to YouTube in chunks, using a resumable upload process to handle interruptions like network failures. It retries the upload automatically if a retriable error occurs, ensuring that the process completes even in unreliable conditions.

Parameters:

• insert_request (googleapiclient.http.HttpRequest): The request object for uploading the video, created using the YouTube Data API.

Returns: None. The function prints the status of the upload and the video ID upon successful completion. It handles retries for transient errors.

resumable_upload

def resumable_upload(insert_request):
    response = None
    error = None
    retry = 0
    while response is None:
        try:
            print("Uploading file...")
            status, response = insert_request.next_chunk()
            if response is not None:
                if 'id' in response:
                    print("Video id '%s' was successfully uploaded." %
                          response['id'])
                else:
                    exit("The upload failed with an unexpected response: %s" % response)
        except HttpError as e:
            if e.resp.status in RETRIABLE_STATUS_CODES:
                error = "A retriable HTTP error %d occurred:\n%s" % (e.resp.status,
                                                                     e.content)
            else:
                raise
        except RETRIABLE_EXCEPTIONS as e:
            error = "A retriable error occurred: %s" % e

        if error is not None:
            print(error)
            retry += 1
            if retry > MAX_RETRIES:
                exit("No longer attempting to retry.")

            max_sleep = 2 ** retry
            sleep_seconds = random.random() * max_sleep
            print("Sleeping %f seconds and then retrying..." % sleep_seconds)
            time.sleep(sleep_seconds)

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)