From f6efd32b9024d785a2d5b7bbe6fcb677cbaffc12 Mon Sep 17 00:00:00 2001 From: S Groesz Date: Thu, 10 Dec 2020 00:52:07 +0000 Subject: [PATCH] initial commit --- .gitignore | 8 + jnc_api_tools.py | 471 +++++++++++++++++++++++++++++++++++++++++++++++ loadenv.sh | 5 + readme.md | 2 + 4 files changed, 486 insertions(+) create mode 100644 .gitignore create mode 100644 jnc_api_tools.py create mode 100755 loadenv.sh create mode 100644 readme.md diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..5f56c44 --- /dev/null +++ b/.gitignore @@ -0,0 +1,8 @@ +__pycache__ +*.bak +*.swp +build/ +dist/ +*.egg-info +*.token +*.private diff --git a/jnc_api_tools.py b/jnc_api_tools.py new file mode 100644 index 0000000..e4252b8 --- /dev/null +++ b/jnc_api_tools.py @@ -0,0 +1,471 @@ +import csv +import os +from datetime import datetime, timezone +from typing import List, Dict, Set + +import requests + + +class JNClient: + """Everything you need to talk to the JNC API""" + + BASE_URL = 'https://api.j-novel.club/api' + LOGIN_URL = BASE_URL + '/users/login?include=user' + LOGOUT_URL = BASE_URL + '/users/logout' + SERIES_INFO_URL = BASE_URL + '/series/findOne' + API_USER_URL_PATTERN = BASE_URL + '/users/%s/' # %s is the user id + DOWNLOAD_BOOK_URL_PATTERN = BASE_URL + '/volumes/%s/getpremiumebook' # %s is the book id + ORDER_URL_PATTERN_NEW = 'https://labs.j-novel.club/app/v1/me/redeem/%s' # %s book id + BUY_CREDITS_URL_PATTERN = BASE_URL + '/users/%s/purchasecredit' # %s is user id + + ACCOUNT_TYPE_PREMIUM = 'PremiumMembership' + + def __init__(self, login_email, login_password) -> None: + login_response = self.__login(login_email, login_password) + + if 'error' in login_response: + raise JNCApiError('Login failed!') + + self.auth_token = login_response['id'] + self.user_id = login_response['user']['id'] + self.user_name = login_response['user']['username'] + self.available_credits = login_response['user']['earnedCredits'] - login_response['user']['usedCredits'] + + subscription = login_response['user']['currentSubscription'] + + self.account_type = subscription['plan']['id'] if 'plan' in subscription else None + + def get_owned_books(self) -> dict: + """Requests the list of owned books from JNC. + + :return dict with information on the owned books (ids, titles, etc.)""" + return requests.get( + self.API_USER_URL_PATTERN % self.user_id, + params={'filter': '{"include":[{"ownedBooks":"serie"}]}'}, + headers={'Authorization': self.auth_token} + ).json()['ownedBooks'] + + def order_book(self, book_id: str) -> None: + """Order book on JNC side, i.e. redeem premium credit + + Notable responses: + 204: Success + 401: Unauthorized + 410: Session token expired + 404: Volume not found + 501: Can't buy manga at this time + 402: No credits left to redeem + 409: Already own this volume + 500: Internal server error (reported to us) + Other: Unknown server error + """ + if self.available_credits <= 0: + raise NoCreditsError('No credits available to order book!') + + response = requests.post( + self.ORDER_URL_PATTERN_NEW % book_id, + headers={'Authorization': f'Bearer {self.auth_token}'} + ) + + if response.status_code == 409: + raise JNCApiError('Book already ordered') + + if not response.ok: + raise JNCApiError(f'Error when ordering book. Response was: {response.status_code}') + + self.available_credits -= 1 + + def download_book(self, book_id: str) -> bytes: + """Will attempt to download a book from JNC + JNC does not respond with a standard 404 error when a book cannot be found (despite being marked as published) + and instead will do a redirect to an error page, which itself reports a http 200 + + :param book_id the id of the book. + :return The response content + :raise JNCApiError when the book is not available for download yet.""" + r = requests.get( + self.DOWNLOAD_BOOK_URL_PATTERN % book_id, + params={ + 'userId': self.user_id, + 'userName': self.user_name, + 'access_token': self.auth_token + }, allow_redirects=False + ) + + if r.status_code != 200: + raise JNCApiError(str(r.status_code) + ': Book not available.') + + return r.content + + def get_series_info(self, series_title_slug: str) -> dict: + """Fetch information about a series from JNC, including the volumes of the series""" + filter_string = '{"where":{"titleslug":"%s"},"include":["volumes"]}' % series_title_slug + return requests.get( + self.SERIES_INFO_URL, + params={'filter': filter_string} + ).json() + + def buy_credits(self, amount: int) -> None: + """Buy premium credits on JNC. Max. amount: 10. Price depends on membership status.""" + if (type(amount) is not int) or (0 >= amount > 10): + raise ArgumentError('It is not possible to buy less than 1 or more than 10 credits.') + + response = requests.post( + self.BUY_CREDITS_URL_PATTERN % self.user_id, + headers={ + 'Accept': 'application/json', + 'content-type': 'application/json', + 'Authorization': self.auth_token + }, + json={'number': amount}, + allow_redirects=False + ) + + if not response.status_code < 300: + raise JNCApiError('Could not purchase credits!') + + self.available_credits += amount + + def get_premium_credit_price(self) -> int: + """Determines the price of premium credits based on account status""" + if self.account_type is None: + return None + elif self.ACCOUNT_TYPE_PREMIUM in self.account_type: + return 6 + else: + return 7 + + def __login(self, login_email, password) -> dict: + """Sends a login request to JNC. This method will not work if the user uses SSO (Google, Facebook)""" + return requests.post( + self.LOGIN_URL, + headers={'Accept': 'application/json', 'content-type': 'application/json'}, + json={'email': login_email, 'password': password} + ).json() + + def __logout(self) -> None: + """Does a logout request to JNC to invalidate the access token. Intended to be called automatically when the + class instance is garbage collected.""" + requests.post( + self.LOGOUT_URL, + headers={'Authorization': self.auth_token} + ) + print('Logged out') + + def __del__(self): + try: + if len(self.auth_token) > 0: + self.__logout() + except AttributeError: + pass + + +class JNCBook: + book_id: str + title: str + title_slug: str + volume_num: int + series_id: str + series_slug: str = None + release_date: datetime + + def __init__(self, book_id: str, title: str, title_slug: str, volume_num: int, series_id: str, + date_string_iso: str, series_slug: str = None): + self.release_date = datetime.fromisoformat(date_string_iso.rstrip('Z')).replace(tzinfo=timezone.utc) + self.series_id = series_id + self.volume_num = volume_num + self.title_slug = title_slug + self.title = title + self.book_id = book_id + self.series_slug = series_slug + + +class JNCSeries: + title_slug: str + followed: bool = None + is_detailed: bool = False + + series_id: str = None + title: str = None + tags: str = None + volumes: Dict[str, JNCBook] = None + + def __init__(self, title_slug: str, followed: bool = None, series_id: str = None, is_detailed: bool = False, + title: str = None, volumes: Dict[str, JNCBook] = None, tags: str = None): + self.tags = tags + self.volumes = volumes + self.title = title + self.is_detailed = is_detailed + self.series_id = series_id + self.followed = followed + self.title_slug = title_slug + + +class JNCDataHandler: + jnclient: JNClient + owned_books: Dict[str, JNCBook] = {} + owned_series: Dict[str, JNCSeries] = {} + owned_series_csv_path: str + owned_books_csv_path: str + download_dir: str + no_confirm_series: bool + no_confirm_credits: bool + no_confirm_order: bool + downloaded_book_ids: Set[str] = set() + new_downloaded_books: Dict[str, str] = {} + + def __init__(self, jnclient: JNClient, owned_series_csv_path: str, owned_books_csv_path: str, download_dir: str, + no_confirm_series: bool = False, no_confirm_credits: bool = False, no_confirm_order: bool = False): + self.jnclient = jnclient + self.no_confirm_series = no_confirm_series + self.no_confirm_credits = no_confirm_credits + self.no_confirm_order = no_confirm_order + self.cur_time = datetime.now(timezone.utc) + self.download_target_dir = os.path.expanduser(download_dir) + self.downloaded_books_list_file = os.path.expanduser(owned_books_csv_path) + self.owned_series_file = os.path.expanduser(owned_series_csv_path) + self.__ensure_files_exist() + self.load_owned_books() + self.load_owned_series() + self.load_followed_series_details() + self.read_downloaded_books_file() + + def load_owned_books(self) -> None: + raw_owned_books = self.jnclient.get_owned_books() + for raw_book in raw_owned_books: + book = JNCUtils.build_jnc_book_from_api_response(raw_book) + self.owned_books[book.book_id] = book + + self.owned_books = JNCUtils.sort_books(self.owned_books) + + def load_owned_series(self) -> None: + for book in self.owned_books.values(): + if book.series_slug is not None and book.series_id not in self.owned_series: + self.owned_series[book.series_id] = JNCSeries(book.series_slug, series_id=book.series_id) + + def handle_new_series(self) -> None: + """"Ask the user if he wants to follow a new series he owns""" + for series in self.owned_series.values(): + if series.followed is None: + series.followed = self.no_confirm_series or JNCUtils.user_confirm( + f'{series.title or series.title_slug} is a new series. Do you want to follow it?' + ) + self.load_followed_series_details() + + def read_owned_series_file(self) -> Dict[str, JNCSeries]: + """Returns dictionary with series with the series title slug as key""" + series = {} + with open(self.owned_series_file, mode='r', newline='') as file: + csv_reader = csv.reader(file, delimiter='\t') + for series_row in csv_reader: + series[series_row[0]] = JNCSeries(series_row[0], True if series_row[1] == 'True' else False) + return series + + def read_downloaded_books_file(self) -> None: + with open(self.downloaded_books_list_file, mode='r', newline='') as f: + self.downloaded_book_ids = set([row[0] for row in csv.reader(f, delimiter='\t')]) + + def download_book(self, book: JNCBook) -> None: + try: + book_content = self.jnclient.download_book(book.book_id) + + book_file_name = book.title_slug + '.epub' + book_file_path = os.path.join(self.download_target_dir, book_file_name) + + with open(book_file_path, mode='wb') as f: + f.write(book_content) + + self.new_downloaded_books[book.book_id] = book.title + except JNCApiError as err: + print(err) + + def load_followed_series_details(self) -> None: + known_series = self.read_owned_series_file() + for owned_series in self.owned_series.values(): + for series_slug, cur_known_series in known_series.items(): + if owned_series.title_slug == series_slug: + owned_series.followed = cur_known_series.followed + + if not owned_series.followed \ + or owned_series.is_detailed is True: + continue + + series_details = self.jnclient.get_series_info(owned_series.title_slug) + series_id = series_details['id'] + self.owned_series[series_id].title = series_details['title'] + self.owned_series[series_id].tags = series_details['tags'] + self.owned_series[series_id].is_detailed = True + self.owned_series[series_id].volumes = {} + + for raw_book in series_details['volumes']: + book = JNCUtils.build_jnc_book_from_api_response(raw_book) + self.owned_series[series_id].volumes[book.book_id] = book + + def get_orderable_books(self) -> List[JNCBook]: + orderable_books = [] + """Check for new volumes in followed series""" + for series in self.owned_series.values(): + if series.followed is False or series.is_detailed is False: + continue + + for book_id in series.volumes: + if book_id not in self.owned_books: + orderable_books.append(series.volumes[book_id]) + return orderable_books + + def get_preorders(self) -> List[JNCBook]: + preorders = [] + for book in self.owned_books.values(): + if book.release_date > self.cur_time: + preorders.append(book) + return preorders + + def download_new_books(self) -> None: + print('\nDownloading new books:') + preorders = self.get_preorders() + for book in self.owned_books.values(): + + if book.book_id in self.downloaded_book_ids: + self.new_downloaded_books[book.book_id] = book.title + continue + + # Skip preorders + if any(book.book_id == preorder.book_id for preorder in preorders): + continue + + print(f'{book.title} \t{book.book_id} \t{book.release_date}') + + self.download_book(book) + + def buy_credits(self, credits_to_buy) -> None: + print(f'\nAttempting to buy {credits_to_buy} credits.') + unit_price = self.jnclient.get_premium_credit_price() + + if unit_price is None: + print('Inactive subscription, cannot buy credits') + return + + print(f'Each premium credit will cost US${unit_price}') + while credits_to_buy > 0: + purchase_batch = 10 if credits_to_buy > 10 else credits_to_buy + price = purchase_batch * unit_price + if self.no_confirm_credits \ + or JNCUtils.user_confirm(f'Do you want to buy {purchase_batch} premium credits for US${price}?'): + self.jnclient.buy_credits(purchase_batch) + print(f'Successfully bought {purchase_batch} premium credits. ') + credits_to_buy -= purchase_batch + print(f'{credits_to_buy} premium credits left to buy.') + else: + # abort when user does not confirm + break + print('\n') + + def order_unowned_books(self, buy_individual_credits) -> None: + print('\nOrdering unowned volumes of followed series:') + new_books_ordered = False + for book in self.get_orderable_books(): + print(f'Order book {book.title}') + if self.no_confirm_order or JNCUtils.user_confirm('Do you want to order?'): + if (self.jnclient.available_credits == 0) and buy_individual_credits: + self.buy_credits(1) + if self.jnclient.available_credits == 0: + print('No premium credits left. Stop order process.') + break + self.jnclient.order_book(book.book_id) + print( + f'Ordered {book.title}! Remaining credits: {self.jnclient.available_credits}\n' + ) + new_books_ordered = True + if new_books_ordered: + # refresh data + self.load_owned_books() + self.load_owned_series() + self.handle_new_series() + + def unfollow_complete_series(self) -> None: + owned_book_ids = self.owned_books.keys() + for series in self.owned_series.values(): + if not series.followed or (series.is_detailed and 'fully translated' not in series.tags): + continue + series_volume_ids = set([volume.book_id for volume in series.volumes.values()]) + + if series_volume_ids.issubset(owned_book_ids): + print(f'{series.title} is fully owned and completed. Series will not be followed anymore.') + series.followed = False + continue + + def write_owned_series_file(self) -> None: + follow_states = {} + for series in self.owned_series.values(): + if series.followed is None: + continue + follow_states[series.title_slug] = series.followed + + with open(self.owned_series_file, mode='w', newline='') as f: + series_csv_writer = csv.writer(f, delimiter='\t') + series_csv_writer.writerows(follow_states.items()) + + def write_downloaded_books_file(self) -> None: + with open(self.downloaded_books_list_file, mode='w', newline='') as f: + csv_writer = csv.writer(f, delimiter='\t') + csv_writer.writerows(self.new_downloaded_books.items()) + + def print_new_volumes(self) -> None: + orderable_books = self.get_orderable_books() + if len(orderable_books) > 0: + print('\nThe following new books of series you follow can be ordered:') + for book in orderable_books: + print(book.title) + + def print_preorders(self) -> None: + preorders = self.get_preorders() + if len(preorders): + print('\nCurrent preorders (Release Date / Title):') + for book in preorders: + print(f'{book.release_date} {book.title}') + + def __ensure_files_exist(self) -> None: + if not os.path.isfile(self.owned_series_file): + open(self.owned_series_file, 'a').close() + if not os.path.isfile(self.downloaded_books_list_file): + open(self.downloaded_books_list_file, 'a').close() + + +class JNCUtils: + @staticmethod + def user_confirm(message: str) -> bool: + answer = input(message + ' (y/n)') + return True if answer == 'y' else False + + @staticmethod + def sort_books(books: Dict[str, JNCBook]) -> Dict[str, JNCBook]: + """Sorts List of JNCBooks by their series slug and volume number and returns result""" + sorted_book_ids = sorted( + books, + key=lambda k: (books[k].series_slug or books[k].title_slug, books[k].volume_num) + ) + return {book_id: books[book_id] for book_id in sorted_book_ids} + + @staticmethod + def build_jnc_book_from_api_response(raw_book_response: dict) -> JNCBook: + return JNCBook( + raw_book_response['id'], + raw_book_response['title'], + raw_book_response['titleslug'], + raw_book_response['volumeNumber'], + raw_book_response['serieId'], + raw_book_response['publishingDate'], # Format: 2020-07-12T05:00:00.000Z + raw_book_response.get('serie', {}).get('titleslug', None) + ) + + +class JNCApiError(Exception): + pass + + +class NoCreditsError(Exception): + pass + + +class ArgumentError(Exception): + pass diff --git a/loadenv.sh b/loadenv.sh new file mode 100755 index 0000000..dc7ec8e --- /dev/null +++ b/loadenv.sh @@ -0,0 +1,5 @@ +#!/bin/sh - +# Get JNC Login info +export jncuser=`cat user.private` +export jncpass=`cat pass.private` + diff --git a/readme.md b/readme.md new file mode 100644 index 0000000..2aaf158 --- /dev/null +++ b/readme.md @@ -0,0 +1,2 @@ +create two files: user.private and pass.private. In these two files should only be the username and password, respectively, with no trailing whitespace or line breaks. +