TikTok to YouTube Pipeline Part 4: Scheduling & Publishing to YouTube

# youtube_auth.py

import os
import pickle
from pathlib import Path

from google.auth.transport.requests import Request
from google_auth_oauthlib.flow import InstalledAppFlow
from googleapiclient.discovery import build

SCOPES = ["https://www.googleapis.com/auth/youtube.upload"]
TOKEN_PATH = Path("credentials/youtube_token.pickle")
CLIENT_SECRET_PATH = Path("credentials/client_secret.json")


def get_authenticated_service():
    """Authenticate and return a YouTube API service object."""
    credentials = None

    if TOKEN_PATH.exists():
        with open(TOKEN_PATH, "rb") as token:
            credentials = pickle.load(token)

    if not credentials or not credentials.valid:
        if credentials and credentials.expired and credentials.refresh_token:
            credentials.refresh(Request())
        else:
            flow = InstalledAppFlow.from_client_secrets_file(
                str(CLIENT_SECRET_PATH), SCOPES
            )
            credentials = flow.run_local_server(port=8090)

        TOKEN_PATH.parent.mkdir(parents=True, exist_ok=True)
        with open(TOKEN_PATH, "wb") as token:
            pickle.dump(credentials, token)

    return build("youtube", "v3", credentials=credentials)

# upload_youtube.py

import sys
from datetime import datetime
from pathlib import Path

from googleapiclient.http import MediaFileUpload

from youtube_auth import get_authenticated_service

OUTPUT_DIR = Path("data/output")


def generate_metadata(keyword: str) -> dict:
    """Generate title, description, and tags for a YouTube upload."""
    topic = keyword.replace("_", " ").title()
    date_str = datetime.now().strftime("%B %d, %Y")

    return {
        "title": f"🔴 {topic} — Latest Updates | {date_str}",
        "description": (
            f"Today's coverage of {topic}. "
            f"This segment compiles the latest on-the-ground footage "
            f"and verified reports.\n\n"
            f"⚠️ Some footage may be graphic. Viewer discretion advised.\n\n"
            f"📌 Subscribe for daily global news updates.\n"
            f"🔔 Turn on notifications to never miss breaking news.\n\n"
            f"#GlobalNews #{keyword.replace('_', '')} #BreakingNews"
        ),
        "tags": [
            topic, "breaking news", "global news",
            keyword.replace("_", " "), "world news",
            "current events", "news today",
        ],
    }


def upload_video(
    youtube,
    video_path: str,
    title: str,
    description: str,
    tags: list[str],
    category_id: str = "25",  # News & Politics
    privacy: str = "private",
):
    """Upload a single video to YouTube."""
    body = {
        "snippet": {
            "title": title,
            "description": description,
            "tags": tags,
            "categoryId": category_id,
        },
        "status": {
            "privacyStatus": privacy,
            "selfDeclaredMadeForKids": False,
        },
    }

    media = MediaFileUpload(
        video_path,
        mimetype="video/mp4",
        resumable=True,
        chunksize=1024 * 1024 * 10,  # 10 MB chunks
    )

    request = youtube.videos().insert(
        part="snippet,status",
        body=body,
        media_body=media,
    )

    response = None
    while response is None:
        status, response = request.next_chunk()
        if status:
            pct = int(status.progress() * 100)
            print(f"   📤 Uploading... {pct}%")

    video_id = response["id"]
    print(f"   ✅ Uploaded: https://youtu.be/{video_id}")

    return video_id


def main():
    youtube = get_authenticated_service()

    video_files = sorted(OUTPUT_DIR.glob("*_news_segment.mp4"))

    if not video_files:
        print("❌ No compiled segments found in data/output/")
        sys.exit(1)

    print(f"📋 Found {len(video_files)} segment(s) to upload.\n")

    for video_path in video_files:
        keyword = video_path.stem.replace("_news_segment", "")
        meta = generate_metadata(keyword)

        print(f"🎬 Uploading: {meta['title']}")

        upload_video(
            youtube,
            str(video_path),
            title=meta["title"],
            description=meta["description"],
            tags=meta["tags"],
            privacy="private",  # Change to "public" when ready
        )

        print()

    print("🎉 All uploads complete!")


if __name__ == "__main__":
    main()

# pipeline.py

import subprocess
import sys
from pathlib import Path


def run_step(script: str, args: list[str] = None):
    """Run a pipeline step and exit on failure."""
    cmd = [sys.executable, script] + (args or [])
    print(f"\n{'='*60}")
    print(f"▶️  Running: {' '.join(cmd)}")
    print(f"{'='*60}\n")

    result = subprocess.run(cmd)
    if result.returncode != 0:
        print(f"\n❌ Step failed: {script}")
        sys.exit(1)


def find_latest_results() -> str:
    """Find the most recent search results JSON file."""
    results = sorted(Path("data/raw").glob("tiktok_results_*.json"))
    if not results:
        raise FileNotFoundError("No results found in data/raw/")
    return str(results[-1])


def main():
    print("🚀 Starting TikTok → YouTube Pipeline\n")

    # Step 1: Search TikTok
    run_step("search_tiktok.py")

    # Step 2: Download videos
    latest_json = find_latest_results()
    run_step("download_videos.py", [latest_json])

    # Step 3: Generate scripts
    run_step("generate_script.py")

    # Step 4: Generate TTS audio
    run_step("generate_tts.py")

    # Step 5: Compile videos
    run_step("compile_video.py")

    # Step 6: Upload to YouTube
    run_step("upload_youtube.py")

    print("\n🎉 Pipeline complete! Check your YouTube Studio.")


if __name__ == "__main__":
    main()