522 lines
21 KiB
Python
522 lines
21 KiB
Python
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)
|