Remove deprecated backend files and documentation, including Docker configurations, environment variables, and various scripts, to streamline the project structure and eliminate unused components.
This commit is contained in:
180
backend/app/models.py
Normal file
180
backend/app/models.py
Normal file
@@ -0,0 +1,180 @@
|
||||
import os
|
||||
import logging
|
||||
from datetime import datetime
|
||||
from typing import Optional, List
|
||||
|
||||
from sqlalchemy import create_engine, Column, Integer, String, Boolean, DateTime, ForeignKey, Float
|
||||
from sqlalchemy.ext.declarative import declarative_base
|
||||
from sqlalchemy.orm import relationship, sessionmaker, Session
|
||||
import bcrypt
|
||||
|
||||
from config.settings import DATABASE_PATH, ensure_database_directory
|
||||
from utils.logging_config import get_logger
|
||||
|
||||
Base = declarative_base()
|
||||
logger = get_logger("app")
|
||||
|
||||
class User(Base):
|
||||
__tablename__ = "users"
|
||||
|
||||
id = Column(Integer, primary_key=True)
|
||||
email = Column(String(120), unique=True, nullable=False)
|
||||
password_hash = Column(String(128), nullable=False)
|
||||
name = Column(String(100), nullable=False)
|
||||
role = Column(String(20), default="user") # "admin" oder "user"
|
||||
created_at = Column(DateTime, default=datetime.now)
|
||||
|
||||
jobs = relationship("Job", back_populates="user", cascade="all, delete-orphan")
|
||||
|
||||
def set_password(self, password: str) -> None:
|
||||
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: str) -> bool:
|
||||
password_bytes = password.encode('utf-8')
|
||||
hash_bytes = self.password_hash.encode('utf-8')
|
||||
return bcrypt.checkpw(password_bytes, hash_bytes)
|
||||
|
||||
def is_admin(self) -> bool:
|
||||
return self.role == "admin"
|
||||
|
||||
def to_dict(self) -> dict:
|
||||
return {
|
||||
"id": self.id,
|
||||
"email": self.email,
|
||||
"name": self.name,
|
||||
"role": self.role,
|
||||
"created_at": self.created_at.isoformat() if self.created_at else None
|
||||
}
|
||||
|
||||
|
||||
class Printer(Base):
|
||||
__tablename__ = "printers"
|
||||
|
||||
id = Column(Integer, primary_key=True)
|
||||
name = Column(String(100), nullable=False)
|
||||
location = Column(String(100))
|
||||
mac_address = Column(String(50), nullable=False, unique=True)
|
||||
plug_ip = Column(String(50), nullable=False)
|
||||
plug_username = Column(String(100), nullable=False)
|
||||
plug_password = Column(String(100), nullable=False)
|
||||
active = Column(Boolean, default=True)
|
||||
created_at = Column(DateTime, default=datetime.now)
|
||||
|
||||
jobs = relationship("Job", back_populates="printer", cascade="all, delete-orphan")
|
||||
|
||||
def to_dict(self) -> dict:
|
||||
return {
|
||||
"id": self.id,
|
||||
"name": self.name,
|
||||
"location": self.location,
|
||||
"mac_address": self.mac_address,
|
||||
"plug_ip": self.plug_ip,
|
||||
"active": self.active,
|
||||
"created_at": self.created_at.isoformat() if self.created_at else None
|
||||
}
|
||||
|
||||
|
||||
class Job(Base):
|
||||
__tablename__ = "jobs"
|
||||
|
||||
id = Column(Integer, primary_key=True)
|
||||
title = Column(String(200), nullable=False)
|
||||
user_id = Column(Integer, ForeignKey("users.id"), nullable=False)
|
||||
printer_id = Column(Integer, ForeignKey("printers.id"), nullable=False)
|
||||
start_time = Column(DateTime, nullable=False)
|
||||
end_time = Column(DateTime, nullable=False)
|
||||
actual_end_time = Column(DateTime)
|
||||
status = Column(String(20), default="scheduled") # scheduled, active, completed, aborted
|
||||
created_at = Column(DateTime, default=datetime.now)
|
||||
notes = Column(String(500))
|
||||
material_used = Column(Float) # in Gramm
|
||||
|
||||
user = relationship("User", back_populates="jobs")
|
||||
printer = relationship("Printer", back_populates="jobs")
|
||||
|
||||
def to_dict(self) -> dict:
|
||||
return {
|
||||
"id": self.id,
|
||||
"title": self.title,
|
||||
"user_id": self.user_id,
|
||||
"printer_id": self.printer_id,
|
||||
"start_time": self.start_time.isoformat() if self.start_time else None,
|
||||
"end_time": self.end_time.isoformat() if self.end_time else None,
|
||||
"actual_end_time": self.actual_end_time.isoformat() if self.actual_end_time else None,
|
||||
"status": self.status,
|
||||
"created_at": self.created_at.isoformat() if self.created_at else None,
|
||||
"notes": self.notes,
|
||||
"material_used": self.material_used,
|
||||
"user": self.user.to_dict() if self.user else None,
|
||||
"printer": self.printer.to_dict() if self.printer else None
|
||||
}
|
||||
|
||||
|
||||
class Stats(Base):
|
||||
__tablename__ = "stats"
|
||||
|
||||
id = Column(Integer, primary_key=True)
|
||||
total_print_time = Column(Integer, default=0) # in Sekunden
|
||||
total_jobs_completed = Column(Integer, default=0)
|
||||
total_material_used = Column(Float, default=0.0) # in Gramm
|
||||
last_updated = Column(DateTime, default=datetime.now)
|
||||
|
||||
|
||||
def init_db() -> None:
|
||||
"""Initialisiert die Datenbank und erstellt alle Tabellen."""
|
||||
ensure_database_directory()
|
||||
engine = create_engine(f"sqlite:///{DATABASE_PATH}")
|
||||
Base.metadata.create_all(engine)
|
||||
logger.info("Datenbank initialisiert.")
|
||||
|
||||
|
||||
def create_initial_admin(email: str, password: str, name: str) -> bool:
|
||||
"""
|
||||
Erstellt einen initialen Admin-Benutzer, falls die Datenbank leer ist.
|
||||
|
||||
Args:
|
||||
email: E-Mail-Adresse des Admins
|
||||
password: Passwort des Admins
|
||||
name: Name des Admins
|
||||
|
||||
Returns:
|
||||
bool: True, wenn der Admin erstellt wurde, False sonst
|
||||
"""
|
||||
engine = create_engine(f"sqlite:///{DATABASE_PATH}")
|
||||
Session_class = sessionmaker(bind=engine)
|
||||
session = Session_class()
|
||||
|
||||
# Prüfen, ob bereits Benutzer existieren
|
||||
user_count = session.query(User).count()
|
||||
if user_count > 0:
|
||||
session.close()
|
||||
return False
|
||||
|
||||
# Ersten Admin anlegen
|
||||
admin = User(
|
||||
email=email,
|
||||
name=name,
|
||||
role="admin"
|
||||
)
|
||||
admin.set_password(password)
|
||||
|
||||
session.add(admin)
|
||||
session.commit()
|
||||
|
||||
# Statistik-Eintrag anlegen
|
||||
stats = Stats()
|
||||
session.add(stats)
|
||||
session.commit()
|
||||
|
||||
session.close()
|
||||
logger.info(f"Initialer Admin-Benutzer {email} wurde angelegt.")
|
||||
return True
|
||||
|
||||
|
||||
def get_db_session() -> Session:
|
||||
"""Gibt eine neue Datenbank-Session zurück."""
|
||||
engine = create_engine(f"sqlite:///{DATABASE_PATH}")
|
||||
Session_class = sessionmaker(bind=engine)
|
||||
return Session_class()
|
Reference in New Issue
Block a user