initial commit

helpers/peertube_api.py

@@ -0,0 +1,132 @@
+# 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