Moves generate_incident_summary() to AI service module and refactors code

src/dispatch/ai/service.py

@@ -5,6 +5,8 @@
 
 from dispatch.case.enums import CaseResolutionReason
 from dispatch.case.models import Case
+from dispatch.enums import Visibility
+from dispatch.incident.models import Incident
 from dispatch.plugin import service as plugin_service
 from dispatch.signal import service as signal_service
 
@@ -191,12 +193,7 @@ def generate_case_signal_summary(case: Case, db_session: Session) -> dict[str, s
     )
 
     try:
-        summary = json.loads(
-            response["choices"][0]["message"]["content"]
-            .replace("```json", "")
-            .replace("```", "")
-            .strip()
-        )
+        summary = json.loads(response.replace("```json", "").replace("```", "").strip())
 
         # we check if the summary is empty
         if not summary:
@@ -209,3 +206,76 @@ def generate_case_signal_summary(case: Case, db_session: Session) -> dict[str, s
         message = "Unable to generate GenAI signal analysis. Error decoding response from the artificial-intelligence plugin."
         log.warning(message)
         raise GenAIException(message) from e
+
+
+def generate_incident_summary(incident: Incident, db_session: Session) -> str:
+    """
+    Generate a summary for an incident.
+
+    Args:
+        incident (Incident): The incident object for which the summary is being generated.
+        db_session (Session): The database session used for querying related data.
+
+    Returns:
+        str: A string containing the summary of the incident, or an error message if summary generation fails.
+    """
+    # Skip summary for restricted incidents
+    if incident.visibility == Visibility.restricted:
+        return "Incident summary not generated for restricted incident."
+
+    # Skip if no incident review document
+    if not incident.incident_review_document or not incident.incident_review_document.resource_id:
+        log.info(
+            f"Incident summary not generated for incident {incident.name}. No review document found."
+        )
+        return "Incident summary not generated. No review document found."
+
+    # Don't generate if no enabled ai plugin or storage plugin
+    genai_plugin = plugin_service.get_active_instance(
+        db_session=db_session, plugin_type="artificial-intelligence", project_id=incident.project.id
+    )
+    if not genai_plugin:
+        message = f"Incident summary not generated for incident {incident.name}. No artificial-intelligence plugin enabled."
+        log.warning(message)
+        return "Incident summary not generated. No artificial-intelligence plugin enabled."
+
+    storage_plugin = plugin_service.get_active_instance(
+        db_session=db_session, plugin_type="storage", project_id=incident.project.id
+    )
+
+    if not storage_plugin:
+        log.info(
+            f"Incident summary not generated for incident {incident.name}. No storage plugin enabled."
+        )
+        return "Incident summary not generated. No storage plugin enabled."
+
+    try:
+        pir_doc = storage_plugin.instance.get(
+            file_id=incident.incident_review_document.resource_id,
+            mime_type="text/plain",
+        )
+        prompt = f"""
+            Given the text of the security post-incident review document below,
+            provide answers to the following questions in a paragraph format.
+            Do not include the questions in your response.
+            Do not use any of these words in your summary unless they appear in the document: breach, unauthorized, leak, violation, unlawful, illegal.
+            1. What is the summary of what happened?
+            2. What were the overall risk(s)?
+            3. How were the risk(s) mitigated?
+            4. How was the incident resolved?
+            5. What are the follow-up tasks?
+
+            {pir_doc}
+        """
+
+        summary = genai_plugin.instance.chat_completion(prompt=prompt)
+
+        incident.summary = summary
+        db_session.add(incident)
+        db_session.commit()
+
+        return summary
+
+    except Exception as e:
+        log.exception(f"Error trying to generate summary for incident {incident.name}: {e}")
+        return "Incident summary not generated. An error occurred."

src/dispatch/incident/flows.py

@@ -4,6 +4,7 @@
 
 from sqlalchemy.orm import Session
 
+from dispatch.ai import service as ai_service
 from dispatch.case import flows as case_flows
 from dispatch.case import service as case_service
 from dispatch.case.enums import CaseResolutionReason, CaseStatus
@@ -568,7 +569,7 @@ def incident_closed_status_flow(incident: Incident, db_session=None):
     send_incident_rating_feedback_message(incident, db_session)
 
     # if an AI plugin is enabled, we send the incident review doc for summary
-    incident_service.generate_incident_summary(incident=incident, db_session=db_session)
+    ai_service.generate_incident_summary(incident=incident, db_session=db_session)
 
 
 def conversation_topic_dispatcher(

src/dispatch/incident/scheduled.py

@@ -1,38 +1,37 @@
 import logging
-
 from collections import defaultdict
+from datetime import date, datetime
 
-from datetime import datetime, date
 from schedule import every
 from sqlalchemy import func
 from sqlalchemy.orm import Session
 
-from dispatch.enums import Visibility
+from dispatch.ai import service as ai_service
 from dispatch.conversation.enums import ConversationButtonActions
 from dispatch.database.core import resolve_attr
 from dispatch.decorators import scheduled_project_task, timer
+from dispatch.enums import Visibility
+from dispatch.incident import service as incident_service
 from dispatch.messaging.strings import (
     INCIDENT,
     INCIDENT_DAILY_REPORT,
     INCIDENT_DAILY_REPORT_TITLE,
+    INCIDENT_SUMMARY_TEMPLATE,
     INCIDENT_WEEKLY_REPORT,
     INCIDENT_WEEKLY_REPORT_NO_INCIDENTS,
     INCIDENT_WEEKLY_REPORT_TITLE,
-    INCIDENT_SUMMARY_TEMPLATE,
     MessageType,
 )
 from dispatch.nlp import build_phrase_matcher, build_term_vocab, extract_terms_from_text
 from dispatch.notification import service as notification_service
-from dispatch.incident import service as incident_service
+from dispatch.participant import flows as participant_flows
+from dispatch.participant_role.models import ParticipantRoleType
 from dispatch.plugin import service as plugin_service
 from dispatch.project.models import Project
 from dispatch.scheduler import scheduler
 from dispatch.search_filter import service as search_filter_service
 from dispatch.tag import service as tag_service
 from dispatch.tag.models import Tag
-from dispatch.participant import flows as participant_flows
-from dispatch.participant_role.models import ParticipantRoleType
-
 
 from .enums import IncidentStatus
 from .messaging import send_incident_close_reminder
@@ -42,7 +41,6 @@
     get_all_last_x_hours_by_status,
 )
 
-
 log = logging.getLogger(__name__)
 
 
@@ -290,7 +288,7 @@ def incident_report_weekly(db_session: Session, project: Project):
             if incident.summary:
                 summary = incident.summary
             else:
-                summary = incident_service.generate_incident_summary(
+                summary = ai_service.generate_incident_summary(
                     db_session=db_session, incident=incident
                 )
 

src/dispatch/incident/service.py

@@ -6,16 +6,14 @@
 """
 
 import logging
-
 from datetime import datetime, timedelta
 from typing import List, Optional
-from pydantic.error_wrappers import ErrorWrapper, ValidationError
 
+from pydantic.error_wrappers import ErrorWrapper, ValidationError
 from sqlalchemy.orm import Session
 
-from dispatch.decorators import timer
 from dispatch.case import service as case_service
-from dispatch.enums import Visibility
+from dispatch.decorators import timer
 from dispatch.event import service as event_service
 from dispatch.exceptions import NotFoundError
 from dispatch.incident.priority import service as incident_priority_service
@@ -34,7 +32,6 @@
 from .enums import IncidentStatus
 from .models import Incident, IncidentCreate, IncidentRead, IncidentUpdate
 
-
 log = logging.getLogger(__name__)
 
 
@@ -438,69 +435,3 @@ def delete(*, db_session: Session, incident_id: int):
     """Deletes an existing incident."""
     db_session.query(Incident).filter(Incident.id == incident_id).delete()
     db_session.commit()
-
-
-def generate_incident_summary(*, db_session: Session, incident: Incident) -> str:
-    """Generates a summary of the incident."""
-    # Skip summary for restricted incidents
-    if incident.visibility == Visibility.restricted:
-        return "Incident summary not generated for restricted incident."
-
-    # Skip if no incident review document
-    if not incident.incident_review_document or not incident.incident_review_document.resource_id:
-        log.info(
-            f"Incident summary not generated for incident {incident.id}. No review document found."
-        )
-        return "Incident summary not generated. No review document found."
-
-    # Don't generate if no enabled ai plugin or storage plugin
-    ai_plugin = plugin_service.get_active_instance(
-        db_session=db_session, plugin_type="artificial-intelligence", project_id=incident.project.id
-    )
-    if not ai_plugin:
-        log.info(
-            f"Incident summary not generated for incident {incident.id}. No AI plugin enabled."
-        )
-        return "Incident summary not generated. No AI plugin enabled."
-
-    storage_plugin = plugin_service.get_active_instance(
-        db_session=db_session, plugin_type="storage", project_id=incident.project.id
-    )
-
-    if not storage_plugin:
-        log.info(
-            f"Incident summary not generated for incident {incident.id}. No storage plugin enabled."
-        )
-        return "Incident summary not generated. No storage plugin enabled."
-
-    try:
-        pir_doc = storage_plugin.instance.get(
-            file_id=incident.incident_review_document.resource_id,
-            mime_type="text/plain",
-        )
-        prompt = f"""
-            Given the text of the security post-incident review document below,
-            provide answers to the following questions in a paragraph format.
-            Do not include the questions in your response.
-            Do not use any of these words in your summary unless they appear in the document: breach, unauthorized, leak, violation, unlawful, illegal.
-            1. What is the summary of what happened?
-            2. What were the overall risk(s)?
-            3. How were the risk(s) mitigated?
-            4. How was the incident resolved?
-            5. What are the follow-up tasks?
-
-            {pir_doc}
-        """
-
-        response = ai_plugin.instance.chat_completion(prompt=prompt)
-        summary = response["choices"][0]["message"]["content"]
-
-        incident.summary = summary
-        db_session.add(incident)
-        db_session.commit()
-
-        return summary
-
-    except Exception as e:
-        log.exception(f"Error trying to generate summary for incident {incident.id}: {e}")
-        return "Incident summary not generated. An error occurred."

src/dispatch/incident/views.py

@@ -1,26 +1,28 @@
 import calendar
-from datetime import date, datetime
-from dateutil.relativedelta import relativedelta
-from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Query, status
 import json
 import logging
+from datetime import date, datetime
 from typing import Annotated, List
-from starlette.requests import Request
+
+from dateutil.relativedelta import relativedelta
+from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Query, status
 from sqlalchemy.exc import IntegrityError
+from starlette.requests import Request
 
+from dispatch.ai import service as ai_service
 from dispatch.auth.permissions import (
     IncidentEditPermission,
+    IncidentEventPermission,
     IncidentJoinOrSubscribePermission,
     IncidentViewPermission,
     PermissionsDependency,
-    IncidentEventPermission,
 )
 from dispatch.auth.service import CurrentUser
 from dispatch.common.utils.views import create_pydantic_include
 from dispatch.database.core import DbSession
 from dispatch.database.service import CommonParameters, search_filter_sort_paginate
 from dispatch.event import flows as event_flows
-from dispatch.event.models import EventUpdate, EventCreateMinimal
+from dispatch.event.models import EventCreateMinimal, EventUpdate
 from dispatch.incident.enums import IncidentStatus
 from dispatch.individual.models import IndividualContactRead
 from dispatch.models import OrganizationSlug, PrimaryKey
@@ -32,11 +34,11 @@
     incident_add_or_reactivate_participant_flow,
     incident_create_closed_flow,
     incident_create_flow,
+    incident_create_resources_flow,
     incident_create_stable_flow,
     incident_delete_flow,
     incident_subscribe_participant_flow,
     incident_update_flow,
-    incident_create_resources_flow,
 )
 from .metrics import create_incident_metric_query, make_forecast
 from .models import (
@@ -47,7 +49,7 @@
     IncidentRead,
     IncidentUpdate,
 )
-from .service import create, delete, get, update, generate_incident_summary
+from .service import create, delete, get, update
 
 log = logging.getLogger(__name__)
 
@@ -509,7 +511,7 @@ def generate_summary(
     db_session: DbSession,
     current_incident: CurrentIncident,
 ):
-    return generate_incident_summary(
+    return ai_service.generate_incident_summary(
         db_session=db_session,
         incident=current_incident,
     )