youtube_tool.py

#!/usr/bin/env python3
"""YouTube CLI tool for Hermes Agent - uses OAuth2 refresh token flow."""

import argparse
import json
import os
import sys
import mimetypes
from urllib.parse import urlencode

try:
    import requests
except ImportError:
    import subprocess
    subprocess.check_call([sys.executable, "-m", "pip", "install", "requests", "-q"])
    import requests


def get_access_token():
    """Get a fresh access token using the refresh token."""
    client_id = os.environ.get("YOUTUBE_CLIENT_ID")
    client_secret = os.environ.get("YOUTUBE_CLIENT_SECRET")
    refresh_token = os.environ.get("YOUTUBE_REFRESH_TOKEN")
    
    if not all([client_id, client_secret, refresh_token]):
        print("ERROR: Missing YouTube credentials. Set YOUTUBE_CLIENT_ID, YOUTUBE_CLIENT_SECRET, YOUTUBE_REFRESH_TOKEN")
        sys.exit(1)
    
    r = requests.post("https://oauth2.googleapis.com/token", data={
        "client_id": client_id,
        "client_secret": client_secret,
        "refresh_token": refresh_token,
        "grant_type": "refresh_token",
    })
    
    if r.status_code != 200:
        print(f"ERROR: Token refresh failed: {r.text}")
        sys.exit(1)
    
    return r.json()["access_token"]


def api(method, endpoint, token, params=None, json_data=None, files=None):
    """Make a YouTube API request."""
    url = f"https://www.googleapis.com/youtube/v3/{endpoint}"
    headers = {"Authorization": f"Bearer {token}"}
    
    if method == "GET":
        r = requests.get(url, headers=headers, params=params, timeout=30)
    elif method == "POST":
        if files:
            r = requests.post(url, headers=headers, params=params, files=files, timeout=600)
        else:
            headers["Content-Type"] = "application/json"
            r = requests.post(url, headers=headers, params=params, json=json_data, timeout=30)
    elif method == "PUT":
        headers["Content-Type"] = "application/json"
        r = requests.put(url, headers=headers, params=params, json=json_data, timeout=30)
    elif method == "DELETE":
        r = requests.delete(url, headers=headers, params=params, timeout=30)
    
    return r


def cmd_channel(args):
    """Show channel info."""
    token = get_access_token()
    r = api("GET", "channels", token, params={"part": "snippet,statistics,contentDetails", "mine": "true"})
    data = r.json()
    
    if not data.get("items"):
        print("No channel found.")
        return
    
    ch = data["items"][0]
    s = ch["snippet"]
    stats = ch["statistics"]
    
    print(f"Channel: {s['title']}")
    print(f"Description: {s.get('description', 'N/A')}")
    print(f"Subscribers: {stats.get('subscriberCount', 'hidden')}")
    print(f"Videos: {stats.get('videoCount', '0')}")
    print(f"Views: {stats.get('viewCount', '0')}")
    print(f"Created: {s.get('publishedAt', 'N/A')}")
    print(f"Channel ID: {ch['id']}")


def cmd_list(args):
    """List recent videos."""
    token = get_access_token()
    
    # Get uploads playlist
    r = api("GET", "channels", token, params={"part": "contentDetails", "mine": "true"})
    data = r.json()
    if not data.get("items"):
        print("No channel found.")
        return
    
    uploads_id = data["items"][0]["contentDetails"]["relatedPlaylists"]["uploads"]
    
    # Get videos from uploads playlist
    params = {"part": "snippet,status", "playlistId": uploads_id, "maxResults": args.count}
    r = api("GET", "playlistItems", token, params=params)
    data = r.json()
    
    if not data.get("items"):
        print("No videos found.")
        return
    
    print(f"{'#':<3} {'Title':<50} {'Status':<12} {'Published':<20}")
    print("-" * 90)
    for i, item in enumerate(data["items"], 1):
        s = item["snippet"]
        vid = s["resourceId"]["videoId"]
        title = s["title"][:48]
        status = item.get("status", {}).get("privacyStatus", "?")
        published = s.get("publishedAt", "?")[:19]
        print(f"{i:<3} {title:<50} {status:<12} {published}")
        print(f"    https://youtu.be/{vid}")


def cmd_upload(args):
    """Upload a video."""
    if not os.path.exists(args.file):
        print(f"ERROR: File not found: {args.file}")
        sys.exit(1)
    
    token = get_access_token()
    
    # Step 1: Initialize resumable upload
    metadata = {
        "snippet": {
            "title": args.title,
            "description": args.description or "",
            "tags": args.tags.split(",") if args.tags else [],
            "categoryId": args.category or "22",  # 22 = People & Blogs
        },
        "status": {
            "privacyStatus": args.privacy or "private",
            "selfDeclaredMadeForKids": False,
        }
    }
    
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json",
        "X-Upload-Content-Type": mimetypes.guess_type(args.file)[0] or "video/*",
        "X-Upload-Content-Length": str(os.path.getsize(args.file)),
    }
    
    r = requests.post(
        "https://www.googleapis.com/upload/youtube/v3/videos?uploadType=resumable&part=snippet,status",
        headers=headers,
        json=metadata,
        timeout=30,
    )
    
    if r.status_code != 200:
        print(f"ERROR: Upload init failed: {r.status_code} {r.text}")
        sys.exit(1)
    
    upload_url = r.headers["Location"]
    
    # Step 2: Upload the file
    file_size = os.path.getsize(args.file)
    print(f"Uploading: {args.file} ({file_size / 1048576:.1f} MB)")
    print(f"Title: {args.title}")
    print(f"Privacy: {args.privacy or 'private'}")
    
    with open(args.file, "rb") as f:
        r = requests.put(
            upload_url,
            headers={
                "Authorization": f"Bearer {token}",
                "Content-Type": mimetypes.guess_type(args.file)[0] or "video/*",
                "Content-Length": str(file_size),
            },
            data=f,
            timeout=3600,
        )
    
    if r.status_code == 200:
        vid = r.json()
        print(f"\n✅ Uploaded successfully!")
        print(f"Video ID: {vid['id']}")
        print(f"URL: https://youtu.be/{vid['id']}")
        print(f"Status: {vid['status']['uploadStatus']}")
    else:
        print(f"ERROR: Upload failed: {r.status_code} {r.text}")


def cmd_search(args):
    """Search videos on the channel."""
    token = get_access_token()
    
    # Get channel ID
    r = api("GET", "channels", token, params={"part": "id", "mine": "true"})
    channel_id = r.json()["items"][0]["id"]
    
    params = {
        "part": "snippet",
        "channelId": channel_id,
        "q": args.query,
        "type": "video",
        "maxResults": args.count,
        "order": "relevance",
    }
    
    r = api("GET", "search", token, params=params)
    data = r.json()
    
    if not data.get("items"):
        print("No results found.")
        return
    
    for i, item in enumerate(data["items"], 1):
        s = item["snippet"]
        vid = item["id"]["videoId"]
        print(f"{i}. {s['title']}")
        print(f"   https://youtu.be/{vid}")
        print(f"   {s.get('publishedAt', '')[:10]}")
        print()


def cmd_delete(args):
    """Delete a video by ID."""
    token = get_access_token()
    r = api("DELETE", "videos", token, params={"id": args.video_id})
    if r.status_code == 204:
        print(f"✅ Video {args.video_id} deleted.")
    else:
        print(f"ERROR: {r.status_code} {r.text}")


def main():
    parser = argparse.ArgumentParser(description="YouTube CLI for Hermes Agent")
    sub = parser.add_subparsers(dest="command", required=True)
    
    # channel
    sub.add_parser("channel", help="Show channel info")
    
    # list
    p_list = sub.add_parser("list", help="List recent videos")
    p_list.add_argument("-n", "--count", type=int, default=10, help="Number of videos")
    
    # upload
    p_upload = sub.add_parser("upload", help="Upload a video")
    p_upload.add_argument("file", help="Video file path")
    p_upload.add_argument("-t", "--title", required=True, help="Video title")
    p_upload.add_argument("-d", "--description", default="", help="Video description")
    p_upload.add_argument("--tags", default="", help="Comma-separated tags")
    p_upload.add_argument("--category", default="22", help="Category ID (default: 22)")
    p_upload.add_argument("--privacy", default="private", choices=["public", "unlisted", "private"])
    
    # search
    p_search = sub.add_parser("search", help="Search channel videos")
    p_search.add_argument("query", help="Search query")
    p_search.add_argument("-n", "--count", type=int, default=5)
    
    # delete
    p_del = sub.add_parser("delete", help="Delete a video")
    p_del.add_argument("video_id", help="Video ID to delete")
    
    args = parser.parse_args()
    
    commands = {
        "channel": cmd_channel,
        "list": cmd_list,
        "upload": cmd_upload,
        "search": cmd_search,
        "delete": cmd_delete,
    }
    
    commands[args.command](args)


if __name__ == "__main__":
    main()