Files
IOT_application/Machine-Backend/app/models/models.py
2025-11-26 20:01:52 +05:30

554 lines
24 KiB
Python

from app import db
import datetime
import time
import json
from werkzeug.security import generate_password_hash, check_password_hash
class RefillerMachine(db.Model):
__tablename__ = 'refiller_machines'
id = db.Column(db.Integer, primary_key=True)
refiller_id = db.Column(db.Integer, db.ForeignKey('users.id', ondelete='CASCADE'), nullable=False)
machine_id = db.Column(db.String(10), db.ForeignKey('machines.machine_id', ondelete='CASCADE'), nullable=False)
assigned_at = db.Column(db.DateTime, default=datetime.datetime.utcnow)
assigned_by = db.Column(db.Integer, db.ForeignKey('users.id'), nullable=True)
# Relationships
assigner = db.relationship('User', foreign_keys=[assigned_by], backref='machine_assignments_made')
# Unique constraint: one refiller can't be assigned to same machine twice
__table_args__ = (
db.UniqueConstraint('refiller_id', 'machine_id', name='unique_refiller_machine'),
)
def to_dict(self):
return {
'id': self.id,
'refiller_id': self.refiller_id,
'machine_id': self.machine_id,
'assigned_at': self.assigned_at.strftime("%Y-%m-%d %H:%M:%S") if self.assigned_at else None,
'assigned_by': self.assigned_by,
'assigned_by_username': self.assigner.username if self.assigner else None
}
# Machine Model - UPDATED
class Machine(db.Model):
__tablename__ = 'machines'
id = db.Column(db.Integer, primary_key=True)
machine_id = db.Column(db.String(10), unique=True, nullable=False)
machine_model = db.Column(db.String(100), nullable=False)
machine_type = db.Column(db.String(100), nullable=False)
client_id = db.Column(db.Integer, db.ForeignKey('users.id'), nullable=False)
client_name = db.Column(db.String(100), nullable=True)
branch_id = db.Column(db.String(10), nullable=False)
branch_name = db.Column(db.String(100), nullable=False)
operation_status = db.Column(db.String(50), nullable=False)
connection_status = db.Column(db.String(50), nullable=False)
created_on = db.Column(db.String(20), nullable=False)
password = db.Column(db.String(128), nullable=False)
created_by = db.Column(db.Integer, db.ForeignKey('users.id'), nullable=True)
# Relationships
slots = db.relationship('VendingSlot', backref='machine', lazy=True)
client = db.relationship('User', foreign_keys=[client_id], backref='client_machines')
creator = db.relationship('User', foreign_keys=[created_by], backref='created_machines')
# ⭐ NEW: Many-to-many relationship with Refillers through RefillerMachine
assigned_refillers = db.relationship(
'User',
secondary='refiller_machines',
primaryjoin='Machine.machine_id == RefillerMachine.machine_id',
secondaryjoin='and_(User.id == RefillerMachine.refiller_id, User.roles == "Refiller")',
backref='assigned_machines_rel',
viewonly=True
)
def to_dict(self):
# Get assigned refillers
refiller_assignments = RefillerMachine.query.filter_by(machine_id=self.machine_id).all()
assigned_refillers = []
for assignment in refiller_assignments:
refiller = User.query.get(assignment.refiller_id)
if refiller:
assigned_refillers.append({
'id': refiller.id,
'username': refiller.username,
'email': refiller.email,
'assigned_at': assignment.assigned_at.strftime("%Y-%m-%d %H:%M:%S") if assignment.assigned_at else None
})
return {
'id': self.id,
'machine_id': self.machine_id,
'machine_model': self.machine_model,
'machine_type': self.machine_type,
'client_id': self.client_id,
'client_name': self.client.username if self.client else self.client_name,
'branch_id': self.branch_id,
'branch_name': self.branch_name,
'operation_status': self.operation_status,
'connection_status': self.connection_status,
'created_on': self.created_on,
'password': self.password,
'created_by': self.created_by,
'created_by_username': self.creator.username if self.creator else None,
'assigned_refillers': assigned_refillers # ⭐ NEW
}
def set_password(self, password):
self.password = generate_password_hash(password)
def check_password(self, password):
return check_password_hash(self.password, password)
# User Model - UPDATED WITH BOTH CLIENT AND MACHINE ASSIGNMENTS
class User(db.Model):
__tablename__ = 'users'
id = db.Column(db.Integer, primary_key=True)
user_id = db.Column(db.String(50), unique=True, nullable=False)
username = db.Column(db.String(50), nullable=False)
email = db.Column(db.String(100), nullable=False, unique=True)
contact = db.Column(db.String(20), nullable=False)
roles = db.Column(db.String(50), nullable=False)
user_status = db.Column(db.String(50), nullable=False)
password = db.Column(db.String(255), nullable=False)
created_by = db.Column(db.Integer, db.ForeignKey('users.id'), nullable=True)
assigned_to = db.Column(db.Integer, db.ForeignKey('users.id'), nullable=True) # Assigned to Client
# File storage fields
photo = db.Column(db.String(255), nullable=True)
company_logo = db.Column(db.String(255), nullable=True)
documents = db.Column(db.Text, nullable=True)
# Timestamps
created_at = db.Column(db.DateTime, default=datetime.datetime.utcnow)
updated_at = db.Column(db.DateTime, default=datetime.datetime.utcnow, onupdate=datetime.datetime.utcnow)
# Relationships
creator = db.relationship('User', remote_side=[id], backref='created_users', foreign_keys=[created_by])
assigned_client = db.relationship('User', remote_side=[id], backref='assigned_refillers', foreign_keys=[assigned_to])
def to_dict(self):
"""Convert user object to dictionary"""
# Get assigned machines for Refillers
assigned_machines = []
if self.roles == 'Refiller':
machine_assignments = RefillerMachine.query.filter_by(refiller_id=self.id).all()
for assignment in machine_assignments:
machine = Machine.query.filter_by(machine_id=assignment.machine_id).first()
if machine:
assigned_machines.append({
'machine_id': machine.machine_id,
'machine_model': machine.machine_model,
'branch_name': machine.branch_name,
'assigned_at': assignment.assigned_at.strftime("%Y-%m-%d %H:%M:%S") if assignment.assigned_at else None
})
return {
'id': self.id,
'user_id': self.user_id,
'username': self.username,
'email': self.email,
'contact': self.contact,
'roles': self.roles,
'user_status': self.user_status,
'photo': self.photo,
'company_logo': self.company_logo,
'documents': json.loads(self.documents) if self.documents else [],
'created_at': self.created_at.strftime("%Y-%m-%d %H:%M:%S") if self.created_at else None,
'updated_at': self.updated_at.strftime("%Y-%m-%d %H:%M:%S") if self.updated_at else None,
'created_by': self.created_by,
'created_by_username': self.creator.username if self.creator else None,
'assigned_to': self.assigned_to,
'assigned_to_username': self.assigned_client.username if self.assigned_client else None,
'assigned_machines': assigned_machines, # ⭐ NEW
'assigned_machines_count': len(assigned_machines) # ⭐ NEW
}
def set_password(self, password):
"""Hash and set the password"""
self.password = generate_password_hash(password, method='pbkdf2:sha256')
print(f"Password hashed for user {self.username}")
def check_password(self, password):
"""Verify password against hash"""
result = check_password_hash(self.password, password)
print(f"Password check for {self.username}: {result}")
return result
@staticmethod
def generate_user_id(username, email):
"""Generate unique user ID from username and email"""
username_part = ''.join(filter(str.isalnum, username[:4])).upper()
email_part = ''.join(filter(str.isalnum, email.split('@')[0][:4])).upper()
timestamp = str(int(time.time()))[-4:]
user_id = f"USR-{username_part}{email_part}{timestamp}"
# Ensure uniqueness
counter = 1
original_id = user_id
while User.query.filter_by(user_id=user_id).first():
user_id = f"{original_id}{counter}"
counter += 1
return user_id
# Product Model
class Product(db.Model):
__tablename__ = 'products'
id = db.Column(db.Integer, primary_key=True)
product_id = db.Column(db.String(10), unique=True, nullable=False)
product_name = db.Column(db.String(100), nullable=False)
price = db.Column(db.Float, nullable=False)
product_image = db.Column(db.String(255), nullable=False)
created_date = db.Column(db.String(20), nullable=False)
created_by = db.Column(db.Integer, db.ForeignKey('users.id'), nullable=True)
# NEW: Billing and expiration dates
billing_date = db.Column(db.DateTime, nullable=True)
expiration_date = db.Column(db.DateTime, nullable=True)
# Relationship
creator = db.relationship('User', backref='created_products')
def to_dict(self):
return {
'id': self.id,
'product_id': self.product_id,
'product_name': self.product_name,
'price': str(self.price),
'product_image': self.product_image,
'created_date': self.created_date,
'billing_date': self.billing_date.strftime("%Y-%m-%d") if self.billing_date else None,
'expiration_date': self.expiration_date.strftime("%Y-%m-%d") if self.expiration_date else None,
'created_by': self.created_by,
'created_by_username': self.creator.username if self.creator else None
}
# VendingSlot Model
class VendingSlot(db.Model):
__tablename__ = 'vending_slots'
id = db.Column(db.Integer, primary_key=True)
machine_id = db.Column(db.String(10), db.ForeignKey('machines.machine_id'), nullable=False)
row_id = db.Column(db.String(1), nullable=False)
slot_name = db.Column(db.String(3), nullable=False)
enabled = db.Column(db.Boolean, default=True)
product_id = db.Column(db.String(10), db.ForeignKey('products.product_id'), nullable=True)
units = db.Column(db.Integer, default=0)
price = db.Column(db.String(10), default="N/A")
def to_dict(self):
return {
'name': self.slot_name,
'enabled': self.enabled,
'productId': self.product_id,
'units': self.units,
'price': self.price
}
# Transaction Model
class Transaction(db.Model):
__tablename__ = 'transactions'
id = db.Column(db.Integer, primary_key=True)
transaction_id = db.Column(db.String(50), unique=True, nullable=False)
machine_id = db.Column(db.String(10), db.ForeignKey('machines.machine_id'), nullable=False)
product_name = db.Column(db.String(100), nullable=False)
quantity = db.Column(db.Integer, nullable=False)
amount = db.Column(db.Float, nullable=False)
payment_type = db.Column(db.String(50), nullable=False) # wallet/qr/card
status = db.Column(db.String(50), nullable=False) # Success/Dispense failed/Unprocessed
amount_receiving_status = db.Column(db.String(50), nullable=True) # Dispense failed refund
return_amount = db.Column(db.Float, nullable=True)
created_at = db.Column(db.DateTime, default=datetime.datetime.utcnow)
machine = db.relationship('Machine', backref='transactions')
def to_dict(self):
return {
'id': self.id,
'transaction_id': self.transaction_id,
'machine_id': self.machine_id,
'product_name': self.product_name,
'quantity': self.quantity,
'amount': self.amount,
'payment_type': self.payment_type,
'status': self.status,
'amount_receiving_status': self.amount_receiving_status,
'return_amount': self.return_amount,
'created_at': self.created_at.strftime("%Y-%m-%d %H:%M:%S") if self.created_at else None
}
# Role Model
class Role(db.Model):
__tablename__ = 'roles'
id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String(50), unique=True, nullable=False)
description = db.Column(db.String(255))
permissions = db.Column(db.Text) # JSON string of permission IDs
created_by = db.Column(db.Integer, db.ForeignKey('users.id'), nullable=True)
created_at = db.Column(db.DateTime, default=datetime.datetime.utcnow)
updated_at = db.Column(db.DateTime, default=datetime.datetime.utcnow, onupdate=datetime.datetime.utcnow)
# Relationship
creator = db.relationship('User', backref='created_roles')
def to_dict(self):
return {
'id': self.id,
'name': self.name,
'description': self.description,
'permissions': json.loads(self.permissions) if self.permissions else [],
'created_by': self.created_by,
'created_by_username': self.creator.username if self.creator else None,
'created_at': self.created_at.strftime("%Y-%m-%d %H:%M:%S") if self.created_at else None,
'updated_at': self.updated_at.strftime("%Y-%m-%d %H:%M:%S") if self.updated_at else None
}
# Branch Model
class Branch(db.Model):
__tablename__ = 'branches'
id = db.Column(db.Integer, primary_key=True)
branch_id = db.Column(db.String(50), unique=True, nullable=False)
code = db.Column(db.String(20), unique=True, nullable=False)
name = db.Column(db.String(100), nullable=False)
location = db.Column(db.String(100), nullable=False)
address = db.Column(db.Text, nullable=False)
contact = db.Column(db.String(20), nullable=False)
created_by = db.Column(db.Integer, db.ForeignKey('users.id'), nullable=True)
created_at = db.Column(db.DateTime, default=datetime.datetime.utcnow)
updated_at = db.Column(db.DateTime, default=datetime.datetime.utcnow, onupdate=datetime.datetime.utcnow)
# Relationship
creator = db.relationship('User', backref='created_branches')
def to_dict(self):
return {
'id': self.id,
'branch_id': self.branch_id,
'code': self.code,
'name': self.name,
'location': self.location,
'address': self.address,
'contact': self.contact,
'created_by': self.created_by,
'created_by_username': self.creator.username if self.creator else None,
'created_at': self.created_at.strftime("%Y-%m-%d %H:%M:%S") if self.created_at else None,
'updated_at': self.updated_at.strftime("%Y-%m-%d %H:%M:%S") if self.updated_at else None
}
@staticmethod
def generate_branch_id():
"""Generate unique branch ID with BR prefix"""
import uuid
while True:
branch_id = f"BR{uuid.uuid4().hex[:8].upper()}"
if not Branch.query.filter_by(branch_id=branch_id).first():
return branch_id
# Brand Model
class Brand(db.Model):
__tablename__ = 'brands'
id = db.Column(db.Integer, primary_key=True)
brand_id = db.Column(db.String(50), unique=True, nullable=False)
name = db.Column(db.String(100), nullable=False)
branch_id = db.Column(db.String(50), db.ForeignKey('branches.branch_id'), nullable=False)
branch_name = db.Column(db.String(100), nullable=False)
image = db.Column(db.String(255), nullable=True)
created_by = db.Column(db.Integer, db.ForeignKey('users.id'), nullable=True)
created_at = db.Column(db.DateTime, default=datetime.datetime.utcnow)
updated_at = db.Column(db.DateTime, default=datetime.datetime.utcnow, onupdate=datetime.datetime.utcnow)
# Relationships
branch = db.relationship('Branch', backref='brands', foreign_keys=[branch_id])
creator = db.relationship('User', backref='created_brands')
def to_dict(self):
return {
'id': self.id,
'brand_id': self.brand_id,
'name': self.name,
'branch_id': self.branch_id,
'branch_name': self.branch_name,
'image': self.image,
'created_by': self.created_by,
'created_by_username': self.creator.username if self.creator else None,
'created_at': self.created_at.strftime("%Y-%m-%d %H:%M:%S") if self.created_at else None,
'updated_at': self.updated_at.strftime("%Y-%m-%d %H:%M:%S") if self.updated_at else None
}
@staticmethod
def generate_brand_id(name):
"""
Generate brand ID from first 3 letters of name + 4-digit sequence
Example: CocaCola -> COC0001, COC0002, etc.
"""
import re
# Extract only letters from name and get first 3
letters_only = ''.join(filter(str.isalpha, name)).upper()
prefix = letters_only[:3].ljust(3, 'X') # Pad with X if less than 3 letters
# Find the highest existing sequence number for this prefix
existing_brands = Brand.query.filter(
Brand.brand_id.like(f"{prefix}%")
).all()
if not existing_brands:
sequence = 1
else:
# Extract sequence numbers and find max
sequences = []
for brand in existing_brands:
try:
seq_part = brand.brand_id[3:] # Get part after prefix
if seq_part.isdigit():
sequences.append(int(seq_part))
except:
continue
sequence = max(sequences) + 1 if sequences else 1
# Format: PREFIX + 4-digit sequence
brand_id = f"{prefix}{sequence:04d}"
return brand_id
# Category Model
class Category(db.Model):
__tablename__ = 'categories'
id = db.Column(db.Integer, primary_key=True)
category_id = db.Column(db.String(50), unique=True, nullable=False)
name = db.Column(db.String(100), nullable=False)
image = db.Column(db.String(255), nullable=True)
brand_id = db.Column(db.String(50), db.ForeignKey('brands.brand_id'), nullable=False)
brand_name = db.Column(db.String(100), nullable=False)
branch_id = db.Column(db.String(50), db.ForeignKey('branches.branch_id'), nullable=False)
branch_name = db.Column(db.String(100), nullable=False)
created_by = db.Column(db.Integer, db.ForeignKey('users.id'), nullable=True)
created_at = db.Column(db.DateTime, default=datetime.datetime.utcnow)
updated_at = db.Column(db.DateTime, default=datetime.datetime.utcnow, onupdate=datetime.datetime.utcnow)
# Relationships
brand = db.relationship('Brand', backref='categories', foreign_keys=[brand_id])
branch = db.relationship('Branch', backref='categories', foreign_keys=[branch_id])
creator = db.relationship('User', backref='created_categories')
def to_dict(self):
return {
'id': self.id,
'category_id': self.category_id,
'name': self.name,
'image': self.image,
'brand_id': self.brand_id,
'brand_name': self.brand_name,
'branch_id': self.branch_id,
'branch_name': self.branch_name,
'created_by': self.created_by,
'created_by_username': self.creator.username if self.creator else None,
'created_at': self.created_at.strftime("%Y-%m-%d %H:%M:%S") if self.created_at else None,
'updated_at': self.updated_at.strftime("%Y-%m-%d %H:%M:%S") if self.updated_at else None
}
@staticmethod
def generate_category_id(name):
"""
Generate category ID from first 3 letters of name + 4-digit sequence
Example: Beverages -> BEV0001, BEV0002, etc.
"""
import re
# Extract only letters from name and get first 3
letters_only = ''.join(filter(str.isalpha, name)).upper()
prefix = letters_only[:3].ljust(3, 'X') # Pad with X if less than 3 letters
# Find the highest existing sequence number for this prefix
existing_categories = Category.query.filter(
Category.category_id.like(f"{prefix}%")
).all()
if not existing_categories:
sequence = 1
else:
# Extract sequence numbers and find max
sequences = []
for category in existing_categories:
try:
seq_part = category.category_id[3:] # Get part after prefix
if seq_part.isdigit():
sequences.append(int(seq_part))
except:
continue
sequence = max(sequences) + 1 if sequences else 1
# Format: PREFIX + 4-digit sequence
category_id = f"{prefix}{sequence:04d}"
return category_id
class SubCategory(db.Model):
__tablename__ = 'sub_categories'
id = db.Column(db.Integer, primary_key=True)
sub_category_id = db.Column(db.String(50), unique=True, nullable=False)
name = db.Column(db.String(100), nullable=False)
image = db.Column(db.String(255), nullable=True)
category_id = db.Column(db.String(50), db.ForeignKey('categories.category_id'), nullable=False)
category_name = db.Column(db.String(100), nullable=False)
brand_id = db.Column(db.String(50), db.ForeignKey('brands.brand_id'), nullable=False)
brand_name = db.Column(db.String(100), nullable=False)
branch_id = db.Column(db.String(50), db.ForeignKey('branches.branch_id'), nullable=False)
branch_name = db.Column(db.String(100), nullable=False)
created_by = db.Column(db.Integer, db.ForeignKey('users.id'), nullable=True)
created_at = db.Column(db.DateTime, default=datetime.datetime.utcnow)
updated_at = db.Column(db.DateTime, default=datetime.datetime.utcnow, onupdate=datetime.datetime.utcnow)
# Relationships
category = db.relationship('Category', backref='sub_categories', foreign_keys=[category_id])
brand = db.relationship('Brand', backref='sub_categories', foreign_keys=[brand_id])
branch = db.relationship('Branch', backref='sub_categories', foreign_keys=[branch_id])
creator = db.relationship('User', backref='created_subcategories')
def to_dict(self):
return {
'id': self.id,
'sub_category_id': self.sub_category_id,
'name': self.name,
'image': self.image,
'category_id': self.category_id,
'category_name': self.category_name,
'brand_id': self.brand_id,
'brand_name': self.brand_name,
'branch_id': self.branch_id,
'branch_name': self.branch_name,
'created_by': self.created_by,
'created_by_username': self.creator.username if self.creator else None,
'created_at': self.created_at.strftime("%Y-%m-%d %H:%M:%S") if self.created_at else None,
'updated_at': self.updated_at.strftime("%Y-%m-%d %H:%M:%S") if self.updated_at else None
}
@staticmethod
def generate_sub_category_id(name):
letters_only = ''.join(filter(str.isalpha, name)).upper()
prefix = letters_only[:3].ljust(3, 'X')
existing = SubCategory.query.filter(SubCategory.sub_category_id.like(f"{prefix}%")).all()
if not existing:
sequence = 1
else:
sequences = [int(s.sub_category_id[3:]) for s in existing if s.sub_category_id[3:].isdigit()]
sequence = max(sequences) + 1 if sequences else 1
return f"{prefix}{sequence:04d}"