import alive_progress import requests import connectors import argparse import logging import config 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: if not args.non_series and not float(key).is_integer(): continue 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_audible_serie(books, serie_name): processed_books = BookCollection(serie_name) for json in books: if json["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) else: logger.debug("Skipping non-series book: %s", json["asin"]) 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: if library["mediaType"] != "book" or library["provider"] != "audible": continue logger.info("==== %s ====", library["name"]) series = abs.get_series_by_library_id(library["id"]) for serie in alive_progress.alive_it(series, title=library["name"]): 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_audible_serie( audible_serie["relationships"], series_name ) if len(abs_book_sequence) >= len(audible_book_sequence): continue missing_keys = set( [ key for key in audible_book_sequence.keys() if key not in abs_book_sequence ] ) # Separate missing and soon-to-be-released books missing_books = [] soon_to_release_books = [] for key in missing_keys: found = False for asin in audible_book_sequence[key]: try: audnexus.get_book_from_asin(asin) missing_books.append(key) logger.debug( "%s Book %.1f is missing - %s", series_name, key, audible_book_sequence[key][0], ) found = True break except requests.exceptions.HTTPError: pass if not found and args.oncoming: soon_to_release_books.append(key) logger.debug( "%s Book %d is yet to be released - %s", series_name, key, audible_book_sequence[key][0], ) msgs = [] if missing_books: msgs.append(f"{len(missing_books)} book.s missing") if soon_to_release_books: msgs.append(f"{len(soon_to_release_books)} book.s yet to be released") for i, msg in enumerate(msgs): logger.info( "%s - %s", series_name if i == 0 else "".ljust(len(series_name)), msg, ) if __name__ == "__main__": parser = argparse.ArgumentParser() # General flags parser.add_argument( "-v", "--verbose", action="store_true", help="Enable verbose logging" ) parser.add_argument( "-d", "--dev", action="store_true", help="Use development/mock connectors" ) # Feature-specific flags parser.add_argument( "--non-series", action="store_true", help="Include non-series books (books not part of a numbered series)", ) parser.add_argument( "--oncoming", action="store_true", help="Show books to be released", ) 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() logging.basicConfig( filename="log", filemode="w", format="%(levelname)s - %(message)s" if args.verbose else "%(message)s", ) logging.getLogger("httpx").setLevel(logging.WARNING) logging.getLogger("audible").setLevel(logging.WARNING) logging.getLogger("urllib3").setLevel(logging.WARNING) logging.getLogger("httpcore").setLevel(logging.WARNING) alive_progress.config_handler.set_global(enrich_print=False) logger = logging.getLogger(__name__) logger.setLevel(logging.DEBUG if args.verbose else logging.INFO) main()