自動化って大事かも・・・！PeerTubeの動画再投稿で実感した話

import os
import json
import requests
from io import BytesIO
import yt_dlp
from datetime import datetime
from PIL import Image
import time

# 設定
YOUTUBE_URL = "https://www.youtube.com/watch?v=xxxxxxxx" #投稿する動画を指定する
PEERTUBE_INSTANCE = "https://prtb.komaniya.work" #投稿先のインスタンス
USERNAME = "kamosika" #peertubeインスタンスのユーザー名
PASSWORD = "xxxxxxxxx" #パスワード
UPLOAD_CHANNEL_ID = 2 # 動画を投稿するチャンネルのID https://prtb.komaniya.work/api/v1/video-channels で確認した

# client_id と client_secret を取得
def get_client_credentials():
    url = f"{PEERTUBE_INSTANCE}/api/v1/oauth-clients/local"
    response = requests.get(url)
    response_data = response.json()
    client_id = response_data.get("client_id")
    client_secret = response_data.get("client_secret")
    return client_id, client_secret

# PeerTubeのアクセストークンを取得
def get_peertube_access_token():
    client_id, client_secret = get_client_credentials()
    url = f"{PEERTUBE_INSTANCE}/api/v1/users/token"
    headers = {
    "Content-Type": "application/x-www-form-urlencoded"
    }
    data = {
        "client_id": client_id,
        "client_secret": client_secret,
        "grant_type": "password",
        "username": USERNAME,
        "password": PASSWORD
    }
    response = requests.post(url, headers=headers, data=data)
    response_data = response.json()
    return response_data.get("access_token")

# YouTubeのメタデータを取得
def get_youtube_metadata(url):
    ydl_opts = {
        'quiet': True,
        'dumpjson': True
    }
    with yt_dlp.YoutubeDL(ydl_opts) as ydl:
        info = ydl.extract_info(url, download=False)
    return info

# YouTubeの動画をダウンロード
def download_youtube_video(url, output_path):
    ydl_opts = {
        'outtmpl': output_path,
        'format': 'bestvideo+bestaudio/best',
        'merge_output_format': 'mp4'
    }
    with yt_dlp.YoutubeDL(ydl_opts) as ydl:
        ydl.download([url])

def download_and_convert_thumbnail(thumbnail_url, output_path):
    response = requests.get(thumbnail_url)
    response.raise_for_status()
    # レスポンスのバイナリデータを読み込み、Pillowで画像オブジェクトに変換
    img = Image.open(BytesIO(response.content))
    img = img.convert("RGB")  # JPEG形式はRGBで保存する必要があります
    img.save(output_path, format="JPEG")

# PeerTubeに動画をアップロード
# カテゴリ一覧
#     "1": "Music",
#     "2": "Films",
#     "3": "Vehicles",
#     "4": "Art",
#     "5": "Sports",
#     "6": "Travels",
#     "7": "Gaming",
#     "8": "People",
#     "9": "Comedy",
#     "10": "Entertainment",
#     "11": "News & Politics",
#     "12": "How To",
#     "13": "Education",
#     "14": "Activism",
#     "15": "Science & Technology",
#     "16": "Animals",
#     "17": "Kids",
#     "18": "Food"
def upload_to_peertube(video_path, thumbnail_path, title, description, tags, publish_date, access_token, max_retries=10, wait_time=30):
    headers = {
        "Authorization": f"Bearer {access_token}"
    }
    
    data = {
        "name": title,
        "description": description,
        "channelId": UPLOAD_CHANNEL_ID,
        "category": 10,  # 適切なカテゴリIDに変更
        "privacy": 1,  # 公開
        "tags": tags,
        "language": "ja",
        "originallyPublishedAt": publish_date  # 投稿されていた日
    }
    
    for attempt in range(max_retries):
        try:
            with open(video_path, 'rb') as video_file, open(thumbnail_path, 'rb') as thumb_file:
                files = {
                    'videofile': (video_path, video_file, 'video/mp4'),
                    'thumbnailfile': (thumbnail_path, thumb_file, 'image/jpeg')
                }
                response = requests.post(f"{PEERTUBE_INSTANCE}/api/v1/videos/upload", headers=headers, files=files, data=data)
            
            response_json = response.json()
            if response.status_code == 200:
                return response_json  # 成功したら結果を返す
            else:
                print(f"Upload attempt {attempt + 1} failed: {response_json}")
        except requests.RequestException as e:
            print(f"Upload attempt {attempt + 1} failed due to request error: {e}")
        
        if attempt < max_retries - 1:
            print(f"Retrying in {wait_time} seconds...")
            time.sleep(wait_time)
    
    return {"error": "Upload failed after multiple attempts"}

# タグをフィルタリング（各タグ2～30文字、最大5個に制限）
def filter_tags(tags):
    filtered = [tag for tag in tags if 2 <= len(tag) <= 30]
    return filtered[:5]

# メイン処理
if __name__ == "__main__":
    access_token = get_peertube_access_token()
    if not access_token:
        print("Failed to retrieve access token")
        exit(1)
    
    metadata = get_youtube_metadata(YOUTUBE_URL)
    
    title = metadata['title']
    original_description = metadata['description']
    upload_date = datetime.strptime(metadata['upload_date'], "%Y%m%d").strftime("%Y-%m-%d")
    description = f"※この動画は YouTube で {upload_date} に投稿したものです\n\n---\n\n" + original_description
    tags = metadata.get('tags', [])
    
    # 動画ファイルのダウンロード
    video_filename = f"{title}.mp4"
    download_youtube_video(YOUTUBE_URL, video_filename)
    
    # サムネイル画像の取得（metadata内の "thumbnail" キーを利用）
    thumbnail_url = metadata.get("thumbnail")
    if not thumbnail_url:
        print("サムネイルURLが見つかりません")
        exit(1)
    
    thumbnail_filename = f"{title}_thumbnail.jpg"
    download_and_convert_thumbnail(thumbnail_url, thumbnail_filename)
    
    publish_date = datetime.strptime(metadata['upload_date'], "%Y%m%d").strftime("%Y-%m-%d")
    # publish_date = datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%SZ")
    response = upload_to_peertube(video_filename, thumbnail_filename, title, description, filter_tags(tags), publish_date, access_token)
    
    print(response)