API tools faq
paste
Guest User

cbc.py for CBC

a guest
May 20th, 2023
68
0
13 days
Add comment
Not a member of Pastebin yet? Sign Up, it unlocks many cool features!
Python 17.97 KB | Source Code | 0 0
raw download clone embed print report
  1. """Module for general CBC stuff"""
  2. from uuid import uuid4
  3. from base64 import b64encode, b64decode
  4.  
  5. import json
  6. import urllib
  7. import urlparse
  8. # import http.client as http_client
  9. from xml.dom.minidom import *
  10. import xml.etree.ElementTree as ET
  11.  
  12. import requests
  13.  
  14. from .utils import save_cookies, loadCookies, saveAuthorization, log
  15.  
  16. # http_client.HTTPConnection.debuglevel = 1
  17.  
  18. from resources.lib.utils import loadAuthorization
  19.  
  20. CALLSIGN = 'cbc$callSign'
  21. API_KEY = '3f4beddd-2061-49b0-ae80-6f1f2ed65b37'
  22. SCOPES = 'openid '\
  23.         'offline_access '\
  24.         'https://rcmnb2cprod.onmicrosoft.com/84593b65-0ef6-4a72-891c-d351ddd50aab/email '\
  25.         'https://rcmnb2cprod.onmicrosoft.com/84593b65-0ef6-4a72-891c-d351ddd50aab/id.account.create '\
  26.         'https://rcmnb2cprod.onmicrosoft.com/84593b65-0ef6-4a72-891c-d351ddd50aab/id.account.delete '\
  27.         'https://rcmnb2cprod.onmicrosoft.com/84593b65-0ef6-4a72-891c-d351ddd50aab/id.account.info '\
  28.         'https://rcmnb2cprod.onmicrosoft.com/84593b65-0ef6-4a72-891c-d351ddd50aab/id.account.modify '\
  29.         'https://rcmnb2cprod.onmicrosoft.com/84593b65-0ef6-4a72-891c-d351ddd50aab/id.account.reset-password '\
  30.         'https://rcmnb2cprod.onmicrosoft.com/84593b65-0ef6-4a72-891c-d351ddd50aab/id.account.send-confirmation-email '\
  31.         'https://rcmnb2cprod.onmicrosoft.com/84593b65-0ef6-4a72-891c-d351ddd50aab/id.write '\
  32.         'https://rcmnb2cprod.onmicrosoft.com/84593b65-0ef6-4a72-891c-d351ddd50aab/media-drmt '\
  33.         'https://rcmnb2cprod.onmicrosoft.com/84593b65-0ef6-4a72-891c-d351ddd50aab/media-meta '\
  34.         'https://rcmnb2cprod.onmicrosoft.com/84593b65-0ef6-4a72-891c-d351ddd50aab/media-validation '\
  35.         'https://rcmnb2cprod.onmicrosoft.com/84593b65-0ef6-4a72-891c-d351ddd50aab/media-validation.read '\
  36.         'https://rcmnb2cprod.onmicrosoft.com/84593b65-0ef6-4a72-891c-d351ddd50aab/metrik '\
  37.         'https://rcmnb2cprod.onmicrosoft.com/84593b65-0ef6-4a72-891c-d351ddd50aab/oidc4ropc '\
  38.         'https://rcmnb2cprod.onmicrosoft.com/84593b65-0ef6-4a72-891c-d351ddd50aab/ott-profiling '\
  39.         'https://rcmnb2cprod.onmicrosoft.com/84593b65-0ef6-4a72-891c-d351ddd50aab/ott-subscription '\
  40.         'https://rcmnb2cprod.onmicrosoft.com/84593b65-0ef6-4a72-891c-d351ddd50aab/profile '\
  41.         'https://rcmnb2cprod.onmicrosoft.com/84593b65-0ef6-4a72-891c-d351ddd50aab/subscriptions.validate '\
  42.         'https://rcmnb2cprod.onmicrosoft.com/84593b65-0ef6-4a72-891c-d351ddd50aab/subscriptions.write '\
  43.         'https://rcmnb2cprod.onmicrosoft.com/84593b65-0ef6-4a72-891c-d351ddd50aab/toutv '\
  44.         'https://rcmnb2cprod.onmicrosoft.com/84593b65-0ef6-4a72-891c-d351ddd50aab/toutv-presentation '\
  45.         'https://rcmnb2cprod.onmicrosoft.com/84593b65-0ef6-4a72-891c-d351ddd50aab/toutv-profiling '\
  46.         'https://rcmnb2cprod.onmicrosoft.com/84593b65-0ef6-4a72-891c-d351ddd50aab/testapiwithjwtendpoint.admin '\
  47.         'https://rcmnb2cprod.onmicrosoft.com/84593b65-0ef6-4a72-891c-d351ddd50aab/id.account.info'
  48. AUTHORIZE_LOGIN = 'https://login.cbc.radio-canada.ca/bef1b538-1950-4283-9b27-b096cbc18070/B2C_1A_ExternalClient_FrontEnd_Login_CBC/oauth2/v2.0/authorize'
  49. SELF_ASSERTED_LOGIN = 'https://login.cbc.radio-canada.ca/bef1b538-1950-4283-9b27-b096cbc18070/B2C_1A_ExternalClient_FrontEnd_Login_CBC/SelfAsserted'
  50. CONFIRM_LOGIN = 'https://login.cbc.radio-canada.ca/bef1b538-1950-4283-9b27-b096cbc18070/B2C_1A_ExternalClient_FrontEnd_Login_CBC/api/SelfAsserted/confirmed'
  51. SIGNIN_LOGIN = 'https://login.cbc.radio-canada.ca/bef1b538-1950-4283-9b27-b096cbc18070/B2C_1A_ExternalClient_FrontEnd_Login_CBC/api/CombinedSigninAndSignup/confirmed'
  52. RADIUS_LOGIN_FMT = 'https://api.loginradius.com/identity/v2/auth/login?{}'
  53. RADIUS_JWT_FMT = 'https://cloud-api.loginradius.com/sso/jwt/api/token?{}'
  54. TOKEN_URL = 'https://services.radio-canada.ca/ott/cbc-api/v2/token'
  55. PROFILE_URL = 'https://services.radio-canada.ca/ott/subscription/v2/gem/subscriber/profile'
  56.  
  57.  
  58. LAYOUT_MAP = {
  59.     'featured': 'https://services.radio-canada.ca/ott/cbc-api/v2/home',
  60.     'shows': 'https://services.radio-canada.ca/ott/cbc-api/v2/hubs/shows',
  61.     'documentaries': 'https://services.radio-canada.ca/ott/cbc-api/v2/hubs/documentaries',
  62.     'kids': 'https://services.radio-canada.ca/ott/cbc-api/v2/hubs/kids'
  63. }
  64. SHOW_BY_ID = 'https://services.radio-canada.ca/ott/cbc-api/v2/shows/{}'
  65. CATEGORY_BY_ID = 'https://services.radio-canada.ca/ott/cbc-api/v2/categories/{}'
  66. ASSET_BY_ID = 'https://services.radio-canada.ca/ott/cbc-api/v2/assets/{}'
  67. SEARCH_BY_NAME = 'https://services.radio-canada.ca/ott/cbc-api/v2/search'
  68.  
  69.  
  70. class CBC:
  71.     """Class for CBC stuff."""
  72.  
  73.     def __init__(self):
  74.         """Initialize the CBC class."""
  75.         # Create requests session object
  76.         self.session = requests.Session()
  77.         session_cookies = loadCookies()
  78.         if session_cookies is not None:
  79.             self.session.cookies = session_cookies
  80.  
  81.     @staticmethod
  82.     def azure_authorize_authorize(sess):
  83.         """
  84.        Make the first authorization call.
  85.        @param sess A requests session
  86.        """
  87.         nonce= str(uuid4())
  88.         guid = str(uuid4())
  89.         #state_str = f'{guid}|{{"action":"login","returnUrl":"/","fromSubscription":false}}'.encode()
  90.         #Convert using CLI f2format and temp file
  91.         state_str = '{}|{{"action":"login","returnUrl":"/","fromSubscription":false}}'.format(guid).encode()
  92.         state = b64encode(state_str).decode('ascii')
  93.         params = {
  94.             'client_id': 'fc05b0ee-3865-4400-a3cc-3da82c330c23',
  95.             'nonce': nonce,
  96.             'redirect_uri': 'https://gem.cbc.ca/auth-changed',
  97.             'scope': SCOPES,
  98.             'response_type': 'id_token token',
  99.             'response_mode': 'fragment',
  100.             'state': state,
  101.             'state_value': state,
  102.             'ui_locales': 'en',
  103.         }
  104.         resp = sess.get(AUTHORIZE_LOGIN, params=params)
  105.         if resp.status_code != 200:
  106.             log('Call to authorize fails', True)
  107.             return False
  108.  
  109.         if not 'x-ms-gateway-requestid' in resp.headers:
  110.             log('authorize authorize response had no x-ms-gateway-requestid header')
  111.             return False
  112.  
  113.         return resp.headers['x-ms-gateway-requestid']
  114.  
  115.     @staticmethod
  116.     def azure_authorize_self_asserted(sess, username, tx_arg, password = None):
  117.         """
  118.        Make the second authorization call.
  119.        @param sess The requests session
  120.        """
  121.         cookies = sess.cookies.get_dict()
  122.         headers = { 'x-csrf-token': cookies['x-ms-cpim-csrf'] }
  123.         params = { 'tx': tx_arg, 'p': 'B2C_1A_ExternalClient_FrontEnd_Login_CBC' }
  124.         data = { 'request_type': 'RESPONSE', 'email': username}
  125.         if password:
  126.             data['password'] = password
  127.  
  128.         resp = sess.post(SELF_ASSERTED_LOGIN, params=params, headers=headers, data=data)
  129.         if not resp.status_code == 200:
  130.             log('Call to SelfAsserted fails', True)
  131.             return False
  132.         return True
  133.  
  134.     @staticmethod
  135.     def azure_authorize_confirmed(sess, tx_arg):
  136.         """
  137.        Make the third authorization call.
  138.        @param sess The requests session
  139.        @param csrf The csrf token
  140.        @param tx_arg the tx parameter
  141.        """
  142.         cookies = sess.cookies.get_dict()
  143.         params = {
  144.             'tx': tx_arg,
  145.             'p': 'B2C_1A_ExternalClient_FrontEnd_Login_CBC',
  146.             'csrf_token': cookies['x-ms-cpim-csrf'],
  147.         }
  148.  
  149.         resp = sess.get(CONFIRM_LOGIN, params=params)
  150.         if resp.status_code != 200:
  151.             log('Call to authorize fails', True)
  152.             return False
  153.  
  154.         if not 'x-ms-gateway-requestid' in resp.headers:
  155.             log('authorize confirmed response had no x-ms-gateway-requestid header')
  156.             return False
  157.  
  158.         return resp.headers['x-ms-gateway-requestid']
  159.  
  160.     @staticmethod
  161.     def azure_authorize_sign_in(sess, tx_arg):
  162.         """
  163.        Make the third authorization call.
  164.        @param sess The requests session
  165.        @param csrf The csrf token
  166.        @param tx_arg the tx parameter
  167.        """
  168.         cookies = sess.cookies.get_dict()
  169.         params = {
  170.             'tx': tx_arg,
  171.             'p': 'B2C_1A_ExternalClient_FrontEnd_Login_CBC',
  172.             'csrf_token': cookies['x-ms-cpim-csrf'],
  173.             'rememberMe': 'true',
  174.         }
  175.  
  176.         resp = sess.get(SIGNIN_LOGIN, params=params, allow_redirects=False)
  177.         if resp.status_code != 302:
  178.             log('Call to authorize fails', True)
  179.             return None
  180.  
  181.         url = urlparse.urlparse(resp.headers['location'])
  182.         frags = urlparse.parse_qs(url.fragment)
  183.         access_token = frags['access_token'][0]
  184.         id_token = frags['id_token'][0]
  185.  
  186.         return (access_token, id_token)
  187.  
  188.     def azure_authorize(self, username=None, password=None, callback=None):
  189.         """
  190.        Perform multi-step authorization with CBC's azure authorization platform.
  191.        ** Azure Active Directory B2C **
  192.        """
  193.         sess = requests.Session()
  194.  
  195.         if callback:
  196.             callback(0)
  197.         gw_req_id = CBC.azure_authorize_authorize(sess)
  198.         if not gw_req_id:
  199.             log('Authorization "authorize" step failed', True)
  200.             return False
  201.  
  202.         if callback:
  203.             callback(20)
  204.         cookies = sess.cookies.get_dict()
  205.         if 'x-ms-cpim-csrf' not in cookies:
  206.             log('Unable to get csrt token for self asserted', True)
  207.             return False
  208.  
  209.         if 'x-ms-cpim-trans' not in cookies:
  210.             log('Unable to get transaction for self asserted', True)
  211.             return False
  212.  
  213.         trans = cookies['x-ms-cpim-trans']
  214.         trans = b64decode(trans).decode()
  215.         trans = json.loads(trans)
  216.         if not 'C_ID' in trans:
  217.             log('Unable to get C_ID from trans', True)
  218.             return False
  219.         tid = trans['C_ID']
  220.  
  221.         #tid_str = f'{{"TID":"{tid}"}}'.encode()
  222.         #Convert using CLI f2format and temp file
  223.         tid_str = '{{"TID":"{}"}}'.format(tid).encode()
  224.         b64_tid = b64encode(tid_str).decode('ascii')
  225.         b64_tid = b64_tid.rstrip('=')
  226.         #tx_arg = f'StateProperties={b64_tid}'
  227.         #Convert using CLI f2format and temp file
  228.         tx_arg = 'StateProperties={}'.format(b64_tid)
  229.         if callback:
  230.             callback(40)
  231.         if not CBC.azure_authorize_self_asserted(sess, username, tx_arg):
  232.             log('Authorization "SelfAsserted" step failed', True)
  233.             return False
  234.  
  235.         if callback:
  236.             callback(60)
  237.         gw_req_id = CBC.azure_authorize_confirmed(sess, tx_arg)
  238.         if not gw_req_id:
  239.             log('Authorization "confirmed" step failed', True)
  240.             return False
  241.  
  242.         if callback:
  243.             callback(80)
  244.         if not CBC.azure_authorize_self_asserted(sess, username, tx_arg, password):
  245.             log('Authorization "SelfAsserted" step failed', True)
  246.             return False
  247.         access_token, id_token = CBC.azure_authorize_sign_in(sess, tx_arg)
  248.         if not access_token or not id_token:
  249.             log('Authorization "confirmed" step failed', True)
  250.             return False
  251.  
  252.         if callback:
  253.             callback(90)
  254.         claims_token = self.get_claims_token(access_token)
  255.  
  256.         saveAuthorization({'token': access_token, 'claims': claims_token})
  257.         if callback:
  258.             callback(100)
  259.  
  260.         return True
  261.  
  262.     def get_claims_token(self, access_token):
  263.         """Get the claims token for tied to the access token."""
  264.         #headers = {'Authorization': f'Bearer {access_token}'}
  265.         #Convert using CLI f2format and temp file
  266.         headers = {'Authorization': 'Bearer {}'.format(access_token)}
  267.         params = {'device': 'web'}
  268.         req = self.session.get(PROFILE_URL, headers=headers, params=params)
  269.         if not req.status_code == 200:
  270.             log('{} returns status {}'.format(req.url, req.status_code), True)
  271.             return None
  272.         return json.loads(req.content)['claimsToken']
  273.  
  274.     def getImage(self, item):
  275.         """Get an image."""
  276.         # ignore 'cbc$liveImage' - the pix don't make sense after the first load
  277.         if 'defaultThumbnailUrl' in item:
  278.             return item['defaultThumbnailUrl']
  279.         if 'cbc$staticImage' in item:
  280.             return item['cbc$staticImage']
  281.         if 'cbc$featureImage' in item:
  282.             return item['cbc$featureImage']
  283.         return None
  284.  
  285.     @staticmethod
  286.     def get_callsign(item):
  287.         """Get the callsign for a channel."""
  288.         return item[CALLSIGN] if CALLSIGN in item else None
  289.  
  290.     @staticmethod
  291.     def get_labels(item):
  292.         """Get labels for a CBC item."""
  293.         labels = {
  294.             'studio': 'Canadian Broadcasting Corporation',
  295.             'country': 'Canada'
  296.         }
  297.         if 'cbc$callSign' in item:
  298.             labels['title'] = '{} {}'.format(item['cbc$callSign'].replace('\u2013', "-"), item['title'].replace('\u2013', "-"))
  299.         else:
  300.             labels['title'] = item['title']
  301.  
  302.         if 'cbc$show' in item:
  303.             labels['tvshowtitle'] = item['cbc$show']
  304.         elif 'clearleap:series' in item:
  305.             labels['tvshowtitle'] = item['clearleap:series']
  306.  
  307.         if 'description' in item:
  308.             labels['plot'] = item['description']
  309.             labels['plotoutline'] = item['description']
  310.  
  311.         if 'cbc$liveDisplayCategory' in item:
  312.             labels['genre'] = item['cbc$liveDisplayCategory']
  313.         elif 'media:keywords' in item:
  314.             labels['genre'] = item['media:keywords']
  315.  
  316.         if 'clearleap:season' in item:
  317.             labels['season'] = item['clearleap:season']
  318.  
  319.         if 'clearleap:episodeInSeason' in item:
  320.             labels['episode'] = item['clearleap:episodeInSeason']
  321.  
  322.         if 'media:rating' in item:
  323.             labels['mpaa'] = item['media:rating']
  324.  
  325.         if 'premiered' in item:
  326.             labels['premiered'] = item['premiered']
  327.  
  328.         if 'video' in item:
  329.             labels['mediatype'] = 'video'
  330.         elif 'cbc$audioVideo' in item:
  331.             if item['cbc$audioVideo'].lower() == 'video':
  332.                 labels['mediatype'] = 'video'
  333.  
  334.         return labels
  335.  
  336.  
  337.     def parseSmil(self, smil):
  338.         """Parse a SMIL file for the video."""
  339.         resp = self.session.get(smil)
  340.  
  341.         if not resp.status_code == 200:
  342.             log('ERROR: {} returns status of {}'.format(smil, r.status_code), True)
  343.             return None
  344.         save_cookies(self.session.cookies)
  345.  
  346.         dom = parseString(r.content)
  347.         seq = dom.getElementsByTagName('seq')[0]
  348.         video = seq.getElementsByTagName('video')[0]
  349.         src = video.attributes['src'].value
  350.         title = video.attributes['title'].value
  351.         abstract = video.attributes['abstract'].value
  352.         return src
  353.  
  354.     @staticmethod
  355.     def get_session():
  356.         """Get a requests session object with CBC cookies."""
  357.         sess = requests.Session()
  358.         cookies = loadCookies()
  359.         if cookies is not None:
  360.             self.session.cookies = cookies
  361.         return sess
  362.  
  363.     # """Module for the V2 Gem API."""
  364.     # import json
  365.     #
  366.     # import requests
  367.     #
  368.     # from resources.lib.utils import log
  369.     # from resources.lib.cbc import CBC
  370.     #
  371.     # Move LAYOUT_MAP to top of file
  372.     #
  373.  
  374.     # class GemV2:
  375.     #     """V2 Gem API class."""
  376.  
  377.     # @staticmethod   # DEBUG
  378.     def GemV2_get_layout(self, name):
  379.         """Get a Gem V2 layout by name."""
  380.         url = LAYOUT_MAP[name]
  381.         # resp = CBC.get_session().get(url)
  382.         resp = self.session.get(url)
  383.         return json.loads(resp.content)
  384.  
  385.     # @staticmethod   # DEBUG
  386.     def GemV2_get_show_layout_by_id(self, show_id):
  387.         """Get a Gem V2 show layout by ID."""
  388.         url = SHOW_BY_ID.format(show_id)
  389.         # resp = CBC.get_session().get(url)
  390.         resp = self.session.get(url)
  391.         return json.loads(resp.content)
  392.  
  393.     # @staticmethod   # DEBUG
  394.     def GemV2_get_asset_by_id(self, asset_id):
  395.         url = ASSET_BY_ID.format(asset_id)
  396.         # resp = CBC.get_session().get(url)
  397.         resp = self.session.get(url)
  398.         return json.loads(resp.content)
  399.  
  400.     # @staticmethod   # DEBUG
  401.     def GemV2_get_episode(self, url):
  402.         """Get a Gem V2 episode by URL."""
  403.         # resp = CBC.get_session().get(url)  # Python 3
  404.         # resp = self.session.get(url)       # Python 2
  405.         auth = loadAuthorization()
  406.  
  407.         # if we have no authorization, return none to for the UI to authorize
  408.         if auth is None:
  409.             return None
  410.  
  411.         headers = {}
  412.         if 'token' in auth:
  413.             headers['Authorization'] = 'Bearer {}'.format(auth['token'])
  414.  
  415.         if 'claims' in auth:
  416.             headers['x-claims-token'] = auth['claims']
  417.  
  418.         resp = requests.get(url, headers=headers)
  419.         return json.loads(resp.content)
  420.  
  421.     # @staticmethod   # DEBUG
  422.     def GemV2_get_category(self, category_id):
  423.         """Get a Gem V2 category by ID."""
  424.         url = CATEGORY_BY_ID.format(category_id)
  425.         # resp = CBC.get_session().get(url)
  426.         resp = self.session.get(url)
  427.         return json.loads(resp.content)
  428.  
  429.     @staticmethod
  430.     def GemV2_get_labels(show, episode):
  431.         """Get labels for a show."""
  432.         labels = {
  433.             'studio': 'Canadian Broadcasting Corporation',
  434.             'country': 'Canada',
  435.             'tvshowtitle': show['title'],
  436.             'title': episode['title'],
  437.             'originaltitle': episode['title'],
  438.             'plot': episode['description'],
  439.             'plotoutline': episode['description'],
  440.             'season': episode['season'],
  441.         }
  442.         if 'episode' in episode:
  443.             labels['episode'] = episode['episode']
  444.  
  445.         if 'duration' in episode:
  446.             labels['duration'] = episode['duration']
  447.  
  448.         return labels
  449.  
  450.     # @staticmethod   # DEBUG
  451.     def GemV2_search_by_term(self, term):
  452.         params = {'term': term}
  453.         # resp = CBC.get_session().get(SEARCH_BY_NAME, params=params)
  454.         resp = self.session.get(SEARCH_BY_NAME, params=params)
  455.         return json.loads(resp.content)
  456.  
Add Comment
Please, Sign In to add comment
Public Pastes
We use cookies for various purposes including analytics. By continuing to use Pastebin, you agree to our use of cookies as described in the Cookies Policy.  OK, I Understand
Not a member of Pastebin yet?
Sign Up, it unlocks many cool features!