1#!/usr/local/bin/python3.8 2# vim:fileencoding=utf-8 3# License: GPL v3 Copyright: 2020, Kovid Goyal <kovid at kovidgoyal.net> 4 5import os 6import re 7 8 9def get_book_library_details(absolute_path_to_ebook): 10 from calibre.srv.library_broker import correct_case_of_last_path_component, library_id_from_path 11 absolute_path_to_ebook = os.path.abspath(os.path.expanduser(absolute_path_to_ebook)) 12 base = os.path.dirname(absolute_path_to_ebook) 13 m = re.search(r' \((\d+)\)$', os.path.basename(base)) 14 if m is None: 15 return 16 book_id = int(m.group(1)) 17 library_dir = os.path.dirname(os.path.dirname(base)) 18 corrected_path = correct_case_of_last_path_component(library_dir) 19 library_id = library_id_from_path(corrected_path) 20 dbpath = os.path.join(library_dir, 'metadata.db') 21 dbpath = os.environ.get('CALIBRE_OVERRIDE_DATABASE_PATH') or dbpath 22 if not os.path.exists(dbpath): 23 return 24 return {'dbpath': dbpath, 'book_id': book_id, 'fmt': absolute_path_to_ebook.rpartition('.')[-1].upper(), 'library_id': library_id} 25 26 27def database_has_annotations_support(cursor): 28 return next(cursor.execute('pragma user_version;'))[0] > 23 29 30 31def load_annotations_map_from_library(book_library_details, user_type='local', user='viewer'): 32 import apsw 33 from calibre.db.backend import annotations_for_book, Connection 34 ans = {} 35 dbpath = book_library_details['dbpath'] 36 try: 37 conn = apsw.Connection(dbpath, flags=apsw.SQLITE_OPEN_READONLY) 38 except Exception: 39 return ans 40 try: 41 conn.setbusytimeout(Connection.BUSY_TIMEOUT) 42 cursor = conn.cursor() 43 if not database_has_annotations_support(cursor): 44 return ans 45 for annot in annotations_for_book( 46 cursor, book_library_details['book_id'], book_library_details['fmt'], 47 user_type=user_type, user=user 48 ): 49 ans.setdefault(annot['type'], []).append(annot) 50 finally: 51 conn.close() 52 return ans 53 54 55def save_annotations_list_to_library(book_library_details, alist, sync_annots_user=''): 56 import apsw 57 from calibre.db.backend import save_annotations_for_book, Connection, annotations_for_book 58 from calibre.gui2.viewer.annotations import annotations_as_copied_list 59 from calibre.db.annotations import merge_annotations 60 dbpath = book_library_details['dbpath'] 61 try: 62 conn = apsw.Connection(dbpath, flags=apsw.SQLITE_OPEN_READWRITE) 63 except Exception: 64 return 65 try: 66 conn.setbusytimeout(Connection.BUSY_TIMEOUT) 67 if not database_has_annotations_support(conn.cursor()): 68 return 69 amap = {} 70 with conn: 71 cursor = conn.cursor() 72 for annot in annotations_for_book(cursor, book_library_details['book_id'], book_library_details['fmt']): 73 amap.setdefault(annot['type'], []).append(annot) 74 merge_annotations((x[0] for x in alist), amap) 75 if sync_annots_user: 76 other_amap = {} 77 for annot in annotations_for_book(cursor, book_library_details['book_id'], book_library_details['fmt'], user_type='web', user=sync_annots_user): 78 other_amap.setdefault(annot['type'], []).append(annot) 79 merge_annotations(amap, other_amap) 80 alist = tuple(annotations_as_copied_list(amap)) 81 save_annotations_for_book(cursor, book_library_details['book_id'], book_library_details['fmt'], alist) 82 if sync_annots_user: 83 alist = tuple(annotations_as_copied_list(other_amap)) 84 save_annotations_for_book(cursor, book_library_details['book_id'], book_library_details['fmt'], alist, user_type='web', user=sync_annots_user) 85 finally: 86 conn.close() 87