1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
|
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)
|
