diff options
author | 2017-05-24 15:04:28 +0200 | |
---|---|---|
committer | 2017-05-24 15:04:28 +0200 | |
commit | 67d03e7daa8fee0a9dff70872d52f12b8da8827c (patch) | |
tree | 0a625745e0b27f698d83f9f11ea53281678b073e | |
download | eshelf-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.py | 435 | ||||
-rw-r--r-- | plugins/backup.py | 75 |
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]') |