from rest_framework import viewsets, status, filters from rest_framework.decorators import action from rest_framework.response import Response from rest_framework.permissions import IsAuthenticated from django_filters.rest_framework import DjangoFilterBackend from django.db.models import Q, Count, Avg from django.utils import timezone from django.shortcuts import get_object_or_404 from ..models import ( Postmortem, KnowledgeBaseArticle, IncidentRecommendation, LearningPattern, KnowledgeBaseUsage, AutomatedPostmortemGeneration ) from ..serializers.knowledge import ( PostmortemSerializer, PostmortemListSerializer, KnowledgeBaseArticleSerializer, KnowledgeBaseArticleListSerializer, IncidentRecommendationSerializer, IncidentRecommendationListSerializer, LearningPatternSerializer, LearningPatternListSerializer, KnowledgeBaseUsageSerializer, AutomatedPostmortemGenerationSerializer, PostmortemGenerationRequestSerializer, RecommendationRequestSerializer, KnowledgeBaseSearchSerializer, KnowledgeBaseArticleRatingSerializer ) from ..services.postmortem_generator import PostmortemGenerator from ..services.recommendation_engine import RecommendationEngine from ..services.knowledge_base_search import KnowledgeBaseSearchService class PostmortemViewSet(viewsets.ModelViewSet): """ViewSet for managing postmortems""" queryset = Postmortem.objects.all() serializer_class = PostmortemSerializer permission_classes = [IsAuthenticated] filter_backends = [DjangoFilterBackend, filters.SearchFilter, filters.OrderingFilter] filterset_fields = ['status', 'severity', 'is_automated', 'owner'] search_fields = ['title', 'executive_summary', 'root_cause_analysis'] ordering_fields = ['created_at', 'updated_at', 'due_date', 'severity'] ordering = ['-created_at'] def get_serializer_class(self): """Return appropriate serializer based on action""" if self.action == 'list': return PostmortemListSerializer return PostmortemSerializer def get_queryset(self): """Filter queryset based on user permissions""" queryset = super().get_queryset() # Filter by user access permissions if not self.request.user.is_staff: # Non-staff users can only see postmortems they own or are involved in queryset = queryset.filter( Q(owner=self.request.user) | Q(reviewers=self.request.user) | Q(approver=self.request.user) ).distinct() return queryset @action(detail=True, methods=['post']) def generate_automated(self, request, pk=None): """Generate automated postmortem for an incident""" postmortem = self.get_object() if postmortem.is_automated: return Response( {'error': 'Postmortem is already automated'}, status=status.HTTP_400_BAD_REQUEST ) generator = PostmortemGenerator() try: generated_content = generator.generate_postmortem(postmortem.incident) # Update postmortem with generated content postmortem.executive_summary = generated_content.get('executive_summary', '') postmortem.timeline = generated_content.get('timeline', []) postmortem.root_cause_analysis = generated_content.get('root_cause_analysis', '') postmortem.impact_assessment = generated_content.get('impact_assessment', '') postmortem.lessons_learned = generated_content.get('lessons_learned', '') postmortem.action_items = generated_content.get('action_items', []) postmortem.is_automated = True postmortem.generation_confidence = generated_content.get('confidence_score', 0.0) postmortem.auto_generated_sections = generated_content.get('generated_sections', []) postmortem.save() return Response({ 'message': 'Postmortem generated successfully', 'confidence_score': postmortem.generation_confidence }) except Exception as e: return Response( {'error': f'Failed to generate postmortem: {str(e)}'}, status=status.HTTP_500_INTERNAL_SERVER_ERROR ) @action(detail=True, methods=['post']) def approve(self, request, pk=None): """Approve a postmortem""" postmortem = self.get_object() if postmortem.status != 'IN_REVIEW': return Response( {'error': 'Postmortem must be in review status to approve'}, status=status.HTTP_400_BAD_REQUEST ) postmortem.status = 'APPROVED' postmortem.approver = request.user postmortem.save() return Response({'message': 'Postmortem approved successfully'}) @action(detail=True, methods=['post']) def publish(self, request, pk=None): """Publish a postmortem""" postmortem = self.get_object() if postmortem.status != 'APPROVED': return Response( {'error': 'Postmortem must be approved before publishing'}, status=status.HTTP_400_BAD_REQUEST ) postmortem.status = 'PUBLISHED' postmortem.published_at = timezone.now() postmortem.save() return Response({'message': 'Postmortem published successfully'}) @action(detail=False, methods=['get']) def overdue(self, request): """Get overdue postmortems""" overdue_postmortems = self.get_queryset().filter( due_date__lt=timezone.now(), status__in=['DRAFT', 'IN_REVIEW'] ) serializer = self.get_serializer(overdue_postmortems, many=True) return Response(serializer.data) @action(detail=False, methods=['get']) def statistics(self, request): """Get postmortem statistics""" queryset = self.get_queryset() stats = { 'total_postmortems': queryset.count(), 'by_status': dict(queryset.values('status').annotate(count=Count('id')).values_list('status', 'count')), 'by_severity': dict(queryset.values('severity').annotate(count=Count('id')).values_list('severity', 'count')), 'automated_percentage': queryset.filter(is_automated=True).count() / max(queryset.count(), 1) * 100, 'overdue_count': queryset.filter( due_date__lt=timezone.now(), status__in=['DRAFT', 'IN_REVIEW'] ).count(), 'avg_completion_time': queryset.filter( published_at__isnull=False ).aggregate( avg_time=Avg('published_at' - 'created_at') )['avg_time'] } return Response(stats) class KnowledgeBaseArticleViewSet(viewsets.ModelViewSet): """ViewSet for managing knowledge base articles""" queryset = KnowledgeBaseArticle.objects.all() serializer_class = KnowledgeBaseArticleSerializer permission_classes = [IsAuthenticated] filter_backends = [DjangoFilterBackend, filters.SearchFilter, filters.OrderingFilter] filterset_fields = ['article_type', 'category', 'subcategory', 'status', 'is_featured', 'difficulty_level'] search_fields = ['title', 'content', 'summary', 'tags', 'search_keywords'] ordering_fields = ['created_at', 'updated_at', 'view_count', 'title'] ordering = ['-updated_at'] lookup_field = 'slug' def get_serializer_class(self): """Return appropriate serializer based on action""" if self.action == 'list': return KnowledgeBaseArticleListSerializer return KnowledgeBaseArticleSerializer def retrieve(self, request, *args, **kwargs): """Retrieve article and increment view count""" instance = self.get_object() instance.increment_view_count() # Log the view KnowledgeBaseUsage.objects.create( user=request.user, usage_type='VIEW', knowledge_article=instance, context={'ip_address': request.META.get('REMOTE_ADDR')} ) serializer = self.get_serializer(instance) return Response(serializer.data) @action(detail=True, methods=['post']) def rate(self, request, slug=None): """Rate a knowledge base article""" article = self.get_object() serializer = KnowledgeBaseArticleRatingSerializer(data=request.data) if serializer.is_valid(): # Log the rating KnowledgeBaseUsage.objects.create( user=request.user, usage_type='RATE', knowledge_article=article, context={ 'rating': serializer.validated_data['rating'], 'feedback': serializer.validated_data.get('feedback', '') } ) return Response({'message': 'Rating recorded successfully'}) return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST) @action(detail=True, methods=['post']) def bookmark(self, request, slug=None): """Bookmark a knowledge base article""" article = self.get_object() # Log the bookmark KnowledgeBaseUsage.objects.create( user=request.user, usage_type='BOOKMARK', knowledge_article=article ) return Response({'message': 'Article bookmarked successfully'}) @action(detail=False, methods=['post']) def search(self, request): """Search knowledge base articles""" serializer = KnowledgeBaseSearchSerializer(data=request.data) if serializer.is_valid(): search_service = KnowledgeBaseSearchService() results = search_service.search( query=serializer.validated_data['query'], article_types=serializer.validated_data.get('article_types'), categories=serializer.validated_data.get('categories'), difficulty_levels=serializer.validated_data.get('difficulty_levels'), limit=serializer.validated_data['limit'], offset=serializer.validated_data['offset'] ) return Response(results) return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST) @action(detail=False, methods=['get']) def due_for_review(self, request): """Get articles due for review""" due_articles = self.get_queryset().filter( next_review_due__lt=timezone.now() ) serializer = self.get_serializer(due_articles, many=True) return Response(serializer.data) @action(detail=False, methods=['get']) def popular(self, request): """Get popular articles""" popular_articles = self.get_queryset().filter( status='PUBLISHED' ).order_by('-view_count')[:10] serializer = self.get_serializer(popular_articles, many=True) return Response(serializer.data) @action(detail=False, methods=['get']) def statistics(self, request): """Get knowledge base statistics""" queryset = self.get_queryset() stats = { 'total_articles': queryset.count(), 'by_type': dict(queryset.values('article_type').annotate(count=Count('id')).values_list('article_type', 'count')), 'by_status': dict(queryset.values('status').annotate(count=Count('id')).values_list('status', 'count')), 'by_difficulty': dict(queryset.values('difficulty_level').annotate(count=Count('id')).values_list('difficulty_level', 'count')), 'total_views': queryset.aggregate(total_views=Count('view_count'))['total_views'], 'due_for_review': queryset.filter(next_review_due__lt=timezone.now()).count(), 'featured_articles': queryset.filter(is_featured=True).count() } return Response(stats) class IncidentRecommendationViewSet(viewsets.ModelViewSet): """ViewSet for managing incident recommendations""" queryset = IncidentRecommendation.objects.all() serializer_class = IncidentRecommendationSerializer permission_classes = [IsAuthenticated] filter_backends = [DjangoFilterBackend, filters.SearchFilter, filters.OrderingFilter] filterset_fields = ['recommendation_type', 'confidence_level', 'is_applied', 'incident'] search_fields = ['title', 'description', 'reasoning'] ordering_fields = ['created_at', 'confidence_score', 'similarity_score'] ordering = ['-confidence_score', '-similarity_score'] def get_serializer_class(self): """Return appropriate serializer based on action""" if self.action == 'list': return IncidentRecommendationListSerializer return IncidentRecommendationSerializer def get_queryset(self): """Filter queryset based on user permissions""" queryset = super().get_queryset() # Filter by incident access permissions if not self.request.user.is_staff: from incident_intelligence.models import Incident accessible_incidents = Incident.objects.filter( Q(assigned_to=self.request.user) | Q(reporter=self.request.user) ) queryset = queryset.filter(incident__in=accessible_incidents) return queryset @action(detail=True, methods=['post']) def apply(self, request, pk=None): """Apply a recommendation""" recommendation = self.get_object() if recommendation.is_applied: return Response( {'error': 'Recommendation has already been applied'}, status=status.HTTP_400_BAD_REQUEST ) recommendation.is_applied = True recommendation.applied_at = timezone.now() recommendation.applied_by = request.user recommendation.save() # Log the application KnowledgeBaseUsage.objects.create( user=request.user, usage_type='APPLY', recommendation=recommendation, incident=recommendation.incident ) return Response({'message': 'Recommendation applied successfully'}) @action(detail=True, methods=['post']) def rate_effectiveness(self, request, pk=None): """Rate the effectiveness of a recommendation""" recommendation = self.get_object() rating = request.data.get('rating') if not rating or not (1 <= rating <= 5): return Response( {'error': 'Rating must be between 1 and 5'}, status=status.HTTP_400_BAD_REQUEST ) recommendation.effectiveness_rating = rating recommendation.save() return Response({'message': 'Effectiveness rating recorded successfully'}) @action(detail=False, methods=['post']) def generate_for_incident(self, request): """Generate recommendations for an incident""" serializer = RecommendationRequestSerializer(data=request.data) if serializer.is_valid(): recommendation_engine = RecommendationEngine() try: recommendations = recommendation_engine.generate_recommendations( incident_id=serializer.validated_data['incident_id'], recommendation_types=serializer.validated_data.get('recommendation_types'), max_recommendations=serializer.validated_data['max_recommendations'], min_confidence=serializer.validated_data['min_confidence'] ) return Response({ 'message': 'Recommendations generated successfully', 'recommendations': recommendations }) except Exception as e: return Response( {'error': f'Failed to generate recommendations: {str(e)}'}, status=status.HTTP_500_INTERNAL_SERVER_ERROR ) return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST) @action(detail=False, methods=['get']) def statistics(self, request): """Get recommendation statistics""" queryset = self.get_queryset() stats = { 'total_recommendations': queryset.count(), 'by_type': dict(queryset.values('recommendation_type').annotate(count=Count('id')).values_list('recommendation_type', 'count')), 'by_confidence': dict(queryset.values('confidence_level').annotate(count=Count('id')).values_list('confidence_level', 'count')), 'applied_count': queryset.filter(is_applied=True).count(), 'avg_effectiveness': queryset.filter( effectiveness_rating__isnull=False ).aggregate(avg_rating=Avg('effectiveness_rating'))['avg_rating'], 'high_confidence_count': queryset.filter(confidence_score__gte=0.8).count() } return Response(stats) class LearningPatternViewSet(viewsets.ModelViewSet): """ViewSet for managing learning patterns""" queryset = LearningPattern.objects.all() serializer_class = LearningPatternSerializer permission_classes = [IsAuthenticated] filter_backends = [DjangoFilterBackend, filters.SearchFilter, filters.OrderingFilter] filterset_fields = ['pattern_type', 'is_validated'] search_fields = ['name', 'description', 'triggers', 'actions'] ordering_fields = ['created_at', 'confidence_score', 'frequency', 'success_rate'] ordering = ['-confidence_score', '-frequency'] def get_serializer_class(self): """Return appropriate serializer based on action""" if self.action == 'list': return LearningPatternListSerializer return LearningPatternSerializer @action(detail=True, methods=['post']) def validate(self, request, pk=None): """Validate a learning pattern""" pattern = self.get_object() if pattern.is_validated: return Response( {'error': 'Pattern has already been validated'}, status=status.HTTP_400_BAD_REQUEST ) pattern.is_validated = True pattern.validated_by = request.user pattern.validation_notes = request.data.get('validation_notes', '') pattern.save() return Response({'message': 'Pattern validated successfully'}) @action(detail=True, methods=['post']) def apply(self, request, pk=None): """Apply a learning pattern""" pattern = self.get_object() pattern.times_applied += 1 pattern.last_applied = timezone.now() pattern.save() return Response({'message': 'Pattern applied successfully'}) @action(detail=False, methods=['get']) def statistics(self, request): """Get learning pattern statistics""" queryset = self.get_queryset() stats = { 'total_patterns': queryset.count(), 'by_type': dict(queryset.values('pattern_type').annotate(count=Count('id')).values_list('pattern_type', 'count')), 'validated_count': queryset.filter(is_validated=True).count(), 'avg_confidence': queryset.aggregate(avg_confidence=Avg('confidence_score'))['avg_confidence'], 'avg_success_rate': queryset.aggregate(avg_success=Avg('success_rate'))['avg_success'], 'total_applications': queryset.aggregate(total_apps=Count('times_applied'))['total_apps'] } return Response(stats) class AutomatedPostmortemGenerationViewSet(viewsets.ReadOnlyModelViewSet): """ViewSet for viewing automated postmortem generation logs""" queryset = AutomatedPostmortemGeneration.objects.all() serializer_class = AutomatedPostmortemGenerationSerializer permission_classes = [IsAuthenticated] filter_backends = [DjangoFilterBackend, filters.OrderingFilter] filterset_fields = ['status', 'incident', 'generation_trigger'] ordering_fields = ['started_at', 'completed_at', 'processing_time'] ordering = ['-started_at'] @action(detail=False, methods=['post']) def generate_postmortem(self, request): """Generate automated postmortem for an incident""" serializer = PostmortemGenerationRequestSerializer(data=request.data) if serializer.is_valid(): generator = PostmortemGenerator() try: result = generator.generate_postmortem_for_incident( incident_id=serializer.validated_data['incident_id'], include_timeline=serializer.validated_data['include_timeline'], include_logs=serializer.validated_data['include_logs'], trigger=serializer.validated_data['generation_trigger'] ) return Response({ 'message': 'Postmortem generation initiated', 'generation_id': result['generation_id'] }) except Exception as e: return Response( {'error': f'Failed to generate postmortem: {str(e)}'}, status=status.HTTP_500_INTERNAL_SERVER_ERROR ) return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)