mirror of
https://github.com/th3r00t/pyShelf.git
synced 2026-04-28 01:59:35 -04:00
Refactored for better packaging
This commit is contained in:
0
src/backend/__init__.py
Executable file
0
src/backend/__init__.py
Executable file
1
src/backend/data/shelf.json
Executable file
1
src/backend/data/shelf.json
Executable file
File diff suppressed because one or more lines are too long
0
src/backend/lib/__init__.py
Executable file
0
src/backend/lib/__init__.py
Executable file
35
src/backend/lib/api_hooks.py
Executable file
35
src/backend/lib/api_hooks.py
Executable file
@@ -0,0 +1,35 @@
|
||||
#!/usr/bin/python
|
||||
import sys
|
||||
|
||||
import requests
|
||||
|
||||
# sys.path.insert(1, 'lib/')
|
||||
|
||||
|
||||
class DuckDuckGo:
|
||||
"""duckduckgo related searching"""
|
||||
|
||||
def __init__(self):
|
||||
self.url = "https://api.duckduckgo.com/?q="
|
||||
|
||||
def image_result(self, query):
|
||||
"""
|
||||
Returns json containing url to image
|
||||
:param _key: &t=h_&iar=images&iax=images&ia=images&format=json&pretty=1
|
||||
"""
|
||||
_key = "&t=h_&iar=images&iax=images&ia=images&format=json&pretty=1"
|
||||
try:
|
||||
query = query.string
|
||||
except AttributeError:
|
||||
query = query
|
||||
search_result = requests.get(self.url + query + _key)
|
||||
try:
|
||||
image_result = search_result.json()["Image"]
|
||||
except ValueError:
|
||||
image_result = ""
|
||||
if search_result.status_code == 200 and image_result != "":
|
||||
image = requests.get(search_result.json()["Image"], stream=True)
|
||||
image.raw.decode_content = True
|
||||
return image.raw
|
||||
else:
|
||||
return False
|
||||
33
src/backend/lib/config.py
Executable file
33
src/backend/lib/config.py
Executable file
@@ -0,0 +1,33 @@
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
|
||||
|
||||
class Config:
|
||||
"""
|
||||
Main System Configuration
|
||||
"""
|
||||
|
||||
_fp = "config.json"
|
||||
print(os.path)
|
||||
|
||||
def __init__(self, root=os.path.abspath("../../")):
|
||||
_data = self.open_file(root)
|
||||
self.book_path = _data["BOOKPATH"]
|
||||
self.TITLE = _data["TITLE"]
|
||||
self.VERSION = _data["VERSION"]
|
||||
self.TITLE = self.TITLE + " ver " + self.VERSION
|
||||
self.book_shelf = _data["BOOKSHELF"]
|
||||
# self.catalogue_db = "data/catalogue.db"
|
||||
self.catalogue_db = root + "/" + _data["DATABASE"]
|
||||
self.file_array = [
|
||||
self.book_shelf,
|
||||
self.catalogue_db,
|
||||
]
|
||||
self.root = root
|
||||
self.auto_scan = True
|
||||
|
||||
def open_file(self, root):
|
||||
with open(root + "/" + self._fp, "r") as read_file:
|
||||
data = json.load(read_file)
|
||||
return data
|
||||
154
src/backend/lib/library.py
Executable file
154
src/backend/lib/library.py
Executable file
@@ -0,0 +1,154 @@
|
||||
#!/usr/bin/python
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import zipfile
|
||||
|
||||
from bs4 import BeautifulSoup
|
||||
from PIL import Image
|
||||
|
||||
from .api_hooks import DuckDuckGo
|
||||
from .config import Config
|
||||
from .storage import Storage
|
||||
|
||||
# config = Config()
|
||||
|
||||
|
||||
class Catalogue:
|
||||
"""Decodes and stores book information"""
|
||||
|
||||
"""Step One: filter_books"""
|
||||
|
||||
def __init__(self, config):
|
||||
self.file_list = []
|
||||
self.opf_regx = re.compile(r"\.opf")
|
||||
self.cover_regx = re.compile(r"\.jpg|\.jpeg|\.png|\.bmp|\.gif")
|
||||
self.html_regx = re.compile(r"\.html")
|
||||
self.root_dir = config.root
|
||||
self.book_folder = config.book_path
|
||||
self.book_shelf = config.book_shelf
|
||||
self._book_list_expanded = None
|
||||
self.books = None
|
||||
|
||||
def scan_folder(self, _path=None):
|
||||
if _path is not None:
|
||||
folder = _path
|
||||
elif os.path.isdir(self.root_dir + "/" + self.book_folder):
|
||||
folder = self.root_dir + "/" + self.book_folder
|
||||
else:
|
||||
folder = self.book_folder
|
||||
for f in os.listdir(folder):
|
||||
_path = os.path.abspath(folder + "/" + f)
|
||||
_is_dir = os.path.isdir(_path.strip() + "/")
|
||||
if _is_dir:
|
||||
self.file_list.append(self.scan_folder(_path))
|
||||
self.file_list.append(_path)
|
||||
|
||||
def filter_books(self):
|
||||
"""
|
||||
Scan book folder recursively for epub files
|
||||
filter_books(0) -> Catalogue.books
|
||||
filter_books(1) -> self.books[]
|
||||
:param ret: 0 -> create class property -> dump json
|
||||
:param ret: 1 -> create & return class property
|
||||
"""
|
||||
self.scan_folder()
|
||||
regx = re.compile(r"\.epub")
|
||||
try:
|
||||
self.books = list(filter(regx.search, filter(None, self.file_list)))
|
||||
except TypeError as e:
|
||||
print(e)
|
||||
self._book_list_expanded = {}
|
||||
with open(self.book_shelf, "w") as f:
|
||||
for book in self.books:
|
||||
self._book_list_expanded[book] = self.process_book(book)
|
||||
json.dump(self._book_list_expanded, f)
|
||||
return self._book_list_expanded
|
||||
|
||||
@staticmethod
|
||||
def process_book(book):
|
||||
"""Return dictionary of epub file contents"""
|
||||
book = zipfile.ZipFile(book, "r")
|
||||
details = {}
|
||||
with book as book_zip:
|
||||
details["files"] = []
|
||||
details["path"] = book.filename
|
||||
expanded = book_zip.infolist()
|
||||
regx = re.compile(r"\.opf|cover")
|
||||
for i in expanded:
|
||||
match = re.search(regx, i.filename)
|
||||
if match:
|
||||
# Returns zip file location of requested files
|
||||
details["files"].append(match.string)
|
||||
return details
|
||||
|
||||
def extract_metadata(self, book):
|
||||
"""
|
||||
Return extracted metadata and cover picture
|
||||
book['path'] == Full path to ebook file
|
||||
book['files'] == list of files from self.process_book(book)
|
||||
"""
|
||||
book_zip = zipfile.ZipFile(book["path"], "r")
|
||||
with book_zip as f:
|
||||
content = self.extract_content(book_zip, book)
|
||||
soup = BeautifulSoup(content, "lxml")
|
||||
title = soup.find("dc:title")
|
||||
if title is None:
|
||||
title = book["path"].split("/")[-1].rsplit(".", 1)[0]
|
||||
else:
|
||||
title = title.contents[0]
|
||||
author = soup.find("dc:creator")
|
||||
if author is not None:
|
||||
author = author.contents[0]
|
||||
try:
|
||||
cover = self.extract_cover_image(book_zip, book)
|
||||
except IndexError:
|
||||
# cover = self.extract_cover_html(book_zip, book)
|
||||
cover = DuckDuckGo().image_result(title)
|
||||
book_details = [title, author, cover, book["path"]]
|
||||
return book_details
|
||||
|
||||
def extract_content(self, book_zip, book):
|
||||
content = book_zip.open(list(filter(self.opf_regx.search, book["files"]))[0])
|
||||
return content
|
||||
|
||||
def extract_cover_html(self, book_zip, book):
|
||||
cover = book_zip.open(list(filter(self.html_regx.search, book["files"]))[0])
|
||||
return cover
|
||||
|
||||
def extract_cover_image(self, book_zip, book):
|
||||
cover = book_zip.open(list(filter(self.cover_regx.search, book["files"]))[0])
|
||||
try:
|
||||
cover = book_zip.read(cover.name)
|
||||
return cover
|
||||
except KeyError:
|
||||
return False
|
||||
|
||||
def compare_shelf_current(self):
|
||||
db = Storage()
|
||||
stored = db.book_paths_list()
|
||||
closed = db.close()
|
||||
if self.books is None:
|
||||
self.filter_books()
|
||||
on_disk, in_storage = [], []
|
||||
for _x in self.books:
|
||||
on_disk.append(_x)
|
||||
for _y in stored:
|
||||
in_storage.append(_y[0])
|
||||
a, b, = set(on_disk), set(in_storage)
|
||||
c = set.difference(a, b)
|
||||
return c
|
||||
|
||||
def import_books(self, list=None):
|
||||
book_list = self.compare_shelf_current()
|
||||
db = Storage()
|
||||
for book in book_list:
|
||||
book = self.process_book(book)
|
||||
extracted = self.extract_metadata(book)
|
||||
db.insert_book(extracted)
|
||||
inserted = db.commit()
|
||||
if inserted is not True:
|
||||
print(inserted)
|
||||
if input("Continue ? y/n") == "y":
|
||||
pass
|
||||
db.close()
|
||||
74
src/backend/lib/pyShelf.py
Executable file
74
src/backend/lib/pyShelf.py
Executable file
@@ -0,0 +1,74 @@
|
||||
#!/usr/bin/python
|
||||
import os
|
||||
|
||||
from .config import Config
|
||||
from .storage import Storage
|
||||
|
||||
# config = Config()
|
||||
# Storage = Storage()
|
||||
|
||||
|
||||
class InitFiles:
|
||||
"""First run file creation operations"""
|
||||
|
||||
def __init__(self, file_array):
|
||||
print("Begining creation of file structure")
|
||||
for _pointer in file_array:
|
||||
if not os.path.isfile(_pointer):
|
||||
self.CreateFile(_pointer)
|
||||
print("Concluded file creation")
|
||||
|
||||
def CreateFile(self, _pointer):
|
||||
"""Create the file"""
|
||||
if not os.path.isdir(os.path.split(_pointer)[0]):
|
||||
os.mkdir(os.path.split(_pointer)[0])
|
||||
f = open(_pointer, "w+")
|
||||
f.close()
|
||||
|
||||
|
||||
class BookDisplay:
|
||||
"""All functions related to displaying book information in the HTML UI"""
|
||||
|
||||
def __init__(self, **kwargs):
|
||||
"""
|
||||
Initialize class variables
|
||||
:return: None
|
||||
"""
|
||||
self.books_per_page = None
|
||||
self.current_page = 0
|
||||
self.thumbnail_size = [200, 300]
|
||||
self.thumbnail_scale = 1
|
||||
self.total_pages = None
|
||||
try:
|
||||
self.screen_size = kwargs["screen_size"]
|
||||
except Exception:
|
||||
self.screen_size = [900, 600]
|
||||
|
||||
def nextPage(self):
|
||||
"""
|
||||
Goto next book page
|
||||
:return: new current_page
|
||||
"""
|
||||
self.current_page += 1
|
||||
return self.current_page
|
||||
|
||||
def previousPage(self):
|
||||
"""
|
||||
Goto previous book page
|
||||
:return: new current_page
|
||||
"""
|
||||
self.current_page -= 1
|
||||
return self.current_page
|
||||
|
||||
def booksPerPage(self, screen_size):
|
||||
"""
|
||||
Set books per page
|
||||
|
||||
:param screen_size: Array containing x,y pixel sizes
|
||||
:return: self.books_per_page
|
||||
"""
|
||||
x = (self.thumbnail_size[0] * self.thumbnail_scale) + 10
|
||||
y = (self.thumbnail_size[1] * self.thumbnail_scale) + 10
|
||||
self.books_per_page = int(self.screen_size[0] // x) * int(
|
||||
self.screen_size[1] // y
|
||||
)
|
||||
86
src/backend/lib/storage.py
Executable file
86
src/backend/lib/storage.py
Executable file
@@ -0,0 +1,86 @@
|
||||
#!/usr/bin/python
|
||||
import sqlite3
|
||||
import sys
|
||||
|
||||
# sys.path.insert(1, '../')
|
||||
from .config import Config
|
||||
|
||||
# db_pointer = Config().catalogue_db
|
||||
|
||||
|
||||
class Storage:
|
||||
"""Contains all methods for system storage"""
|
||||
|
||||
def __init__(self, db_pointer=None):
|
||||
# Optionaly pass db_file to specify another db or for testing
|
||||
if db_pointer is None:
|
||||
db_pointer = Config().catalogue_db
|
||||
self.db_file = db_pointer
|
||||
self.database()
|
||||
# self.create_tables()
|
||||
|
||||
def database(self):
|
||||
"""Create database cursor"""
|
||||
try:
|
||||
self.db = sqlite3.connect(self.db_file)
|
||||
self.cursor = self.db.cursor()
|
||||
return True
|
||||
except Exception as e:
|
||||
print(self.db_file)
|
||||
print(e)
|
||||
return False
|
||||
|
||||
def create_tables(self):
|
||||
"""Create table structure"""
|
||||
q_check = "SELECT * FROM books"
|
||||
q_create = """CREATE TABLE books(title text, author text,
|
||||
categories text null, cover blob null, pages int null, progress int null,
|
||||
file_name text)"""
|
||||
try:
|
||||
self.cursor.execute(q_check)
|
||||
except sqlite3.OperationalError as e:
|
||||
self.cursor.execute(q_create)
|
||||
|
||||
def insert_book(self, book):
|
||||
"""
|
||||
Insert book in database
|
||||
:returns: True if succeeds False if not
|
||||
"""
|
||||
q_x = """SELECT title FROM books WHERE EXISTS(SELECT * from books WHERE `title` = ?)"""
|
||||
q = """INSERT INTO books (title, author, cover, progress, file_name, pages) values (?, ?, ?, 0, ?, 0)"""
|
||||
try:
|
||||
try:
|
||||
cover_image = book[2].data
|
||||
except:
|
||||
cover_image = book[2]
|
||||
x = self.cursor.execute(q_x, (book[0],))
|
||||
try:
|
||||
len(x.fetchone()) > 0
|
||||
except Exception:
|
||||
if not book[2]: # If cover image is missing unset entry
|
||||
cover_image = None
|
||||
self.cursor.execute(q, (book[0], book[1], cover_image, book[3]))
|
||||
return True
|
||||
except Exception as e:
|
||||
print(e)
|
||||
return False
|
||||
|
||||
def book_paths_list(self):
|
||||
q = """SELECT file_name FROM books"""
|
||||
x = self.cursor.execute(q)
|
||||
try:
|
||||
x = x.fetchall()
|
||||
except Exception:
|
||||
x = []
|
||||
return x
|
||||
|
||||
def commit(self):
|
||||
try:
|
||||
self.db.commit()
|
||||
return True
|
||||
except Exception as e:
|
||||
return e
|
||||
|
||||
def close(self):
|
||||
self.db.close()
|
||||
return True
|
||||
14
src/backend/pyShelf_ScanLibrary.py
Executable file
14
src/backend/pyShelf_ScanLibrary.py
Executable file
@@ -0,0 +1,14 @@
|
||||
#!/usr/bin/python
|
||||
import os
|
||||
import sys
|
||||
|
||||
from lib.config import Config
|
||||
from lib.library import Catalogue
|
||||
from lib.pyShelf import InitFiles
|
||||
|
||||
ROOT_DIR = os.path.abspath("../..")
|
||||
sys.path.append(ROOT_DIR)
|
||||
config = Config(ROOT_DIR) # Get configuration settings
|
||||
InitFiles(config.file_array) # Initialize file system
|
||||
Catalogue = Catalogue(config) # Open the Catalogue
|
||||
Catalogue.import_books() # Filter Your books
|
||||
10
src/backend/pyproject.toml
Executable file
10
src/backend/pyproject.toml
Executable file
@@ -0,0 +1,10 @@
|
||||
[tool.isort]
|
||||
force_grid_wrap = 0
|
||||
include_trailing_comma = true
|
||||
line_length = 88
|
||||
multi_line_output = 3
|
||||
use_parentheses = true
|
||||
# NOTE: the known_third_party setting is managed by
|
||||
# seed-isort-config and should not be modified directly.
|
||||
# Any changes made to this setting will be overwritten.
|
||||
known_third_party = ["PIL", "bs4", "requests"]
|
||||
0
src/backend/tests/__init__.py
Executable file
0
src/backend/tests/__init__.py
Executable file
16
src/backend/tests/config_test.py
Executable file
16
src/backend/tests/config_test.py
Executable file
@@ -0,0 +1,16 @@
|
||||
import os
|
||||
|
||||
from ..lib.config import Config
|
||||
|
||||
|
||||
class TestConfig:
|
||||
config = Config(os.path.abspath(os.path.curdir))
|
||||
|
||||
def test_book_dir(self):
|
||||
assert os.path.isdir(self.config.book_path)
|
||||
|
||||
def test_title(self):
|
||||
assert "pyShelf" in self.config.TITLE
|
||||
|
||||
def test_version(self):
|
||||
assert self.config.VERSION is not None
|
||||
34
src/backend/tests/library_test.py
Executable file
34
src/backend/tests/library_test.py
Executable file
@@ -0,0 +1,34 @@
|
||||
import json
|
||||
import os
|
||||
|
||||
from ..lib.config import Config
|
||||
from ..lib.library import Catalogue
|
||||
|
||||
|
||||
class Test_Config(Config):
|
||||
def __init__(self):
|
||||
Config.__init__(self, "config.json")
|
||||
_data = self.open_file()
|
||||
|
||||
def open_file(self, root="config.json"):
|
||||
with open("config.json") as read_file:
|
||||
data = json.load(read_file)
|
||||
return data
|
||||
|
||||
|
||||
class Test_Catalogue(Catalogue):
|
||||
def __init__(self):
|
||||
Catalogue.__init__(self, root=os.path.abspath("."))
|
||||
|
||||
def filter_books(self):
|
||||
self.book_shelf = "app/" + self.book_shelf
|
||||
return super().filter_books()
|
||||
|
||||
|
||||
class TestCatalogue:
|
||||
root = os.path.abspath(os.path.curdir)
|
||||
config = Test_Config()
|
||||
|
||||
def test_filter_books(self):
|
||||
book_list = Test_Catalogue().filter_books()
|
||||
assert len(book_list) > 0
|
||||
Reference in New Issue
Block a user