quinten/npo
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fixed shit

Browse Source
This commit is contained in:
Quinten0508 2024-04-22 17:27:39 +02:00
parent 93f5b14420
commit 47a81dca29
7 changed files with 42 additions and 1078 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
.gitignore vendored
View File

@ -2,7 +2,7 @@
Thumbs.db
desktop.ini
# OS X
# MacOS
.DS_Store
.Spotlight-V100
.Trashes
@ -16,5 +16,9 @@ desktop.ini
N_m3u8DL-RE*
mp4decrypt*
env/
avondshow.txt
urls.txt
urls/
cdm/
# misc
test.py

BIN
cdm/__pycache__/wks.cpython-311.pyc
View File

Binary file not shown.

842
cdm/wks.py
View File

File diff suppressed because one or more lines are too long

52
npo all-in-one.py
View File

@ -1,14 +1,8 @@
# Pre-requisites:
# * N_m3u8DL-RE and mp4decrypt in current directory
# * ffmpeg in PATH
# * pip install -r requirements.txt
# PIP Requirements:
# * protobuf
# * bs4
# * xmltodict
# * browser_cookie3
# * requests
# * pycryptodomex
import argparse
import requests
@ -16,14 +10,17 @@ import subprocess
import os
from bs4 import BeautifulSoup
import json
import platform # check for windows OS
import shutil # check for ffmpeg in PATH, part of python std
import browser_cookie3 # cookies for premium accs
import platform # check for windows OS
import shutil # check for ffmpeg in PATH
import browser_cookie3 # cookies for premium accs
from fake_useragent import UserAgent # sets useragent
import concurrent.futures # concurrent downloads when using a -file
import time # just for lols
from cdm.wks import WvDecrypt, device_android_generic, PsshExtractor, KeyExtractor
import concurrent.futures
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.3',
'User-Agent': UserAgent(platforms='pc', min_version=120.0).random,
'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,*/*;q=0.8',
'Accept-Language': 'en-US,en;q=0.5',
'Cache-Control': 'no-cache',
@ -71,13 +68,15 @@ def find_cookies():
if not userinput or userinput.lower() != 'y':
return
cookies = browser_cookie3.load(domain_name='npo.nl')
# browser_cookie3.load() should use ALL browsers' cookies. If this doesn't work, replace browser_cookie3.load with browser_cookie3.<browser>.
# See notes at the end of this script for possible options. Example: browser_cookie3.chrome or browser_cookie3.librewolf.
cookies = browser_cookie3.librewolf(domain_name='npo.nl')
return cookies
def find_targetId(url):
# Get full HTML and extract productId and episode number
# "future proof"
# "future proof" :)
response_targetId = requests.get(url)
content = response_targetId.content
@ -97,7 +96,6 @@ def find_targetId(url):
script_content = script_tag.contents[0]
else:
print("Script tag not found.")
print("Hint: Use the -token <token> argument to supply your own.")
def search(data, target_slug):
if isinstance(data, list):
@ -123,10 +121,20 @@ def find_targetId(url):
def find_CSRF(targetId, plus_cookie):
response_CSRF = requests.get('https://npo.nl/start/api/auth/session', headers=headers, cookies=plus_cookie)
response_cookies = response_CSRF.cookies.get_dict()
csrf = response_cookies["__Host-next-auth.csrf-token"]
if plus_cookie:
csrf_label = "__Secure-next-auth.session-token"
else:
csrf_label = "__Host-next-auth.csrf-token"
# LOGGED IN: __Secure-next-auth.session-token
# LOGGED OUT: __Host-next-auth.csrf-token
# sigh.
csrf = response_cookies[csrf_label]
csrf_cookies = {
'__Host-next-auth.csrf-token': csrf,
csrf_label: csrf,
'__Secure-next-auth.callback-url': 'https%3A%2F%2Fnpo.nl',
}
@ -137,11 +145,13 @@ def find_CSRF(targetId, plus_cookie):
'productId': targetId,
}
response_token = requests.post('https://npo.nl/start/api/domain/player-token', cookies=plus_cookie, headers=headers, json=json_productId)
response_token = requests.get('https://npo.nl/start/api/domain/player-token', cookies=plus_cookie, headers=headers, params=json_productId)
token = response_token.json()["token"]
return token
def find_MPD(token, url, plus_cookie):
headers['Authorization'] = token
@ -209,13 +219,14 @@ def check_prereq():
def create_filename(url, programKey):
# season title
# 1 2 3 4 5 6 7 8 (optional)
# create filename based on input URL: https://npo.nl/start/serie /wie-is-de-mol /seizoen-24 /wie-is-de-mol_56 /afspelen
# https://npo.nl/start/serie /de-avondshow-met-arjen-lubach /seizoen-8_1 /de-avondshow-met-arjen-lubach_93 /afspelen
# https://npo.nl/start/serie /taarten-van-abel /seizoen-17 /joto /afspelen
url_split = url.split("/")
title = url_split[7].split("_")[0]
season = url_split[6].split("_")[0]
# filename_enc = title + "_" + season + "_ep-" + programKey + "_encrypted"
filename_enc = title + "_" + season + "_ep-" + programKey + "_encrypted"
filename = filename_enc.replace("_encrypted", "")
return filename_enc, filename
@ -285,6 +296,7 @@ def execute(url, plus_cookie, process_no):
plus_cookie = find_cookies()
start_time = time.time()
max_workers = min(os.cpu_count(), len(urls))
with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
@ -296,6 +308,8 @@ with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
completed_videos += 1
print(f"{completed_videos}/{len(urls)} video{'s'[:len(urls) != 1]} completed")
execution_time = time.time() - start_time
print("Execution time in seconds: " + str(execution_time))
#########
# NOTES #

135
npo.py
View File

@ -1,135 +0,0 @@
import argparse
import requests
from bs4 import BeautifulSoup
import json
from cdm.wks import WvDecrypt, device_android_generic, PsshExtractor, KeyExtractor
# Parse URL input
parser = argparse.ArgumentParser(description='PYWKS-NPO')
parser.add_argument('-url', dest='url', required=True, help='NPO Video URL')
args = parser.parse_args()
# Get HTML and extract productId
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:109.0) Gecko/20100101 Firefox/119.0',
'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,*/*;q=0.8',
'Accept-Language': 'en-US,en;q=0.5',
'Connection': 'keep-alive',
'Upgrade-Insecure-Requests': '1',
'Sec-Fetch-Dest': 'document',
'Sec-Fetch-Mode': 'navigate',
'Sec-Fetch-Site': 'same-site',
'Pragma': 'no-cache',
'Cache-Control': 'no-cache',
}
response_targetId = requests.get(args.url, headers=headers)
content = response_targetId.content
try:
url_split = args.url.split("/")
target_slug = url_split[7]
except:
print("URL invalid.")
print("URL format: https://npo.nl/start/serie/wie-is-de-mol/seizoen-24/wie-is-de-mol_56/afspelen")
print(f"Your URL: {args.url}")
exit()
soup = BeautifulSoup(content, 'html.parser')
script_tag = soup.find('script', {'id': '__NEXT_DATA__'})
if script_tag:
script_content = script_tag.contents[0]
else:
print("Script tag not found.")
def search(data, target_slug):
if isinstance(data, list):
for item in data:
result = search(item, target_slug)
if result:
return result
elif isinstance(data, dict):
for key, value in data.items():
if key == "slug" and value == target_slug:
return data.get("productId")
else:
result = search(value, target_slug)
if result:
return result
return None
data_dict = json.loads(script_content)
target_product_id = search(data_dict, target_slug)
# Get CSRF token
response_CSRF = requests.get('https://npo.nl/start/api/auth/session', headers=headers)
response_cookies = response_CSRF.cookies.get_dict()
csrf = response_cookies["__Host-next-auth.csrf-token"]
# Get player token
cookies = {
'__Host-next-auth.csrf-token': csrf,
'__Secure-next-auth.callback-url': 'https%3A%2F%2Fnpo.nl',
}
json_productId = {
'productId': target_product_id,
}
response_token = requests.post('https://npo.nl/start/api/domain/player-token', cookies=cookies, headers=headers, json=json_productId)
token = response_token.json()["token"]
# Get MPD URL
headers['authorization'] = token
json_auth = {
'profileName': 'dash',
'drmType': 'widevine',
'referrerUrl': args.url,
}
response = requests.post('https://prod.npoplayer.nl/stream-link', headers=headers, json=json_auth)
response_data = response.json()
stream_data = response_data.get('stream', {})
if stream_data.get('streamURL'):
print('MPD URL:', stream_data.get('streamURL'))
else:
print("NO MPD URL - BAD TOKEN")
exit()
# Get PSSH
mpd_url = stream_data.get('streamURL')
license_url = "https://npo-drm-gateway.samgcloud.nepworldwide.nl/authentication"
response = requests.get(mpd_url, headers=headers)
pssh_extractor = PsshExtractor(response.text)
pssh_value = pssh_extractor.extract_pssh()
print("PSSH:", pssh_value)
headers_license = {
'x-custom-data': stream_data.get('drmToken'),
'origin': 'https://start-player.npo.nl',
'referer': 'https://start-player.npo.nl/',
}
# Get Key
cert_b64 = None
key_extractor = KeyExtractor(pssh_value, cert_b64, license_url, headers_license)
keys = key_extractor.get_keys()
wvdecrypt = WvDecrypt(init_data_b64=pssh_value, cert_data_b64=cert_b64, device=device_android_generic)
raw_challenge = wvdecrypt.get_challenge()
data = raw_challenge
for key in keys:
if isinstance(key, list):
if key:
for key_str in key:
print(f"KEY: {key_str}")

78
npo_new.py
View File

@ -1,78 +0,0 @@
import argparse
import requests
from bs4 import BeautifulSoup
import json
import logging
import coloredlogs
from cdm.wks import WvDecrypt, device_android_generic, PsshExtractor, KeyExtractor
# def search(data, target_slug):
# if isinstance(data, list):
# for item in data:
# result = search(item, target_slug)
# if result:
# return result
# elif isinstance(data, dict):
# for key, value in data.items():
# if key == "slug" and value == target_slug:
# return data.get("productId")
# else:
# result = search(value, target_slug)
# if result:
# return result
# return None
parser = argparse.ArgumentParser(description='PYWKS-NPO')
parser.add_argument('-url', dest='url', required=True, help='NPO Video URL')
parser.add_argument("-logger", default="INFO", choices=["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"], help="Logger level")
args = parser.parse_args()
LOG_FORMAT = "{asctime} [{levelname[0]}] {name} : {message}"
LOG_DATE_FORMAT = "%Y-%m-%d %H:%M:%S"
LOG_STYLE = "{"
coloredlogs.install(level=args.logger, fmt=LOG_FORMAT, datefmt=LOG_DATE_FORMAT, style=LOG_STYLE)
logger = logging.getLogger("NPO")
response_target_id = requests.get(args.url)
content = response_target_id.content
try:
target_slug = args.url.split("/")[7]
except IndexError:
logger.error("Invalid URL format. Example: https://npo.nl/start/serie/wie-is-de-mol/seizoen-24/wie-is-de-mol_56/afspelen")
exit()
soup = BeautifulSoup(content, 'html.parser')
script_tag = soup.find('script', {'id': '__NEXT_DATA__'})
script_content = script_tag.contents[0] if script_tag else logger.error("Script tag not found.")
data_dict = json.loads(script_content)
target_product_id = search(data_dict, target_slug)
if not target_product_id:
logger.error("Failed to retrieve target product ID.")
exit()
response_csrf = requests.get('https://npo.nl/start/api/auth/session')
cookies = {'__Host-next-auth.csrf-token': response_csrf.cookies.get_dict()["__Host-next-auth.csrf-token"],'__Secure-next-auth.callback-url': 'https://npo.nl'}
json_product_id = {'productId': target_product_id}
response_token = requests.post('https://npo.nl/start/api/domain/player-token', cookies=cookies, json=json_product_id)
headers = {'authorization': response_token.json()["token"]}
json_auth = {'profileName': 'dash', 'drmType': 'widevine', 'referrerUrl': args.url}
response = requests.post('https://prod.npoplayer.nl/stream-link', headers=headers, json=json_auth)
stream_data = response.json().get('stream', {})
if not stream_data.get('streamURL'):
logger.error("Failed to retrieve MPD URL. Invalid or expired authentication token.")
exit()
mpd_url = stream_data.get('streamURL')
logger.info(f"MPD URL: {mpd_url}")
license_url = "https://npo-drm-gateway.samgcloud.nepworldwide.nl/authentication"
response = requests.get(mpd_url, headers=headers)
pssh_extractor = PsshExtractor(response.text)
pssh_value = pssh_extractor.extract_pssh()
logger.info(f"PSSH: {pssh_value}")
headers_license = {'x-custom-data': stream_data.get('drmToken'),'origin': 'https://start-player.npo.nl','referer': 'https://start-player.npo.nl/'}
cert_b64 = None
key_extractor = KeyExtractor(pssh_value, cert_b64, license_url, headers_license)
keys = key_extractor.get_keys()
wvdecrypt = WvDecrypt(init_data_b64=pssh_value, cert_data_b64=cert_b64, device=device_android_generic)
raw_challenge = wvdecrypt.get_challenge()
for key in keys:
if isinstance(key, list) and key:
for key_str in key:
logger.info(f"\u251C KEY: {key_str}")

1
requirements.txt
View File

@ -4,3 +4,4 @@ xmltodict
browser_cookie3
requests
pycryptodomex
fake-useragent