feat: add log management dashboard and API endpoints with dynamic database fallback and JWT support

api/authentication.py

@@ -8,6 +8,10 @@ class CsrfExemptSessionAuthentication(SessionAuthentication):
 
 class DummyAuthentication(BaseAuthentication):
     def authenticate(self, request):
+        auth_header = request.headers.get('Authorization', '')
+        if auth_header.startswith('Bearer '):
+            # Defer to JWT authentication when an explicit token is provided
+            return None
         user = User.objects.first()
         if not user:
             user = User.objects.create_user(username='dummy', email='dummy@example.com', password='password')
@@ -15,3 +19,4 @@ class DummyAuthentication(BaseAuthentication):
         elif not hasattr(user, 'profile'):
             SupplierProfile.objects.create(user=user)
         return (user, None)
+

api/logs_views.py

@@ -0,0 +1,289 @@
+import os
+import re
+from pathlib import Path
+from django.conf import settings
+from django.http import HttpResponse
+from django.shortcuts import render
+from rest_framework.views import APIView
+from rest_framework.response import Response
+from rest_framework.permissions import AllowAny
+from rest_framework import status
+
+ALLOWED_LOG_FILES = {
+    'app.log': 'Application Logs (INFO+)',
+    'error.log': 'Error Logs (ERROR+)',
+    'requests.log': 'HTTP Request/Response Logs',
+    'gunicorn-access.log': 'Gunicorn Access Logs',
+    'gunicorn-error.log': 'Gunicorn Error Logs',
+}
+
+LOG_LEVEL_REGEX = re.compile(r'\b(DEBUG|INFO|WARNING|WARN|ERROR|CRITICAL)\b', re.IGNORECASE)
+TIMESTAMP_REGEX = re.compile(r'\[?(\d{4}-\d{2}-\d{2}[ T]\d{2}:\d{2}:\d{2}(?:\.\d+)?)\]?')
+
+def get_safe_log_path(filename):
+    """Ensure requested file is strictly located inside settings.LOGS_DIR and in whitelist."""
+    if not filename:
+        filename = 'app.log'
+    # Security: basename only, no directory traversal
+    clean_filename = os.path.basename(filename)
+    if clean_filename not in ALLOWED_LOG_FILES and not clean_filename.endswith('.log'):
+        return None
+    full_path = Path(settings.LOGS_DIR).resolve() / clean_filename
+    # Prevent traversal
+    if not str(full_path).startswith(str(Path(settings.LOGS_DIR).resolve())):
+        return None
+    return full_path
+
+def parse_log_line(line, line_num=0, source_file=''):
+    """Extract structured details from a log line."""
+    clean_line = line.rstrip('\r\n')
+    level = 'INFO'
+    timestamp = ''
+    logger_name = ''
+    message = clean_line
+
+    ts_match = TIMESTAMP_REGEX.search(clean_line)
+    if ts_match:
+        timestamp = ts_match.group(1)
+
+    lvl_match = LOG_LEVEL_REGEX.search(clean_line)
+    if lvl_match:
+        raw_lvl = lvl_match.group(1).upper()
+        level = 'WARNING' if raw_lvl == 'WARN' else raw_lvl
+
+    # Try to parse pipe-separated format: [ts] LEVEL name ... | message
+    if ' | ' in clean_line:
+        parts = clean_line.split(' | ', 1)
+        header = parts[0]
+        message = parts[1]
+        header_parts = header.split()
+        for p in header_parts:
+            if p in ('api', 'django.request', 'django.security', 'root') or '.' in p:
+                logger_name = p
+                break
+    else:
+        # Fallback message
+        message = clean_line
+
+    return {
+        'id': line_num,
+        'source': source_file,
+        'timestamp': timestamp,
+        'level': level,
+        'logger': logger_name,
+        'message': message,
+        'raw': clean_line,
+    }
+
+class LogsView(APIView):
+    permission_classes = [AllowAny]
+
+    def get(self, request):
+        filename = request.query_params.get('file', 'app.log')
+        level_filter = request.query_params.get('level', 'ALL').upper()
+        search_query = request.query_params.get('search', '').strip()
+        tail = request.query_params.get('tail', 'true').lower() in ('true', '1')
+        raw_format = request.query_params.get('raw', 'false').lower() in ('true', '1')
+
+        try:
+            limit = int(request.query_params.get('limit', 150))
+            limit = max(1, min(limit, 2000))
+        except ValueError:
+            limit = 150
+
+        # Handle 'all' to read across multiple files
+        target_files = []
+        if filename == 'all':
+            target_files = ['app.log', 'error.log', 'requests.log']
+        else:
+            safe_path = get_safe_log_path(filename)
+            if not safe_path:
+                return Response({'error': f'Invalid or disallowed log file name: {filename}'}, status=status.HTTP_400_BAD_REQUEST)
+            target_files = [safe_path.name]
+
+        all_entries = []
+        line_counter = 0
+        file_stats = {}
+        level_counts = {'DEBUG': 0, 'INFO': 0, 'WARNING': 0, 'ERROR': 0, 'CRITICAL': 0}
+
+        for fname in target_files:
+            file_path = Path(settings.LOGS_DIR) / fname
+            if not file_path.exists():
+                file_stats[fname] = {'exists': False, 'size_bytes': 0, 'lines': 0}
+                continue
+
+            size_bytes = file_path.stat().st_size
+            file_stats[fname] = {
+                'exists': True,
+                'size_bytes': size_bytes,
+                'size_kb': round(size_bytes / 1024, 2),
+            }
+
+            try:
+                with open(file_path, 'r', encoding='utf-8', errors='replace') as f:
+                    lines = f.readlines()
+                    file_stats[fname]['lines'] = len(lines)
+
+                    for raw_line in lines:
+                        if not raw_line.strip():
+                            continue
+                        line_counter += 1
+                        parsed = parse_log_line(raw_line, line_num=line_counter, source_file=fname)
+                        lvl = parsed['level']
+                        if lvl in level_counts:
+                            level_counts[lvl] += 1
+                        all_entries.append(parsed)
+            except Exception as e:
+                return Response({'error': f'Failed reading {fname}: {str(e)}'}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
+
+        # Apply filtering
+        filtered = all_entries
+
+        # Level filter
+        if level_filter and level_filter != 'ALL':
+            filtered = [entry for entry in filtered if entry['level'] == level_filter]
+
+        # Search filter
+        if search_query:
+            q = search_query.lower()
+            filtered = [
+                entry for entry in filtered
+                if q in entry['raw'].lower() or q in entry['message'].lower() or q in entry['logger'].lower()
+            ]
+
+        # Order & Limit
+        total_matched = len(filtered)
+        if tail:
+            # Take the most recent entries
+            display_entries = filtered[-limit:]
+        else:
+            display_entries = filtered[:limit]
+
+        if raw_format:
+            raw_text = '\n'.join([e['raw'] for e in display_entries])
+            return HttpResponse(raw_text, content_type='text/plain; charset=utf-8')
+
+        return Response({
+            'success': True,
+            'file': filename,
+            'available_files': list(ALLOWED_LOG_FILES.keys()),
+            'file_stats': file_stats,
+            'total_lines_read': line_counter,
+            'total_matched': total_matched,
+            'returned_count': len(display_entries),
+            'level_counts': level_counts,
+            'filters': {
+                'file': filename,
+                'level': level_filter,
+                'search': search_query,
+                'limit': limit,
+                'tail': tail,
+            },
+            'logs': display_entries,
+        })
+
+class LogStatsView(APIView):
+    permission_classes = [AllowAny]
+
+    def get(self, request):
+        stats = {}
+        total_size = 0
+        total_errors = 0
+        total_warnings = 0
+        total_lines = 0
+
+        for filename, desc in ALLOWED_LOG_FILES.items():
+            file_path = Path(settings.LOGS_DIR) / filename
+            if not file_path.exists():
+                stats[filename] = {
+                    'description': desc,
+                    'exists': False,
+                    'size_bytes': 0,
+                    'size_formatted': '0 KB',
+                    'line_count': 0,
+                    'error_count': 0,
+                    'last_modified': None,
+                }
+                continue
+
+            st = file_path.stat()
+            size = st.st_size
+            total_size += size
+            mtime = st.st_mtime
+
+            # Count lines and errors
+            lines_cnt = 0
+            err_cnt = 0
+            warn_cnt = 0
+            try:
+                with open(file_path, 'r', encoding='utf-8', errors='replace') as f:
+                    for line in f:
+                        lines_cnt += 1
+                        up = line.upper()
+                        if 'ERROR' in up or 'CRITICAL' in up:
+                            err_cnt += 1
+                        elif 'WARN' in up:
+                            warn_cnt += 1
+            except Exception:
+                pass
+
+            total_lines += lines_cnt
+            total_errors += err_cnt
+            total_warnings += warn_cnt
+
+            stats[filename] = {
+                'description': desc,
+                'exists': True,
+                'size_bytes': size,
+                'size_formatted': f"{round(size / 1024, 2)} KB" if size < 1024 * 1024 else f"{round(size / (1024 * 1024), 2)} MB",
+                'line_count': lines_cnt,
+                'error_count': err_cnt,
+                'warning_count': warn_cnt,
+                'last_modified': mtime,
+            }
+
+        return Response({
+            'files': stats,
+            'summary': {
+                'total_files': len(ALLOWED_LOG_FILES),
+                'total_lines': total_lines,
+                'total_errors': total_errors,
+                'total_warnings': total_warnings,
+                'total_size_bytes': total_size,
+                'total_size_formatted': f"{round(total_size / 1024, 2)} KB" if total_size < 1024 * 1024 else f"{round(total_size / (1024 * 1024), 2)} MB",
+            }
+        })
+
+class LogClearView(APIView):
+    permission_classes = [AllowAny]
+
+    def post(self, request):
+        filename = request.data.get('file', 'app.log')
+        if filename == 'all':
+            cleared = []
+            for fname in ALLOWED_LOG_FILES.keys():
+                file_path = Path(settings.LOGS_DIR) / fname
+                if file_path.exists():
+                    with open(file_path, 'w', encoding='utf-8') as f:
+                        f.write('')
+                    cleared.append(fname)
+            return Response({'success': True, 'cleared': cleared, 'message': f'Cleared {len(cleared)} log files'})
+
+        safe_path = get_safe_log_path(filename)
+        if not safe_path:
+            return Response({'error': f'Invalid log file: {filename}'}, status=status.HTTP_400_BAD_REQUEST)
+
+        if not safe_path.exists():
+            return Response({'error': f'Log file does not exist: {filename}'}, status=status.HTTP_404_NOT_FOUND)
+
+        with open(safe_path, 'w', encoding='utf-8') as f:
+            f.write('')
+
+        return Response({'success': True, 'file': filename, 'message': f'Log file {filename} truncated successfully'})
+
+def dashboard_view(request):
+    """Serve the single-page testing and log viewer dashboard."""
+    return render(request, 'dashboard.html', {
+        'debug': settings.DEBUG,
+        'environment': os.environ.get('ENVIRONMENT', 'local'),
+    })

api/test_logs.py

@@ -0,0 +1,55 @@
+import pytest
+from django.urls import reverse
+from rest_framework import status
+from rest_framework.test import APIClient
+
+@pytest.fixture
+def api_client():
+    return APIClient()
+
+@pytest.mark.django_db
+def test_dashboard_view(api_client):
+    response = api_client.get('/dashboard/')
+    assert response.status_code == status.HTTP_200_OK
+    assert b"Seller Central" in response.content
+    assert b"Log Dashboard" in response.content
+
+@pytest.mark.django_db
+def test_logs_endpoint_defaults(api_client):
+    url = reverse('logs-list')
+    response = api_client.get(url)
+    assert response.status_code == status.HTTP_200_OK
+    assert response.data['success'] is True
+    assert 'logs' in response.data
+    assert 'level_counts' in response.data
+    assert 'filters' in response.data
+
+@pytest.mark.django_db
+def test_logs_endpoint_level_filter(api_client):
+    url = reverse('logs-list')
+    response = api_client.get(f"{url}?file=app.log&level=ERROR")
+    assert response.status_code == status.HTTP_200_OK
+    assert response.data['filters']['level'] == 'ERROR'
+    for entry in response.data['logs']:
+        assert entry['level'] == 'ERROR'
+
+@pytest.mark.django_db
+def test_logs_endpoint_security_path_traversal(api_client):
+    url = reverse('logs-list')
+    response = api_client.get(f"{url}?file=../../etc/passwd")
+    assert response.status_code == status.HTTP_400_BAD_REQUEST
+    assert 'error' in response.data
+
+@pytest.mark.django_db
+def test_logs_stats_endpoint(api_client):
+    url = reverse('logs-stats')
+    response = api_client.get(url)
+    assert response.status_code == status.HTTP_200_OK
+    assert 'files' in response.data
+    assert 'summary' in response.data
+
+@pytest.mark.django_db
+def test_logs_clear_invalid_file(api_client):
+    url = reverse('logs-clear')
+    response = api_client.post(url, {'file': '../../secret.txt'})
+    assert response.status_code == status.HTTP_400_BAD_REQUEST

api/urls.py

@@ -7,6 +7,7 @@ from .views import (
     WalletView, WithdrawView, SubmitGstinView, PresignedUrlView, MockUploadView, MockDownloadView,
     UploadEAadhaarView, VerifyPANView
 )
+from .logs_views import LogsView, LogStatsView, LogClearView
 
 router = DefaultRouter()
 router.register(r'products', ProductViewSet, basename='product')
@@ -29,5 +30,8 @@ urlpatterns = [
     path('products/bulk-upload/', BulkUploadView.as_view(), name='bulk-upload'),
     path('wallet/', WalletView.as_view(), name='wallet'),
     path('wallet/withdraw/', WithdrawView.as_view(), name='withdraw'),
+    path('logs/', LogsView.as_view(), name='logs-list'),
+    path('logs/stats/', LogStatsView.as_view(), name='logs-stats'),
+    path('logs/clear/', LogClearView.as_view(), name='logs-clear'),
     path('', include(router.urls)),
 ]

config/settings.py

@@ -35,7 +35,7 @@ if env_path.exists():
 SECRET_KEY = 'django-insecure-w97g8j2h1=r%4$i1qc4xu%#c3)iw^4@#0-^sr3h&*okow^ax!@'
 
 # SECURITY WARNING: don't run with debug turned on in production!
-DEBUG = True
+DEBUG = os.environ.get('DEBUG', 'True').lower() not in ('0', 'false', 'f', 'no')
 
 ALLOWED_HOSTS = ['*']
 
@@ -79,7 +79,7 @@ ROOT_URLCONF = 'config.urls'
 TEMPLATES = [
     {
         'BACKEND': 'django.template.backends.django.DjangoTemplates',
-        'DIRS': [],
+        'DIRS': [BASE_DIR / 'templates'],
         'APP_DIRS': True,
         'OPTIONS': {
             'context_processors': [
@@ -98,32 +98,64 @@ WSGI_APPLICATION = 'config.wsgi.application'
 # Database Configuration
 # ---------------------------------------------------------------------------
 import os
+import sys
+import socket
 
-db_credentials = {
-    'ENGINE':   'django.db.backends.postgresql',
-    'USER':     os.environ.get('DB_USER', 'vignesh'),
-    'PASSWORD': os.environ.get('DB_PASSWORD', 'vtechnosoft@123A'),
-    'HOST':     os.environ.get('DB_HOST', '16.113.106.152'),
-    'PORT':     os.environ.get('DB_PORT', '5432'),
-}
+is_testing = 'test' in sys.argv or any('pytest' in arg for arg in sys.argv) or 'pytest' in sys.modules
+use_sqlite = is_testing or os.environ.get('USE_SQLITE', '').lower() in ('true', '1') or os.environ.get('DB_ENGINE') == 'sqlite3'
 
-env_name = os.environ.get('ENVIRONMENT', 'prod')
-db_suffix = f'_{env_name}'
+db_host = os.environ.get('DB_HOST', '16.113.106.152')
+db_port = int(os.environ.get('DB_PORT', '5432'))
 
-DATABASES = {
-    'default': {
-        'NAME': f'sellerprofile{db_suffix}',
-        **db_credentials
-    },
-    'customerprofile': {
-        'NAME': f'customerprofile{db_suffix}',
-        **db_credentials
-    },
-    'productprofile': {
-        'NAME': f'productprofile{db_suffix}',
-        **db_credentials
+# If in local debug mode and no explicit DB_HOST was forced in env, verify Postgres connectivity
+if not use_sqlite and 'DB_HOST' not in os.environ and DEBUG:
+    try:
+        s = socket.create_connection((db_host, db_port), timeout=0.6)
+        s.close()
+    except OSError:
+        use_sqlite = True
+
+if use_sqlite:
+    DATABASES = {
+        'default': {
+            'ENGINE': 'django.db.backends.sqlite3',
+            'NAME': BASE_DIR / 'db.sqlite3',
+        },
+        'customerprofile': {
+            'ENGINE': 'django.db.backends.sqlite3',
+            'NAME': BASE_DIR / 'db.sqlite3',
+        },
+        'productprofile': {
+            'ENGINE': 'django.db.backends.sqlite3',
+            'NAME': BASE_DIR / 'db.sqlite3',
+        }
+    }
+else:
+    db_credentials = {
+        'ENGINE':   'django.db.backends.postgresql',
+        'USER':     os.environ.get('DB_USER', 'vignesh'),
+        'PASSWORD': os.environ.get('DB_PASSWORD', 'vtechnosoft@123A'),
+        'HOST':     db_host,
+        'PORT':     str(db_port),
+    }
+
+    env_name = os.environ.get('ENVIRONMENT', 'prod')
+    db_suffix = f'_{env_name}'
+
+    DATABASES = {
+        'default': {
+            'NAME': f'sellerprofile{db_suffix}',
+            **db_credentials
+        },
+        'customerprofile': {
+            'NAME': f'customerprofile{db_suffix}',
+            **db_credentials
+        },
+        'productprofile': {
+            'NAME': f'productprofile{db_suffix}',
+            **db_credentials
+        }
     }
-}
 
 DATABASE_ROUTERS = ['config.db_router.DatabaseRouter']
 
@@ -184,7 +216,10 @@ from datetime import timedelta
 
 REST_FRAMEWORK = {
     'DEFAULT_AUTHENTICATION_CLASSES': (
-        'api.authentication.DummyAuthentication' if DEBUG else 'rest_framework_simplejwt.authentication.JWTAuthentication',
+        'rest_framework_simplejwt.authentication.JWTAuthentication',
+        'api.authentication.DummyAuthentication',
+    ) if DEBUG else (
+        'rest_framework_simplejwt.authentication.JWTAuthentication',
     ),
     'DEFAULT_PAGINATION_CLASS': 'rest_framework.pagination.PageNumberPagination',
     'PAGE_SIZE': 20,

config/urls.py

@@ -16,17 +16,16 @@ Including another URLconf
 """
 from django.contrib import admin
 from django.urls import path, include
-from django.http import HttpResponse
-
-def welcome_view(request):
-    return HttpResponse("welcome vignesh")
+from django.shortcuts import redirect
+from api.logs_views import dashboard_view
 
 from django.conf import settings
 from django.conf.urls.static import static
 
 urlpatterns = [
     path('admin/', admin.site.urls),
+    path('dashboard/', dashboard_view, name='dashboard'),
     path('api/', include('api.urls')),
-    path('', welcome_view, name='welcome'),
+    path('', lambda request: redirect('dashboard'), name='root'),
 ] + static(settings.MEDIA_URL, document_root=settings.MEDIA_ROOT)