123 lines
4.4 KiB
Python
123 lines
4.4 KiB
Python
"""
|
|
Manage all of the inputs from a web source
|
|
"""
|
|
|
|
import os
|
|
from werkzeug.utils import secure_filename
|
|
from task import TextTask, ImageTask, CutTask
|
|
|
|
|
|
class Web:
|
|
"""Web is the class that gets all of the information from web calls
|
|
( API and Web page ) and provides checks before sending stuff to printing"""
|
|
|
|
def __init__(self, app, print_queue):
|
|
super(Web).__init__()
|
|
self.print_queue = print_queue
|
|
self.app = app
|
|
|
|
def print_sms(self, texte, sign: str) -> bool:
|
|
"""
|
|
Get text and a signature, prints the text and cuts after that.
|
|
"""
|
|
self.app.logger.debug("Printing : " + str(texte) + " from " + str(sign))
|
|
|
|
try:
|
|
# We create two new tasks and add them directly to the queue
|
|
# TODO: this might need to be improved because
|
|
# !! there is no garantee !! that both the SMS task and the Cut task
|
|
# are added back to back, another task could be
|
|
# inserted between the two.
|
|
sms = self.print_queue.enqueue(TextTask(content=texte, signature=sign))
|
|
cut = self.print_queue.enqueue(CutTask())
|
|
except Exception as e:
|
|
self.app.logger.error(e)
|
|
raise RuntimeError("Could not add SMS to queue, " + str(e)) from e
|
|
self.app.logger.info("Added two new tasks at position %s and %s", sms, cut)
|
|
return True
|
|
|
|
def print_image(self, image, sign: str) -> bool:
|
|
"""
|
|
Get an image and a signature, prints the image and cuts after that.
|
|
"""
|
|
try:
|
|
file_uploaded = self.upload_file(image)
|
|
except Exception as e:
|
|
self.app.logger.error(e)
|
|
raise RuntimeError("Could not upload file : " + str(e)) from e
|
|
|
|
if file_uploaded:
|
|
self.app.logger.debug("File has been uploaded, printing...")
|
|
try:
|
|
img = self.print_queue.enqueue(
|
|
ImageTask(
|
|
os.path.join(
|
|
self.app.config["UPLOAD_FOLDER"],
|
|
secure_filename(image.filename),
|
|
),
|
|
signature=sign,
|
|
process=True,
|
|
)
|
|
)
|
|
|
|
cut = self.print_queue.enqueue(CutTask())
|
|
except Exception as e:
|
|
raise RuntimeError("Could not add IMG to queue" + str(e)) from e
|
|
|
|
self.app.logger.info("Added two new tasks at position %s and %s", img, cut)
|
|
|
|
return True
|
|
|
|
# def login(self, username: str, password: str) -> bool:
|
|
# """Not implemented"""
|
|
# return
|
|
|
|
# def logout(self, username: str, password: str) -> bool:
|
|
# """Not implemented"""
|
|
# return
|
|
|
|
def allowed_file(self, filename) -> bool:
|
|
"""
|
|
Check if the file extension is allowed
|
|
"""
|
|
self.app.logger.debug("Checking if the file extension is allowed")
|
|
return (
|
|
"." in filename
|
|
and filename.rsplit(".", 1)[1].lower()
|
|
in self.app.config["ALLOWED_EXTENSIONS"]
|
|
)
|
|
|
|
def upload_file(self, image) -> bool:
|
|
"""
|
|
Save the file after executing checks on it
|
|
"""
|
|
self.app.logger.debug("Validating file")
|
|
if not image is None or not image == "":
|
|
if self.allowed_file(image.filename):
|
|
filename = secure_filename(image.filename)
|
|
self.app.logger.debug("File valid")
|
|
try:
|
|
image.save(os.path.join(self.app.config["UPLOAD_FOLDER"], filename))
|
|
except OSError as e:
|
|
self.app.logger.error("Could not save file %s", e)
|
|
raise RuntimeError("An OS error occured while uploading this file : " + str(e)) from e
|
|
|
|
self.app.logger.debug(
|
|
"File saved to "
|
|
+ str(os.path.join(self.app.config["UPLOAD_FOLDER"], filename))
|
|
)
|
|
return True
|
|
|
|
self.app.logger.error(
|
|
"Could not save file because the filename is forbidden"
|
|
)
|
|
raise RuntimeError("This file type is forbidden.")
|
|
|
|
def get_queue_state(self):
|
|
"""Return current queue state"""
|
|
return self.print_queue.get_queue_state()
|
|
|
|
def get_queue_completed(self):
|
|
"""Return completed queue elements"""
|
|
return self.print_queue.get_queue_completed()
|