owncast-clipper/main.py

# Standard Imports
import os
import sys
from urllib.parse import urlparse
from pathlib import Path

# Module Imports
import config
from helpers.websocket import owncast_connect


def main():
    config.CLIP_DIRECTORY = Path(os.environ.get(
        'CLIP_DIRECTORY', '/app/data/hls/0/'))
    config.TEMP_DIRECTORY = Path(os.environ.get('TEMP_DIRECTORY', '/tmp/'))
    try:
        config.CLIP_QUANTITY = int(os.environ.get('CLIP_QUANTITY', '20'))
    except ValueError:
        print("Error: config.CLIP_QUANTITY must be an integer.", file=sys.stderr)
        sys.exit(1)

    try:
        # full base, e.g. https://peertube.coolsite.com or http://peertube:8080
        config.PEERTUBE_HOST = os.environ['PEERTUBE_HOST'].rstrip('/')
        config.PEERTUBE_USER = os.environ['PEERTUBE_USER']
        config.PEERTUBE_PASS = os.environ['PEERTUBE_PASS']
        config.PEERTUBE_CHANNEL_ID = os.environment['PEERTUBE_CHANNEL_ID']
    except KeyError as e:
        print(
            f'Error: PeerTube host and credentials must be specified: {e}', file=sys.stderr)
        sys.exit(1)
    config.PT_API_BASE = f'{config.PEERTUBE_HOST}/api/v1'

    try:
        # full base, e.g. https://owncast.site or http://app:8080
        config.OWNCAST_HOST = os.environ['OWNCAST_HOST'].rstrip('/')
        config.OWNCAST_AUTH = os.environ['OWNCAST_AUTH']
    except KeyError as e:
        print(
            f'Error: Owncast host and access token must be specified: {e}', file=sys.stderr)
        sys.exit(1)
    config.AUTHENTICATION_REQUIRED = os.environ.get(
        'AUTHENTICATION_REQUIRED', False)

    for d in [config.CLIP_DIRECTORY, config.TEMP_DIRECTORY]:
        if not d.is_dir():
            print(f'Error: {d} is not a directory.', file=sys.stderr)
            sys.exit(1)

    owncast_parsed = urlparse(config.OWNCAST_HOST)
    owncast_scheme, owncast_domain = owncast_parsed.scheme, owncast_parsed.netloc

    if owncast_scheme.lower() == 'http':
        owncast_ws_scheme = 'ws'
    elif owncast_scheme.lower() == 'https':
        owncast_ws_scheme = 'wss'
    owncast_connect_url = f'{owncast_ws_scheme}://{owncast_domain}/ws?accessToken={config.OWNCAST_AUTH}'
    print(owncast_connect_url)
    owncast_connect(owncast_connect_url, config.WEBSOCKET_MAX_RETRIES,
                    config.WEBSOCKET_INITIAL_BACKOFF)


if __name__ == "__main__":
    main()