shopdb-flask/shopdb/core/models/auditlog.py

"""Audit log model for tracking changes."""

from datetime import datetime, timezone
from shopdb.extensions import db

def _utcnow():
    # naive UTC for DB columns (stored without tzinfo)
    return datetime.now(timezone.utc).replace(tzinfo=None)


class AuditLog(db.Model):
    """
    Audit log for tracking user actions.

    Records who did what, when, from where, and what changed.
    """
    __tablename__ = 'auditlogs'

    auditlogid = db.Column(db.Integer, primary_key=True)

    # Who
    userid = db.Column(db.Integer, db.ForeignKey('users.userid'), nullable=True)
    username = db.Column(db.String(100), nullable=True)  # Denormalized for history

    # When
    timestamp = db.Column(db.DateTime, default=_utcnow, nullable=False, index=True)

    # Where (client info)
    ipaddress = db.Column(db.String(45), nullable=True)  # IPv6 max length
    useragent = db.Column(db.String(255), nullable=True)

    # What
    action = db.Column(db.String(20), nullable=False, index=True)  # created, updated, deleted
    entitytype = db.Column(db.String(50), nullable=False, index=True)  # Asset, Printer, Setting, etc.
    entityid = db.Column(db.Integer, nullable=True)  # ID of the affected record
    entityname = db.Column(db.String(255), nullable=True)  # Human-readable identifier

    # Changes (stored as JSON text)
    _changes = db.Column('changes', db.Text, nullable=True)  # {"field": {"old": x, "new": y}, ...}

    @property
    def changes(self):
        """Get changes as dict."""
        if self._changes:
            import json
            try:
                return json.loads(self._changes)
            except (json.JSONDecodeError, TypeError):
                return None
        return None

    @changes.setter
    def changes(self, value):
        """Set changes from dict."""
        if value is not None:
            import json
            self._changes = json.dumps(value)
        else:
            self._changes = None

    # Additional context
    details = db.Column(db.Text, nullable=True)  # Optional description

    # Relationship
    user = db.relationship('User', backref=db.backref('auditlogs', lazy='dynamic'))

    def to_dict(self):
        return {
            'auditlogid': self.auditlogid,
            'userid': self.userid,
            'username': self.username,
            'timestamp': self.timestamp.isoformat() + 'Z' if self.timestamp else None,
            'ipaddress': self.ipaddress,
            'useragent': self.useragent,
            'action': self.action,
            'entitytype': self.entitytype,
            'entityid': self.entityid,
            'entityname': self.entityname,
            'changes': self.changes,
            'details': self.details
        }

    @classmethod
    def log(cls, action: str, entitytype: str, entityid: int = None,
            entityname: str = None, changes: dict = None, details: str = None,
            user=None, request=None):
        """
        Create an audit log entry.

        Args:
            action: 'created', 'updated', 'deleted'
            entitytype: Type of entity (e.g., 'Asset', 'Printer', 'Setting')
            entityid: ID of the affected record
            entityname: Human-readable name/identifier
            changes: Dict of field changes {"field": {"old": x, "new": y}}
            details: Optional description
            user: Current user object (or will try to get from flask-jwt-extended)
            request: Flask request object (or will try to get current request)
        """
        from flask import request as flask_request
        from flask_jwt_extended import current_user, verify_jwt_in_request

        # Get user info
        if user is None:
            try:
                verify_jwt_in_request(optional=True)
                user = current_user
            except:
                pass

        userid = user.userid if user else None
        username = user.username if user else None

        # Get request info
        request_obj = request or flask_request
        ipaddress = None
        useragent = None

        if request_obj:
            # Handle proxy forwarding
            ipaddress = request_obj.headers.get('X-Forwarded-For', request_obj.remote_addr)
            if ipaddress and ',' in ipaddress:
                ipaddress = ipaddress.split(',')[0].strip()
            useragent = request_obj.headers.get('User-Agent', '')[:255]

        entry = cls(
            userid=userid,
            username=username,
            ipaddress=ipaddress,
            useragent=useragent,
            action=action,
            entitytype=entitytype,
            entityid=entityid,
            entityname=entityname,
            changes=changes,
            details=details
        )

        db.session.add(entry)
        # Don't commit here - let the caller handle transaction
        return entry

    @classmethod
    def log_create(cls, entitytype: str, entity, name_field: str = 'name'):
        """Log a create action."""
        entityname = getattr(entity, name_field, None) or str(entity)
        entityid = getattr(entity, f'{entitytype.lower()}id', None) or getattr(entity, 'id', None)
        return cls.log('created', entitytype, entityid=entityid, entityname=entityname)

    @classmethod
    def log_update(cls, entitytype: str, entity, changes: dict, name_field: str = 'name'):
        """Log an update action with changes."""
        entityname = getattr(entity, name_field, None) or str(entity)
        entityid = getattr(entity, f'{entitytype.lower()}id', None) or getattr(entity, 'id', None)
        return cls.log('updated', entitytype, entityid=entityid, entityname=entityname, changes=changes)

    @classmethod
    def log_delete(cls, entitytype: str, entityid: int, entityname: str = None):
        """Log a delete action."""
        return cls.log('deleted', entitytype, entityid=entityid, entityname=entityname)