Updated for all the things

Chris Pomeroy

9 days ago 2bca657babf8bec3f9cee673c92eb21f68a68a68

Updated for all the things

 Dockerfile

New file
@@ -0,0 +1,24 @@
FROM python:3.11-slim


RUN export DEBIAN_FRONTEND=noninteractive \
    && apt-get update \
    && apt-get install -y \
        # lsb-release \
        ffmpeg \
        vim-common \
        curl \
    && apt-get install -y libssl-dev libffi-dev \
    && useradd -U -s /bin/bash -u 506 -m default \
    && apt-get autoremove -y \
    && apt-get clean -y \
    && mkdir /audiobooks/ \
    && chown default /audiobooks \
    && rm -rf /var/lib/apt/lists/*

COPY ./aaxConvert.py /home/default/
COPY ./getMissingFromAudible.py /home/default/
COPY ./audible_auth* /home/default/

RUN pip install httpx audible 
USER default

 fix-asin.py

New file
@@ -0,0 +1,78 @@
import httpx
import audible
import json
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_')
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_local_items(limit, library_id):
    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=20.0).json()
    return items['results']


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 find_asin_for_book(title, a_books):
    search = [book for book in a_books if book['title'] in title]
    if search:
        return search[0]['asin']
    else:
        print(f"Title not found {title}")


def update_bookshelf(book_id, book_title, book_asin):
    print(f"Will update {book_title}/{book_id} with asin: {book_asin}")
    data = {"metadata": {"asin": book_asin}}
    with httpx.Client(base_url="https://bookshelf.darkurthe.net/api/",
                      headers=_header) as c:
        update = c.patch(f'items/{book_id}/media', json=data)
        if update.status_code == 200:
            print(f"Updated {book_title}")
        else:
            print(f"There was a problem updating {book_title}:  Status code: {update.status_code}")
    return


audible_books = get_audible_books()
books = get_local_items(0, '0507d3fe-db57-4e6f-bb18-a0c09cd6f196')


for x in books:
    if x['media']['metadata']['asin'] is None:
        # print({x['media']['metadata']['title']})
        asin = find_asin_for_book(x['media']['metadata']['title'], audible_books)
        x_title = x['media']['metadata']['title']
        x_id = x['id']
        if asin:
            update_bookshelf(x_id, x_title, asin)

 getMissingFromAudible.py

New file
@@ -0,0 +1,190 @@
#!/usr/local/bin/python3
import httpx
import aaxConvert
import argparse
import audible
import json
import time
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_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
    '''
    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_books(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)
    print(ab)
    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 main():
    args = get_args()
    asin_manual = args.asin
    audible_books = get_audible_books()
    if asin_manual is not None:
        download_books(asin_manual)
    else:
        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:
                        print(f"{book['title']}, {book['asin']}")
                        download_books(book['asin'])
                        time.sleep(20)


if __name__ == "__main__":
    main()