Automatically share and unshare libraries for Plex users

README.md

Requirements

Have python 3 installed somewhere

Install

Download gist zip file and extract to a directory of your choosing then open terminal / cmd prompt at that directory

For mac and linux

python3 -m venv ./venv
source ./venv/bin/activate
pip3 install -r requirements.txt

For windows

python3 -m venv /path/to/new/virtual/environment
C:\path\to\new\virtual\environment\bin\activate.bat
pip3 install -r requirements.txt

Configuration

User libraries

Update the file called user_libraries_file.json with the appropriate data Get the User IDs and Library IDs from https://plex.tv/api/servers/SERVER_ID/shared_servers See Example: https://i.imgur.com/yt26Uni.png Enter the User IDs and Library IDs in this format below:

{
    UserID1: [LibraryID1, LibraryID2],
    UserID2: [LibraryID1, LibraryID2]
}

Server id and plex token

The server id and plex token can be passed via the command line or with environment variables. Passing via the command line takes precedence and the examples below show it being used this way.

Sharing

plex_util.py --share -l ./user_libraries_file.json -t "YOUR_PLEX_API_TOKEN_HERE" -s "YOUR_SERVER_ID_HERE"

Removing shares

plex_util.py --unshare -l ./user_libraries_file.json -t "YOUR_PLEX_API_TOKEN_HERE" -s "YOUR_SERVER_ID_HERE"

plex_util.py

#!/usr/bin/env python3

import argparse
import json
import os
import sys
from collections import defaultdict
from xml.etree import ElementTree
import requests
import logging

logger = logging.getLogger(__name__)
logging.basicConfig(stream=sys.stdout, format='%(levelname)s:%(message)s', level=logging.DEBUG)


def add_share(server_id, plex_token, json_data):
    """
    Posts to the shared servers to enable the share
    :param server_id: the server id
    :param plex_token: the plex token so that api calls work
    :param json_data: the payload of data as json
    :return: the posted response or None
    """
    url = f'https://plex.tv/api/servers/{server_id}/shared_servers'
    headers = {'X-Plex-Token': plex_token, 'Accept': 'application/json'}
    resp = None
    try:
        resp = requests.post(url, headers=headers, json=json_data)
        resp.raise_for_status()
    except ConnectionError as e:
        logger.error(f'A connection error occurred, {e}')
    except requests.exceptions.RequestException as e:
        logger.error(f'A request exception occurred, {e}')
    return resp


def get_shares(server_id, plex_token):
    """
    Gets the current shares from the api, expected response is xml
    :param server_id: the server id
    :param plex_token: the plex token so that api calls work
    :return: the get response or None
    """
    url = f'https://plex.tv/api/servers/{server_id}/shared_servers'
    headers = {'X-Plex-Token': plex_token, 'Accept': 'application/json'}
    resp = None
    try:
        resp = requests.get(url, headers=headers)
        resp.raise_for_status()
    except ConnectionError as e:
        logger.error(f'A connection error occurred, {e}')
    except requests.exceptions.RequestException as e:
        logger.error(f'A request exception occurred, {e}')
    return resp


def delete_share(server_id, plex_token, user_server_id):
    url = f'https://plex.tv/api/servers/{server_id}/shared_servers/{user_server_id}'
    headers = {'X-Plex-Token': plex_token, 'Accept': 'application/json'}
    resp = None
    try:
        resp = requests.delete(url, headers=headers)
        resp.raise_for_status()
    except ConnectionError as e:
        logger.error(f'A connection error occurred, {e}')
    except requests.exceptions.RequestException as e:
        logger.error(f'A request exception occurred, {e}')
    return resp


def parse_shared_servers(content):
    """
    Takes the shares xml feed of data and returns a dict
    :param content: the raw text content, expecting xml string
    :return: a dict of key being user_id and value being list of server_id
    """
    data = defaultdict(list)
    tree = ElementTree.fromstring(content)
    shared_servers = tree.findall('SharedServer')
    for shared_server in shared_servers:
        user_id = int(shared_server.get('userID'))
        for server in shared_server.findall('Section'):
            data[user_id].append(int(server.get('id')))
    return data


def share(server_id, plex_token, user_libraries):
    """
    Enables the share on all users
    :param server_id: the server id
    :param plex_token: the plex token so that api calls work
    """
    for user_id, library_ids in user_libraries.items():
        payload = {"server_id": server_id, "shared_server": {"library_section_ids": library_ids, "invited_id": user_id}}
        response = add_share(server_id, plex_token, payload)
        if response.ok:
            logger.info("Successfully added share")


def un_share(server_id, plex_token, user_libraries):
    """
    Removes the share on all users
    :param server_id: the server id
    :param plex_token: the plex token so that api calls work
    """
    get_response = get_shares(server_id, plex_token)
    shared_servers = parse_shared_servers(get_response.content)
    for user_id, library_ids in user_libraries.items():
        user_server_id = shared_servers.get(user_id)
        if server_id:
            delete_response = delete_share(server_id, plex_token, user_server_id)
            if delete_response.ok:
                logger.info(f'Unshared libraries with user {user_id}')
        else:
            logger.info(f'No libraries shared with user {user_id}')


def init_args():
    """
    Sets up argument parsing and returns the arguments
    :return: argparse values
    """
    parser = argparse.ArgumentParser(description='Share / Un share Plex')
    parser.add_argument('-t', action="store", dest="plex_token", type=str, default=os.environ.get('PLEX_TOKEN', ''))
    parser.add_argument('-s', action="store", dest="server_id", type=str, default=os.environ.get('SERVER_ID', ''))
    parser.add_argument('-l', action="store", dest="user_libraries_file", type=argparse.FileType('r'), required=True)
    group = parser.add_mutually_exclusive_group()
    group.add_argument('--share', action='store_true', default=False, dest='share')
    group.add_argument('--unshare', action='store_true', default=False, dest='unshare')
    args = parser.parse_args()
    return args


def main():
    """
    Entry function that uses the arguments passed to it and calls share or unshare accordingly
    """
    args = init_args()
    user_libraries = json.loads(args.user_libraries_file)
    if args.share:
        share(args.server_id, args.plex_token, user_libraries)
    else:
        un_share(args.server_id, args.plex_token, user_libraries)


if __name__ == "__main__":
    main()

test_plex_util.py

import argparse
import io
import unittest
from unittest import mock

import requests
from plex_util import add_share, get_shares, delete_share, parse_shared_servers, init_args, main


class TestAddShare(unittest.TestCase):

    def test_bad_credentials(self):
        resp = add_share('12345', '54321', {})
        actual = resp.status_code
        expected = 401
        self.assertEqual(actual, expected)

    def test_returns_response_object(self):
        resp = add_share('12345', '54321', {})
        actual = isinstance(resp, requests.models.Response)
        expected = True
        self.assertEqual(actual, expected)


class TestGetShares(unittest.TestCase):

    def test_bad_credentials(self):
        resp = get_shares('12345', '54321')
        actual = resp.status_code
        expected = 401
        self.assertEqual(actual, expected)

    def test_returns_response_object(self):
        resp = get_shares('12345', '54321')
        actual = isinstance(resp, requests.models.Response)
        expected = True
        self.assertEqual(actual, expected)


class TestDeleteShare(unittest.TestCase):

    def test_bad_credentials(self):
        resp = delete_share('12345', '54321', '9999')
        actual = resp.status_code
        expected = 401
        self.assertEqual(actual, expected)

    def test_returns_response_object(self):
        resp = delete_share('12345', '54321', '9999')
        actual = isinstance(resp, requests.models.Response)
        expected = True
        self.assertEqual(actual, expected)


class TestParseSharedServersXML(unittest.TestCase):

    def test_correct_parse(self):
        xml_content = '''
        <MediaContainer friendlyName="myPlex" machineIdentifier="" identifier="con.plexapp.plugins.myplex">
            <SharedServer userID="12345">
                <Section id="9999" />
                <Section id="1111" />
            </SharedServer>
        </MediaContainer>
        '''
        shared_server = parse_shared_servers(xml_content)
        actual = shared_server
        expected = {12345: [9999, 1111]}
        self.assertEqual(actual, expected)


class TestShare(unittest.TestCase):

    def test_share(self):
        self.assertEqual(True, True)


class TestUnShare(unittest.TestCase):

    def test_unshare(self):
        self.assertEqual(True, True)


class TestInitArgs(unittest.TestCase):

    @mock.patch('sys.argv', ['-l', 'user_libraries_file.json', '--share', '--unshare'])
    def test_share_unshare_both_exist_error(self):
        with self.assertRaises(SystemExit):
            init_args()

    @mock.patch('sys.argv', ['--share'])
    def test_required_json_file_missing_error(self):
        with self.assertRaises(SystemExit):
            init_args()


class TestMain(unittest.TestCase):

    @mock.patch('argparse.ArgumentParser.parse_args',
                return_value=argparse.Namespace(plex_token='12345', server_id='54321', share=True, unshare=False,
                                                user_libraries_file=io.StringIO("some initial text data")))
    def test_invalid_json_raises(self, mock_args):
        with self.assertRaises(TypeError):
            main()


if __name__ == '__main__':
    unittest.main()

user_libraries_file.json

{
	"1234567": [
		1234567, 1234567
	]
}