Merge branch 'main' into jw/231

Author: jakewheeler

refiner/app/core/exceptions.py

@@ -221,3 +221,20 @@ class DatabaseDataError(DatabaseError):
     """
 
     pass
+
+
+class RefinementException(Exception):
+    """
+    Exception raised during a failed refinement run.
+    """
+
+    def __init__(self, message: str, detail: str):
+        """
+        RefinementException constructor.
+
+        Args:
+            message (str): High-level error message
+            detail (str): Additional detail describing the issue encountered
+        """
+        super().__init__(message)
+        self.detail = detail

refiner/app/lambda/lambda_function.py

@@ -7,7 +7,7 @@
 import json
 import os
 from collections import defaultdict
-from dataclasses import dataclass, field
+from dataclasses import asdict, dataclass, field
 from typing import Any, TypedDict
 
 import boto3
@@ -26,8 +26,8 @@
 from app.services.ecr.refine import get_file_size_in_mib
 from app.services.pipeline import (
     AugmentationRun,
+    RefinementContext,
     RefinementResult,
-    RefinementTrace,
     create_augmentation_run_from_xml_files,
     discover_reportable_conditions,
     produce_remainder_rr_for_jurisdiction,
@@ -83,20 +83,13 @@ class RefinementState:
     jurisdiction/per-condition refinement traces, the AIMS-facing
     metadata dict, and the set of codes per jurisdiction that were
     NOT refined (used to drive remainder RR production).
-
-    TODO:
-    * skipped_condition_codes_by_jurisdiction holds information that
-      is also present on the per-condition RefinementTrace objects
-      (traces with refinement_outcome == "skipped")
-    * that means skipped codes are tracked in two places
     """
 
     output_files: set[str] = field(default_factory=set)
     metadata: RefinerMetadata = field(default_factory=dict)
     skipped_condition_codes_by_jurisdiction: dict[str, set[str]] = field(
         default_factory=lambda: defaultdict(set)
     )
-    traces: list[RefinementTrace] = field(default_factory=list)
 
 
 @dataclass
@@ -528,12 +521,6 @@ def run_refinement(input: RefinementInput) -> RefinementOutput:
         run=run,
     )
 
-    log_refinement_summary(
-        persistence_id=input.persistence_id,
-        output_files=state.output_files,
-        traces=state.traces,
-    )
-
     return RefinementOutput(
         output_file_keys=list(state.output_files), metadata=state.metadata
     )
@@ -588,10 +575,6 @@ def process_condition(
     Process a single reportable condition for a jurisdiction.
     """
     rsg_code = reportable_condition.code
-    trace = RefinementTrace(
-        jurisdiction_code=jurisdiction_code,
-        rsg_code=rsg_code,
-    )
 
     cg_metadata = rsg_cg_payload.mappings.get(rsg_code)
 
@@ -604,28 +587,23 @@ def process_condition(
             operation="skipped",
         )
         mark_condition_skipped(
-            trace=trace,
             jurisdiction_code=jurisdiction_code,
             condition_code=rsg_code,
             reason="rsg_not_in_mapping",
             state=state,
         )
         return
 
-    trace.canonical_url = cg_metadata.canonical_url
-
-    processed_configuration = load_active_configuration(
+    active_configuration = load_active_configuration(
         s3_client=refiner_input.s3_client,
         config_bucket=refiner_input.config_bucket_name,
         jurisdiction_code=jurisdiction_code,
         cg_metadata=cg_metadata,
         rsg_metadata=reportable_condition,
-        trace=trace,
     )
 
-    if processed_configuration is None:
+    if active_configuration is None:
         mark_condition_skipped(
-            trace=trace,
             jurisdiction_code=jurisdiction_code,
             condition_code=rsg_code,
             reason="no_active_configuration",
@@ -635,25 +613,35 @@ def process_condition(
 
     result = refine_for_condition(
         xml_files=refiner_input.xml_files,
-        processed_configuration=processed_configuration,
-        trace=trace,
+        processed_configuration=active_configuration.configuration,
+        context=RefinementContext(
+            canonical_url=cg_metadata.canonical_url,
+            jurisdiction_id=jurisdiction_code,
+            configuration_version=active_configuration.version,
+        ),
         run=run,
     )
 
-    state.traces.append(trace)
-
     write_refined_outputs(
         refiner_input=refiner_input,
         jurisdiction_code=jurisdiction_code,
         condition_grouper_name=cg_metadata.name,
         result=result,
-        trace=trace,
         condition_code=rsg_code,
         state=state,
     )
 
     state.metadata[jurisdiction_code][rsg_code] = True
 
+    logger.info(
+        "Refinement complete for condition.",
+        rsg_code=rsg_code,
+        jurisidiction_code=jurisdiction_code,
+        metrics=asdict(result.metrics),
+        report=asdict(result.report),
+        operation="log_summary",
+    )
+
 
 def load_condition_mapping_for_jurisdiction(
     s3_client,
@@ -693,21 +681,20 @@ def skip_all_conditions_for_missing_mapping(
     Mark every condition in a jurisdiction as skipped when the mapping file is missing.
     """
     for condition in jurisdiction_group.conditions:
-        trace = RefinementTrace(
+        logger.info(
+            "Mapping file not found for condition",
             jurisdiction_code=jurisdiction_code,
             rsg_code=condition.code,
             refinement_outcome="skipped",
-            skip_reason="no_mapping_file",
+            reason="no_mapping_file",
         )
-        state.traces.append(trace)
         state.skipped_condition_codes_by_jurisdiction[jurisdiction_code].add(
             condition.code
         )
         state.metadata[jurisdiction_code][condition.code] = False
 
 
 def mark_condition_skipped(
-    trace: RefinementTrace,
     jurisdiction_code: str,
     condition_code: str,
     reason: str,
@@ -716,22 +703,31 @@ def mark_condition_skipped(
     """
     Helper to mark a condition as "skipped" during refinement.
     """
-    trace.refinement_outcome = "skipped"
-    trace.skip_reason = reason
+    logger.info(
+        "Refiner skipped condition.",
+        jurisdiction=jurisdiction_code,
+        condition_code=condition_code,
+        reason=reason,
+        operation="skipped",
+    )
 
-    state.traces.append(trace)
     state.metadata[jurisdiction_code][condition_code] = False
     state.skipped_condition_codes_by_jurisdiction[jurisdiction_code].add(condition_code)
 
 
+@dataclass
+class ActiveConfiguration:
+    configuration: ProcessedConfiguration
+    version: int
+
+
 def load_active_configuration(
     s3_client,
     config_bucket: str,
     jurisdiction_code: str,
     cg_metadata: ConditionMapValue,
     rsg_metadata: ReportableCondition,
-    trace: RefinementTrace,
-) -> ProcessedConfiguration | None:
+) -> ActiveConfiguration | None:
     """
     Attempts to find and load the active configuration.
     """
@@ -756,8 +752,6 @@ def load_active_configuration(
         )
         return None
 
-    trace.configuration_version = config_version_to_use
-
     serialized_configuration_key = get_active_file_key(
         jurisdiction_id=jurisdiction_code,
         canonical_url=cg_metadata.canonical_url,
@@ -780,7 +774,10 @@ def load_active_configuration(
         operation="activation_file_read",
     )
 
-    return ProcessedConfiguration.from_dict(serialized_configuration)
+    return ActiveConfiguration(
+        configuration=ProcessedConfiguration.from_dict(serialized_configuration),
+        version=config_version_to_use,
+    )
 
 
 def write_refined_outputs(
@@ -789,7 +786,6 @@ def write_refined_outputs(
     condition_code: str,
     condition_grouper_name: str,
     result: RefinementResult,
-    trace: RefinementTrace,
     state: RefinementState,
 ) -> None:
     """
@@ -804,7 +800,7 @@ def write_refined_outputs(
     refiner_input.s3_client.put_object(
         Bucket=refiner_input.output_bucket_name,
         Key=eicr_output_key,
-        Body=result.refined_eicr.encode("utf-8"),
+        Body=result.documents.eicr.encode("utf-8"),
         ContentType="application/xml",
     )
 
@@ -814,17 +810,17 @@ def write_refined_outputs(
     refiner_input.s3_client.put_object(
         Bucket=refiner_input.output_bucket_name,
         Key=rr_output_key,
-        Body=result.refined_rr.encode("utf-8"),
+        Body=result.documents.rr.encode("utf-8"),
         ContentType="application/xml",
     )
 
     state.output_files.add(rr_output_key)
 
     logger.info(
-        "Condition refinement complete.",
+        "Writing refined output files.",
         eicr_key=eicr_output_key,
         rr_key=rr_output_key,
-        eicr_size_reduction_percentage=trace.eicr_size_reduction_percentage,
+        eicr_size_reduction_percentage=result.metrics.eicr.size_reduction_percentage,
         jurisdiction_code=jurisdiction_code,
         condition_code=condition_code,
         operation="condition_refinement_complete",
@@ -908,32 +904,3 @@ def write_remainder_rrs(
             augmented_rr_doc_id=remainder.augmented_result.augmented_doc_id,
             operation="remainder_rr_written",
         )
-
-
-def log_refinement_summary(
-    persistence_id: str,
-    output_files: set[str],
-    traces: list[RefinementTrace],
-) -> None:
-    """
-    Log a final summary for the entire refinement run.
-    """
-    logger.info(
-        "Refinement complete.",
-        persistence_id=persistence_id,
-        output_file_urls=list(output_files),
-        traces=[
-            {
-                "jurisdiction": t.jurisdiction_code,
-                "rsg_code": t.rsg_code,
-                "canonical_url": t.canonical_url,
-                "outcome": t.refinement_outcome,
-                "skip_reason": t.skip_reason,
-                "config_version": t.configuration_version,
-                "eicr_size_reduction": t.eicr_size_reduction_percentage,
-                "eicr_size_mb": t.eicr_size_mib,
-            }
-            for t in traces
-        ],
-        operation="refinement_complete",
-    )