IT_Ticket_System/app/models.py

from datetime import datetime
from flask_login import UserMixin
from werkzeug.security import generate_password_hash, check_password_hash
from app import db


# ─── Enumerations ────────────────────────────────────────────────────────────

class UserRole:
    EMPLOYEE = 'employee'
    IT_STAFF  = 'it_staff'
    ADMIN     = 'admin'

class TicketStatus:
    OPEN        = 'open'
    IN_PROGRESS = 'in_progress'
    PENDING     = 'pending'
    RESOLVED    = 'resolved'
    CLOSED      = 'closed'

class TicketPriority:
    LOW      = 'low'
    MEDIUM   = 'medium'
    HIGH     = 'high'
    CRITICAL = 'critical'

class TicketCategory:
    HARDWARE       = 'hardware'
    SOFTWARE       = 'software'
    NETWORK        = 'network'
    ACCESS         = 'access'
    EMAIL          = 'email'
    PRINTER        = 'printer'
    PHONE          = 'phone'
    SECURITY       = 'security'
    OTHER          = 'other'

class NotificationType:
    TICKET_CREATED   = 'ticket_created'
    TICKET_UPDATED   = 'ticket_updated'
    TICKET_ASSIGNED  = 'ticket_assigned'
    COMMENT_ADDED    = 'comment_added'
    STATUS_CHANGED   = 'status_changed'
    TICKET_RESOLVED  = 'ticket_resolved'
    TICKET_CLOSED    = 'ticket_closed'
    MENTION          = 'mention'


# ─── Models ──────────────────────────────────────────────────────────────────

class User(UserMixin, db.Model):
    __tablename__ = 'users'

    id           = db.Column(db.Integer, primary_key=True)
    email        = db.Column(db.String(120), unique=True, nullable=False, index=True)
    username     = db.Column(db.String(64), unique=True, nullable=False, index=True)
    full_name    = db.Column(db.String(128), nullable=False)
    password_hash= db.Column(db.String(256), nullable=False)
    role         = db.Column(db.String(20), default=UserRole.EMPLOYEE, nullable=False)
    department   = db.Column(db.String(100))
    phone        = db.Column(db.String(20))
    avatar_url   = db.Column(db.String(256))
    is_active    = db.Column(db.Boolean, default=True, nullable=False)
    email_notif  = db.Column(db.Boolean, default=True, nullable=False)
    web_notif    = db.Column(db.Boolean, default=True, nullable=False)
    created_at   = db.Column(db.DateTime, default=datetime.utcnow)
    last_login   = db.Column(db.DateTime)

    # Relationships
    created_tickets  = db.relationship('Ticket', foreign_keys='Ticket.created_by_id', backref='creator', lazy='dynamic')
    assigned_tickets = db.relationship('Ticket', foreign_keys='Ticket.assigned_to_id', backref='assignee', lazy='dynamic')
    comments         = db.relationship('Comment', backref='author', lazy='dynamic')
    notifications    = db.relationship('Notification', backref='recipient', lazy='dynamic')
    activity_logs    = db.relationship('ActivityLog', backref='user', lazy='dynamic')

    def set_password(self, password):
        self.password_hash = generate_password_hash(password)

    def check_password(self, password):
        return check_password_hash(self.password_hash, password)

    @property
    def is_it_staff(self):
        return self.role in (UserRole.IT_STAFF, UserRole.ADMIN)

    @property
    def is_admin(self):
        return self.role == UserRole.ADMIN

    def __repr__(self):
        return f'<User {self.username}>'


class Ticket(db.Model):
    __tablename__ = 'tickets'

    id            = db.Column(db.Integer, primary_key=True)
    ticket_number = db.Column(db.String(20), unique=True, nullable=False, index=True)
    title         = db.Column(db.String(200), nullable=False)
    description   = db.Column(db.Text, nullable=False)
    category      = db.Column(db.String(50), default=TicketCategory.OTHER, nullable=False)
    priority      = db.Column(db.String(20), default=TicketPriority.MEDIUM, nullable=False)
    status        = db.Column(db.String(20), default=TicketStatus.OPEN, nullable=False, index=True)
    location      = db.Column(db.String(100))
    asset_tag     = db.Column(db.String(50))

    created_by_id  = db.Column(db.Integer, db.ForeignKey('users.id'), nullable=False)
    assigned_to_id = db.Column(db.Integer, db.ForeignKey('users.id'))

    created_at     = db.Column(db.DateTime, default=datetime.utcnow, index=True)
    updated_at     = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
    resolved_at    = db.Column(db.DateTime)
    closed_at      = db.Column(db.DateTime)
    due_date       = db.Column(db.DateTime)

    ai_generated   = db.Column(db.Boolean, default=False)
    internal_notes = db.Column(db.Text)
    resolution_notes = db.Column(db.Text)

    # Relationships
    comments     = db.relationship('Comment', backref='ticket', lazy='dynamic', cascade='all, delete-orphan')
    attachments  = db.relationship('Attachment', backref='ticket', lazy='dynamic', cascade='all, delete-orphan')
    notifications= db.relationship('Notification', backref='ticket', lazy='dynamic', cascade='all, delete-orphan')
    history      = db.relationship('TicketHistory', backref='ticket', lazy='dynamic', cascade='all, delete-orphan')

    def generate_ticket_number(self):
        """Generate a unique ticket number like TKT-20240101-0001.

        Concurrency safety
        ------------------
        The naive approach — query the last ticket, increment its sequence,
        then insert — has a TOCTOU race: two concurrent requests can both read
        the same "last" ticket and generate the same next number, causing an
        IntegrityError at commit time.

        We eliminate the race by appending FOR UPDATE to the SELECT. This
        acquires a row-level write lock on the last matching ticket for the
        duration of the current transaction, serialising concurrent callers
        through the database rather than through application code. The lock is
        released automatically when the transaction commits or rolls back.

        MySQL / MariaDB: with_for_update() emits SELECT … FOR UPDATE.
        SQLite (dev/test): SELECT … FOR UPDATE is silently ignored, which is
        acceptable because SQLite's connection-level locking already prevents
        concurrent writes in practice.
        """
        date_str = datetime.utcnow().strftime('%Y%m%d')
        last = (
            Ticket.query
            .filter(Ticket.ticket_number.like(f'TKT-{date_str}-%'))
            .order_by(Ticket.id.desc())
            .with_for_update()   # acquires a write lock; serialises concurrent callers
            .first()
        )
        if last:
            seq = int(last.ticket_number.split('-')[-1]) + 1
        else:
            seq = 1
        return f'TKT-{date_str}-{seq:04d}'

    @property
    def priority_color(self):
        return {
            TicketPriority.LOW:      'success',
            TicketPriority.MEDIUM:   'warning',
            TicketPriority.HIGH:     'danger',
            TicketPriority.CRITICAL: 'dark',
        }.get(self.priority, 'secondary')

    @property
    def status_color(self):
        return {
            TicketStatus.OPEN:        'primary',
            TicketStatus.IN_PROGRESS: 'info',
            TicketStatus.PENDING:     'warning',
            TicketStatus.RESOLVED:    'success',
            TicketStatus.CLOSED:      'secondary',
        }.get(self.status, 'secondary')

    def __repr__(self):
        return f'<Ticket {self.ticket_number}>'


class Comment(db.Model):
    __tablename__ = 'comments'

    id          = db.Column(db.Integer, primary_key=True)
    ticket_id   = db.Column(db.Integer, db.ForeignKey('tickets.id'), nullable=False)
    author_id   = db.Column(db.Integer, db.ForeignKey('users.id'), nullable=False)
    body        = db.Column(db.Text, nullable=False)
    is_internal = db.Column(db.Boolean, default=False)  # IT-only internal note
    created_at  = db.Column(db.DateTime, default=datetime.utcnow)
    updated_at  = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)

    attachments = db.relationship('Attachment', backref='comment', lazy='dynamic')

    def __repr__(self):
        return f'<Comment {self.id} on Ticket {self.ticket_id}>'


class Attachment(db.Model):
    __tablename__ = 'attachments'

    id          = db.Column(db.Integer, primary_key=True)
    ticket_id   = db.Column(db.Integer, db.ForeignKey('tickets.id'))
    comment_id  = db.Column(db.Integer, db.ForeignKey('comments.id'))
    filename    = db.Column(db.String(256), nullable=False)
    stored_name = db.Column(db.String(256), nullable=False)
    file_size   = db.Column(db.Integer)
    mime_type   = db.Column(db.String(100))
    uploaded_by = db.Column(db.Integer, db.ForeignKey('users.id'))
    uploaded_at = db.Column(db.DateTime, default=datetime.utcnow)

    uploader    = db.relationship('User', foreign_keys=[uploaded_by])

    def __repr__(self):
        return f'<Attachment {self.filename}>'


class Notification(db.Model):
    __tablename__ = 'notifications'

    id          = db.Column(db.Integer, primary_key=True)
    user_id     = db.Column(db.Integer, db.ForeignKey('users.id'), nullable=False)
    ticket_id   = db.Column(db.Integer, db.ForeignKey('tickets.id'))
    type        = db.Column(db.String(50), nullable=False)
    title       = db.Column(db.String(200), nullable=False)
    message     = db.Column(db.Text)
    is_read     = db.Column(db.Boolean, default=False, index=True)
    created_at  = db.Column(db.DateTime, default=datetime.utcnow, index=True)
    link        = db.Column(db.String(256))

    def __repr__(self):
        return f'<Notification {self.id} for User {self.user_id}>'


class TicketHistory(db.Model):
    __tablename__ = 'ticket_history'

    id          = db.Column(db.Integer, primary_key=True)
    ticket_id   = db.Column(db.Integer, db.ForeignKey('tickets.id'), nullable=False)
    changed_by  = db.Column(db.Integer, db.ForeignKey('users.id'), nullable=False)
    field_name  = db.Column(db.String(50), nullable=False)
    old_value   = db.Column(db.Text)
    new_value   = db.Column(db.Text)
    changed_at  = db.Column(db.DateTime, default=datetime.utcnow)

    changer     = db.relationship('User', foreign_keys=[changed_by])

    def __repr__(self):
        return f'<TicketHistory {self.id}>'


class ActivityLog(db.Model):
    __tablename__ = 'activity_logs'

    id          = db.Column(db.Integer, primary_key=True)
    user_id     = db.Column(db.Integer, db.ForeignKey('users.id', ondelete='SET NULL'))
    action      = db.Column(db.String(100), nullable=False)
    entity_type = db.Column(db.String(50))
    entity_id   = db.Column(db.Integer)
    details     = db.Column(db.Text)
    ip_address  = db.Column(db.String(45))
    created_at  = db.Column(db.DateTime, default=datetime.utcnow, index=True)

    def __repr__(self):
        return f'<ActivityLog {self.action} by User {self.user_id}>'


class KnowledgeBase(db.Model):
    __tablename__ = 'knowledge_base'

    id          = db.Column(db.Integer, primary_key=True)
    title       = db.Column(db.String(200), nullable=False)
    body        = db.Column(db.Text, nullable=False)
    category    = db.Column(db.String(50))
    tags        = db.Column(db.String(256))
    author_id   = db.Column(db.Integer, db.ForeignKey('users.id'))
    is_published= db.Column(db.Boolean, default=True)
    view_count  = db.Column(db.Integer, default=0)
    created_at  = db.Column(db.DateTime, default=datetime.utcnow)
    updated_at  = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)

    author      = db.relationship('User', foreign_keys=[author_id])

    def __repr__(self):
        return f'<KnowledgeBase {self.title}>'


class KBAttachment(db.Model):
    """Files and images attached to a Knowledge Base article."""
    __tablename__ = 'kb_attachments'

    id          = db.Column(db.Integer, primary_key=True)
    article_id  = db.Column(db.Integer, db.ForeignKey('knowledge_base.id'), nullable=False)
    filename    = db.Column(db.String(256), nullable=False)   # original name
    stored_name = db.Column(db.String(256), nullable=False)   # UUID-based on disk
    mime_type   = db.Column(db.String(100))
    file_size   = db.Column(db.Integer)
    uploaded_by = db.Column(db.Integer, db.ForeignKey('users.id'))
    uploaded_at = db.Column(db.DateTime, default=datetime.utcnow)

    article     = db.relationship('KnowledgeBase', backref=db.backref('attachments', lazy='dynamic', cascade='all, delete-orphan'))
    uploader    = db.relationship('User', foreign_keys=[uploaded_by])

    @property
    def url(self):
        return f'/admin/kb/files/{self.stored_name}'

    def __repr__(self):
        return f'<KBAttachment {self.filename}>'


class SystemSetting(db.Model):
    """Key-value store for application-wide configuration flags.

    Values are persisted as strings; helper class methods handle
    typed access so callers never touch raw strings directly.
    """
    __tablename__ = 'system_settings'

    id          = db.Column(db.Integer, primary_key=True)
    key         = db.Column(db.String(100), unique=True, nullable=False, index=True)
    value       = db.Column(db.String(500), nullable=False)
    description = db.Column(db.String(256))
    updated_at  = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)

    @classmethod
    def get(cls, key, default=None):
        """Return the raw string value for *key*, or *default* if absent."""
        row = cls.query.filter_by(key=key).first()
        return row.value if row else default

    @classmethod
    def get_bool(cls, key, default=True):
        """Return the value for *key* coerced to bool."""
        raw = cls.get(key)
        if raw is None:
            return default
        return raw.lower() in ('1', 'true', 'yes', 'on')

    @classmethod
    def set(cls, key, value, description=None):
        """Upsert *key* = *value*.  Caller is responsible for committing."""
        row = cls.query.filter_by(key=key).first()
        if row:
            row.value = str(value)
            if description is not None:
                row.description = description
        else:
            row = cls(key=key, value=str(value), description=description)
            db.session.add(row)
        return row

    def __repr__(self):
        return f'<SystemSetting {self.key}={self.value}>'