Merge pull request #11 from MEHRSHAD-MIRSHEKARY/feature/send-logs-com…

…mand

Feature/send logs command

.github/dependabot.yml

@@ -0,0 +1,20 @@
+# To get started with Dependabot version updates, you'll need to specify which
+# package ecosystems to update and where the package manifests are located.
+# Please see the documentation for all configuration options:
+# https://docs.github.com/code-security/dependabot/dependabot-version-updates/configuration-options-for-the-dependabot.yml-file
+
+version: 2
+updates:
+  - package-ecosystem: "pip"
+    directory: "/packages" # Location of the requirements.txt file
+    schedule:
+      interval: "weekly"
+    # If you want to target the specific file
+    target-branch: "main" # Replace with your default branch if different
+
+  - package-ecosystem: "pip"
+    directory: "/packages" # Location of the requirements-dev.txt file
+    schedule:
+      interval: "weekly"
+    # If you want to target the specific file
+    target-branch: "main" # Replace with your default branch if different

django_logging/apps.py

@@ -18,6 +18,12 @@ def ready(self):
             "LOG_FILE_LEVELS", ["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"]
         )
         log_dir = log_settings.get("LOG_DIR", os.path.join(os.getcwd(), "logs"))
+        log_email_notifier = log_settings.get("LOG_EMAIL_NOTIFIER", {})
+        log_email_notifier_enable = log_email_notifier.get("ENABLE", False)
+        log_email_notifier_log_levels = [
+            "ERROR" if log_email_notifier.get("NOTIFY_ERROR", False) else None,
+            "CRITICAL" if log_email_notifier.get("NOTIFY_CRITICAL", False) else None,
+        ]
 
         # Set the logging configuration
-        set_logging(log_levels, log_dir)
+        set_logging(log_levels, log_dir, log_email_notifier_enable, log_email_notifier_log_levels)

django_logging/filters/level_filter.py

@@ -0,0 +1,38 @@
+import logging
+
+
+class LoggingLevelFilter(logging.Filter):
+    """
+    Filters log records based on their logging level.
+
+    This filter is used to prevent log records from being written to log files
+    intended for lower log levels. For example, if we have separate log
+    files for DEBUG, INFO, WARNING, and ERROR levels, this filter ensures that
+    a log record with level ERROR is only written to the ERROR log file, and not
+    to the DEBUG, INFO or WARNING log files.
+    """
+
+    def __init__(self, logging_level: int):
+        """
+        Initializes a LoggingLevelFilter instance.
+
+        Args:
+            logging_level: The logging level to filter on (e.g. logging.DEBUG, logging.INFO, etc.).
+
+        Returns:
+            None
+        """
+        super().__init__()
+        self.logging_level = logging_level
+
+    def filter(self, record: logging.LogRecord) -> bool:
+        """
+        Filters a log record based on its level.
+
+        Args:
+            record: The log record to filter.
+
+        Returns:
+            True if the log record's level matches the specified logging level, False otherwise.
+        """
+        return record.levelno == self.logging_level

django_logging/handlers/__init__.py

@@ -0,0 +1 @@
+from .email_handler import EmailHandler

django_logging/handlers/email_handler.py

@@ -0,0 +1,60 @@
+import logging
+
+from django.conf import settings
+from django.template import engines
+from django.utils.timezone import now
+from django_logging.utils.email.notifier import send_email_async
+from django_logging.middleware import RequestLogMiddleware
+
+
+class EmailHandler(logging.Handler):
+    def __init__(self, include_html=False, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+
+        self.include_html = include_html
+
+        try:
+            # Attempt to retrieve the logging settings from the Django settings
+            logging_settings = settings.DJANGO_LOGGING
+
+            self.include_html = logging_settings.get("LOG_EMAIL_NOTIFIER", {}).get("USE_TEMPLATE", include_html)
+
+        except AttributeError:
+            logging.warning(
+                f"DJANGO_LOGGING settings not found. Using default include_html value: {self.include_html}.")
+        except Exception as e:
+            logging.error(f"An unexpected error occurred while initializing EmailHandler: {e}")
+
+    def emit(self, record):
+        try:
+            request = getattr(record, "request", None)
+            log_entry = self.format(record)
+
+            if self.include_html:
+                email_body = self.render_template(log_entry, request)
+            else:
+                email_body = log_entry
+
+            subject = f"New Log Record: {record.levelname}"
+            send_email_async(subject, email_body, [settings.ADMIN_EMAIL])
+
+        except Exception as e:
+            self.handleError(record)
+
+    @staticmethod
+    def render_template(log_entry, request=None, template_path="email_notifier_template.html"):
+        django_engine = engines["django"]
+        template = django_engine.get_template(template_path)
+
+        # Fetch IP address and user agent using middleware methods
+        ip_address = RequestLogMiddleware.get_ip_address(request) if request else "Unknown"
+        user_agent = RequestLogMiddleware.get_user_agent(request) if request else "Unknown"
+
+        context = {
+            "message": log_entry,
+            "time": now(),
+            "browser_type": user_agent,
+            "ip_address": ip_address,
+        }
+
+        return template.render(context)

django_logging/management/commands/send_logs.py

@@ -0,0 +1,107 @@
+import os
+import shutil
+import tempfile
+import logging
+from django.core.mail import EmailMessage
+from django.core.management.base import BaseCommand
+from django.conf import settings
+
+logger = logging.getLogger(__name__)
+
+
+class Command(BaseCommand):
+    """
+    A Django management command that zips the log directory and sends it to
+    the specified email address.
+
+    This command is used to send the log files to a specified email address.
+    It zips the log directory, creates an email with the zipped file as an attachment,
+    and sends it to the specified email address.
+    """
+
+    help = "Send log folder to the specified email address"
+
+    def add_arguments(self, parser):
+        """
+        Add custom command arguments.
+
+        Parameters:
+            parser (ArgumentParser): The argument parser to add arguments to.
+        """
+        parser.add_argument(
+            "email", type=str, help="The email address to send the logs to"
+        )
+
+    def handle(self, *args, **kwargs):
+        """
+        The main entry point for the command.
+
+        Parameters:
+            args (tuple): Positional arguments.
+            kwargs (dict): Keyword arguments.
+        """
+        email = kwargs["email"]
+
+        log_dir = settings.DJANGO_LOGGING.get("LOG_DIR", os.path.join(os.getcwd(), "logs"))
+
+        if not os.path.exists(log_dir):
+            self.stdout.write(
+                self.style.ERROR(f'Log directory "{log_dir}" does not exist.')
+            )
+            logger.error(f'Log directory "{log_dir}" does not exist.')
+            return
+
+        self.check_email_settings()
+
+        # Create a temporary file to store the zipped logs
+        with tempfile.NamedTemporaryFile(delete=False) as tmp_file:
+            zip_path = f"{tmp_file.name}.zip"
+            tmp_file.close()
+
+            # Zip the log directory
+            shutil.make_archive(tmp_file.name, "zip", log_dir)
+
+        # Send the email with the zipped logs
+        email_subject = "Log Files"
+        email_body = "Please find the attached log files."
+        email_message = EmailMessage(
+            subject=email_subject,
+            body=email_body,
+            from_email=settings.DEFAULT_FROM_EMAIL,
+            to=[email],
+        )
+        email_message.attach_file(zip_path)
+
+        try:
+            email_message.send()
+            self.stdout.write(self.style.SUCCESS(f"Logs sent successfully to {email}."))
+            logger.info(f"Logs sent successfully to {email}.")
+        except Exception as e:
+            self.stdout.write(self.style.ERROR(f"Failed to send logs: {e}"))
+            logger.error(f"Failed to send logs: {e}")
+        finally:
+            # Clean up the temporary file
+            os.remove(zip_path)
+            logger.info("Temporary zip file cleaned up successfully.")
+
+    def check_email_settings(self):
+        """
+        Check if all required email settings are present in the settings file.
+
+        Raises an exception if any of the required email settings are missing.
+        """
+        required_settings = [
+            "EMAIL_HOST",
+            "EMAIL_PORT",
+            "EMAIL_HOST_USER",
+            "EMAIL_HOST_PASSWORD",
+            "EMAIL_USE_TLS",
+            "DEFAULT_FROM_EMAIL",
+        ]
+
+        for setting in required_settings:
+            if not getattr(settings, setting, None):
+                error_message = f"Missing required email setting: {setting}"
+                self.stdout.write(self.style.ERROR(error_message))
+                logger.error(f"Missing required email setting: {setting}")
+                raise Exception(error_message)

django_logging/middleware/__init__.py

@@ -0,0 +1 @@
+from .request_middleware import RequestLogMiddleware

django_logging/middleware/request_middleware.py

@@ -0,0 +1,84 @@
+import logging
+from django.contrib.auth import get_user_model
+from django.http import HttpResponse, HttpRequest
+from typing import Callable
+
+logger = logging.getLogger(__name__)
+
+
+class RequestLogMiddleware:
+    """
+    Middleware to log information about each incoming request.
+
+    This middleware logs the request path, the user making the request (if authenticated),
+    and the user's IP address.
+    """
+
+    def __init__(self, get_response: Callable[[HttpRequest], HttpResponse]) -> None:
+        """
+        Initializes the RequestLogMiddleware instance.
+
+        Args:
+            get_response: A callable that returns an HttpResponse object.
+        """
+        self.get_response = get_response
+        user_model = get_user_model()
+        self.username_field = user_model.USERNAME_FIELD
+
+    def __call__(self, request: HttpRequest) -> HttpResponse:
+        """
+        Processes an incoming request and logs relevant information.
+
+        Args:
+            request: The incoming request object.
+
+        Returns:
+            The response object returned by the view function.
+        """
+        # Before view (and later middleware) are called.
+        response = self.get_response(request)
+
+        # After view is called.
+        if hasattr(request, "user") and request.user.is_authenticated:
+            user = getattr(request.user, self.username_field, "Anonymous")
+        else:
+            user = "Anonymous"
+
+        # Get the user's IP address
+        ip_address = self.get_ip_address(request)
+
+        # Get the user agent
+        user_agent = self.get_user_agent(request)
+
+        # Attach IP and user agent to the request
+        request.ip_address = ip_address
+        request.browser_type = user_agent
+
+        logger.info(
+            f"Request Info: (request_path: {request.path}, user: {user},"
+            f"\nIP: {ip_address}, user_agent: {user_agent})"
+        )
+
+        return response
+
+    @staticmethod
+    def get_ip_address(request: HttpRequest) -> str:
+        """
+        Retrieves the client's IP address from the request object.
+        """
+        ip_address = request.META.get("HTTP_X_FORWARDED_FOR")
+        if ip_address:
+            ip_address = ip_address.split(",")[0]
+        else:
+            ip_address = request.META.get("LIMITED_ACCESS")
+            if not ip_address:
+                ip_address = request.META.get("REMOTE_ADDR")
+
+        return ip_address
+
+    @staticmethod
+    def get_user_agent(request: HttpRequest) -> str:
+        """
+        Retrieves the client's user agent from the request object.
+        """
+        return request.META.get("HTTP_USER_AGENT", "Unknown User Agent")

django_logging/settings/conf.py

@@ -3,6 +3,8 @@
 import os
 from typing import List, Dict, Optional
 
+from django_logging.filters.level_filter import LoggingLevelFilter
+
 
 class LogConfig:
     """
@@ -17,9 +19,14 @@ def __init__(
             self,
             log_levels: List[str],
             log_dir: str,
+            log_email_notifier_enable: bool,
+            log_email_notifier_log_levels: List[str],
     ) -> None:
+
         self.log_levels = log_levels
         self.log_dir = log_dir
+        self.email_notifier_enable = log_email_notifier_enable
+        self.email_notifier_log_levels = log_email_notifier_log_levels
 
 
 class LogManager:
@@ -67,6 +74,7 @@ def set_conf(self) -> None:
                 "filename": log_file,
                 "formatter": "default",
                 "level": level,
+                "filters": [level.lower()]
             }
             for level, log_file in self.log_files.items()
         }
@@ -76,6 +84,27 @@ def set_conf(self) -> None:
             "level": "DEBUG",
         }
 
+        email_handler = {
+            f"email_{level.lower()}": {
+                "class": "django_logging.handlers.EmailHandler",
+                "level": level,
+                "filters": [level.lower()],
+            }
+            for level in self.log_config.email_notifier_log_levels
+            if level
+        }
+
+        if self.log_config.email_notifier_enable:
+            handlers.update(email_handler)
+
+        filters = {
+            level.lower(): {
+                "()": LoggingLevelFilter,
+                "logging_level": getattr(logging, level),
+            }
+            for level in self.log_config.log_levels
+        }
+
         loggers = {
             level.lower(): {
                 "level": level,
@@ -88,6 +117,7 @@ def set_conf(self) -> None:
         config = {
             "version": 1,
             "handlers": handlers,
+            "filters": filters,
             "loggers": loggers,
             "root": {"level": "DEBUG", "handlers": list(handlers.keys())},
             "disable_existing_loggers": False,