Skip to content

Instantly share code, notes, and snippets.

@blacktwin
Last active December 19, 2017 22:43
Show Gist options
  • Save blacktwin/bcee9eae4fcb3e474a338d54961d01a4 to your computer and use it in GitHub Desktop.
Save blacktwin/bcee9eae4fcb3e474a338d54961d01a4 to your computer and use it in GitHub Desktop.
Pull all user IP data from PlexPy and using get_geoip_lookup grab the long, lat, city name, IP, username. Use the long and lat to map location on map on global. Legend includes City and Username. Duplicates are removed in legend. Legend can be dragged. Client location plot size is based on play count.
import requests
import sys
import matplotlib.pyplot as plt
from mpl_toolkits.basemap import Basemap
import matplotlib as mpl
from collections import OrderedDict
import math
import numpy as np
## EDIT THESE SETTINGS ##
PLEXPY_APIKEY = 'XXXXXXX' # Your PlexPy API key
PLEXPY_URL = 'http://localhost:8181/' # Your PlexPy URL
# Replace LAN IP addresses that start with the LAN_SUBNET with a WAN IP address
# to retrieve geolocation data. Leave REPLACEMENT_WAN_IP blank for no replacement.
LAN_SUBNET = ('127.0.0')
REPLACEMENT_WAN_IP = ''
# Find your user_id number and play at least one file from the server.
# This will be used to mark the server location.
SERVER_USER_ID = ''
# Enter Friendly name for Server ie 'John Smith'
SERVER_FRIENDLY = ''
# Server location information. Find this information on your own.
# If server plot is out of scope add print(geo_lst) after ut.user_id loop. ~line 143 to find the error
SERVER_LON = ''
SERVER_LAT = ''
SERVER_CITY = ''
SERVER_STATE = ''
## END EDIT ##
## Map stuff ##
plt.figure(figsize=(16,8))
m = Basemap(projection='robin', lat_0=0, lon_0=-100, resolution='l', area_thresh=100000.0)
mpl.rcParams['legend.handlelength'] = 0
m.drawmapboundary(fill_color='#1F1F1F')
m.drawcoastlines()
m.drawstates()
m.drawcountries()
m.fillcontinents(color='#3C3C3C', lake_color='#1F1F1F')
m.drawmeridians(np.arange(0, 360, 30))
m.drawparallels(np.arange(-90, 90, 30))
title_string = "Location of Plex users based on ISP data"
class GeoData(object):
def __init__(self, data=None):
data = data or {}
self.continent = data.get('continent', 'N/A')
self.country = data.get('country', 'N/A')
self.region = data.get('region', 'N/A')
self.city = data.get('city', 'N/A')
self.postal_code = data.get('postal_code', 'N/A')
self.timezone = data.get('timezone', 'N/A')
self.latitude = data.get('latitude', 'N/A')
self.longitude = data.get('longitude', 'N/A')
self.accuracy = data.get('accuracy', 'N/A')
class UserTB(object):
def __init__(self, data=None):
data = data or []
self.user_id = [d['user_id'] for d in data]
class UserIPs(object):
def __init__(self, data=None):
data = data or []
self.ip_address = [d['ip_address'] for d in data]
self.friendly_name = [d['friendly_name'] for d in data]
self.play_count = [d['play_count'] for d in data]
def get_get_users_tables():
# Get the user IP list from PlexPy
payload = {'apikey': PLEXPY_APIKEY,
'cmd': 'get_users_table'}
try:
r = requests.get(PLEXPY_URL.rstrip('/') + '/api/v2', params=payload)
response = r.json()
res_data = response['response']['data']['data']
return UserTB(data=res_data)
except Exception as e:
sys.stderr.write("PlexPy API 'get_get_users_tables' request failed: {0}.".format(e))
def get_get_users_ips(user_id=''):
# Get the user IP list from PlexPy
payload = {'apikey': PLEXPY_APIKEY,
'cmd': 'get_user_ips',
'user_id': user_id,
'length': 25} #length is number of returns, default is 25
try:
r = requests.get(PLEXPY_URL.rstrip('/') + '/api/v2', params=payload)
response = r.json()
recordsTotal = response['response']['data']['recordsTotal']
res_data = response['response']['data']['data']
return UserIPs(data=res_data)
except Exception as e:
sys.stderr.write("PlexPy API 'get_get_users_ips' request failed: {0}.".format(e))
def get_geoip_info(ip_address=''):
# Get the geo IP lookup from PlexPy
payload = {'apikey': PLEXPY_APIKEY,
'cmd': 'get_geoip_lookup',
'ip_address': ip_address}
try:
r = requests.get(PLEXPY_URL.rstrip('/') + '/api/v2', params=payload)
response = r.json()
if response['response']['result'] == 'success':
data = response['response']['data']
if data.get('error'):
raise Exception(data['error'])
else:
sys.stdout.write("Successfully retrieved geolocation data.")
return GeoData(data=data)
else:
raise Exception(response['response']['message'])
except Exception as e:
sys.stderr.write("PlexPy API 'get_geoip_lookup' request failed: {0}.".format(e))
return GeoData()
if __name__ == '__main__':
geo_lst = [[SERVER_LON, SERVER_LAT, SERVER_CITY, SERVER_STATE, SERVER_USER_ID, REPLACEMENT_WAN_IP, SERVER_FRIENDLY, 0]]
ut = get_get_users_tables()
for i in ut.user_id:
ip = get_get_users_ips(user_id=i)
c = 0
if ip is not None:
fn = [x.encode('UTF8') for x in ip.friendly_name]
fn = list(set(fn))
ulst = [str(i)]
for x in ip.ip_address:
if x.startswith(LAN_SUBNET) and REPLACEMENT_WAN_IP:
x = REPLACEMENT_WAN_IP
g = get_geoip_info(ip_address=x)
if str(g.longitude) == 'N/A' or str(g.latitude) == 'N/A':
pass
else:
glst = [str(g.longitude)] + [str(g.latitude)] + [str(g.city)] + [str(g.region)]
geo_lst += [glst + ulst + [str(x)] + fn + [ip.play_count[c]]]
c += 1
for (lon, lat, city, reg, user_id, ip, fn, pc) in geo_lst:
if user_id == SERVER_USER_ID and ip == REPLACEMENT_WAN_IP:
color = '#FFAC05'
marker = '*'
markersize = 10
else:
color = '#A96A1C'
marker = '.'
markersize = 5 + pc * .1
x, y = m(lon, lat)
labels = 'Location: ' + city + ', ' + reg + ' User: ' + fn
m.drawgreatcircle(float(SERVER_LON), float(SERVER_LAT), float(lon), float(lat), linewidth=1, alpha=0.4,
color='#AC7420')
m.plot(x, y, marker=marker, color=color, markersize=markersize, label=labels, alpha=0.3)
handles, labels = plt.gca().get_legend_handles_labels()
by_label = OrderedDict(zip(labels, handles))
leg = plt.legend(by_label.values(), by_label.keys(), fancybox=True, fontsize='x-small', numpoints=1, title="Legend")
if leg:
lleng = len(leg.legendHandles)
for i in range(0, lleng):
leg.legendHandles[i]._legmarker.set_markersize(5)
leg.legendHandles[i]._legmarker.set_alpha(1)
leg.get_title().set_color('#7B777C')
leg.draggable()
leg.get_frame().set_facecolor('#2C2C2C')
for text in leg.get_texts():
plt.setp(text, color='#A5A5A7')
plt.title(title_string)
plt.show()
import requests
import sys
import matplotlib.pyplot as plt
from mpl_toolkits.basemap import Basemap
import matplotlib as mpl
from collections import OrderedDict
import math
import numpy as np
## EDIT THESE SETTINGS ##
PLEXPY_APIKEY = 'XXXXXXX' # Your PlexPy API key
PLEXPY_URL = 'http://localhost:8181/' # Your PlexPy URL
# Replace LAN IP addresses that start with the LAN_SUBNET with a WAN IP address
# to retrieve geolocation data. Leave REPLACEMENT_WAN_IP blank for no replacement.
LAN_SUBNET = ('127.0.0')
REPLACEMENT_WAN_IP = ''
# Find your user_id number and play at least one file from the server.
# This will be used to mark the server location.
SERVER_USER_ID = ''
# Enter Friendly name for Server ie 'John Smith'
SERVER_FRIENDLY = ''
# Server location information. Find this information on your own.
# If server plot is out of scope add print(geo_lst) after ut.user_id loop. ~line 143 to find the error
SERVER_LON = ''
SERVER_LAT = ''
SERVER_CITY = ''
SERVER_STATE = ''
## END EDIT ##
## Map stuff ##
plt.figure(figsize=(16,8))
m = Basemap(projection='robin', lat_0=0, lon_0=-100, resolution='l', area_thresh=100000.0)
mpl.rcParams['legend.handlelength'] = 0
m.drawmapboundary(fill_color='#1F1F1F')
m.drawcoastlines()
m.drawstates()
m.drawcountries()
m.fillcontinents(color='#3C3C3C', lake_color='#1F1F1F')
m.drawmeridians(np.arange(0, 360, 30))
m.drawparallels(np.arange(-90, 90, 30))
title_string = "Location of Plex users based on ISP data"
class GeoData(object):
def __init__(self, data=None):
data = data or {}
self.continent = data.get('continent', 'N/A')
self.country = data.get('country', 'N/A')
self.region = data.get('region', 'N/A')
self.city = data.get('city', 'N/A')
self.postal_code = data.get('postal_code', 'N/A')
self.timezone = data.get('timezone', 'N/A')
self.latitude = data.get('latitude', 'N/A')
self.longitude = data.get('longitude', 'N/A')
self.accuracy = data.get('accuracy', 'N/A')
class UserIPs(object):
def __init__(self, data=None):
d = data or {}
self.ip_address = d['ip_address']
self.friendly_name = d['friendly_name']
self.play_count = d['play_count']
def get_get_users_tables():
# Get the user IP list from PlexPy
payload = {'apikey': PLEXPY_APIKEY,
'cmd': 'get_users_table'}
try:
r = requests.get(PLEXPY_URL.rstrip('/') + '/api/v2', params=payload)
response = r.json()
res_data = response['response']['data']['data']
return [d['user_id'] for d in res_data]
except Exception as e:
sys.stderr.write("PlexPy API 'get_get_users_tables' request failed: {0}.".format(e))
def get_get_users_ips(user_id):
# Get the user IP list from PlexPy
payload = {'apikey': PLEXPY_APIKEY,
'cmd': 'get_user_ips',
'user_id': user_id,
'length': 25} #length is number of returns, default is 25
try:
r = requests.get(PLEXPY_URL.rstrip('/') + '/api/v2', params=payload)
response = r.json()
res_data = response['response']['data']['data']
return [UserIPs(data=d) for d in res_data]
except Exception as e:
sys.stderr.write("PlexPy API 'get_get_users_ips' request failed: {0}.".format(e))
def get_geoip_info(ip_address=''):
# Get the geo IP lookup from PlexPy
payload = {'apikey': PLEXPY_APIKEY,
'cmd': 'get_geoip_lookup',
'ip_address': ip_address}
try:
r = requests.get(PLEXPY_URL.rstrip('/') + '/api/v2', params=payload)
response = r.json()
if response['response']['result'] == 'success':
data = response['response']['data']
if data.get('error'):
raise Exception(data['error'])
else:
sys.stdout.write("Successfully retrieved geolocation data.")
return GeoData(data=data)
else:
raise Exception(response['response']['message'])
except Exception as e:
sys.stderr.write("PlexPy API 'get_geoip_lookup' request failed: {0}.".format(e))
return GeoData()
def add_to_dictlist(d, key, val):
if key not in d:
d[key] = [val]
else:
d[key].append(val)
if __name__ == '__main__':
geo_dict = {SERVER_FRIENDLY:[[SERVER_LON, SERVER_LAT, SERVER_CITY, SERVER_STATE,
REPLACEMENT_WAN_IP, 0]]}
for i in get_get_users_tables():
try:
user_ip = get_get_users_ips(user_id=i)
for a in user_ip:
ip = a.ip_address
if ip.startswith(LAN_SUBNET) and REPLACEMENT_WAN_IP:
ip = REPLACEMENT_WAN_IP
g = get_geoip_info(ip_address=ip)
add_to_dictlist(geo_dict, a.friendly_name, [str(g.longitude), str(g.latitude),
str(g.city), str(g.region), ip, a.play_count])
except Exception as e:
print(e)
pass
#print(geo_dict)
for key, value in geo_dict.items():
for (lon, lat, city, reg, ip, pc) in value:
if key == SERVER_FRIENDLY:
color = '#FFAC05'
marker = '*'
markersize = 10
else:
color = '#A96A1C'
marker = '.'
markersize = 5 + (pc * .1)
x, y = m(lon, lat)
labels = 'Location: ' + city + ', ' + reg + ' User: ' + key
m.drawgreatcircle(float(SERVER_LON), float(SERVER_LAT), float(lon), float(lat), linewidth=1, alpha=0.4,
color='#AC7420')
m.plot(x, y, marker=marker, color=color, markersize=markersize, label=labels, alpha=0.3)
handles, labels = plt.gca().get_legend_handles_labels()
by_label = OrderedDict(zip(reversed(labels), reversed(handles)))
leg = plt.legend(by_label.values(), by_label.keys(), fancybox=True, fontsize='x-small',
numpoints=1, title="Legend")
if leg:
lleng = len(leg.legendHandles)
for i in range(0, lleng):
leg.legendHandles[i]._legmarker.set_markersize(5)
leg.legendHandles[i]._legmarker.set_alpha(1)
leg.get_title().set_color('#7B777C')
leg.draggable()
leg.get_frame().set_facecolor('#2C2C2C')
for text in leg.get_texts():
plt.setp(text, color='#A5A5A7')
plt.title(title_string)
plt.show()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment