133 lines
4.7 KiB
Python
133 lines
4.7 KiB
Python
# Standard Imports
|
|
import sys
|
|
import json
|
|
from pprint import pprint
|
|
|
|
# Package Imports
|
|
import requests
|
|
import peertube
|
|
|
|
# Local Imports
|
|
from peertube.rest import ApiException
|
|
from pathlib import Path
|
|
|
|
|
|
def peertube_request(endpoint: str, verb: str, **kwargs):
|
|
"""
|
|
A helper function that facilitates basic requests to PeerTube and decodes.
|
|
|
|
Args:
|
|
endpoint (str): A URL to act against.
|
|
verb (str): either "GET" or "POST".
|
|
**kwargs: Passed through to underlying request call (for e.g. POSTs).
|
|
|
|
Returns:
|
|
dict: The JSON-decoded response from the request, or None if there was an error.
|
|
"""
|
|
|
|
data = kwargs.get('data', None)
|
|
try:
|
|
if verb == 'GET':
|
|
response = requests.get(endpoint)
|
|
elif verb == 'POST':
|
|
response = requests.post(endpoint, data=data)
|
|
else:
|
|
return None
|
|
status_code = response.status_code
|
|
data = response.json()
|
|
if status_code != 200:
|
|
print(
|
|
f'Received a {status_code} response from PeerTube: {response}', file=sys.stderr)
|
|
return None
|
|
except json.JSONDecodeError as e:
|
|
print(
|
|
f'Unable to parse response JSON from PeerTube: {e}: {response}', file=sys.stderr)
|
|
return None
|
|
except requests.exceptions.RequestException as e:
|
|
print(
|
|
f'An error occurred while communicating with PeerTube: {e}', file=sys.stderr)
|
|
return None
|
|
else:
|
|
return data
|
|
|
|
|
|
def authenticate_to_peertube(api_base: str, username: str, password: str):
|
|
"""
|
|
Collects an access token from PeerTube via password authentication
|
|
|
|
Args:
|
|
api_base (str): The API base url for the given PeerTube instance, e.g. https://video.site/api/v1
|
|
username (str): The username of the PeerTube account to be used in authentication
|
|
password (str): The password of the PeerTube account to be used in authentication
|
|
|
|
Returns:
|
|
str: A string containing the access_token following authentication, or None if unable to auth.
|
|
"""
|
|
|
|
if (data := peertube_request(api_base + '/oauth-clients/local', 'GET')) is None:
|
|
print('Error initializing OAuth flow with PeerTube.', file=sys.stderr)
|
|
return None
|
|
client_id = data['client_id']
|
|
client_secret = data['client_secret']
|
|
data = {
|
|
'client_id': client_id,
|
|
'client_secret': client_secret,
|
|
'grant_type': 'password',
|
|
'response_type': 'code',
|
|
'username': username,
|
|
'password': password
|
|
}
|
|
if (data := peertube_request(api_base + '/users/token', 'POST', data=data)) is None:
|
|
print('Error authenticating with PeerTube via OAuth.', file=sys.stderr)
|
|
return None
|
|
return data['access_token']
|
|
|
|
|
|
def upload_to_peertube(api_base: str, api_token: str, file: Path, channel_id: int, title: str, **kwargs):
|
|
"""
|
|
Uploads a file from the local filesystem to PeerTube and returns a ready-to-consume URL to the video.
|
|
|
|
Args:
|
|
api_base (str): The API base url for the given PeerTube instance, e.g. https://video.site/api/v1
|
|
api_token (str): A valid OAuth access token obtained from previous authentication.
|
|
file (Path): An absolute path to the video file to be uploaded.
|
|
channel_id (int): The internal ID number of the PeerTube channel to be uploaded to.
|
|
title (str): The public-facing name of the video on PeerTube
|
|
**kwargs: Any other values to pass to the PeerTube Video Upload API: https://framagit.org/framasoft/peertube/clients/python/-/blob/master/docs/VideoApi.md#videos_upload_post
|
|
|
|
Returns:
|
|
str: A string containing the URL to the video on PeerTube, or None if unable to publish.
|
|
"""
|
|
configuration = peertube.Configuration(
|
|
host=api_base
|
|
)
|
|
configuration.access_token = api_token
|
|
|
|
with peertube.ApiClient(configuration) as api_client:
|
|
api_instance = peertube.VideoApi(api_client)
|
|
try:
|
|
options = {
|
|
**kwargs,
|
|
'privacy': 1,
|
|
'wait_transcoding': True
|
|
}
|
|
api_response = api_instance.videos_upload_post(
|
|
file, channel_id, title, **options)
|
|
except ApiException as e:
|
|
print(
|
|
f'Exception while uploading to PeerTube: {e}', file=sys.stderr)
|
|
return None
|
|
|
|
try:
|
|
video_id = api_response.video.uuid
|
|
api_response = api_instance.videos_id_get(video_id)
|
|
except ApiException as e:
|
|
print(
|
|
f'Exception while getting Clip URL from PeerTube: {e}', file=sys.stderr)
|
|
return None
|
|
except KeyError as e:
|
|
print(
|
|
f'Could not find Video ID in response from PeerTube: {e}', file=sys.stderr)
|
|
return None
|
|
return api_response.embed_path
|