Scribd
Upload
11 views5 pages

Example Py

This Python script is designed to handle various actions related to user authentication, video on demand (VOD), channels, and events using a session-based API. It utilizes the Google Translator for subtitle translation and manages user pairing and login processes, while also handling requests for media manifests and heartbeat signals. The script outputs JSON responses based on the specified action, including error handling for failed requests.

Uploaded by

darllandrm
Copyright
© © All Rights Reserved
We take content rights seriously. If you suspect this is your content, claim it here.
Available Formats
Download as TXT, PDF, TXT or read online on Scribd
11 views5 pages

Example Py

This Python script is designed to handle various actions related to user authentication, video on demand (VOD), channels, and events using a session-based API. It utilizes the Google Translator for subtitle translation and manages user pairing and login processes, while also handling requests for media manifests and heartbeat signals. The script outputs JSON responses based on the specified action, including error handling for failed requests.

Uploaded by

darllandrm
Copyright
© © All Rights Reserved
We take content rights seriously. If you suspect this is your content, claim it here.
Available Formats
Download as TXT, PDF, TXT or read online on Scribd
You are on page 1/ 5

#!

/usr/bin/python3
import sys
import os
import o11
import base64
import json
import datetime
import pytz
import time
from deep_translator import GoogleTranslator

user=o11.parse_params(sys.argv, 'user')
password=o11.parse_params(sys.argv, 'password')
device=o11.parse_params(sys.argv, 'device')
pin=o11.parse_params(sys.argv, 'pin')

id=o11.parse_params(sys.argv, 'id')
text=o11.parse_params(sys.argv, 'text')
srclang=o11.parse_params(sys.argv, 'srclang')
dstlang=o11.parse_params(sys.argv, 'dstlang')
action=o11.parse_params(sys.argv, 'action')

bind=o11.parse_params(sys.argv, 'bind')
proxy=o11.parse_params(sys.argv, 'proxy')
doh=o11.parse_params(sys.argv, 'doh')
dns=o11.parse_params(sys.argv, 'dns')
worker=o11.parse_params(sys.argv, 'worker')

cdm=o11.parse_params(sys.argv, 'cdm')
drm=o11.parse_params(sys.argv, 'drm')
kid=o11.parse_params(sys.argv, 'kid')
pssh=o11.parse_params(sys.argv, 'pssh')
challenge=o11.parse_params(sys.argv, 'challenge')
licenseurl=o11.parse_params(sys.argv, 'licenseurl')
licenseparams=o11.parse_params(sys.argv, 'licenseparams')

heartbeaturl=o11.parse_params(sys.argv, 'heartbeaturl')
heartbeatparams=o11.parse_params(sys.argv, 'heartbeatparams')

o11Session = o11.session(bind=bind, proxy=proxy, worker=worker)


req = o11Session.get_session()
if doh != "":
o11.dns(doh)
elif dns != "":
o11.dns(dns)

if action == "subtitles":
if srclang == "":
srclang = "auto"
output = {
"Text": GoogleTranslator(source=srclang,
target=dstlang).translate(text)
}
print(json.dumps(output))
sys.exit(1)

if challenge == "cert":
challenge = "CAQ="
authFile = '/example_' + user + '.tokens'

headers = {}

def pair():
print("pairing...", flush=True)

response = req.get()
try:
code = response.json()['code']
print("Please go to ... and enter code " + code, flush=True)
except:
print(response.text)
sys.exit()

while True:
response = req.get()
try:
response.json()['token']
json.dump(response.json(),
open(os.path.abspath(os.path.dirname(__file__)) + authFile, 'w'))
print("pairing done", flush=True)
break
except:
pass

time.sleep(1)

def login():
print("logging in...", file=sys.stderr)

# Get token
response = req.get()
try:
response.json()['token']
json.dump(response.json(), open(os.path.abspath(os.path.dirname(__file__))
+ authFile, 'w'))
except:
output = {
"ErrorCode": 401,
"ErrorMessage": "auth failed"
}
print(json.dumps(output))
sys.exit(0)

print("logged in successfully", file=sys.stderr)

def do_action():

if action == "pair":
pair()
sys.exit()
elif action == "login":
login()
sys.exit()

try:
auth = json.load(open(os.path.abspath(os.path.dirname(__file__)) +
authFile))
except:
return "error"

if action == "channels":
output = {}
output['Channels'] = []

response = req.get()
try:
for chan in response.json():
channel = {}
channel['Name'] = chan['Name']
channel['Mode'] = "live"
channel['SessionManifest'] = True
channel['ManifestScript'] = 'id=' + str(chan['Id'])
channel['CdmType'] = "widevine"
channel['UseCdm'] = True
channel['Cdm'] = 'id=' + str(chan['Id'])
channel['Video'] = 'best'
channel['OnDemand'] = True
channel['SpeedUp'] = True
channel['Info'] = "unsubscribed"
output['Channels'].append(channel)

print(json.dumps(output, indent=2))
except:
print(response.text, file=sys.stderr)
return "error"

elif action == "vod":


output = {}
output['Vod'] = []

response = req.get()
try:
for item in response.json():
vod = {}
vod['Name'] = item['Name']
vod['Category'] = ''
vod['Description'] = ''
vod['Mode'] = "vod"
vod['SessionManifest'] = True
vod['ManifestScript'] = 'id=' + str(item['Id'])
vod['CdmType'] = "widevine"
vod['UseCdm'] = True
vod['Cdm'] = 'id=' + str(item['Id'])
vod['Video'] = 'best'
output['Vod'].append(vod)

print(json.dumps(output, indent=2))
except:
print(response.text, file=sys.stderr)
return "error"

elif action == "events":


output = {}
output['Events'] = []

response = req.get()
try:
for ev in response.json():
event = {}
event['Name'] = ev['Name']
event['Mode'] = "live"
event['SessionManifest'] = True
event['ManifestScript'] = 'id=' + str(ev['Id'])
event['CdmType'] = "widevine"
event['UseCdm'] = True
event['Cdm'] = 'id=' + str(ev['Id'])
event['Video'] = 'best'
event['Autostart'] = True
date_object = datetime.datetime.strptime(ev['startDate'], "%Y-%m-
%dT%H:%M:%S.000Z")
date_utc = date_object.replace(tzinfo=pytz.UTC)
event['Start'] = int(date_utc.timestamp())
date_object = datetime.datetime.strptime(ev['endDate'], "%Y-%m-%dT
%H:%M:%S.000Z")
date_utc = date_object.replace(tzinfo=pytz.UTC)
event['End'] = int(date_utc.timestamp())

output['Events'].append(event)

print(json.dumps(output, indent=2))
except:
print(response.text, file=sys.stderr)
return "error"

elif action == "heartbeat":


output = {
"ManifestUrl": response.json()['mpdUrl'],
"Headers": {
"Manifest": {
'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X
10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36',
},
"Media": {
'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X
10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36',
}
},
"Heartbeat": {
"Url": '',
"Params": '',
"PeriodMs": 5*60*1000
},
}
print(json.dumps(output))

elif action == "manifest":


try:
output = {
"Cdn": [],
"ManifestUrl": response.json()['mpdUrl'],
"Headers": {
"Manifest": {
'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X
10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36',
},
"Media": {
'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X
10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36',
}
},
"Heartbeat": {
"Url": '',
"Params": '',
"PeriodMs": 5*60*1000
},
"License": {
"Url": '',
"Params": ''
}
}
for cdn in [ '' ]:
response = req.get("", headers=headers)
output['Cdn'].append({ "Name": cdn, "ManifestUrl": mpd_url })
output['ManifestUrl'] = mpd_url
print(json.dumps(output))
except:
print(response.text, file=sys.stderr)
return "error"

elif action == "cdm" and cdm == "internal":


# return license from challenge
response = req.post(licenseurl, headers=headers,
data=base64.b64decode(challenge))
response_b64 = str(base64.b64encode(response.content), 'ascii')
if response_b64.startswith('CA'):
output = {
"CdmAnswer": response_b64
}
print(json.dumps(output))
else:
print(response.text, file=sys.stderr)
return "error"

elif action == "cdm" and cdm == "external":


output = {
"CdmAnswer": keys
}
print(json.dumps(output))

elif action == "pssh":


output = {
"ProcessedPssh": pssh
}
print(json.dumps(output))

else:
print("invalid action: " + action)

if do_action() == "error":
login()
if do_action() == "error":
sys.exit(1)

You might also like