124 lines
4.6 KiB
Python
Executable File
124 lines
4.6 KiB
Python
Executable File
from app import db
|
|
import uuid
|
|
from datetime import datetime, timedelta
|
|
import jwt
|
|
from config import Config
|
|
import bcrypt
|
|
|
|
class User(db.Model):
|
|
__tablename__ = 'user'
|
|
|
|
id = db.Column(db.String(36), primary_key=True, default=lambda: str(uuid.uuid4()))
|
|
username = db.Column(db.String(64), index=True, unique=True, nullable=False)
|
|
display_name = db.Column(db.String(120))
|
|
email = db.Column(db.String(120), index=True, unique=True, nullable=False)
|
|
password_hash = db.Column(db.String(128), nullable=False)
|
|
role = db.Column(db.String(20), default='user')
|
|
|
|
print_jobs = db.relationship('PrintJob', backref='user', lazy='dynamic', cascade='all, delete-orphan')
|
|
sessions = db.relationship('Session', backref='user', lazy='dynamic', cascade='all, delete-orphan')
|
|
|
|
def set_password(self, password):
|
|
"""Hash and set the user's password"""
|
|
password_bytes = password.encode('utf-8')
|
|
salt = bcrypt.gensalt()
|
|
self.password_hash = bcrypt.hashpw(password_bytes, salt).decode('utf-8')
|
|
|
|
def check_password(self, password):
|
|
"""Check if the provided password matches the stored hash"""
|
|
password_bytes = password.encode('utf-8')
|
|
stored_hash = self.password_hash.encode('utf-8')
|
|
return bcrypt.checkpw(password_bytes, stored_hash)
|
|
|
|
def generate_token(self):
|
|
"""Generate a JWT token for this user"""
|
|
payload = {
|
|
'user_id': self.id,
|
|
'username': self.username,
|
|
'email': self.email,
|
|
'role': self.role,
|
|
'exp': datetime.utcnow() + timedelta(seconds=Config.JWT_ACCESS_TOKEN_EXPIRES)
|
|
}
|
|
return jwt.encode(payload, Config.JWT_SECRET, algorithm='HS256')
|
|
|
|
@staticmethod
|
|
def verify_token(token):
|
|
"""Verify and decode a JWT token"""
|
|
try:
|
|
payload = jwt.decode(token, Config.JWT_SECRET, algorithms=['HS256'])
|
|
return payload
|
|
except:
|
|
return None
|
|
|
|
|
|
class Session(db.Model):
|
|
__tablename__ = 'session'
|
|
|
|
id = db.Column(db.String(36), primary_key=True, default=lambda: str(uuid.uuid4()))
|
|
user_id = db.Column(db.String(36), db.ForeignKey('user.id', ondelete='CASCADE'), nullable=False)
|
|
expires_at = db.Column(db.Integer, nullable=False)
|
|
|
|
|
|
class Printer(db.Model):
|
|
__tablename__ = 'printer'
|
|
|
|
id = db.Column(db.String(36), primary_key=True, default=lambda: str(uuid.uuid4()))
|
|
name = db.Column(db.String(120), nullable=False)
|
|
description = db.Column(db.Text, nullable=False)
|
|
status = db.Column(db.Integer, nullable=False, default=0) # 0: OPERATIONAL, 1: OUT_OF_ORDER
|
|
|
|
print_jobs = db.relationship('PrintJob', backref='printer', lazy='dynamic', cascade='all, delete-orphan')
|
|
|
|
|
|
class PrintJob(db.Model):
|
|
__tablename__ = 'printJob'
|
|
|
|
id = db.Column(db.String(36), primary_key=True, default=lambda: str(uuid.uuid4()))
|
|
printer_id = db.Column(db.String(36), db.ForeignKey('printer.id', ondelete='CASCADE'), nullable=False)
|
|
user_id = db.Column(db.String(36), db.ForeignKey('user.id', ondelete='CASCADE'), nullable=False)
|
|
start_at = db.Column(db.DateTime, nullable=False, default=datetime.utcnow)
|
|
duration_in_minutes = db.Column(db.Integer, nullable=False)
|
|
comments = db.Column(db.Text)
|
|
aborted = db.Column(db.Boolean, nullable=False, default=False)
|
|
abort_reason = db.Column(db.Text)
|
|
|
|
def get_end_time(self):
|
|
return self.start_at + timedelta(minutes=self.duration_in_minutes)
|
|
|
|
def is_active(self):
|
|
now = datetime.utcnow()
|
|
return (not self.aborted and
|
|
self.start_at <= now and
|
|
now < self.get_end_time())
|
|
|
|
def get_remaining_time(self):
|
|
if self.aborted:
|
|
return 0
|
|
|
|
now = datetime.utcnow()
|
|
if now < self.start_at:
|
|
# Job hasn't started yet
|
|
return self.duration_in_minutes * 60
|
|
|
|
end_time = self.get_end_time()
|
|
if now >= end_time:
|
|
# Job has ended
|
|
return 0
|
|
|
|
# Job is ongoing
|
|
remaining_seconds = (end_time - now).total_seconds()
|
|
return int(remaining_seconds)
|
|
|
|
def to_dict(self):
|
|
return {
|
|
'id': self.id,
|
|
'printer_id': self.printer_id,
|
|
'user_id': self.user_id,
|
|
'start_at': self.start_at.isoformat(),
|
|
'duration_in_minutes': self.duration_in_minutes,
|
|
'comments': self.comments,
|
|
'aborted': self.aborted,
|
|
'abort_reason': self.abort_reason,
|
|
'remaining_time': self.get_remaining_time(),
|
|
'is_active': self.is_active()
|
|
} |