diff --git a/api/test_views.py b/api/test_views.py index 6f215aedc..a54da83a1 100644 --- a/api/test_views.py +++ b/api/test_views.py @@ -31,8 +31,9 @@ DrefFactory, DrefFinalReportFactory, DrefOperationalUpdateFactory, + DrefSummaryFactory, ) -from dref.models import Dref, DrefFile +from dref.models import Dref, DrefFile, DrefSummary from main.test_case import APITestCase from per.factories import OpsLearningFactory @@ -1535,3 +1536,37 @@ def test_dref_operational_update_with_timeline_ops_updates(self): self.assertIsNotNone(dref_data) self.assertIsNotNone(dref_data["final_report_details"]) self.assertEqual(len(dref_data["timeline_operational_updates"]), 3) + + def test_dref_summary_fields_present_when_summary_exists(self): + event = EventFactory.create(dtype=self.disaster_type) + dref = self._approved_dref(event) + DrefSummaryFactory.create( + dref=dref, + status=DrefSummary.SummaryStatus.SUCCESS, + situational_overview="overview text", + operational_strategy="strategy text", + people_centered_approach="approach text", + challenges_identified="challenges text", + lessons_learned="lessons text", + ) + + data = self._get(event).data + + self.assertEqual(data["stage"], EventStage.DREF_APPLICATION) + summary = data["dref"]["summary"] + self.assertIsNotNone(summary) + self.assertEqual(summary["status"], DrefSummary.SummaryStatus.SUCCESS) + self.assertEqual(summary["situational_overview"], "overview text") + self.assertEqual(summary["operational_strategy"], "strategy text") + self.assertEqual(summary["people_centered_approach"], "approach text") + self.assertEqual(summary["challenges_identified"], "challenges text") + self.assertEqual(summary["lessons_learned"], "lessons text") + + def test_dref_summary_is_none_when_no_summary_exists(self): + event = EventFactory.create(dtype=self.disaster_type) + self._approved_dref(event) + + data = self._get(event).data + + self.assertEqual(data["stage"], EventStage.DREF_APPLICATION) + self.assertIsNone(data["dref"]["summary"]) diff --git a/assets b/assets index 780b7a6c0..03bb6f64a 160000 --- a/assets +++ b/assets @@ -1 +1 @@ -Subproject commit 780b7a6c0efc7b6e353015308c701be2d7c2c419 +Subproject commit 03bb6f64ab4e668fff21b0c6d10b799b53742284 diff --git a/docker-compose.yml b/docker-compose.yml index 6ff7c3f57..a9a8faf89 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -33,6 +33,8 @@ x-server: &base_server_setup CACHE_TEST_REDIS_URL: ${CACHE_TEST_REDIS_URL:-redis://redis:6379/11} CACHE_MIDDLEWARE_SECONDS: ${CACHE_MIDDLEWARE_SECONDS:-600} ELASTIC_SEARCH_HOST: ${ELASTIC_SEARCH_HOST:-elasticsearch://elasticsearch:9200} + # LLM + USE_DUMMY_LLM_CLIENT: ${USE_DUMMY_LLM_CLIENT:-true} # Appeal API APPEALS_USER: ${APPEALS_USER:-} APPEALS_PASS: ${APPEALS_PASS:-} diff --git a/dref/admin.py b/dref/admin.py index 02e84fa7e..390b4c4d9 100644 --- a/dref/admin.py +++ b/dref/admin.py @@ -1,13 +1,14 @@ -from django.contrib import admin +from django.contrib import admin, messages from reversion_compare.admin import CompareVersionAdmin -from lang.admin import TranslationAdmin +from lang.admin import TranslationAdmin, TranslationInlineModelAdmin from .models import ( Dref, DrefFile, DrefFinalReport, DrefOperationalUpdate, + DrefSummary, IdentifiedNeed, NationalSocietyAction, PlannedIntervention, @@ -15,6 +16,7 @@ RiskSecurity, SourceInformation, ) +from .tasks import generate_dref_summary class ReadOnlyMixin: @@ -125,8 +127,31 @@ class SourceInformationAdmin(admin.ModelAdmin): search_fields = ("source_name",) +class DrefSummaryInline(admin.StackedInline, TranslationInlineModelAdmin): + model = DrefSummary + extra = 0 + readonly_fields = ( + "status", + "source_hash", + "created_at", + "updated_at", + ) + fields = ( + "status", + "source_hash", + "situational_overview", + "operational_strategy", + "people_centered_approach", + "challenges_identified", + "lessons_learned", + "created_at", + "updated_at", + ) + + @admin.register(Dref) class DrefAdmin(CompareVersionAdmin, TranslationAdmin, admin.ModelAdmin): + inlines = [DrefSummaryInline] search_fields = ("title", "appeal_code") list_display = ( "title", @@ -301,3 +326,37 @@ def save_model(self, request, obj, form, change): @admin.register(ProposedAction) class ProposedActionAdmin(ReadOnlyMixin, admin.ModelAdmin): search_fields = ["action"] + + +@admin.register(DrefSummary) +class DrefSummaryAdmin(TranslationAdmin, admin.ModelAdmin): + list_display = ("dref", "status", "source", "source_id", "created_at", "updated_at") + list_filter = ("status",) + search_fields = ("dref__title", "dref__appeal_code") + readonly_fields = ("source_hash", "created_at", "updated_at") + autocomplete_fields = ("dref",) + actions = ["regenerate_summary"] + fields = ( + "dref", + "status", + "source_hash", + "situational_overview", + "operational_strategy", + "people_centered_approach", + "challenges_identified", + "lessons_learned", + "created_at", + "updated_at", + ) + + @admin.action(description="Regenerate summary for selected DREFs") + def regenerate_summary(self, request, queryset): + """Re-trigger summary generation from the DREF's current latest approved source.""" + queryset.filter(status=DrefSummary.SummaryStatus.PROCESSING).update(status=DrefSummary.SummaryStatus.FAILED) + for summary in queryset: + generate_dref_summary.delay(dref_id=summary.dref_id, overwrite=True) + self.message_user( + request, + f"Queued summary regeneration for {queryset.count()} DREF(s).", + messages.SUCCESS, + ) diff --git a/dref/factories/dref.py b/dref/factories/dref.py index ceabd59d0..4cc61d27a 100644 --- a/dref/factories/dref.py +++ b/dref/factories/dref.py @@ -9,6 +9,7 @@ DrefFile, DrefFinalReport, DrefOperationalUpdate, + DrefSummary, IdentifiedNeed, NationalSocietyAction, PlannedIntervention, @@ -198,3 +199,19 @@ class Meta: model = ProposedAction proposed_type = fuzzy.FuzzyChoice(ProposedAction.Action) + + +class DrefSummaryFactory(factory.django.DjangoModelFactory): + class Meta: + model = DrefSummary + + dref = factory.SubFactory(DrefFactory) + source_hash = factory.Sequence(lambda n: f"{n:064d}") + source = DrefSummary.SourceModel.DREF + source_id = factory.SelfAttribute("dref.id") + status = DrefSummary.SummaryStatus.SUCCESS + situational_overview = fuzzy.FuzzyText(length=100) + operational_strategy = fuzzy.FuzzyText(length=100) + people_centered_approach = fuzzy.FuzzyText(length=100) + challenges_identified = fuzzy.FuzzyText(length=100) + lessons_learned = fuzzy.FuzzyText(length=100) diff --git a/dref/migrations/0090_drefsummary.py b/dref/migrations/0090_drefsummary.py new file mode 100644 index 000000000..6d9b171ef --- /dev/null +++ b/dref/migrations/0090_drefsummary.py @@ -0,0 +1,54 @@ +# Generated by Django 5.2.14 on 2026-07-03 00:58 + +import django.db.models.deletion +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ('dref', '0089_remove_dref_field_report_dref_event'), + ] + + operations = [ + migrations.CreateModel( + name='DrefSummary', + fields=[ + ('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')), + ('created_at', models.DateTimeField(auto_now_add=True)), + ('updated_at', models.DateTimeField(auto_now=True)), + ('source', models.IntegerField(choices=[(100, 'Dref'), (200, 'Dref Operational Update'), (300, 'Dref Final Report')], help_text='The model this summary was generated from.', verbose_name='source')), + ('source_id', models.PositiveBigIntegerField(verbose_name='source id')), + ('source_hash', models.CharField(max_length=64, unique=True)), + ('situational_overview', models.TextField(blank=True, null=True)), + ('situational_overview_en', models.TextField(blank=True, null=True)), + ('situational_overview_es', models.TextField(blank=True, null=True)), + ('situational_overview_fr', models.TextField(blank=True, null=True)), + ('situational_overview_ar', models.TextField(blank=True, null=True)), + ('operational_strategy', models.TextField(blank=True, null=True)), + ('operational_strategy_en', models.TextField(blank=True, null=True)), + ('operational_strategy_es', models.TextField(blank=True, null=True)), + ('operational_strategy_fr', models.TextField(blank=True, null=True)), + ('operational_strategy_ar', models.TextField(blank=True, null=True)), + ('people_centered_approach', models.TextField(blank=True, null=True)), + ('people_centered_approach_en', models.TextField(blank=True, null=True)), + ('people_centered_approach_es', models.TextField(blank=True, null=True)), + ('people_centered_approach_fr', models.TextField(blank=True, null=True)), + ('people_centered_approach_ar', models.TextField(blank=True, null=True)), + ('challenges_identified', models.TextField(blank=True, null=True)), + ('challenges_identified_en', models.TextField(blank=True, null=True)), + ('challenges_identified_es', models.TextField(blank=True, null=True)), + ('challenges_identified_fr', models.TextField(blank=True, null=True)), + ('challenges_identified_ar', models.TextField(blank=True, null=True)), + ('lessons_learned', models.TextField(blank=True, null=True)), + ('lessons_learned_en', models.TextField(blank=True, null=True)), + ('lessons_learned_es', models.TextField(blank=True, null=True)), + ('lessons_learned_fr', models.TextField(blank=True, null=True)), + ('lessons_learned_ar', models.TextField(blank=True, null=True)), + ('status', models.IntegerField(choices=[(100, 'Pending'), (200, 'Processing'), (300, 'Success'), (400, 'Failed')], default=100)), + ('translation_module_original_language', models.CharField(choices=[('en', 'English'), ('es', 'Spanish'), ('fr', 'French'), ('ar', 'Arabic')], default='en', help_text='Language used to create this entity', max_length=2, verbose_name='Entity Original language')), + ('translation_module_skip_auto_translation', models.BooleanField(default=False, help_text='Skip auto translation operation for this entity?', verbose_name='Skip auto translation')), + ('dref', models.OneToOneField(on_delete=django.db.models.deletion.CASCADE, related_name='summary', to='dref.dref')), + ], + ), + ] diff --git a/dref/models.py b/dref/models.py index 4be868395..84f63b7fd 100644 --- a/dref/models.py +++ b/dref/models.py @@ -1233,6 +1233,10 @@ class DrefOperationalUpdate(models.Model): source_information = models.ManyToManyField(SourceInformation, blank=True, verbose_name=_("Source Information")) __budget_file_id = None + # TYPING + id: int + budget_file_id: int | None + class Meta: verbose_name = _("Dref Operational Update") verbose_name_plural = _("Dref Operational Updates") @@ -1657,6 +1661,9 @@ class DrefFinalReport(models.Model): ) __financial_report_id = None + # TYPING + id: int + class Meta: verbose_name = _("Dref Final Report") verbose_name_plural = _("Dref Final Reports") @@ -1700,3 +1707,70 @@ def get_for(user, status=None): if status == Dref.Status.APPROVED: return queryset.filter(status=Dref.Status.APPROVED) return queryset + + +class DrefSummary(models.Model): + + class SummaryStatus(models.IntegerChoices): + PENDING = 100, _("Pending") + PROCESSING = 200, _("Processing") + SUCCESS = 300, _("Success") + FAILED = 400, _("Failed") + + class SourceModel(models.IntegerChoices): + DREF = 100, _("Dref") + DREF_OPERATIONAL_UPDATE = 200, _("Dref Operational Update") + DREF_FINAL_REPORT = 300, _("Dref Final Report") + + dref = models.OneToOneField( + Dref, + on_delete=models.CASCADE, + related_name="summary", + ) + + created_at = models.DateTimeField(auto_now_add=True) + updated_at = models.DateTimeField(auto_now=True) + + source = models.IntegerField( + verbose_name=_("source"), + help_text=_("The model this summary was generated from."), + choices=SourceModel.choices, + ) + source_id = models.PositiveBigIntegerField( + verbose_name=_("source id"), + ) + + source_hash = models.CharField( + max_length=64, + unique=True, + ) + + situational_overview = models.TextField( + blank=True, + null=True, + ) + + operational_strategy = models.TextField( + blank=True, + null=True, + ) + + people_centered_approach = models.TextField( + blank=True, + null=True, + ) + + challenges_identified = models.TextField( + blank=True, + null=True, + ) + + lessons_learned = models.TextField( + blank=True, + null=True, + ) + + status = models.IntegerField( + choices=SummaryStatus.choices, + default=SummaryStatus.PENDING, + ) diff --git a/dref/serializers.py b/dref/serializers.py index 2a6525e57..c8c8bd1ad 100644 --- a/dref/serializers.py +++ b/dref/serializers.py @@ -25,6 +25,7 @@ DrefFile, DrefFinalReport, DrefOperationalUpdate, + DrefSummary, IdentifiedNeed, NationalSocietyAction, PlannedIntervention, @@ -2427,6 +2428,25 @@ def get_dref_contacts(self, obj): return get_dref_emergency_contacts(obj) +class DrefSummarySerializer(ModelSerializer): + status_display = serializers.CharField(source="get_status_display", read_only=True) + + class Meta: + model = DrefSummary + fields = ( + "id", + "status", + "status_display", + "situational_overview", + "operational_strategy", + "people_centered_approach", + "challenges_identified", + "lessons_learned", + "created_at", + "updated_at", + ) + + class EmergencyDrefSerializer(serializers.ModelSerializer): status_display = serializers.CharField(source="get_status_display", read_only=True) country_details = MiniCountrySerializer(source="country", read_only=True) @@ -2436,6 +2456,7 @@ class EmergencyDrefSerializer(serializers.ModelSerializer): cover_image_file = DrefFileSerializer(source="cover_image", required=False, allow_null=True) disaster_type_details = DisasterTypeSerializer(source="disaster_type", read_only=True) type_of_dref_display = serializers.CharField(source="get_type_of_dref_display", read_only=True) + summary = DrefSummarySerializer(read_only=True) # Dref operational update operational_update_details = serializers.SerializerMethodField() @@ -2498,6 +2519,8 @@ class Meta: "timeline_operational_updates", # Contacts "dref_contacts", + # Summary + "summary", ) @extend_schema_field(TimelineEmergencyDrefOperationalUpdateSerializer(many=True)) diff --git a/dref/summary.py b/dref/summary.py new file mode 100644 index 000000000..4fae765ca --- /dev/null +++ b/dref/summary.py @@ -0,0 +1,414 @@ +""" +DREF AI summary generation. +""" + +import hashlib +import json +import logging +from typing import Callable, Dict, List, Optional, Union + +import tiktoken +from django.db.models import F + +from api.utils import get_model_name +from dref.models import Dref, DrefFinalReport, DrefOperationalUpdate, DrefSummary +from main.llm import get_dref_summary_llm_client + +logger = logging.getLogger(__name__) + +ENCODING_NAME = "cl100k_base" + +MAX_OUTPUT_CHARS_PER_FIELD = 1500 +MAX_INPUT_TOKENS = 10000 + + +# The models a DrefSummary can be generated from. +DrefSummarySource = Union[Dref, DrefOperationalUpdate, DrefFinalReport] + +SOURCE_BY_MODEL: Dict[type, DrefSummary.SourceModel] = { + Dref: DrefSummary.SourceModel.DREF, + DrefOperationalUpdate: DrefSummary.SourceModel.DREF_OPERATIONAL_UPDATE, + DrefFinalReport: DrefSummary.SourceModel.DREF_FINAL_REPORT, +} + +# DrefSummary fields — order is the iteration order for prompt assembly. +SUMMARY_FIELDS: List[str] = [ + "situational_overview", + "operational_strategy", + "people_centered_approach", + "challenges_identified", + "lessons_learned", +] + +SYSTEM_MESSAGE = ( + "You are an IFRC expert analyst specializing in DREF (Disaster Response Emergency Fund) " + "operations. Analyze the provided DREF data and produce clear, professional humanitarian " + "summaries suitable for IFRC staff and National Society personnel. Use only the information " + "provided in the data; do not invent facts, figures, or details. Where supporting information " + "for a section is absent, return an empty string for that section rather than speculating." +) + +# Section prompt builders + + +def _build_situational_overview_prompt(**kwargs) -> str: + data = {k: v for k, v in kwargs.items() if v not in (None, "", [], {})} + data_json = json.dumps(data, indent=2, ensure_ascii=False, default=str) + return ( + 'Data for "situational_overview" — disaster context, affected areas, ' + f"scale of impact and strategic rationale for the operation:\n{data_json}" + ) + + +def _build_operational_strategy_prompt(**kwargs) -> str: + data = {k: v for k, v in kwargs.items() if v not in (None, "", [], {})} + data_json = json.dumps(data, indent=2, ensure_ascii=False, default=str) + return ( + 'Data for "operational_strategy" — operation objective, response strategy, ' + f"target population, timeframe and budget:\n{data_json}" + ) + + +def _build_people_centered_approach_prompt(**kwargs) -> str: + data = {k: v for k, v in kwargs.items() if v not in (None, "", [], {})} + data_json = json.dumps(data, indent=2, ensure_ascii=False, default=str) + return ( + 'Data for "people_centered_approach" — targeting criteria, beneficiary selection, ' + f"engagement and protection approach:\n{data_json}" + ) + + +def _build_challenges_identified_prompt(**kwargs) -> str: + data = {k: v for k, v in kwargs.items() if v not in (None, "", [], {})} + data_json = json.dumps(data, indent=2, ensure_ascii=False, default=str) + return ( + 'Data for "challenges_identified" — operational challenges, gaps, coordination ' + f"issues and security/safety risks recorded per planned intervention:\n{data_json}" + ) + + +def _build_lessons_learned_prompt(**kwargs) -> str: + data = {k: v for k, v in kwargs.items() if v not in (None, "", [], {})} + data_json = json.dumps(data, indent=2, ensure_ascii=False, default=str) + return ( + 'Data for "lessons_learned" — lessons learnt recorded per planned intervention ' f"(empty if none recorded):\n{data_json}" + ) + + +# Registry +SECTION_PROMPT_BUILDERS: Dict[str, Callable[..., str]] = { + "situational_overview": _build_situational_overview_prompt, + "operational_strategy": _build_operational_strategy_prompt, + "people_centered_approach": _build_people_centered_approach_prompt, + "challenges_identified": _build_challenges_identified_prompt, + "lessons_learned": _build_lessons_learned_prompt, +} + +GLOBAL_PROMPT = ( + "The DREF data above is organised by summary section. Using ONLY that data, write five concise " + "summary sections. Return a single JSON object (and nothing else) with exactly these keys:\n" + "\n" + ' "situational_overview": The disaster situation, affected areas and scale of impact, plus the ' + 'strategic rationale for the operation. Use the data under the "situational_overview" key. ' + "Write a coherent paragraph; include specific figures where available (e.g. people affected, displaced).\n" + ' "operational_strategy": The overall objective and strategic approach of the response. Use the ' + 'data under the "operational_strategy" key. State the target population, timeframe and headline ' + "interventions where available.\n" + ' "people_centered_approach": Who is targeted and how they are selected, engaged and protected. ' + 'Use the data under the "people_centered_approach" key, including the women/men/girls/boys ' + "breakdown where given.\n" + ' "challenges_identified": The challenges, gaps, coordination issues and security/safety risks. ' + 'Use the data under the "challenges_identified" key.\n' + ' "lessons_learned": The lessons learned from this operation. Use the data under the ' + '"lessons_learned" key. If none are recorded, return an empty string.\n' + "\n" + "Requirements:\n" + "- Each value must be plain text (no markdown, no bullet lists, no nested JSON).\n" + "- Each section should be one well-structured paragraph in professional humanitarian language.\n" + "- Be precise with any facts or numbers; never fabricate them.\n" + "- Return ONLY the JSON object, with no surrounding prose or code fences." +) + + +def count_tokens(text: str) -> int: + try: + return len(tiktoken.get_encoding(ENCODING_NAME).encode(text)) + except Exception as e: + logger.error(f"Error counting tokens for DREF summary: {e}") + return len(text) // 4 + + +def _field_val(obj, field_name): + """Return a model field's value, preferring human-readable display for choice fields.""" + v = getattr(obj, field_name, None) + if v in (None, ""): + return None + display = getattr(obj, f"get_{field_name}_display", None) + return display() if callable(display) else v + + +def _extract_fields(obj, field_names: List[str]) -> dict: + return {name: v for name in field_names if (v := _field_val(obj, name)) is not None} + + +# Shared field lists. + +SITUATIONAL_COMMON_FIELDS: List[str] = ["title", "event_description", "date_of_approval"] + +OPERATIONAL_COMMON_FIELDS: List[str] = [ + "operation_objective", + "response_strategy", + "total_targeted_population", + "people_in_need", + "event_date", +] + +PEOPLE_COMMON_FIELDS: List[str] = ["people_assisted", "selection_criteria"] + +DEMOGRAPHIC_FIELDS: List[str] = ["women", "men", "girls", "boys"] + + +class DrefSummaryGenerator: + """Assembles per-section prompts from a source document and produces all summaries.""" + + def __init__(self): + self.client = get_dref_summary_llm_client() + + # Shared helpers — called by multiple extractors + + @staticmethod + def _situational_overview_kwargs(source_doc) -> dict: + """Build situational_overview kwargs — common across all document types. + + ``event_scope`` is omitted for Imminent DREFs since the scope is not + yet known at that stage. + """ + is_imminent = source_doc.type_of_dref == Dref.DrefType.IMMINENT + kwargs = _extract_fields(source_doc, SITUATIONAL_COMMON_FIELDS) + if source_doc.country: + kwargs["country"] = {"name": source_doc.country.name, "iso": source_doc.country.iso} + if source_doc.disaster_type: + kwargs["disaster_type"] = {"name": source_doc.disaster_type.name} + # event_scope is not yet known for Imminent DREFs, so it is excluded there. + if not is_imminent: + kwargs.update(_extract_fields(source_doc, ["event_scope"])) + return kwargs + + @staticmethod + def _challenges_and_lessons_kwargs(source_doc) -> Dict[str, dict]: + """Collect challenges and lessons from ``planned_interventions`` M2M. + + Only called for ``DrefFinalReport`` + """ + + def pi_title(pi): + display = getattr(pi, "get_title_display", None) + return display() if callable(display) else pi.title + + # Order explicitly: planned_interventions has no Meta.ordering, so an + # unordered .all() can return rows in different orders across queries, + # which would change the source hash and trigger needless regeneration. + planned = list(source_doc.planned_interventions.order_by("id")) + return { + "challenges_identified": { + "planned_interventions": [{"title": pi_title(pi), "challenges": pi.challenges} for pi in planned if pi.challenges] + or None, + }, + "lessons_learned": { + "planned_interventions": [ + {"title": pi_title(pi), "lessons_learnt": pi.lessons_learnt} for pi in planned if pi.lessons_learnt + ] + or None, + }, + } + + @classmethod + def _extract_dref_kwargs(cls, dref) -> Dict[str, dict]: + """Dref Application / Assessment / Imminent — three sections only. + + Challenges and lessons are not applicable at the application stage; + they are formally recorded only in the Final Report. + """ + is_imminent = dref.type_of_dref == Dref.DrefType.IMMINENT + operational = _extract_fields(dref, OPERATIONAL_COMMON_FIELDS + ["amount_requested", "end_date"]) + # Normalise to a single "operation_timeframe" key regardless of source field. + timeframe = _field_val(dref, "operation_timeframe_imminent" if is_imminent else "operation_timeframe") + if timeframe is not None: + operational["operation_timeframe"] = timeframe + return { + "situational_overview": cls._situational_overview_kwargs(dref), + "operational_strategy": operational, + "people_centered_approach": _extract_fields(dref, PEOPLE_COMMON_FIELDS), + } + + @classmethod + def _extract_dref_ops_kwargs(cls, ops) -> Dict[str, dict]: + """DrefOperationalUpdate — three sections; different field names from Dref. + + Includes demographic breakdown (women/men/girls/boys) in people_centered_approach. + Challenges and lessons are not generated for Operational Updates. + """ + operational = _extract_fields(ops, OPERATIONAL_COMMON_FIELDS + ["total_dref_allocation"]) + # Ops Update names these differently; normalise to the shared key names. + operational.update( + { + k: v + for k, v in { + "operation_end_date": _field_val(ops, "new_operational_end_date"), + "operation_timeframe": _field_val(ops, "total_operation_timeframe"), + }.items() + if v is not None + } + ) + return { + "situational_overview": cls._situational_overview_kwargs(ops), + "operational_strategy": operational, + "people_centered_approach": _extract_fields(ops, PEOPLE_COMMON_FIELDS + DEMOGRAPHIC_FIELDS), + } + + @classmethod + def _extract_dref_final_kwargs(cls, final) -> Dict[str, dict]: + """DrefFinalReport — all five sections. + + Challenges and lessons come from ``planned_interventions`` M2M via + ``_challenges_and_lessons_kwargs``; this is the only document type + where those two sections are generated. + """ + is_imminent = final.type_of_dref == Dref.DrefType.IMMINENT + operational = _extract_fields(final, OPERATIONAL_COMMON_FIELDS + ["total_dref_allocation", "operation_end_date"]) + # Normalise to a single "operation_timeframe" key regardless of source field. + timeframe = _field_val(final, "total_operation_timeframe_imminent" if is_imminent else "total_operation_timeframe") + if timeframe is not None: + operational["operation_timeframe"] = timeframe + kwargs = { + "situational_overview": cls._situational_overview_kwargs(final), + "operational_strategy": operational, + "people_centered_approach": _extract_fields(final, PEOPLE_COMMON_FIELDS + DEMOGRAPHIC_FIELDS), + } + kwargs.update(cls._challenges_and_lessons_kwargs(final)) + return kwargs + + @classmethod + def get_section_kwargs(cls, source_doc: DrefSummarySource) -> Dict[str, dict]: + """Dispatch to the model-specific extractor. + + Public so callers that need both the hash and the generated summary + for the same ``source_doc`` (see ``compute_source_hash``/``generate_all``) + can compute this once and pass it to both instead of extracting twice. + """ + if isinstance(source_doc, Dref): + return cls._extract_dref_kwargs(source_doc) + if isinstance(source_doc, DrefOperationalUpdate): + return cls._extract_dref_ops_kwargs(source_doc) + if isinstance(source_doc, DrefFinalReport): + return cls._extract_dref_final_kwargs(source_doc) + return {} + + @staticmethod + def get_latest_approved_source(dref: Dref) -> Optional[tuple[DrefSummary.SourceModel, DrefSummarySource]]: + """Most authoritative (source, source object) pair for ``dref``. + + Priority: Final Report > latest Operational Update > Dref itself. + """ + final_report = ( + DrefFinalReport.objects.select_related("country", "disaster_type") + .filter(dref=dref, status=Dref.Status.APPROVED) + .order_by("-created_at") + .first() + ) + if final_report: + return SOURCE_BY_MODEL[DrefFinalReport], final_report + + latest_ops_update = ( + DrefOperationalUpdate.objects.select_related("country", "disaster_type") + .filter(dref=dref, status=Dref.Status.APPROVED) + .order_by(F("operational_update_number").desc(nulls_last=True), "-created_at") + .first() + ) + if latest_ops_update: + return SOURCE_BY_MODEL[DrefOperationalUpdate], latest_ops_update + + if dref.status == Dref.Status.APPROVED: + return SOURCE_BY_MODEL[Dref], dref + + return None + + @classmethod + def build_source_text(cls, source_doc: DrefSummarySource) -> str: + """JSON representation of all section kwargs for inspection and debugging.""" + return json.dumps(cls.get_section_kwargs(source_doc), indent=2, ensure_ascii=False, default=str) + + @classmethod + def compute_source_hash(cls, source_doc: DrefSummarySource, section_kwargs: Optional[Dict[str, dict]] = None) -> str: + """Hash of all source content feeding the summary, for change detection.""" + model_label = get_model_name(type(source_doc)) + if section_kwargs is None: + section_kwargs = cls.get_section_kwargs(source_doc) + payload = {"model": model_label, "id": source_doc.id, "source": section_kwargs} + content = json.dumps(payload, sort_keys=True, ensure_ascii=False, default=str) + return hashlib.sha256(content.encode("utf-8")).hexdigest() + + @staticmethod + def _parse_response(response: str) -> Dict[str, str]: + """Parse the model's JSON object, tolerating ```json code fences.""" + text = response.strip() + if text.startswith("```"): + text = text.split("\n", 1)[1] if "\n" in text else text + text = text.rsplit("```", 1)[0] + data = json.loads(text) + if not isinstance(data, dict): + raise ValueError("Expected a JSON object of summary fields") + return data + + def generate_all(self, source_doc: DrefSummarySource, section_kwargs: Optional[Dict[str, dict]] = None) -> Dict[str, str]: + """Generate every summary for ``source_doc`` in a single LLM call. + + Always returns a value for every ``SUMMARY_FIELDS`` key (empty string + when a section has no content), so callers overwrite stale values + instead of leaving them in place on regeneration. ``section_kwargs`` + can be passed in by a caller that already computed it (e.g. via + ``compute_source_hash``) to avoid re-extracting it from ``source_doc``. + """ + empty_results: Dict[str, str] = {field_name: "" for field_name in SUMMARY_FIELDS} + + if section_kwargs is None: + section_kwargs = self.get_section_kwargs(source_doc) + if not section_kwargs: + logger.info(f"No source content for ({type(source_doc).__name__}) ({source_doc.id}) summary; skipping generation.") + return empty_results + + # Call each builder with only its section's kwargs, then join blocks. + section_blocks = [SECTION_PROMPT_BUILDERS[section](**section_kwargs.get(section, {})) for section in SUMMARY_FIELDS] + user_content = "\n\n".join(section_blocks) + "\n\n" + GLOBAL_PROMPT + + # Respect the token budget: trim the data blocks if the prompt is too long. + if count_tokens(SYSTEM_MESSAGE + user_content) > MAX_INPUT_TOKENS: + logger.warning(f"({type(source_doc).__name__}) ({source_doc.id}) summary prompt too long; truncating source text.") + char_budget = MAX_INPUT_TOKENS * 4 - len(GLOBAL_PROMPT) - len(SYSTEM_MESSAGE) + data_blocks = "\n\n".join(section_blocks) + user_content = data_blocks[: max(char_budget, 0)] + "\n\n" + GLOBAL_PROMPT + + messages = [ + {"role": "system", "content": SYSTEM_MESSAGE}, + {"role": "user", "content": user_content}, + ] + + response = self.client.get_response(messages) + if not response: + raise RuntimeError(f"No LLM response for ({type(source_doc).__name__}) ({source_doc.id}) summary") + + try: + parsed = self._parse_response(response) + except (json.JSONDecodeError, ValueError) as e: + raise RuntimeError(f"Could not parse ({type(source_doc).__name__}) ({source_doc.id}) summary response: {e}") from e + + results = dict(empty_results) + for field_name in SUMMARY_FIELDS: + value = parsed.get(field_name) + if not isinstance(value, str): + continue + summary = value.strip() + if len(summary) > MAX_OUTPUT_CHARS_PER_FIELD: + summary = summary[:MAX_OUTPUT_CHARS_PER_FIELD].rstrip() + results[field_name] = summary + return results diff --git a/dref/tasks.py b/dref/tasks.py index 0b0d08c96..fde4ef745 100644 --- a/dref/tasks.py +++ b/dref/tasks.py @@ -1,17 +1,25 @@ import logging +from datetime import timedelta +from enum import Enum +from typing import Optional from celery import shared_task from django.apps import apps +from django.db import transaction from django.template.loader import render_to_string +from django.utils import timezone from api.utils import get_model_name from lang.tasks import translate_model_fields +from main.lock import RedisLockKey, redis_lock from main.translation import TRANSLATOR_ORIGINAL_LANGUAGE_FIELD_NAME +from main.utils import logger_context from notifications.notification import send_notification from .models import ( Dref, DrefFile, + DrefSummary, IdentifiedNeed, NationalSocietyAction, PlannedIntervention, @@ -21,10 +29,27 @@ RiskSecurity, SourceInformation, ) +from .summary import DrefSummaryGenerator from .utils import get_email_context logger = logging.getLogger(__name__) +# The PROCESSING status has no TTL of its own (see comment in generate_dref_summary +# below), so a row can get stuck PROCESSING forever if a worker dies mid-generation. +# Treat it as stale past this, so a later trigger can take over. +PROCESSING_STALE_AFTER = timedelta(minutes=10) + + +class DrefSummaryGenerationResult(str, Enum): + """Outcome of a ``generate_dref_summary`` run.""" + + SUCCESS = "success" + SOURCE_NOT_FOUND = "source_not_found" + ALREADY_IN_PROGRESS = "already_in_progress" + UP_TO_DATE = "up_to_date" + FAILED = "failed" + SUPERSEDED = "superseded" + @shared_task def send_dref_email(dref_id, users_emails, new_or_updated=""): @@ -57,6 +82,112 @@ def send_dref_email(dref_id, users_emails, new_or_updated=""): ] +@shared_task(soft_time_limit=600, time_limit=630) +def generate_dref_summary(dref_id: int, overwrite: bool = False) -> DrefSummaryGenerationResult: + """Generate and store the AI-assisted summaries for a DREF. + + Always (re)generates from whichever approved source is currently latest + for the DREF (see ``get_latest_approved_source``), rather than a + specific source passed in, so it self-corrects no matter which + approval triggered it or the order concurrent runs execute in. + """ + dref = Dref.objects.select_related("country", "disaster_type").filter(id=dref_id).first() + if not dref: + logger.error("Dref not found for summary generation", extra=logger_context({"dref_id": dref_id})) + return DrefSummaryGenerationResult.SOURCE_NOT_FOUND + + # The Redis lock only needs to guard the brief read-check-mark-PROCESSING + # section below against a concurrent trigger for the same DREF (double- + # approve, admin retrigger, ...); it is released before the LLM call. + # The PROCESSING status written inside the lock is what actually blocks + # a second run for as long as generation takes - unlike the lock, it has + # no TTL, so it still holds even if generation outlives lock_expire. + with redis_lock(key=RedisLockKey.DREF_SUMMARY, id=dref_id) as acquired: + if not acquired: + logger.warning(f"DREF summary generation already in progress for DREF ({dref_id}); skipping.") + return DrefSummaryGenerationResult.ALREADY_IN_PROGRESS + + latest_source = DrefSummaryGenerator.get_latest_approved_source(dref) + if not latest_source: + logger.error(f"No approved source found for DREF ({dref_id}) summary") + return DrefSummaryGenerationResult.SOURCE_NOT_FOUND + source_type, source_obj = latest_source + + section_kwargs = DrefSummaryGenerator.get_section_kwargs(source_obj) + source_hash = DrefSummaryGenerator.compute_source_hash(source_obj, section_kwargs=section_kwargs) + summary_instance: Optional[DrefSummary] = DrefSummary.objects.filter(dref=dref).first() + + if ( + summary_instance + and summary_instance.status == DrefSummary.SummaryStatus.PROCESSING + and timezone.now() - summary_instance.updated_at < PROCESSING_STALE_AFTER + and summary_instance.source == source_type + and summary_instance.source_id == source_obj.id + ): + # Same source already in flight; a different one falls through to regenerate below. + logger.warning(f"DREF summary already in progress for DREF ({dref_id}); skipping.") + return DrefSummaryGenerationResult.ALREADY_IN_PROGRESS + + if ( + summary_instance + and not overwrite + and summary_instance.source_hash == source_hash + and summary_instance.status == DrefSummary.SummaryStatus.SUCCESS + ): + logger.info(f"DREF summary up to date for DREF ({dref_id}); skipping generation.") + return DrefSummaryGenerationResult.UP_TO_DATE + + if summary_instance is None: + summary_instance = DrefSummary(dref=dref) + + try: + summary_instance.source_hash = source_hash + summary_instance.source = source_type + summary_instance.source_id = source_obj.id + summary_instance.status = DrefSummary.SummaryStatus.PROCESSING + summary_instance.save() + except Exception: + logger.warning(f"Failed to mark DREF summary as processing for DREF ({dref_id})", exc_info=True) + if summary_instance.pk: + summary_instance.status = DrefSummary.SummaryStatus.FAILED + summary_instance.save() + return DrefSummaryGenerationResult.FAILED + + # Lock released. The (possibly slow) LLM call runs unlocked; the + # PROCESSING status set above is what a concurrent trigger checks. + own_marker = { + "pk": summary_instance.pk, + "source": source_type, + "source_id": source_obj.id, + "status": DrefSummary.SummaryStatus.PROCESSING, + } + try: + logger.info(f"Generating DREF summaries for DREF ({dref_id}) from ({source_type.label}) ({source_obj.id})") + results = DrefSummaryGenerator().generate_all(source_obj, section_kwargs=section_kwargs) + with transaction.atomic(): + latest_summary_instance = DrefSummary.objects.select_for_update().filter(**own_marker).first() + if latest_summary_instance is None: + logger.warning(f"DREF summary run for DREF ({dref_id}) was superseded by a newer trigger; discarding result.") + return DrefSummaryGenerationResult.SUPERSEDED + for field_name, value in results.items(): + setattr(latest_summary_instance, field_name, value) + latest_summary_instance.status = DrefSummary.SummaryStatus.SUCCESS + latest_summary_instance.save() + logger.info(f"Successfully generated DREF summaries for DREF ({dref_id})") + transaction.on_commit(lambda: translate_model_fields.delay(get_model_name(DrefSummary), latest_summary_instance.pk)) + return DrefSummaryGenerationResult.SUCCESS + except Exception: + with transaction.atomic(): + latest_summary_instance = DrefSummary.objects.select_for_update().filter(**own_marker).first() + if latest_summary_instance is None: + logger.warning(f"DREF summary run for DREF ({dref_id}) failed but was already superseded; leaving it as is.") + else: + latest_summary_instance.status = DrefSummary.SummaryStatus.FAILED + latest_summary_instance.save() + logger.warning(f"DREF summary generation failed for DREF ({dref_id})", exc_info=True) + return DrefSummaryGenerationResult.FAILED + + @shared_task def process_dref_translation(model_name, instance_pk): """ diff --git a/dref/test_summary.py b/dref/test_summary.py new file mode 100644 index 000000000..8c725356b --- /dev/null +++ b/dref/test_summary.py @@ -0,0 +1,327 @@ +from datetime import date + +from django.test import TestCase, override_settings + +from dref.factories.dref import ( + DrefFactory, + DrefFinalReportFactory, + DrefOperationalUpdateFactory, + PlannedInterventionFactory, +) +from dref.models import Dref, DrefSummary +from dref.summary import SUMMARY_FIELDS, DrefSummaryGenerator +from dref.tasks import DrefSummaryGenerationResult, generate_dref_summary +from main.llm import ( + AzureOpenAiChat, + DummyDrefSummaryLLMClient, + get_dref_summary_llm_client, +) + + +class DrefSummaryLLMClientTest(TestCase): + def test_uses_dummy_client_during_tests_by_default(self): + # settings.TESTING is always true under the test runner, so the + # factory should return the dummy client even though + # USE_DUMMY_LLM_CLIENT itself defaults to False. + self.assertIsInstance(get_dref_summary_llm_client(), DummyDrefSummaryLLMClient) + + @override_settings( + TESTING=False, + USE_DUMMY_LLM_CLIENT=False, + AZURE_OPENAI_ENDPOINT="https://example.com", + AZURE_OPENAI_KEY="fake-key", + AZURE_OPENAI_DEPLOYMENT_NAME="fake-deployment", + ) + def test_uses_real_client_when_flag_off_and_not_testing(self): + self.assertIsInstance(get_dref_summary_llm_client(), AzureOpenAiChat) + + @override_settings(TESTING=False, USE_DUMMY_LLM_CLIENT=True) + def test_uses_dummy_client_when_flag_on(self): + self.assertIsInstance(get_dref_summary_llm_client(), DummyDrefSummaryLLMClient) + + +class DrefSummaryGeneratorTest(TestCase): + def test_generate_all_succeeds_end_to_end_for_dref(self): + dref = DrefFactory.create(title="Test Dref") + + results = DrefSummaryGenerator().generate_all(dref) + + self.assertEqual(set(results.keys()), set(SUMMARY_FIELDS)) + self.assertIn("DUMMY RESPONSE", results["situational_overview"]) + + def test_generate_all_succeeds_end_to_end_for_dref_operational_update(self): + ops_update = DrefOperationalUpdateFactory.create(dref=DrefFactory.create(), title="Test Ops Update") + + results = DrefSummaryGenerator().generate_all(ops_update) + + self.assertEqual(set(results.keys()), set(SUMMARY_FIELDS)) + self.assertIn("DUMMY RESPONSE", results["situational_overview"]) + + def test_generate_all_succeeds_end_to_end_for_dref_final_report(self): + planned_intervention = PlannedInterventionFactory.create(challenges="Some challenge", lessons_learnt="Some lesson") + final_report = DrefFinalReportFactory.create(title="Test Final Report", planned_interventions=[planned_intervention]) + + results = DrefSummaryGenerator().generate_all(final_report) + + self.assertEqual(set(results.keys()), set(SUMMARY_FIELDS)) + self.assertIn("DUMMY RESPONSE", results["situational_overview"]) + self.assertIn("DUMMY RESPONSE", results["challenges_identified"]) + self.assertIn("DUMMY RESPONSE", results["lessons_learned"]) + + def test_get_section_kwargs_for_dref(self): + dref = DrefFactory.create( + title="Test Dref", + event_description="Severe flooding", + type_of_dref=Dref.DrefType.RESPONSE, + event_scope="Affected 3 districts", + operation_objective="Provide shelter", + response_strategy="Cash and shelter support", + total_targeted_population=5000, + people_in_need=8000, + people_assisted="5000 people", + selection_criteria="Most vulnerable households", + ) + + kwargs = DrefSummaryGenerator.get_section_kwargs(dref) + + self.assertEqual(set(kwargs.keys()), {"situational_overview", "operational_strategy", "people_centered_approach"}) + self.assertEqual(kwargs["situational_overview"]["title"], "Test Dref") + self.assertEqual(kwargs["situational_overview"]["event_description"], "Severe flooding") + self.assertEqual(kwargs["situational_overview"]["event_scope"], "Affected 3 districts") + self.assertEqual(kwargs["operational_strategy"]["operation_objective"], "Provide shelter") + self.assertEqual(kwargs["operational_strategy"]["response_strategy"], "Cash and shelter support") + self.assertEqual(kwargs["operational_strategy"]["total_targeted_population"], 5000) + self.assertEqual(kwargs["operational_strategy"]["people_in_need"], 8000) + self.assertEqual(kwargs["people_centered_approach"]["people_assisted"], "5000 people") + self.assertEqual(kwargs["people_centered_approach"]["selection_criteria"], "Most vulnerable households") + self.assertNotIn("women", kwargs["people_centered_approach"]) + + def test_get_section_kwargs_for_dref_operational_update(self): + ops_update = DrefOperationalUpdateFactory.create( + dref=DrefFactory.create(), + title="Test Ops Update", + event_description="Flooding continues", + operation_objective="Extend shelter support", + response_strategy="Extended cash support", + total_dref_allocation=100000, + people_assisted="6000 people", + selection_criteria="Vulnerable households", + women=1000, + men=900, + girls=500, + boys=600, + new_operational_end_date=date(2025, 6, 1), + total_operation_timeframe=6, + ) + + kwargs = DrefSummaryGenerator.get_section_kwargs(ops_update) + + self.assertEqual(set(kwargs.keys()), {"situational_overview", "operational_strategy", "people_centered_approach"}) + self.assertEqual(kwargs["situational_overview"]["title"], "Test Ops Update") + self.assertEqual(kwargs["situational_overview"]["event_description"], "Flooding continues") + self.assertEqual(kwargs["operational_strategy"]["operation_objective"], "Extend shelter support") + self.assertEqual(kwargs["operational_strategy"]["total_dref_allocation"], 100000) + self.assertEqual(kwargs["operational_strategy"]["operation_end_date"], date(2025, 6, 1)) + self.assertEqual(kwargs["operational_strategy"]["operation_timeframe"], 6) + self.assertEqual(kwargs["people_centered_approach"]["people_assisted"], "6000 people") + self.assertEqual(kwargs["people_centered_approach"]["women"], 1000) + self.assertEqual(kwargs["people_centered_approach"]["men"], 900) + self.assertEqual(kwargs["people_centered_approach"]["girls"], 500) + self.assertEqual(kwargs["people_centered_approach"]["boys"], 600) + + def test_get_section_kwargs_for_dref_final_report(self): + planned_intervention = PlannedInterventionFactory.create( + challenges="Access constraints due to weather", + lessons_learnt="Earlier pre-positioning of stock helps", + ) + final_report = DrefFinalReportFactory.create( + title="Test Final Report", + event_description="Response concluded", + operation_objective="Deliver relief items", + response_strategy="Direct distribution", + total_dref_allocation=150000, + people_assisted="7000 people", + selection_criteria="Most affected households", + women=1200, + men=1100, + girls=600, + boys=700, + planned_interventions=[planned_intervention], + ) + + kwargs = DrefSummaryGenerator.get_section_kwargs(final_report) + + self.assertEqual( + set(kwargs.keys()), + { + "situational_overview", + "operational_strategy", + "people_centered_approach", + "challenges_identified", + "lessons_learned", + }, + ) + self.assertEqual(kwargs["situational_overview"]["title"], "Test Final Report") + self.assertEqual(kwargs["operational_strategy"]["total_dref_allocation"], 150000) + self.assertEqual(kwargs["people_centered_approach"]["women"], 1200) + + challenges = kwargs["challenges_identified"]["planned_interventions"] + self.assertEqual(len(challenges), 1) + self.assertEqual(challenges[0]["challenges"], "Access constraints due to weather") + + lessons = kwargs["lessons_learned"]["planned_interventions"] + self.assertEqual(len(lessons), 1) + self.assertEqual(lessons[0]["lessons_learnt"], "Earlier pre-positioning of stock helps") + + def test_get_section_kwargs_excludes_event_scope_for_imminent_dref(self): + imminent_dref = DrefFactory.create(type_of_dref=Dref.DrefType.IMMINENT, event_scope="scope text") + self.assertNotIn("event_scope", DrefSummaryGenerator.get_section_kwargs(imminent_dref)["situational_overview"]) + + response_dref = DrefFactory.create(type_of_dref=Dref.DrefType.RESPONSE, event_scope="scope text") + self.assertIn("event_scope", DrefSummaryGenerator.get_section_kwargs(response_dref)["situational_overview"]) + + def test_compute_source_hash_changes_with_content_and_is_deterministic(self): + dref = DrefFactory.create(title="Original Title", type_of_dref=Dref.DrefType.RESPONSE) + + hash_a = DrefSummaryGenerator.compute_source_hash(dref) + hash_b = DrefSummaryGenerator.compute_source_hash(dref) + self.assertEqual(hash_a, hash_b) + + dref.title = "Changed Title" + dref.save(update_fields=["title"]) + hash_c = DrefSummaryGenerator.compute_source_hash(dref) + self.assertNotEqual(hash_a, hash_c) + + def test_compute_source_hash_uses_provided_section_kwargs(self): + dref = DrefFactory.create(title="Test Dref", type_of_dref=Dref.DrefType.RESPONSE) + default_hash = DrefSummaryGenerator.compute_source_hash(dref) + + actual_kwargs = DrefSummaryGenerator.get_section_kwargs(dref) + self.assertEqual(default_hash, DrefSummaryGenerator.compute_source_hash(dref, section_kwargs=actual_kwargs)) + + custom_kwargs = {"situational_overview": {"title": "overridden"}} + self.assertNotEqual(default_hash, DrefSummaryGenerator.compute_source_hash(dref, section_kwargs=custom_kwargs)) + + def test_get_latest_approved_source_returns_none_when_dref_not_approved(self): + dref = DrefFactory.create(status=Dref.Status.FINALIZED) + self.assertIsNone(DrefSummaryGenerator.get_latest_approved_source(dref)) + + def test_get_latest_approved_source_follows_priority_order(self): + dref = DrefFactory.create(status=Dref.Status.APPROVED) + self.assertEqual(DrefSummaryGenerator.get_latest_approved_source(dref), (DrefSummary.SourceModel.DREF, dref)) + + # Unapproved updates don't count, so the Dref itself still wins. + DrefOperationalUpdateFactory.create(dref=dref, status=Dref.Status.FINALIZED, operational_update_number=1) + self.assertEqual(DrefSummaryGenerator.get_latest_approved_source(dref), (DrefSummary.SourceModel.DREF, dref)) + + # An approved Operational Update supersedes the Dref; the higher + # operational_update_number wins, regardless of insertion order. + DrefOperationalUpdateFactory.create(dref=dref, status=Dref.Status.APPROVED, operational_update_number=1) + ops_update_2 = DrefOperationalUpdateFactory.create(dref=dref, status=Dref.Status.APPROVED, operational_update_number=2) + self.assertEqual( + DrefSummaryGenerator.get_latest_approved_source(dref), + (DrefSummary.SourceModel.DREF_OPERATIONAL_UPDATE, ops_update_2), + ) + + # An approved Final Report supersedes everything else. + final_report = DrefFinalReportFactory.create(dref=dref, status=Dref.Status.APPROVED) + self.assertEqual( + DrefSummaryGenerator.get_latest_approved_source(dref), + (DrefSummary.SourceModel.DREF_FINAL_REPORT, final_report), + ) + + def test_returns_source_not_found_when_dref_missing_or_nothing_approved(self): + self.assertEqual(generate_dref_summary(dref_id=999999), DrefSummaryGenerationResult.SOURCE_NOT_FOUND) + + dref = DrefFactory.create(status=Dref.Status.FINALIZED) + self.assertEqual(generate_dref_summary(dref_id=dref.id), DrefSummaryGenerationResult.SOURCE_NOT_FOUND) + + def test_generates_from_dref_and_is_up_to_date_on_rerun(self): + dref = DrefFactory.create(status=Dref.Status.APPROVED) + + self.assertEqual(generate_dref_summary(dref_id=dref.id), DrefSummaryGenerationResult.SUCCESS) + summary = DrefSummary.objects.get(dref=dref) + self.assertEqual(summary.source, DrefSummary.SourceModel.DREF) + self.assertEqual(summary.source_id, dref.id) + + # Nothing changed since - re-running should not regenerate. + self.assertEqual(generate_dref_summary(dref_id=dref.id), DrefSummaryGenerationResult.UP_TO_DATE) + + def test_retrigger_with_overwrite_regenerates_even_when_up_to_date(self): + dref = DrefFactory.create(status=Dref.Status.APPROVED) + + self.assertEqual(generate_dref_summary(dref_id=dref.id), DrefSummaryGenerationResult.SUCCESS) + summary = DrefSummary.objects.get(dref=dref) + first_hash = summary.source_hash + first_updated_at = summary.updated_at + + # Nothing changed since, so a plain retrigger would be UP_TO_DATE ... + self.assertEqual(generate_dref_summary(dref_id=dref.id), DrefSummaryGenerationResult.UP_TO_DATE) + + # ... but overwrite=True must bypass that check and regenerate anyway. + self.assertEqual(generate_dref_summary(dref_id=dref.id, overwrite=True), DrefSummaryGenerationResult.SUCCESS) + + self.assertEqual(DrefSummary.objects.filter(dref=dref).count(), 1) + summary.refresh_from_db() + self.assertEqual(summary.status, DrefSummary.SummaryStatus.SUCCESS) + self.assertEqual(summary.source, DrefSummary.SourceModel.DREF) + self.assertEqual(summary.source_id, dref.id) + self.assertEqual(summary.source_hash, first_hash) + self.assertGreater(summary.updated_at, first_updated_at) + + def test_processing_in_flight_skips_same_source_but_regenerates_for_a_newer_one(self): + dref = DrefFactory.create(status=Dref.Status.APPROVED) + # Simulate a run already in flight for the Dref itself. + DrefSummary.objects.create( + dref=dref, + source=DrefSummary.SourceModel.DREF, + source_id=dref.id, + source_hash="in-flight-hash", + status=DrefSummary.SummaryStatus.PROCESSING, + ) + + # A duplicate trigger for the same in-flight source is dropped. + self.assertEqual(generate_dref_summary(dref_id=dref.id), DrefSummaryGenerationResult.ALREADY_IN_PROGRESS) + + # A newer approval lands while that run is still (supposedly) processing. + ops_update = DrefOperationalUpdateFactory.create(dref=dref, status=Dref.Status.APPROVED, operational_update_number=1) + + self.assertEqual(generate_dref_summary(dref_id=dref.id), DrefSummaryGenerationResult.SUCCESS) + summary = DrefSummary.objects.get(dref=dref) + self.assertEqual(summary.source, DrefSummary.SourceModel.DREF_OPERATIONAL_UPDATE) + self.assertEqual(summary.source_id, ops_update.id) + + def test_generate_all_succeeds_for_newer_ops_update_while_older_one_in_flight(self): + """Two Operational Updates on one Dref: a run in flight for the older + one must not block generation once a newer one is approved.""" + dref = DrefFactory.create(status=Dref.Status.APPROVED) + ops_update_1 = DrefOperationalUpdateFactory.create(dref=dref, status=Dref.Status.APPROVED, operational_update_number=1) + DrefSummary.objects.create( + dref=dref, + source=DrefSummary.SourceModel.DREF_OPERATIONAL_UPDATE, + source_id=ops_update_1.id, + source_hash="in-flight-hash", + status=DrefSummary.SummaryStatus.PROCESSING, + ) + + # A duplicate trigger for the same in-flight source (ops_update_1) is dropped. + self.assertEqual(generate_dref_summary(dref_id=dref.id), DrefSummaryGenerationResult.ALREADY_IN_PROGRESS) + + # A second, newer Operational Update is approved while that run is still (supposedly) processing. + ops_update_2 = DrefOperationalUpdateFactory.create(dref=dref, status=Dref.Status.APPROVED, operational_update_number=2) + + # The in-flight record is for a now-stale source (ops_update_1), so this + # trigger falls through and regenerates from ops_update_2 instead of + # being blocked as "already in progress". + self.assertEqual(generate_dref_summary(dref_id=dref.id), DrefSummaryGenerationResult.SUCCESS) + + # Exactly one DrefSummary row for the Dref - updated in place, not duplicated. + self.assertEqual(DrefSummary.objects.filter(dref=dref).count(), 1) + summary = DrefSummary.objects.get(dref=dref) + self.assertEqual(summary.source, DrefSummary.SourceModel.DREF_OPERATIONAL_UPDATE) + self.assertEqual(summary.source_id, ops_update_2.id) + self.assertEqual(summary.status, DrefSummary.SummaryStatus.SUCCESS) + self.assertNotEqual(summary.source_hash, "in-flight-hash") + + # Nothing changed since - re-running should not regenerate. + self.assertEqual(generate_dref_summary(dref_id=dref.id), DrefSummaryGenerationResult.UP_TO_DATE) diff --git a/dref/translation.py b/dref/translation.py index 1c3b29d50..58a7fc986 100644 --- a/dref/translation.py +++ b/dref/translation.py @@ -5,6 +5,7 @@ DrefFile, DrefFinalReport, DrefOperationalUpdate, + DrefSummary, IdentifiedNeed, NationalSocietyAction, PlannedIntervention, @@ -116,6 +117,17 @@ class DrefOperationalUpdateTO(TranslationOptions): ) +@register(DrefSummary) +class DrefSummaryTO(TranslationOptions): + fields = ( + "situational_overview", + "operational_strategy", + "people_centered_approach", + "challenges_identified", + "lessons_learned", + ) + + @register(IdentifiedNeed) class IdentifiedNeedTO(TranslationOptions): fields = ("description",) diff --git a/dref/views.py b/dref/views.py index 2633f9798..7dfe58e34 100644 --- a/dref/views.py +++ b/dref/views.py @@ -50,7 +50,7 @@ DrefShareUserSerializer, MiniDrefSerializer, ) -from dref.tasks import process_dref_translation +from dref.tasks import generate_dref_summary, process_dref_translation from dref.utils import create_event_from_dref from lang.serializers import TranslatedModelSerializerMixin from main.permissions import DenyGuestUserPermission @@ -137,7 +137,7 @@ def get_approved(self, request, pk=None, version=None): dref.status = Dref.Status.APPROVED dref.save(update_fields=["event", "status"]) - + transaction.on_commit(lambda: generate_dref_summary.delay(dref_id=dref.id)) return response.Response(DrefSerializer(dref, context={"request": request}).data) @extend_schema(request=None, responses=DrefSerializer) @@ -234,6 +234,7 @@ def get_approved(self, request, pk=None, version=None): operational_update.status = Dref.Status.APPROVED operational_update.date_of_approval = timezone.now().date() operational_update.save(update_fields=["status", "date_of_approval"]) + transaction.on_commit(lambda: generate_dref_summary.delay(dref_id=operational_update.dref_id)) serializer = DrefOperationalUpdateSerializer(operational_update, context={"request": request}) return response.Response(serializer.data) @@ -302,6 +303,7 @@ def get_approved(self, request, pk=None, version=None): final_report.dref.is_active = False final_report.date_of_approval = timezone.now().date() final_report.dref.save(update_fields=["is_active", "date_of_approval"]) + transaction.on_commit(lambda: generate_dref_summary.delay(dref_id=final_report.dref_id)) serializer = DrefFinalReportSerializer(final_report, context={"request": request}) return response.Response(serializer.data) diff --git a/main/llm.py b/main/llm.py new file mode 100644 index 000000000..87a33e4ff --- /dev/null +++ b/main/llm.py @@ -0,0 +1,183 @@ +import json +import logging +import re +from typing import Dict, List, Optional + +from django.conf import settings +from django.utils.functional import cached_property +from openai import AzureOpenAI + +logger = logging.getLogger(__name__) + +# A chat completion prompt: the ``[{"role": ..., "content": ...}, ...]`` list +# every real call site builds and passes in. +Message = Dict[str, str] +Messages = List[Message] + + +PLACEHOLDER_TEXT = ( + "This is placeholder text standing in for a real AI-generated summary. It is returned by " + "the dummy LLM client Lorem ipsum dolor sit amet, consectetur adipiscing elit. Sed do eiusmod tempor incididunt " + "ut labore et dolore magna aliqua. Ut enim ad minim veniam, quis nostrud exercitation ullamco " + "laboris nisi ut aliquip ex ea commodo consequat. Duis aute irure dolor in reprehenderit in " + "voluptate velit esse cillum dolore eu fugiat nulla pariatur." +) + + +class BaseLLMClient: + def get_response(self, messages: Messages) -> Optional[str]: + raise NotImplementedError + + @staticmethod + def _last_user_message(messages: Messages) -> str: + return next((message["content"] for message in reversed(messages) if message["role"] == "user"), "") + + def _fake_content(self) -> str: + """Placeholder text every dummy response uses in place of real LLM output.""" + return f"DUMMY RESPONSE, {PLACEHOLDER_TEXT}" + + +class AzureOpenAiChat(BaseLLMClient): + """The real Azure OpenAI-backed client, shared by every feature.""" + + def __init__(self, temperature: float = 0.5): + self.temperature = temperature + + if settings.TESTING: + # Tests never hit Azure, so a missing/placeholder config shouldn't fail construction. + return + + if not settings.AZURE_OPENAI_ENDPOINT or not settings.AZURE_OPENAI_KEY or not settings.AZURE_OPENAI_DEPLOYMENT_NAME: + raise Exception("Azure OpenAI configuration missing") + + @cached_property + def client(self): + return AzureOpenAI( + azure_endpoint=settings.AZURE_OPENAI_ENDPOINT, + api_key=settings.AZURE_OPENAI_KEY, + api_version="2023-05-15", + ) + + def get_response(self, messages: Messages) -> Optional[str]: + if settings.USE_DUMMY_LLM_CLIENT or settings.TESTING: + logger.info(f"{type(self).__name__}: dummy mode is on; returning a fake response instead of calling Azure OpenAI.") + return self._fake_content() + + try: + response = self.client.chat.completions.create( + model=settings.AZURE_OPENAI_DEPLOYMENT_NAME, + messages=messages, + temperature=self.temperature, + ) + return response.choices[0].message.content + except Exception as e: + logger.error(f"Error while generating LLM response: {e}", exc_info=True) + return None + + +class BaseDummyLLMClient(BaseLLMClient): + """Summaries generated for local development/testing, without hitting Azure OpenAI. Each subclass implements + ``_fake_response()`` to return a JSON object with the expected fields for that feature + """ + + def get_response(self, messages: Messages) -> Optional[str]: + return self._fake_response(messages) + + def _fake_response(self, messages: Messages) -> str: + raise NotImplementedError + + +class DummyDrefSummaryLLMClient(BaseDummyLLMClient): + """Fake response for DREF summary generation: a JSON object with all of + DREF's summary fields, each set to the same fake content, so + ``dref.summary.DrefSummaryGenerator`` parses/saves it unchanged. + """ + + def _fake_response(self, messages: Messages) -> str: + from dref.summary import SUMMARY_FIELDS + + logger.info("DummyDrefSummaryLLMClient: generating fake response for DREF summary fields") + fake_content = self._fake_content() + payload: Dict[str, str] = {field_name: fake_content for field_name in SUMMARY_FIELDS} + return json.dumps(payload) + + +class DummyOpsLearningLLMClient(BaseDummyLLMClient): + """Fake response for PER Ops Learning summaries.""" + + _EXCERPT_ID_RE = re.compile(r"^(\d+)\. In ", re.MULTILINE) + + def _prompt_excerpt_ids(self, user_content: str) -> List[int]: + return [int(excerpt_id) for excerpt_id in self._EXCERPT_ID_RE.findall(user_content)] + + def _fake_response(self, messages: Messages) -> str: + from per.models import OpsLearning + from per.ops_learning_summary import OpsLearningSummaryTask + + logger.info("DummyOpsLearningLLMClient: generating fake response for Ops Learning summary") + user_content = self._last_user_message(messages) + fake_content = self._fake_content() + excerpt_ids = self._prompt_excerpt_ids(user_content) + logger.info(f"DummyOpsLearningLLMClient: found {len(excerpt_ids)} excerpt id(s) in the prompt") + + all_excerpts_id = ", ".join(str(excerpt_id) for excerpt_id in excerpt_ids) + + payload: Dict[str, Dict[str, str]] + if OpsLearningSummaryTask.component_prompt in user_content: + logger.info("DummyOpsLearningLLMClient: detected the component prompt shape.") + excerpts = list( + OpsLearning.objects.filter(id__in=excerpt_ids, is_validated=True, per_component_validated__isnull=False) + .prefetch_related("per_component_validated") + .distinct() + ) + component = excerpts[0].per_component_validated.first() if excerpts else None + subtype = component.title if component else "Logistics" + excerpts_id = ", ".join(str(excerpt.id) for excerpt in excerpts) or all_excerpts_id + logger.info( + f"DummyOpsLearningLLMClient: {len(excerpts)} excerpt(s) matched a validated component; " + f"using subtype '{subtype}'." + ) + payload = {"0": {"type": "component", "subtype": subtype, "excerpts id": excerpts_id, "content": fake_content}} + + elif OpsLearningSummaryTask.sector_prompt in user_content: + logger.info("DummyOpsLearningLLMClient: detected the sector prompt shape.") + excerpts = list( + OpsLearning.objects.filter(id__in=excerpt_ids, is_validated=True, sector_validated__isnull=False) + .prefetch_related("sector_validated") + .distinct() + ) + sector = excerpts[0].sector_validated.first() if excerpts else None + subtype = sector.title if sector else "Shelter" + excerpts_id = ", ".join(str(excerpt.id) for excerpt in excerpts) or all_excerpts_id + logger.info( + f"DummyOpsLearningLLMClient: {len(excerpts)} excerpt(s) matched a validated sector; " + f"using subtype '{subtype}'." + ) + payload = {"0": {"type": "sector", "subtype": subtype, "excerpts id": excerpts_id, "content": fake_content}} + else: + logger.info( + f"DummyOpsLearningLLMClient: detected the primary prompt shape; " + f"returning 3 fake findings with excerpts id(s) [{all_excerpts_id}]." + ) + payload = { + str(i): { + "title": f"Dummy Finding {i + 1}", + "excerpts id": all_excerpts_id, + "content": fake_content, + "confidence level": "1/5", + } + for i in range(3) + } + return json.dumps(payload) + + +def get_dref_summary_llm_client() -> BaseLLMClient: + if settings.USE_DUMMY_LLM_CLIENT or settings.TESTING: + return DummyDrefSummaryLLMClient() + return AzureOpenAiChat(temperature=0.5) + + +def get_ops_learning_llm_client() -> BaseLLMClient: + if settings.USE_DUMMY_LLM_CLIENT or settings.TESTING: + return DummyOpsLearningLLMClient() + return AzureOpenAiChat(temperature=0.7) diff --git a/main/lock.py b/main/lock.py index 3d1fd2de1..9de5b7aaf 100644 --- a/main/lock.py +++ b/main/lock.py @@ -20,6 +20,7 @@ class RedisLockKey(models.TextChoices): OPERATION_LEARNING_SUMMARY = _BASE + "-operation-learning-summary-{0}" OPERATION_LEARNING_SUMMARY_EXPORT = _BASE + "-operation-learning-summary-export-{0}" MODEL_TRANSLATION = _BASE + "-{model_name}-translation-{id}" + DREF_SUMMARY = _BASE + "-dref-summary-{0}" @contextmanager diff --git a/main/settings.py b/main/settings.py index a500bef8c..b82853c20 100644 --- a/main/settings.py +++ b/main/settings.py @@ -146,6 +146,8 @@ AZURE_OPENAI_ENDPOINT=(str, None), AZURE_OPENAI_KEY=(str, None), AZURE_OPENAI_DEPLOYMENT_NAME=(str, None), + # Use a fake LLM client instead of calling Azure OpenAI + USE_DUMMY_LLM_CLIENT=(bool, False), # ReliefWeb appname RELIEF_WEB_APP_NAME=(str, None), # PowerBI @@ -868,6 +870,7 @@ def decode_base64(env_key, fallback_env_key): AZURE_OPENAI_ENDPOINT = env("AZURE_OPENAI_ENDPOINT") AZURE_OPENAI_KEY = env("AZURE_OPENAI_KEY") AZURE_OPENAI_DEPLOYMENT_NAME = env("AZURE_OPENAI_DEPLOYMENT_NAME") +USE_DUMMY_LLM_CLIENT = env("USE_DUMMY_LLM_CLIENT") OIDC_ENABLE = env("OIDC_ENABLE") OIDC_RSA_PRIVATE_KEY = None diff --git a/per/models.py b/per/models.py index 4cd7e7488..3b792c426 100644 --- a/per/models.py +++ b/per/models.py @@ -705,6 +705,9 @@ class OpsLearning(models.Model): modified_at = models.DateTimeField(verbose_name=_("modified_at"), auto_now=True) created_at = models.DateTimeField(verbose_name=_("created at"), auto_now_add=True) + # TYPING + id: int + class Meta: ordering = ("learning",) verbose_name = _("Operational Learning") diff --git a/per/ops_learning_summary.py b/per/ops_learning_summary.py index d2d3d6cd4..5a5d150ed 100644 --- a/per/ops_learning_summary.py +++ b/per/ops_learning_summary.py @@ -5,17 +5,15 @@ import pandas as pd import tiktoken -from django.conf import settings from django.db import transaction from django.db.models import F -from django.utils.functional import cached_property -from openai import AzureOpenAI from api.logger import logger from api.models import Country from api.utils import get_model_name from deployments.models import SectorTag from lang.tasks import translate_model_fields +from main.llm import get_ops_learning_llm_client from per.cache import OpslearningSummaryCacheHelper from per.models import ( FormComponent, @@ -29,24 +27,6 @@ ) -class AzureOpenAiChat: - - @cached_property - def client(self): - return AzureOpenAI( - azure_endpoint=settings.AZURE_OPENAI_ENDPOINT, api_key=settings.AZURE_OPENAI_KEY, api_version="2023-05-15" - ) - - def get_response(self, message): - try: - response = self.client.chat.completions.create( - model=settings.AZURE_OPENAI_DEPLOYMENT_NAME, messages=message, temperature=0.7 - ) - return response.choices[0].message.content - except Exception as e: - logger.error(f"Error while generating response: {e}", exc_info=True) - - class OpsLearningSummaryTask: PROMPT_DATA_LENGTH_LIMIT = 5000 @@ -815,9 +795,9 @@ def _summarize(prompt, type: OpsLearningPromptResponseCache.PromptType, system_m logger.warning("The length of the prompt might be too long.") return "{}" - # Using Azure OpenAI to summarize the prompt - client = AzureOpenAiChat() - response = client.get_response(message=messages) + # Using Azure OpenAI (or the dummy, per settings.USE_DUMMY_LLM_CLIENT) to summarize the prompt + client = get_ops_learning_llm_client() + response = client.get_response(messages) return response def _validate_format(summary, MAX_RETRIES=3):