Refactor and improve code quality

Refactor and improve code quality

This commit includes several code improvements and refactorings:

1.  **Refactored `app/core/models.py` for log counting:**
    *   Introduced a private helper method `_count_logs_since` in the `SystemLog` model to consolidate logic for counting logs based on time, level, and security status.
    *   Updated `get_error_count_last_24h` and `get_security_events_count_last_24h` to use this new helper method, reducing code duplication.

2.  **Made security log retention configurable:**
    *   Added a `SECURITY_LOG_RETENTION_DAYS` configuration variable in `app/config.py` (defaulting to 90 days, settable via environment variable).
    *   Modified the `cleanup_old_logs` method in `SystemLog` to use this configuration value instead of a hardcoded one.

3.  **Improved `app/core/models.py` documentation and readability:**
    *   Added a docstring to `LogSearchQuery.to_dict()`.
    *   Ensured dictionary definitions and SQLAlchemy queries in `SystemLog` methods are formatted for better readability.
    *   Removed some redundant local imports of `datetime` and `timedelta`.

4.  **Refactored exception handling in `app/auth/routes.py`:**
    *   Introduced a private helper function `_handle_auth_exception` to standardize logging and flash messages for exceptions occurring during user login and registration.
    *   Updated the `login` and `register` routes to use this helper, reducing boilerplate.

These changes aim to improve code maintainability, readability, and configurability.

Author: sPROFFEs

app/auth/routes.py

@@ -9,6 +9,12 @@
 
 auth_bp = Blueprint('auth', __name__, url_prefix='/auth')
 
+def _handle_auth_exception(e, operation_name):
+    """Helper function to log and flash authentication errors."""
+    current_app.logger.error(f"{operation_name} error: {str(e)}")
+    current_app.logger.error(traceback.format_exc())
+    flash(f'An error occurred during {operation_name.lower()}. Please try again.', 'danger')
+
 @auth_bp.route('/login', methods=['GET', 'POST'])
 def login():
     if current_user.is_authenticated:
@@ -29,9 +35,7 @@ def login():
                 next_page = url_for('core.index')
             return redirect(next_page)
         except Exception as e:
-            current_app.logger.error(f"Login error: {str(e)}")
-            current_app.logger.error(traceback.format_exc())
-            flash('An error occurred during login. Please try again.', 'danger')
+            _handle_auth_exception(e, "Login")
     return render_template('auth/login.html', title='Sign In', form=form)
 
 @auth_bp.route('/logout')
@@ -59,10 +63,8 @@ def register():
             flash('Congratulations, you are now a registered user!', 'success')
             return redirect(url_for('auth.login'))
         except Exception as e:
-            db.session.rollback()
-            current_app.logger.error(f"Registration error: {str(e)}")
-            current_app.logger.error(traceback.format_exc())
-            flash('An error occurred during registration. Please try again.', 'danger')
+            db.session.rollback() # Keep rollback here as it's specific to DB operations
+            _handle_auth_exception(e, "Registration")
     return render_template('auth/register.html', title='Register', form=form)
 
 @auth_bp.route('/profile')

app/config.py

@@ -21,6 +21,9 @@ class Config:
     # Modules repository URL
     MODULES_REPOSITORY_URL = os.environ.get('MODULES_REPOSITORY_URL') or \
         'https://github.com/CoreSecFrame/CoreSecFrame-Modules'
+
+    # Log retention settings
+    SECURITY_LOG_RETENTION_DAYS = int(os.environ.get('SECURITY_LOG_RETENTION_DAYS', 90))
 
     # Security settings
     SESSION_COOKIE_SECURE = False  # Set to True in production

app/core/models.py

@@ -1,6 +1,7 @@
 # app/core/models.py
+from flask import current_app
 from app import db
-from datetime import datetime
+from datetime import datetime, timedelta
 from sqlalchemy import Index
 
 class SystemLog(db.Model):
@@ -102,6 +103,20 @@ def is_error(self):
     def is_warning(self):
         """Check if this is a warning log"""
         return self.level == 'WARNING'
+
+    @classmethod
+    def _count_logs_since(cls, since_hours, levels=None, is_security_event=None):
+        """Helper method to count logs within a time window with optional filters."""
+        since = datetime.utcnow() - timedelta(hours=since_hours)
+        query = cls.query.filter(cls.timestamp >= since)
+
+        if levels:
+            query = query.filter(cls.level.in_(levels))
+        
+        if is_security_event is not None:
+            query = query.filter(cls.is_security_event == is_security_event)
+            
+        return query.count()
 
     @classmethod
     def get_recent_logs(cls, limit=50, level=None, is_security=None, user_id=None):
@@ -122,31 +137,16 @@ def get_recent_logs(cls, limit=50, level=None, is_security=None, user_id=None):
     @classmethod
     def get_error_count_last_24h(cls):
         """Get count of errors in the last 24 hours"""
-        from datetime import datetime, timedelta
-        
-        since = datetime.utcnow() - timedelta(hours=24)
-        
-        return cls.query.filter(
-            cls.timestamp >= since,
-            cls.level.in_(['ERROR', 'CRITICAL'])
-        ).count()
+        return cls._count_logs_since(since_hours=24, levels=['ERROR', 'CRITICAL'])
 
     @classmethod
     def get_security_events_count_last_24h(cls):
         """Get count of security events in the last 24 hours"""
-        from datetime import datetime, timedelta
-        
-        since = datetime.utcnow() - timedelta(hours=24)
-        
-        return cls.query.filter(
-            cls.timestamp >= since,
-            cls.is_security_event == True
-        ).count()
+        return cls._count_logs_since(since_hours=24, is_security_event=True)
 
     @classmethod
     def get_log_level_stats(cls, hours=24):
         """Get statistics by log level for the specified time period"""
-        from datetime import datetime, timedelta
         from sqlalchemy import func
 
         since = datetime.utcnow() - timedelta(hours=hours)
@@ -163,7 +163,6 @@ def get_log_level_stats(cls, hours=24):
     @classmethod
     def get_module_stats(cls, hours=24, limit=10):
         """Get top modules by log count"""
-        from datetime import datetime, timedelta
         from sqlalchemy import func
 
         since = datetime.utcnow() - timedelta(hours=hours)
@@ -182,12 +181,11 @@ def get_module_stats(cls, hours=24, limit=10):
     @classmethod
     def cleanup_old_logs(cls, days_to_keep=30):
         """Clean up old log entries"""
-        from datetime import datetime, timedelta
-        
         cutoff_date = datetime.utcnow() - timedelta(days=days_to_keep)
 
-        # Keep security events longer (90 days)
-        security_cutoff = datetime.utcnow() - timedelta(days=90)
+        # Keep security events longer, based on config
+        security_retention_days = current_app.config['SECURITY_LOG_RETENTION_DAYS']
+        security_cutoff = datetime.utcnow() - timedelta(days=security_retention_days)
 
         # Delete old non-security logs
         deleted_count = cls.query.filter(
@@ -227,7 +225,10 @@ class LogSearchQuery(db.Model):
     use_count = db.Column(db.Integer, default=0)
 
     def to_dict(self):
-        """Convert to dictionary for API"""
+        """
+        Returns a dictionary representation of the log search query, 
+        suitable for API responses.
+        """
         return {
             'id': self.id,
             'name': self.name,