Last active
March 13, 2026 05:18
-
-
Save halkeye/5f3e4e7b23ba9e9daa962a6afc54d54b to your computer and use it in GitHub Desktop.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| import os | |
| from sqlalchemy import and_, or_ | |
| import cps.ub as rawdb | |
| from cps import calibre_db, cli_param, config, config_sql, db, logger, ub | |
| from cps.services import hardcover | |
| from cps.ub import get_new_session_instance | |
| def all_read_books(self, offset=0, limit=25): | |
| variables = {} | |
| variables["offset"] = offset | |
| variables["limit"] = limit | |
| query = """ | |
| query ($offset: Int = 0, $limit: Int = 10) { | |
| me { | |
| user_books( | |
| where: {status_id: {_eq: 3}} | |
| order_by: {book_id: asc} | |
| limit: $limit | |
| offset: $offset | |
| distinct_on: book_id | |
| ) { | |
| edition_id | |
| book { | |
| id | |
| slug | |
| } | |
| } | |
| } | |
| } | |
| """ | |
| response = self.execute(query, variables) | |
| return next(iter(response.get("me")), dict).get("user_books") | |
| if __name__ == "__main__": | |
| log = logger.create() | |
| if os.path.isfile("/config/app.db"): | |
| rawdb.app_DB_path = "/config/app.db" # type: ignore[assignment] | |
| else: | |
| rawdb.app_DB_path = "./app.db" # type: ignore[assignment] | |
| ub.init_db_thread() | |
| ub.session = get_new_session_instance() | |
| cli_param.init() | |
| encrypt_key, error = config_sql.get_encryption_key( | |
| os.path.dirname(cli_param.settings_path) | |
| ) | |
| config.init_config(ub.session, encrypt_key, cli_param) | |
| calibre_db.init_db() | |
| if not os.path.isfile("/config/app.db"): | |
| config.config_calibre_dir = os.getcwd() | |
| calibre_db.reconnect_db(config, ub.app_DB_path) | |
| hardcover.HardcoverClient.all_read_books = all_read_books # type: ignore[attr-defined] | |
| for user in ub.session.query(ub.User).all(): | |
| if user.hardcover_token is None: | |
| continue | |
| hardcoverClient = hardcover.HardcoverClient(user.hardcover_token) | |
| has_next = True | |
| offset = 0 | |
| while has_next: | |
| hardcover_books = hardcoverClient.all_read_books(offset, 25) # type: ignore[attr-defined] | |
| has_next = len(hardcover_books) == 25 | |
| offset += len(hardcover_books) | |
| for hardcover_book in hardcover_books: | |
| cw_book = calibre_db.session.query(db.Books).join(db.Identifiers).filter( | |
| or_( | |
| and_(db.Identifiers.type == "hardcover-id", db.Identifiers.val == hardcover_book["book"]["id"]), | |
| and_(db.Identifiers.type == "hardcover-slug", db.Identifiers.val == hardcover_book["book"]["slug"]), | |
| and_(db.Identifiers.type == "hardcover", db.Identifiers.val == hardcover_book["book"]["slug"]), | |
| and_(db.Identifiers.type == "hardcover-edition", db.Identifiers.val == hardcover_book["edition_id"]) | |
| ) | |
| ).first() | |
| if cw_book is None: | |
| print("no book found for hardcover book", hardcover_book) | |
| continue | |
| book_read = ( | |
| ub.session.query(ub.ReadBook) | |
| .filter(ub.ReadBook.book_id == cw_book.id, ub.ReadBook.user_id == user.id) | |
| .one_or_none() | |
| ) | |
| if not book_read: | |
| book_read = ub.ReadBook(user_id=user.id, book_id=cw_book.id) | |
| if book_read.read_status == ub.ReadBook.STATUS_FINISHED: | |
| # already marked as read, so can skip it | |
| continue | |
| print("book not read", cw_book) | |
| book_read.read_status = ub.ReadBook.STATUS_FINISHED | |
| try: | |
| ub.session.merge(book_read) | |
| ub.session_commit() | |
| except Exception as e: | |
| log.error(f"Error queuing book {cw_book.id} for review: {e}") | |
| ub.session.rollback() | |
| raise |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment