-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathPython implementation
107 lines (94 loc) · 4 KB
/
Python implementation
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
import os
import requests
import time
from tqdm import tqdm
# Vimeo API access token
access_token = '{TOKEN}' # Replace with your actual access token
video_filename = 'myfile.mp4' # Replace with your video file name
# Step 1: Create the video
filename_without_extension = os.path.splitext(video_filename)[0]
headers = {
'Authorization': f'Bearer {access_token}',
'Content-Type': 'application/json',
'Accept': 'application/vnd.vimeo.*+json;version=3.4'
}
params = {
'upload': {
'approach': 'tus',
'size': os.path.getsize(video_filename)
},
'name': filename_without_extension
}
response = requests.post('https://api.vimeo.com/me/videos?fields=uri,upload.upload_link', headers=headers, json=params)video_data = response.json()
upload_link = video_data.get('upload', {}).get('upload_link')
video_id = video_data['uri'].split('/')[-1]
print(f"Video ID: {video_id}") # Display the video ID
# Set the chunk size to 256 MB for efficient data transmission
chunk_size = 256 * 1024 * 1024
# Step 2: Upload the video file
with open(video_filename, 'rb') as f:
headers = {
'Tus-Resumable': '1.0.0',
'Upload-Offset': '0',
'Content-Type': 'application/offset+octet-stream'
}
while True:
data = f.read(chunk_size)
if not data:
break
response = requests.patch(upload_link, headers=headers, data=data)
headers['Upload-Offset'] = response.headers.get('Upload-Offset', '0')
upload_offset = int(headers['Upload-Offset'])
if upload_offset == os.path.getsize(video_filename):
print("Upload complete!")
break
# Step 3: Verify the upload
response = requests.head(upload_link, headers={'Tus-Resumable': '1.0.0', 'Accept': 'application/vnd.vimeo.*+json;version=3.4'})
upload_length = int(response.headers.get('Upload-Length', 0))
upload_offset = int(response.headers.get('Upload-Offset', 0))
if upload_length == upload_offset:
print("Upload verified: Video file received completely.")
else:
print("Upload verification failed: Video file upload incomplete.")
# Step 4: Poll video transcode status and log to CLI with progress bar
def get_video_status(video_id, access_token):
url = f"https://api.vimeo.com/videos/{video_id}?fields=uri,transcode.status"
headers = {
'Authorization': f'Bearer {access_token}'
}
response = requests.get(url, headers=headers)
if response.status_code == 200:
video_data = response.json()
transcode_status = video_data.get('transcode', {}).get('status', 'unknown')
return transcode_status
else:
print(f"Error fetching video status: {response.json()}")
return None
def poll_video_status(video_id, access_token, interval=10):
progress_bar = tqdm(total=100, desc=f"Transcoding Progress (ID: {video_id})", bar_format="{l_bar}{bar}| {n_fmt}/{total_fmt} [{elapsed}<{remaining}]")
start_time = time.time()
while True:
status = get_video_status(video_id, access_token)
elapsed_time = time.time() - start_time
if status == "complete":
progress_bar.n = 100
progress_bar.set_description("Transcoding Complete")
progress_bar.close()
print(f"Video {video_id} transcoding complete.")
break
elif status == "error":
progress_bar.set_description("Transcoding Error")
progress_bar.close()
print(f"Video {video_id} transcoding error.")
break
elif status == "in_progress":
progress_percentage = min(int(elapsed_time / interval * 10), 99) # Estimate progress
progress_bar.n = progress_percentage
progress_bar.set_description(f"Transcoding in Progress ({progress_percentage}%)")
else:
progress_bar.set_description("Unexpected Status")
print(f"Unexpected status: {status}")
progress_bar.refresh()
time.sleep(interval)
progress_bar.close()
poll_video_status(video_id, access_token, interval=10)