#!/usr/local/bin/python3
|
import httpx
|
import aaxConvert
|
import argparse
|
import audible
|
import json
|
import time
|
import os
|
from audible.aescipher import decrypt_voucher_from_licenserequest
|
from audible.activation_bytes import (
|
extract_activation_bytes,
|
fetch_activation_sign_auth
|
)
|
|
_audible_auth = audible.Authenticator.from_file('./audible_auth_Jeremy')
|
ab = fetch_activation_sign_auth(auth=_audible_auth)
|
ab = extract_activation_bytes(ab)
|
_actbytes = ab
|
_bearer_token = 'eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJ1c2VySWQiOiJjOWZlOGUyOS0xZjM3LTQ1OWEtOWJiOS02NDg3ZjNjNDNiNGUiLCJ1c2VybmFtZSI6InJvb3QiLCJpYXQiOjE3NTAwNDA1MDl9.kDueJYRlXna-IRZu5jDjTUjwGYp0y01P9TdNw4d5PeE' # noqa E501
|
_header = {
|
"Authorization": f"Bearer {_bearer_token}"
|
}
|
_metadata = {}
|
|
|
def get_auth_files():
|
'''
|
Gets the auth files for audible and activation bytes
|
|
Returns:
|
audible auth, activation bytes
|
'''
|
|
with open('./audible_auth_', 'r') as f:
|
auth_data = json.load(f)
|
audible_auth = audible.Authenticator.from_dict(auth_data)
|
act_bytes = fetch_activation_sign_auth(auth=audible_auth)
|
act_bytes = extract_activation_bytes(act_bytes)
|
return audible_auth, act_bytes
|
|
|
def get_args():
|
'''
|
Parses the args for runtime
|
|
Returns:
|
args class type
|
'''
|
parser = argparse.ArgumentParser(exit_on_error=False)
|
parser.add_argument("-a", "--asin",
|
help="ASIN of the book to download and convert",
|
default=None, required=False)
|
return parser.parse_args()
|
|
|
def get_library_ids() -> list:
|
'''
|
Returns a list of library id's
|
|
Args:
|
url (str): url to query
|
|
Returns:
|
list: A list of libraries avaliable to query.
|
'''
|
libs = []
|
with httpx.Client(base_url="https://bookshelf.darkurthe.net/api/",
|
headers=_header) as c:
|
libraries = c.get('libraries').json()
|
for library in libraries['libraries']:
|
libs.append(library["id"])
|
c.close()
|
return libs
|
|
|
def get_bookshelf_books(library_id, limit=0):
|
with httpx.Client(base_url="https://bookshelf.darkurthe.net/api/",
|
headers=_header) as c:
|
items = c.get(f'libraries/{library_id}/items?limit={limit}',
|
timeout=30.0).json()
|
return items['results']
|
|
|
def search_local_books(a_book, l_books) -> list:
|
search_title = [book for book in l_books if a_book['title'] in
|
book['media']['metadata']['title']]
|
search_asin = [
|
book for book in l_books if a_book['asin'] == book['media']['metadata']['asin']
|
]
|
if search_title or search_asin:
|
return False
|
else:
|
return True
|
|
|
def get_audible_books() -> list:
|
'''
|
Downloads a list of audiobooks from audible.
|
|
Returns:
|
list: A list of books and details in dictionary format
|
'''
|
with audible.Client(auth=_audible_auth) as client:
|
library = client.get(
|
"1.0/library/",
|
num_results=1000,
|
response_groups="product_desc, product_attrs",
|
sort_by="-PurchaseDate"
|
)
|
books = library['items']
|
return books
|
|
|
def get_license_request(asin: str) -> str:
|
'''
|
Gets the url for the book to download
|
|
Args:
|
asin (str): The audiobook asin("Amazon Standard Identification Number")
|
|
Returns:
|
dict: The dict of the license request
|
'''
|
body = {
|
"supported_drm_types": [
|
"Mpeg",
|
"Adrm"
|
],
|
"quality": "High",
|
"consumption_type": "Download"
|
}
|
with audible.Client(auth=_audible_auth) as client:
|
content_url = client.post(
|
f"1.0/content/{asin}/licenserequest",
|
body=body
|
)
|
return content_url
|
|
|
def get_content_url(licenserequest: dict) -> str:
|
'''
|
Gets the url out of the license request.
|
|
Arg:
|
license request (dict):
|
|
Ret:
|
str: The url to download the file from
|
'''
|
# print(licenserequest)
|
if 'Denied' in licenserequest{'content_license']['status'}:
|
raise Exception("License request was denied. For ")
|
return licenserequest['content_license']['content_metadata'][
|
'content_url']['offline_url']
|
|
|
def get_content(asin: str, content_url: str) -> None:
|
'''Downloads the content from audible
|
|
Args:
|
asin (str): The audiobook asin("Amazon Standard Identification Number")
|
content_url (str): The url to download the book from
|
'''
|
headers = {"User-Agent": "Audible/671 CFNetwork/1240.0.4 Darwin/20.6.0"}
|
dl = httpx.get(content_url, timeout=30, headers=headers)
|
with open(f"/tmp/{asin}.aax", 'wb') as f:
|
f.write(dl.content)
|
|
|
def download_book(asin: str) -> None:
|
'''
|
Download the book and hand it off to the converter
|
|
Args:
|
title (str): The title of the book to be downloaded
|
asin (str): The amazon number of the book being downloaded
|
'''
|
lr = get_license_request(asin)
|
# see if we already downloaded the file
|
if os.path.exists(f"/tmp/{asin}.aax"):
|
print(f"{asin} already downloaded, skipping download.")
|
else:
|
get_content(asin, get_content_url(lr))
|
decrypted_voucher = decrypt_voucher_from_licenserequest(_audible_auth, lr)
|
key_iv = {
|
'key': decrypted_voucher['key'],
|
'iv': decrypted_voucher['iv']
|
}
|
print(f"{asin} ready for conversion.")
|
aaxConvert.convert_aax(f"/tmp/{asin}.aax", _actbytes, key_iv)
|
|
|
def list_missing_books():
|
audible_books = get_audible_books()
|
missing_books = []
|
library_ids = get_library_ids()
|
for id in library_ids:
|
bookshelf_books = get_bookshelf_books(id)
|
for book in audible_books:
|
if book['content_type'] == 'Product':
|
search = search_local_books(book, bookshelf_books)
|
if search:
|
missing_books.append(f"{book['title']}: {book['asin']}")
|
return missing_books
|
|
|
def main():
|
args = get_args()
|
asin_manual = args.asin
|
if asin_manual is not None:
|
download_book(asin_manual)
|
else:
|
missing_books = list_missing_books()
|
print(missing_books)
|
answer = input("Would you like to download the missing books? (y/n)\n")
|
if answer.lower() == 'y':
|
for book in missing_books:
|
download_book(book.split(": ")[1])
|
time.sleep(20)
|
|
|
if __name__ == "__main__":
|
main()
|
