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