forked from LiveCarta/BookConverter
208 lines
8.5 KiB
Python
208 lines
8.5 KiB
Python
import os
|
|
import json
|
|
import codecs
|
|
import logging
|
|
import pathlib
|
|
from typing import List, Dict, Union
|
|
from abc import abstractmethod, ABCMeta
|
|
|
|
from src.livecarta_config import LiveCartaConfig
|
|
from src.util.helpers import BookLogger, BookStatusWrapper
|
|
|
|
|
|
class BookSolver:
|
|
"""
|
|
This is Main Abstract class for solving a task of a book conversion
|
|
Having an id of coming book, gets book from server, runs conversion.
|
|
In parallel, it updates status of a book conversion on admin panel.
|
|
Finally, sends result to server.
|
|
Result is a json, JSON schema in book_schema.json
|
|
"""
|
|
|
|
__metaclass__ = ABCMeta
|
|
|
|
def __init__(self, book_id: int = 0, access=None, main_logger=None):
|
|
self.book_type = None
|
|
self.book_id = book_id
|
|
self.access = access
|
|
self.preset_path = None
|
|
self.book_path = None # path to book file, appears after downloading from server
|
|
self.book_output_path = None # path to json file
|
|
self.book_logger = BookLogger(name=f"{__name__}_{self.book_id}")
|
|
self.book_logger.configure_book_logger(book_id=book_id)
|
|
self.status_wrapper = BookStatusWrapper(
|
|
access, self.book_logger, book_id)
|
|
|
|
assert LiveCartaConfig.NUM_SUPPORTED_LEVELS == len(LiveCartaConfig.SUPPORTED_HEADER_TAGS), \
|
|
"Length of headers doesn't match allowed levels."
|
|
|
|
def save_file(self, content: bytes, path_to_save: str, file_type: str) -> str:
|
|
"""
|
|
Function saves binary content of file to folder(path_to_save)
|
|
Parameters
|
|
----------
|
|
|
|
content: bytes str
|
|
binary content of the file
|
|
path_to_save: str
|
|
path to the folder
|
|
file_type: str
|
|
|
|
Returns
|
|
----------
|
|
file_path: str
|
|
path to file on local
|
|
"""
|
|
folder_path: str = os.path.dirname(
|
|
os.path.dirname(os.path.abspath(__file__)))
|
|
folder_path = os.path.join(
|
|
folder_path, path_to_save)
|
|
pathlib.Path(folder_path).mkdir(parents=True, exist_ok=True)
|
|
|
|
file_path: str = os.path.join(
|
|
folder_path, f"{self.book_id}.{file_type}")
|
|
try:
|
|
with open(file_path, "wb+") as file:
|
|
file.write(content)
|
|
self.book_logger.log(
|
|
f"File was saved to folder: {folder_path}.")
|
|
except Exception as exc:
|
|
self.book_logger.log(
|
|
f"Error in writing {self.book_type} file.", logging.ERROR)
|
|
self.book_logger.log_error_to_main_log()
|
|
raise exc
|
|
return file_path
|
|
|
|
def get_preset_file(self):
|
|
"""Method for getting and saving preset from server"""
|
|
try:
|
|
self.book_logger.log(f"Start receiving \033[4mpreset\033[0m file from server. URL:"
|
|
f" {self.access.url}/doc-convert/{self.book_id}/presets")
|
|
content = self.access.get_file(
|
|
file_path=f"{self.access.url}/doc-convert/{self.book_id}/presets")
|
|
self.book_logger.log("\033[4mPreset\033[0m file was received from server.")
|
|
self.preset_path = pathlib.Path(
|
|
str(self.save_file(content, path_to_save="preset", file_type="json")))
|
|
except FileNotFoundError as f_err:
|
|
self.book_logger.log(
|
|
"Can't get preset file from server.", logging.ERROR)
|
|
self.book_logger.log_error_to_main_log()
|
|
raise f_err
|
|
except Exception as exc:
|
|
raise exc
|
|
|
|
def get_book_file(self):
|
|
"""Method for getting and saving book from server"""
|
|
try:
|
|
self.book_logger.log(f"Start receiving \033[4mbook\033[0m file from server. URL:"
|
|
f" {self.access.url}/doc-convert/{self.book_id}/file")
|
|
content = self.access.get_file(
|
|
file_path=f"{self.access.url}/doc-convert/{self.book_id}/file")
|
|
self.book_logger.log("\033[4mBook\033[0m file was received from server.")
|
|
self.book_path = pathlib.Path(self.save_file(
|
|
content, path_to_save=f"books/{self.book_type}", file_type=self.book_type))
|
|
except FileNotFoundError as f_err:
|
|
self.book_logger.log(
|
|
"Can't get book file from server.", logging.ERROR)
|
|
self.book_logger.log_error_to_main_log()
|
|
raise f_err
|
|
except Exception as exc:
|
|
raise exc
|
|
|
|
def check_output_directory(self):
|
|
if self.book_output_path is None:
|
|
folder_path = os.path.dirname(
|
|
os.path.dirname(os.path.abspath(__file__)))
|
|
output_path = os.path.join(
|
|
folder_path, f"books/json/{self.book_id}.json")
|
|
self.book_output_path = output_path
|
|
|
|
self.book_output_path = pathlib.Path(self.book_output_path)
|
|
self.book_logger.log(f"Output file path: {self.book_output_path}")
|
|
|
|
pathlib.Path(self.book_output_path).parent.mkdir(
|
|
parents=True, exist_ok=True)
|
|
self.book_output_path.touch(exist_ok=True)
|
|
|
|
def write_to_json(self, content: Dict[str, List[Dict[str, Union[List, str]]]]):
|
|
self.check_output_directory()
|
|
try:
|
|
with codecs.open(self.book_output_path, "w", encoding="utf-8") as f:
|
|
json.dump(content, f, ensure_ascii=False)
|
|
self.book_logger.log(
|
|
f"Data has been saved to .json file: {self.book_output_path}")
|
|
except Exception as exc:
|
|
self.book_logger.log(
|
|
"Error has occurred while writing .json file." + str(exc), logging.ERROR)
|
|
|
|
def send_json_content_to_server(self, content: Dict[str, List[Dict[str, Union[List, str]]]]):
|
|
"""Function sends json_content to site"""
|
|
try:
|
|
self.access.send_book(self.book_id, content)
|
|
self.book_logger.log(f"JSON data has been sent to server.")
|
|
except Exception as exc:
|
|
self.book_logger.log(
|
|
"Error has occurred while sending json content.", logging.ERROR)
|
|
self.book_logger.log_error_to_main_log()
|
|
self.status_wrapper.set_error()
|
|
raise exc
|
|
|
|
@abstractmethod
|
|
def get_converted_book(self) -> Dict[str, List[Dict[str, Union[List, str]]]]:
|
|
self.book_logger.log("Beginning of processing .json output.")
|
|
self.status_wrapper.set_generating()
|
|
return {}
|
|
|
|
def conversion(self):
|
|
"""
|
|
Function
|
|
- with downloading book from server
|
|
- with sending to server
|
|
|
|
"""
|
|
try:
|
|
self.get_preset_file()
|
|
self.get_book_file()
|
|
self.book_logger.log(
|
|
f"Beginning of conversion from .{self.book_type} to .json.")
|
|
self.status_wrapper.set_processing()
|
|
content_dict: Dict[str, List[Dict[Union[str, List]]]] = self.get_converted_book()
|
|
[os.remove(path) for path in [self.book_path,
|
|
# self.preset_path
|
|
]]
|
|
self.book_logger.log("Beginning of processing .json output.")
|
|
self.status_wrapper.set_generating()
|
|
self.write_to_json(content_dict)
|
|
self.send_json_content_to_server(content_dict)
|
|
self.book_logger.log(
|
|
f"End of the conversion to LiveCarta format. Check {self.book_output_path}.")
|
|
except Exception as exc:
|
|
self.status_wrapper.set_error()
|
|
self.book_logger.log(
|
|
"Error has occurred while conversion.", logging.ERROR)
|
|
self.book_logger.log_error_to_main_log(str(exc))
|
|
raise exc
|
|
|
|
def conversion_local(self, file_path: str):
|
|
"""
|
|
Function
|
|
- without downloading book from server (local)
|
|
- with sending to server
|
|
|
|
"""
|
|
try:
|
|
self.book_logger.log(
|
|
f"Data has been downloaded from {file_path} file")
|
|
self.status_wrapper.set_processing()
|
|
with codecs.open(file_path, "r", encoding="utf-8") as f_json:
|
|
content_dict = json.load(f_json)
|
|
self.book_logger.log("Beginning of processing .json output.")
|
|
self.status_wrapper.set_generating()
|
|
self.send_json_content_to_server(content_dict)
|
|
self.book_logger.log(f"Sent a file to server. Check LiveCarta.")
|
|
except Exception as exc:
|
|
self.status_wrapper.set_error()
|
|
self.book_logger.log(
|
|
"Error has occurred while reading json file." + str(exc), logging.ERROR)
|
|
self.book_logger.log_error_to_main_log(str(exc))
|