Skip to content
Permalink
Branch: master
Find file Copy path
Find file Copy path
Fetching contributors…
Cannot retrieve contributors at this time
executable file 267 lines (229 sloc) 9.93 KB
#!/usr/bin/env python
'''Utilities for reading and writing a Kobo ebook reader's database.
'''
from __future__ import print_function
import os, sys
import sqlite3
def escape_quotes(s):
return s.replace("'", "''")
class KoboDB:
'''Interact with a Kobo e-reader's database:
either one that's mounted live from the device, or a local copy.
'''
def __init__(self, mountpath):
'''Initialize with the path to where your Kobo is mounted,
or where you keep local copies of the files.
'''
self.mountpath = mountpath
self.dbpath = None
self.conn = None
self.cursor = None
def connect(self, dbpath=None):
'''Open the database at the specified path. Defaults to
.kobo/KoboReader.sqlite in the mountpath you've provided.
'''
if dbpath:
self.dbpath = dbpath
elif self.mountpath:
self.dbpath = os.path.join(mountpath, ".kobo/KoboReader.sqlite")
else:
print("No DB path specified")
return
self.conn = sqlite3.connect(self.dbpath)
self.cursor = self.conn.cursor()
def close(self):
'''Commit any changes and close the database.'''
self.conn.commit()
self.conn.close()
self.conn = None
self.cursor = None
def get_field_names(self, tablename):
'''Get names of fields within a specified table.
I haven't found documentation, but PRAGMA table_info returns:
(index, fieldname, type, None, 0)
I don't know what the None and 0 represent.
'''
self.cursor.execute('PRAGMA table_info(%s);' % tablename)
return [ row[1] for row in self.cursor.fetchall() ]
def get_list(self, tablename, **kwargs):
'''Usage: get_list(tablename, selectors='*', modifiers='', order='')
'''
selectors = '*'
modifiers = ''
order = ''
if kwargs:
if 'selectors' in kwargs and kwargs['selectors']:
if type(kwargs['selectors']) is list:
selectors = ','.join(kwargs['selectors'])
else:
selectors = kwargs['selectors']
if 'modifiers' in kwargs and kwargs['modifiers']:
if type(kwargs['modifiers']) is list:
modifiers = " WHERE " + 'AND'.join(kwargs['modifiers'])
else:
modifiers = " WHERE " + kwargs['modifiers']
if 'order' in kwargs and kwargs['order']:
order = " ORDER BY " + kwargs['order']
sql = "SELECT %s FROM %s%s%s;" % (selectors, tablename,
modifiers, order)
print(sql)
self.cursor.execute(sql)
return self.cursor.fetchall()
def get_dlist(self, tablename, **kwargs):
'''Usage: get_dlist(tablename, selectors='*', modifiers='', order='')
'''
l = self.get_list(tablename, **kwargs)
if kwargs and 'selectors' in kwargs:
fields = kwargs['selectors']
else:
fields = self.get_field_names(tablename)
return [ dict(list(zip(fields, values))) for values in l ]
def get_book_by_id(self, id):
sql = "SELECT Title,Attribution FROM content WHERE ContentID='%s';" \
% escape_quotes(id);
# print sql
self.cursor.execute(sql)
books = self.cursor.fetchall()
if not books:
return None
return books[0]
def list_books(self):
'''List all books in the database.
'''
books = self.get_dlist("content",
selectors=[ 'ContentID', 'Title', 'Attribution',
'Description', 'NumShortcovers',
'IsEncrypted', 'IsDownloaded',
'adobe_location' ],
modifiers="content.BookTitle is null",
order="content.Title")
for book in books:
print("%s (%s)" % (book["Title"], book["Attribution"]))
print(" ContentID: %s" % book["ContentID"])
if book["NumShortcovers"]:
print(" Chapters: %s" % book["NumShortcovers"])
print(" Encrypted? %s" % book["IsEncrypted"], end=' ')
print(" Downloaded? %s" % book["IsDownloaded"], end=' ')
if book["adobe_location"]:
if book["adobe_location"] == book["ContentID"]:
print(" adobe_location: Yes")
else:
print("\n adobe_location:", book["adobe_location"])
else:
print()
# Description is very long; make this optional.
# print " Description:", book["Description"]
print()
def list_shelves(self, names=None):
'''List all shelves (collections) in the database.
'''
allshelves = {}
if names:
modifiers = " AND ".join(["ShelfName=%s" % name for name in names])
else:
modifiers = None
sc = self.get_dlist("ShelfContent", modifiers=modifiers)
for item in sc:
if item["ShelfName"] not in allshelves:
allshelves[item["ShelfName"]] = [ item["ContentId"] ]
else:
allshelves[item["ShelfName"]].append(item["ContentId"])
for shelf in allshelves:
print("\n===", shelf, "===")
for id in allshelves[shelf]:
book = self.get_book_by_id(id)
if not book:
print("Eek, book", id, "doesn't exist")
else:
print(" %s (%s)" % book)
def has_shelf(self, shelfname):
'''Does a given shelfname exist? Helpful when checking whether
to add a new shelf based on a tag.
'''
shelves = self.get_dlist("Shelf", selectors=[ "Name" ],
modifiers=[ "Name='%s'" % shelfname ])
print("Has shelf %s?" % shelfname, bool(shelves))
return bool(shelves)
def print_table(self, tablename, **kwargs):
'''Usage: print_table(tablename, selectors='*', modifiers='', order='')
'''
if kwargs and 'selectors' in kwargs and kwargs['selectors']:
fields = kwargs['selectors']
print("kwargs: fields =", fields)
else:
fields = self.get_field_names(tablename)
print("no kwargs: fields =", fields)
for row in self.get_list(tablename, **kwargs):
for i, f in enumerate(fields):
# Must coerce row[i] to unicode before encoding,
# even though it should be unicode already,
# because it could be null.
print(f.encode('UTF-8'), ":", str(row[i]).encode('UTF-8'))
# Adding entries to shelves:
def make_new_shelf(self, shelfname):
'''Create a new shelf/collection.
'''
print("=== Current shelves:")
self.print_table("Shelf", selectors=[ "Name" ])
print("===")
print("Making a new shelf called", shelfname)
# Skip type since it's not clear what it is and it's never set.
# For the final three, PRAGMA table_info(Shelf); says they're
# type BOOL, and querying that table shows true and false there,
# but absolutely everyone on the web says you have to use
# 1 and 0 for sqlite3 and that there is no boolean type.
# XXX DATETIME('now') inserts something like "2015-11-20 16:36:34"
# but Kobo-created shelves look like "2015-08-21T01:47:15Z".
query = '''INSERT INTO Shelf(CreationDate, Id, InternalName,
LastModified, Name, _IsDeleted, _IsVisible, _IsSynced)
VALUES (DATETIME('now'), '%s', '%s', DATETIME('now'), '%s', 0, 1, 1);
''' % (shelfname, shelfname, shelfname)
print(query)
self.cursor.execute(query)
def add_to_shelf(self, kobobook, shelfname):
print("===")
print("Adding", kobobook["Title"], "to shelf", shelfname)
query = '''INSERT INTO ShelfContent(ShelfName, ContentId, DateModified,
_IsDeleted, _IsSynced)
VALUES ('%s', '%s', DATE('now'), 0, 0);''' % (shelfname,
escape_quotes(kobobook['ContentID']))
print(query)
self.cursor.execute(query)
self.conn.commit()
if __name__ == '__main__':
import argparse
parser = argparse.ArgumentParser(description="""Show details about a Kobo ebook reader.
By default, show a list of books.
Copyright 2015 by Akkana Peck; share and enjoy under the GPLv2 or later.""",
formatter_class=argparse.RawDescriptionHelpFormatter)
# Options:
parser.add_argument("-m", "--mountdir", default="/kobo",
help="""Path where the Kobo is mounted. Default: /kobo""")
parser.add_argument("-d", "--db",
default="$mountdir/.kobo/KoboReader.sqlite",
help="""Path to the database.
Default: $mountdir/.kobo/KoboReader.sqlite""")
# Things we can do:
parser.add_argument("-s", "--shelves", action='store_true', default=False,
help="""Show shelves""")
parser.add_argument("-S", "--shelfnames", action='store_true',
default=False,
help="""Show shelf names but not their contents""")
args = parser.parse_args()
args.db = args.db.replace('$mountdir', args.mountdir)
try:
koboDB = KoboDB(args.mountdir)
koboDB.connect(args.db)
except Exception as e:
print("Couldn't open database at %s for Kobo mounted at %s" % \
(args.db, args.mountdir))
print(e)
sys.exit(1)
if args.shelfnames:
shelves = koboDB.get_dlist("Shelf", selectors=[ "Name" ])
for shelf in shelves:
print(shelf["Name"])
elif args.shelves:
koboDB.list_shelves()
else:
koboDB.list_books()
You can’t perform that action at this time.