AudibleSeriesChecker/main.py

import connectors
import argparse
import logging
import config

logging.basicConfig(
    filename="log",
    filemode="w",
    format="%(levelname)s - %(message)s",
)


class Book(dict):
    def __init__(self, asin=""):
        self.asin = asin
        self.title = ""
        self.series = dict()

    def __bool__(self):
        return self.asin != ""

    def __repr__(self):
        return f"Book(asin='{self.asin}', series='{self.series}')"


class BookCollection(dict):
    def __init__(self, series_name, books: list[Book] = []):
        self.__name__ = series_name
        for book in books:
            self.add(book)

    def add(self, book: Book):
        sequence = book.series[self.__name__]
        keys = expand_range(sequence)

        for key in keys:
            self.setdefault(key, [])
            self[key].append(book.asin)

    def get_first_book(self):
        firt_key = list(self.keys())[0]
        return self[firt_key][0]

    def __bool__(self):
        return len(self.keys()) > 0 and self.get_first_book() != None


def expand_range(part):
    """Expands a range string (e.g., "1-10") or float sequence into a list of numbers."""
    try:
        if "-" in part and not part.startswith("-"):
            start, end = map(int, part.split("-"))
            if start >= end:
                return []  # Handle invalid ranges (start >= end)
            return list(range(start, end + 1))
        else:
            float_val = float(part)
            return [float_val]
    except ValueError:
        return []  # Handle non-numeric input or invalid format


def process_sequence(books):
    """Groups books by ASIN, handling sequence ranges (including floats)."""
    books_sequence = {}
    for book in books:
        asin = book["asin"]
        sequence = book.get("sequence", "")

        if sequence:
            keys = expand_range(sequence.split(", ")[0])
        else:
            keys = [float(book.get("sort", "1")) * -1]

        for key in keys:
            if key not in books_sequence:
                books_sequence[key] = []
            books_sequence[key].append(asin)

    keys = sorted(books_sequence.keys(), key=lambda x: float(x))
    ordered_sequence = {}
    for key in keys:
        ordered_sequence[key] = books_sequence[key]
    return ordered_sequence


def process_audible_serie(books, serie_name):
    processed_books = BookCollection(serie_name)

    for json in books:
        if book["relationship_type"] == "series":
            book = Book(json["asin"])
            book.series.setdefault(serie_name, json["sequence"])
            book.series.setdefault(serie_name, f"-{json['sort']}")
            processed_books.add(book)

    return processed_books


def process_abs_serie(books, series_name):
    processed_books = BookCollection(series_name)

    for index, json in enumerate(books):
        meta = json["media"]["metadata"]

        if meta["asin"] == None:
            logger.debug(
                "ASIN missing: %s (%s by %s)",
                meta["title"],
                series_name,
                meta["authorName"],
            )

        book = Book(meta["asin"])
        for name in meta["seriesName"].split(", "):
            try:
                [name, sequence] = name.split(" #")
            except ValueError:
                logger.debug("Serie missing sequence: %s (%s)", meta["title"], name)
                sequence = f"-{index + 1}"

            book.series[name] = sequence
        processed_books.add(book)
    return processed_books


def get_serie_asin(first_book_asin, series_name):
    audnexus_first_book = audnexus.get_book_from_asin(first_book_asin)

    primary = audnexus_first_book.get("seriesPrimary", {"name": ""})
    secondary = audnexus_first_book.get("seriesSecondary", {"name": ""})

    if primary["name"].casefold() == series_name.casefold():
        return primary["asin"]
    elif secondary["name"].casefold() == series_name.casefold():
        return secondary["asin"]
    else:
        audible_first_book = audible.get_produce_from_asin(first_book_asin)

        if "series" not in audible_first_book:
            return None

        series = audible_first_book.get("series", [])
        series_matching_sequence = [x for x in series if x["sequence"] == "1"]

        if len(series_matching_sequence) == 1:
            return series_matching_sequence[0]["asin"]

        # TODO: search by name
        return series[0]["asin"]


def main():

    libraries = abs.get_library_ids()

    for library in libraries:
        series = abs.get_series_by_library_id(library["id"])

        for serie in series:
            series_name = serie["name"]
            abs_book_sequence = process_abs_serie(serie["books"], series_name)

            if not abs_book_sequence:
                logger.debug("No ASINs found for series: %s", series_name)
                continue

            first_book_asin = abs_book_sequence.get_first_book()
            series_asin = get_serie_asin(first_book_asin, series_name)

            if not series_asin:
                logger.debug("Serie does not exist: %s", series_name)
                continue

            audible_serie = audible.get_produce_from_asin(series_asin)
            audible_book_sequence = process_sequence(audible_serie["relationships"])

            if len(abs_book_sequence) >= len(audible_book_sequence):
                continue

            logger.info(
                "%s - %d out of %d",
                series_name,
                len(abs_book_sequence),
                len(audible_book_sequence),
            )

            # TODO: list missing tomes and show their delivery date if not yet out

        # TODO: add input to choose which library is to be scaned
        break


if __name__ == "__main__":

    logging.getLogger("httpx").setLevel(logging.WARNING)
    logging.getLogger("audible").setLevel(logging.WARNING)
    logging.getLogger("urllib3").setLevel(logging.WARNING)
    logging.getLogger("httpcore").setLevel(logging.WARNING)

    parser = argparse.ArgumentParser()
    parser.add_argument("-d", "--dev", action="store_true")
    parser.add_argument("-v", "--verbose", action="store_true")

    args = parser.parse_args()

    if args.dev:
        abs = connectors.ABSConnectorMock(config.ABS_API_URL, config.ABS_API_TOKEN)
        audible = connectors.AudibleConnectorMock(config.AUDIBLE_AUTH_FILE)
        audnexus = connectors.AudNexusConnectorMock()
    else:
        abs = connectors.ABSConnector(config.ABS_API_URL, config.ABS_API_TOKEN)
        audible = connectors.AudibleConnector(config.AUDIBLE_AUTH_FILE)
        audnexus = connectors.AudNexusConnector()

    logger = logging.getLogger(__name__)
    logger.setLevel(logging.DEBUG if args.verbose else logging.INFO)
    main()