summaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
authorGravatar Étienne Deparis 2017-05-24 15:04:28 +0200
committerGravatar Étienne Deparis 2017-05-24 15:04:28 +0200
commit67d03e7daa8fee0a9dff70872d52f12b8da8827c (patch)
tree0a625745e0b27f698d83f9f11ea53281678b073e
downloadeshelf-67d03e7daa8fee0a9dff70872d52f12b8da8827c.tar.gz
eshelf-67d03e7daa8fee0a9dff70872d52f12b8da8827c.tar.xz
eshelf-67d03e7daa8fee0a9dff70872d52f12b8da8827c.zip
Initial import from https://projects.depar.is/divers/dir?ci=f8e368e1e9550db8&name=eshelf
-rw-r--r--eshelf.py435
-rw-r--r--plugins/backup.py75
2 files changed, 510 insertions, 0 deletions
diff --git a/eshelf.py b/eshelf.py
new file mode 100644
index 0000000..1fb825b
--- /dev/null
+++ b/eshelf.py
@@ -0,0 +1,435 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+
+import os
+import re
+import sys
+import sqlite3
+import argparse
+from configparser import ConfigParser
+from xdg.BaseDirectory import xdg_config_home
+
+
+
+def file_listing_helper(current_dir, padding=None):
+ if padding == None:
+ padding = ''
+
+ decorator = '├──'
+ decorator_prefix = padding + '│ '
+
+ curfilelist = os.listdir(current_dir)
+ curfilelistlen = len(curfilelist) - 1
+ curpointer = 0
+ for curfile in curfilelist:
+ if curpointer == curfilelistlen:
+ decorator = '└──'
+ decorator_prefix = padding + ' '
+
+ curpointer += 1
+ print('{0}{1} {2}'.format(padding, decorator, curfile))
+
+ curfile_path = os.path.join(current_dir, curfile)
+ if os.path.isdir(curfile_path):
+ file_listing_helper(curfile_path, decorator_prefix)
+
+
+def read_file(f):
+ if os.path.isfile(f):
+ with open(f, 'r') as s:
+ return s.read().strip()
+ return ''
+
+
+def bookeen_identify():
+ import subprocess
+ try:
+ usblist = str(subprocess.check_output(['lsusb']))
+ except subprocess.CalledProcessError:
+ usblist = ''
+
+ vendorlist = ['0525:a4a5', '0bda:0703', '3034:1795']
+ for prefix in vendorlist:
+ if usblist.find(prefix) >= 0:
+ if prefix == '0525:a4a5':
+ return 'Bookeen Cybook Muse'
+
+ try:
+ usbdetails = str(subprocess.check_output(['lsusb', '-d{0}'.format(prefix), '-v']))
+ m = re.search('bcdDevice\s*([0-9.]+)\\\\n', usbdetails)
+ if m != None and m.group(1) == '3.19':
+ return 'Bookeen Cybook Orizon'
+
+ except subprocess.CalledProcessError:
+ pass
+
+ return 'Bookeen Cybook Opus or Gen 3'
+
+ return None
+
+
+
+class Shelf:
+ def __init__(self, library_path):
+ if library_path == None:
+ return
+
+ if not os.path.isdir(library_path):
+ print('{0} does not exists, please provide us a valid path.'.format(library_path))
+ sys.exit(1)
+
+ self.library_path = library_path
+ config_file = os.path.join(self.library_path, '.eshelf.conf')
+ if not os.path.isfile(config_file):
+ print('{0} is not a valid library folder. Please do first `eshelf init`'.format(self.library_path))
+ sys.exit(1)
+
+ confparser = ConfigParser()
+ confparser.read(config_file)
+ self.config = confparser['DEFAULT']
+
+ if not 'reader_mount_point' in self.config:
+ print('{0} has been corrupted. Please redo `eshelf init`'.format(self.library_path))
+ sys.exit(1)
+
+ self.reader_mount_point = self.config['reader_mount_point']
+ self.reader_db_path = os.path.join(self.reader_mount_point, 'system', 'library')
+ self.reader_db = None
+
+ self.remote_physical_path = '/mnt/fat'
+ if 'remote_physical_path' in self.config:
+ self.remote_physical_path = self.config['remote_physical_path']
+
+
+ def get_config(self):
+ return self.config
+
+
+ def get_reader_db(self):
+ if self.reader_db == None:
+ try:
+ self.reader_db = sqlite3.connect(self.reader_db_path)
+ self.reader_db.row_factory = sqlite3.Row
+ self.reader_db.isolation_level = None
+
+ except sqlite3.OperationalError:
+ print('Something wrong happen. Maybe you do not have mounted your ereader.')
+ sys.exit(1)
+
+ return self.reader_db
+
+
+ def get_cursor(self):
+ return self.get_reader_db().cursor()
+
+
+ def collections(self, opts=[]):
+ finaldata = []
+ cur = self.get_cursor()
+ for row in cur.execute('select f_id_shelf, f_name, f_readonly from t_shelf order by f_name'):
+ loc_data = dict(row)
+ if loc_data['f_readonly'] == 0:
+ finaldata.append('[{0}] {1}'.format(loc_data['f_id_shelf'], loc_data['f_name']))
+
+ if len(finaldata) == 0:
+ return
+
+ print('Collections:')
+ decorator = '├──'
+ curfilelistlen = len(finaldata) - 1
+ curpointer = 0
+ for line in finaldata:
+ if curpointer == curfilelistlen:
+ decorator = '└──'
+
+ curpointer += 1
+ print('{0} {1}'.format(decorator, line))
+
+
+ def books(self, opts=[]):
+ cur = self.get_cursor()
+ finaldata = []
+ values = ()
+ shelf_name = None
+ request = None
+ orphans = False
+ if len(opts) == 1:
+ if opts[0] == 'orphans':
+ orphans = True
+
+ else:
+ if str(opts[0]).isdecimal():
+ shelf_infos_req = 'select f_id_shelf, f_name from t_shelf where f_id_shelf = ?'
+ else:
+ shelf_infos_req = 'select f_id_shelf, f_name from t_shelf where f_name = ? collate nocase'
+
+ shelf_infos = cur.execute(shelf_infos_req, (opts[0],)).fetchone()
+ if shelf_infos == None:
+ print('No such shelf {0}'.format(opts[0]))
+ return
+
+ shelf_infos = dict(shelf_infos)
+ shelf_name = shelf_infos['f_name']
+ values = (shelf_infos['f_id_shelf'],)
+ request = 'select b.f_id_item, b.f_title, b.f_internal_uri from t_item b, t_shelf_link tb where tb.f_item_id = b.f_id_item and tb.f_shelf_id = ? order by b.f_title'
+
+ if request == None:
+ request = 'select f_id_item, f_title, f_internal_uri from t_item order by f_title'
+
+ for row in cur.execute(request, values):
+ loc_data = dict(row)
+ real_internal_path = re.sub(
+ self.remote_physical_path + '/',
+ self.reader_mount_point,
+ loc_data['f_internal_uri'])
+
+ local_file_path = re.sub(
+ os.path.join(self.remote_physical_path, self.config['remote_library_path']) + '/',
+ '',
+ loc_data['f_internal_uri'])
+ # Twice to be sure to coverage all cases
+ local_file_path = re.sub(
+ self.remote_physical_path + '/',
+ '',
+ local_file_path)
+
+ if orphans:
+ if not os.path.isfile(os.path.join(self.library_path, local_file_path)):
+ local_file = real_internal_path
+ else:
+ continue
+ else:
+ local_file = '(not on your computer: {0})'.format(local_file_path)
+ if os.path.isfile(os.path.join(self.library_path, local_file_path)):
+ local_file = ''
+
+ if os.path.isfile(real_internal_path):
+ finaldata.append('[{0}] {1} {2}'.format(loc_data['f_id_item'], loc_data['f_title'], local_file))
+
+
+
+ if shelf_name != None:
+ print('Books in {0}:'.format(shelf_name))
+ else:
+ print('Books in your eReader library:')
+
+
+ if len(finaldata) == 0:
+ return
+
+ decorator = '├──'
+ curfilelistlen = len(finaldata) - 1
+ curpointer = 0
+ for line in finaldata:
+ if curpointer == curfilelistlen:
+ decorator = '└──'
+
+ curpointer += 1
+ print('{0} {1}'.format(decorator, line))
+
+
+ def prepare_add_remove(self, shelf_id_or_name, book_id_or_title):
+ cur = self.get_cursor()
+
+ # Ok now try to find corresponding shelf
+ if str(shelf_id_or_name).isdecimal():
+ shelf_infos_req = 'select f_id_shelf from t_shelf where f_id_shelf = ?'
+ else:
+ shelf_infos_req = 'select f_id_shelf from t_shelf where f_name = ? collate nocase'
+
+ shelf_infos = cur.execute(shelf_infos_req, (shelf_id_or_name,)).fetchone()
+ if shelf_infos == None:
+ print('No such shelf {0}'.format(shelf_id_or_name))
+ sys.exit(1)
+ shelf_infos = dict(shelf_infos)
+
+ # Now check if book exists
+ if str(book_id_or_title).isdecimal():
+ book_infos_req = 'select f_id_item from t_item where f_id_item = ?'
+ else:
+ book_infos_req = 'select f_id_item from t_item where f_title = ? collate nocase'
+
+ book_infos = cur.execute(book_infos_req, (book_id_or_title,)).fetchone()
+ if book_infos == None:
+ print('No such book {0}'.format(book_id_or_title))
+ sys.exit(1)
+ book_infos = dict(book_infos)
+
+ # Internal requirements
+ b_uuid = dict(cur.execute('select f_db_uuid from t_uuiddb limit 1').fetchone())
+
+ return shelf_infos['f_id_shelf'], book_infos['f_id_item'], b_uuid['f_db_uuid']
+
+
+ def add(self, opts=[]):
+ if len(opts) < 2:
+ print('Action must follow the form: add <id_or_title> <shelf_id_or_name>')
+ sys.exit(1)
+
+ cur_shelf, cur_book, b_uuid = self.prepare_add_remove(opts[1], opts[0])
+
+ cur = self.get_cursor()
+ cur.execute(
+ "insert into t_shelf_link values (?, ?, ?)",
+ (cur_book, cur_shelf, b_uuid))
+ print('Successfully added a book to a shelf.')
+ self.books([cur_shelf])
+
+
+ def remove(self, opts=[]):
+ if len(opts) < 2:
+ print('Action must follow the form: remove <id_or_title> <shelf_id_or_name>')
+ sys.exit(1)
+
+ cur_shelf, cur_book, b_uuid = self.prepare_add_remove(opts[1], opts[0])
+
+ cur = self.get_cursor()
+ cur.execute(
+ "delete from t_shelf_link where f_item_id = ? and f_shelf_id = ? and f_db_uuid = ?",
+ (cur_book, cur_shelf, b_uuid))
+ print('Successfully removed a book to a shelf.')
+ self.books([cur_shelf])
+
+
+ def edit(self, opts=[]):
+ if len(opts) < 2:
+ print('Action must follow the form: edit <id> <new_title>')
+ sys.exit(1)
+
+ # Now check if book exists
+ if str(opts[0]).isdecimal():
+ book_infos_req = 'select f_id_item from t_item where f_id_item = ?'
+ else:
+ print('Book id not recognized. Are you sure it is a number?')
+ sys.exit(1)
+
+ cur = self.get_cursor()
+ book_infos = cur.execute(book_infos_req, (opts[0],)).fetchone()
+ if book_infos == None:
+ print('No such book {0}'.format(opts[0]))
+ sys.exit(1)
+ book_infos = dict(book_infos)
+
+ cur.execute(
+ "update t_item set f_title = ? where f_id_item = ?",
+ (opts[1], book_infos['f_id_item']))
+ print(cur.rowcount)
+ print('Title successfully updated to {0}.'.format(opts[1]))
+
+
+ def remote(self, opts=[]):
+ if len(opts) == 1 and opts[1] == 'files':
+ print('Here is your reader content:')
+ file_listing_helper(os.path.join(self.reader_mount_point, self.config['remote_library_path']))
+
+ else:
+ self.device()
+
+
+ def files(self, opts=[]):
+ print('Here is your library content:')
+ file_listing_helper(self.library_path)
+
+
+ def device(self, opts=[]):
+ import xml.etree.ElementTree as ET
+ adobe_info = os.path.join(self.reader_mount_point, '.adobe-digital-editions', 'device.xml')
+ adobe_sn = ''
+ adobe_name = ''
+ if os.path.isfile(adobe_info):
+ tree = ET.parse(adobe_info)
+ device_info = tree.find('./adept:deviceClass', {'adept': 'http://ns.adobe.com/adept'})
+ if device_info != None:
+ adobe_name = device_info.text
+ adobe_serial = tree.find('./adept:deviceSerial', {'adept': 'http://ns.adobe.com/adept'})
+ if adobe_serial != None:
+ adobe_sn = adobe_serial.text
+
+ device_name = bookeen_identify()
+ if device_name == None:
+ device_name = adobe_name
+ print(device_name)
+
+ cur = self.get_cursor()
+ b_ver = dict(cur.execute('select f_desc from t_version order by f_versiondate desc limit 1').fetchone())
+ print('Internal software: {0}'.format(b_ver['f_desc']))
+
+ print('Product ID: {0}'.format(read_file(os.path.join(self.reader_mount_point, 'system', 'product.dat'))))
+ print('Device serial: {0}'.format(read_file(os.path.join(self.reader_mount_point, 'system', 'sn.txt'))))
+ print('Adobe serial: {0}'.format(adobe_sn))
+
+
+ def init(self, opts=[]):
+ library_path = os.path.expanduser(opts[0])
+ if not os.path.isdir(library_path):
+ print('Please give us a valid path to your local library')
+ sys.exit(1)
+
+ config_path = os.path.join(library_path, '.eshelf.conf')
+ confparser = ConfigParser()
+
+ print('Please tell me where is your eReader mount point:')
+ reader_mount_point = input('?> ')
+
+ reader_db_path = os.path.join(reader_mount_point, 'system', 'library')
+ if not os.path.isfile(reader_db_path):
+ print('We cannot locate your eReader. Please check you have plugged it.')
+ sys.exit(1)
+
+ confparser['DEFAULT']['reader_mount_point'] = reader_mount_point
+
+ remote_physical_path = '/mnt/fat'
+ print('Default internal path of your eReader will be /mnt/fat. You are free to change it now or just press return:')
+ int_path = input('?> ')
+ if int_path != None and int_path != '':
+ confparser['DEFAULT']['remote_physical_path'] = int_path
+ else:
+ confparser['DEFAULT']['remote_physical_path'] = '/mnt/fat'
+
+ with open(config_path, 'w') as configfile:
+ confparser.write(configfile)
+
+
+if __name__ == "__main__":
+ predefined_actions = ['help', 'books', 'collections', 'add',
+ 'remove', 'edit', 'init', 'files',
+ 'remote', 'device']
+ plugins_files = []
+ plugins_path = os.path.join(xdg_config_home, 'eshelf', 'plugins')
+ if os.path.isdir(plugins_path):
+ if not plugins_path in sys.path:
+ sys.path.append(plugins_path)
+ plugins_files = [fname[:-3] for fname in os.listdir(plugins_path) if fname.endswith(".py")]
+ predefined_actions += plugins_files
+
+ parser = argparse.ArgumentParser(prog='eShelf')
+ parser.add_argument('command', help='What to do', choices=predefined_actions)
+ parser.add_argument('arguments', nargs='*', help='Supplementary options', default=[])
+ parser.add_argument('--path', '-p', default=os.getcwd(), help='Path to your library.')
+ args = parser.parse_args()
+
+ if args.command == 'help':
+ parser.print_help()
+ sys.exit(0)
+
+ library_path = args.path
+ if args.command == 'init':
+ library_path = None
+
+ my_library = Shelf(library_path)
+
+ if args.command in plugins_files:
+ loc_plugin = __import__(args.command)
+ the_library = loc_plugin.register(library_path, my_library.get_config())
+ if len(args.arguments) == 0:
+ loc_plugin.show_help()
+ exit(2)
+ the_command = args.arguments.pop(0)
+
+ else:
+ the_library = my_library
+ the_command = args.command
+
+ getattr(the_library, the_command)(args.arguments)
+
+ the_library.get_reader_db().close()
diff --git a/plugins/backup.py b/plugins/backup.py
new file mode 100644
index 0000000..5c35e3c
--- /dev/null
+++ b/plugins/backup.py
@@ -0,0 +1,75 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+
+import os
+import shutil
+from datetime import datetime
+
+class Backup:
+ def __init__(self, library_path, shelf_config):
+ self.config = shelf_config
+ self.library_path = os.path.expanduser(library_path)
+ self.reader_path = self.config['reader_path']
+ self.internal_path = '/mnt/fat'
+ if 'internal_path' in self.config:
+ self.internal_path = self.config['internal_path']
+
+
+
+ def pull(self, opts=[]):
+ reader_db_path = os.path.join(self.reader_path, 'system', 'library')
+ restore_path = os.path.join(self.library_path, '.eshelf-archive')
+
+ if not os.path.isdir(restore_path):
+ os.mkdir(restore_path, mode=0o755)
+
+ backupfile = shutil.copy(reader_db_path, os.path.join(restore_path, datetime.now().strftime('%Y%m%d%H%M%S')))
+ if os.path.isfile(backupfile):
+ print('Library backup saved as {0}'.format(backupfile))
+ else:
+ print('Library do not seem to have been saved. Please investigate.')
+
+
+ def push(self, opts=[]):
+ if len(opts) == 0:
+ print('Please provide a backup timestamp.')
+ return
+
+ reader_db_path = os.path.join(self.reader_path, 'system', 'library')
+ restore_path = os.path.join(self.library_path, '.eshelf-archive')
+
+ if not os.path.isdir(restore_path):
+ print('No backup directory. Nothing to do.')
+ return
+
+ backupfile = os.path.join(restore_path, opts[0])
+ if not os.path.isfile(backupfile):
+ print('No backup file found for {0}. Please investigate'.format(backupfile))
+ return
+
+ if not os.path.isfile(reader_db_path):
+ print('Something wrong with the eReader library file. Nothing to do.')
+ return
+
+ os.remove(reader_db_path)
+ shutil.copy(backupfile, reader_db_path)
+ print('Backup has been restored.')
+
+
+ def list(self, opts=[]):
+ restore_path = os.path.join(self.library_path, '.eshelf-archive')
+ if not os.path.isdir(restore_path):
+ print('No backup directory. Nothing to do.')
+ return
+
+ for backup in os.listdir(restore_path):
+ print(' - {0}'.format(backup))
+
+
+
+def register(library_path, config):
+ return Backup(library_path, config)
+
+
+def show_help():
+ print('Valid `backup` commands are [list, push, pull]')