Fix state issue regarding Emdat extraaction

tnagorra · Rup-Narayan-Rajbanshi · commit f6b8170d8356 · 2025-03-20T14:25:11.000+05:45
- Re-format graphql query
- Define query variables for emdat and glide
- Simplify logic for glide task
diff --git a/apps/etl/etl_tasks/emdat.py b/apps/etl/etl_tasks/emdat.py
@@ -3,92 +3,102 @@
 from celery import chain, shared_task
 from django.conf import settings
 
-from apps.etl.extraction.sources.emdat.extract import EMDATExtraction
+from apps.etl.extraction.sources.emdat.extract import EMDATExtraction, EMDATQueryVars
 from apps.etl.transform.sources.emdat import EMDATTransformHandler
 
 QUERY = """
-    query monty ($limit: Int, $offset: Int, $include_hist: Boolean, $from: Int, $to: Int) {
-      api_version
-      public_emdat(
-        cursor: {
-            offset: $offset,
-            limit: $limit
-        }
-        filters: {
-           include_hist: $include_hist
-           from: $from
-           to: $to
-     }
-      ) {
+query monty(
+    $limit: Int
+    $offset: Int
+    $include_hist: Boolean
+    $from: Int
+    $to: Int
+) {
+    api_version
+    public_emdat(
+        cursor: { offset: $offset, limit: $limit }
+        filters: { include_hist: $include_hist, from: $from, to: $to }
+    ) {
         total_available
         info {
-          timestamp
-          filters
-          cursor
-          version
+            timestamp
+            filters
+            cursor
+            version
         }
         data {
-          disno
-          classif_key
-          group
-          subgroup
-          type
-          subtype
-          external_ids
-          name
-          iso
-          country
-          subregion
-          region
-          location
-          origin
-          associated_types
-          ofda_response
-          appeal
-          declaration
-          aid_contribution
-          magnitude
-          magnitude_scale
-          latitude
-          longitude
-          river_basin
-          start_year
-          start_month
-          start_day
-          end_year
-          end_month
-          end_day
-          total_deaths
-          no_injured
-          no_affected
-          no_homeless
-          total_affected
-          reconstr_dam
-          reconstr_dam_adj
-          insur_dam
-          insur_dam_adj
-          total_dam
-          total_dam_adj
-          cpi
-          admin_units
-          entry_date
-          last_update
+            disno
+            classif_key
+            group
+            subgroup
+            type
+            subtype
+            external_ids
+            name
+            iso
+            country
+            subregion
+            region
+            location
+            origin
+            associated_types
+            ofda_response
+            appeal
+            declaration
+            aid_contribution
+            magnitude
+            magnitude_scale
+            latitude
+            longitude
+            river_basin
+            start_year
+            start_month
+            start_day
+            end_year
+            end_month
+            end_day
+            total_deaths
+            no_injured
+            no_affected
+            no_homeless
+            total_affected
+            reconstr_dam
+            reconstr_dam_adj
+            insur_dam
+            insur_dam_adj
+            total_dam
+            total_dam_adj
+            cpi
+            admin_units
+            entry_date
+            last_update
         }
-      }
     }
-    """
+}
+"""
 
 
 @shared_task
 def ext_and_transform_emdat_latest_data(**kwargs):
-    to_year = datetime.now().year
-    from_year = int(settings.EMDAT_START_YEAR)
-    variables = {"limit": -1, "from": from_year, "to": to_year}
+    # FIXME: Why are we getting data from settings.EMDAT_START_YEAR to get the latest data?
+    # Also, the filtering only filters using year so we might have lot of duplicate data
+    variables: EMDATQueryVars = {
+        "limit": -1,
+        "from": int(settings.EMDAT_START_YEAR),
+        "to": datetime.now().year,
+        "include_hist": None,
+    }
 
     chain(EMDATExtraction.task.s(QUERY, variables), EMDATTransformHandler.task.s()).apply_async()
 
 
 @shared_task
 def ext_and_transform_emdat_historical_data(**kwargs):
-    variables = {"limit": -1, "include_hist": True}
+    variables: EMDATQueryVars = {
+        "limit": -1,
+        "from": None,
+        "to": None,
+        "include_hist": True,
+    }
+
     chain(EMDATExtraction.task.s(QUERY, variables), EMDATTransformHandler.task.s()).apply_async()
diff --git a/apps/etl/etl_tasks/glide.py b/apps/etl/etl_tasks/glide.py
@@ -3,41 +3,41 @@
 from celery import chain, shared_task
 from django.conf import settings
 
-from apps.etl.extraction.sources.glide.extract import GlideExtraction
+from apps.etl.extraction.sources.glide.extract import GlideExtraction, GlideQueryVars
 from apps.etl.models import ExtractionData, HazardType
 from apps.etl.transform.sources.glide import GlideTransformHandler
 
 GLIDE_HAZARDS = [
-    ("EQ", HazardType.EARTHQUAKE),
-    ("TC", HazardType.CYCLONE),
-    ("FL", HazardType.FLOOD),
-    ("DR", HazardType.DROUGHT),
-    ("WF", HazardType.WILDFIRE),
-    ("VO", HazardType.VOLCANO),
-    ("TS", HazardType.TSUNAMI),
-    ("CW", HazardType.COLDWAVE),
-    ("EP", HazardType.EPIDEMIC),
-    ("EC", HazardType.EXTRATROPICAL_CYCLONE),
-    ("ET", HazardType.EXTREME_TEMPERATURE),
-    ("FR", HazardType.FIRE),
-    ("FF", HazardType.FLASH_FLOOD),
-    ("HT", HazardType.HEAT_WAVE),
-    ("IN", HazardType.INSECT_INFESTATION),
-    ("LS", HazardType.LANDSLIDE),
-    ("MS", HazardType.MUD_SLIDE),
-    ("ST", HazardType.SEVERE_LOCAL_STROM),
-    ("SL", HazardType.SLIDE),
-    ("AV", HazardType.SNOW_AVALANCHE),
-    ("SS", HazardType.STORM),
-    ("AC", HazardType.TECH_DISASTER),
-    ("TO", HazardType.TORNADO),
-    ("VW", HazardType.VIOLENT_WIND),
-    ("WV", HazardType.WAVE_SURGE),
+    HazardType.EARTHQUAKE,
+    HazardType.FLOOD,
+    HazardType.CYCLONE,
+    HazardType.EPIDEMIC,
+    HazardType.STORM,
+    HazardType.DROUGHT,
+    HazardType.TSUNAMI,
+    HazardType.WILDFIRE,
+    HazardType.VOLCANO,
+    HazardType.COLDWAVE,
+    HazardType.EXTRATROPICAL_CYCLONE,
+    HazardType.EXTREME_TEMPERATURE,
+    HazardType.FIRE,
+    HazardType.FLASH_FLOOD,
+    HazardType.HEAT_WAVE,
+    HazardType.INSECT_INFESTATION,
+    HazardType.LANDSLIDE,
+    HazardType.MUD_SLIDE,
+    HazardType.SEVERE_LOCAL_STROM,
+    HazardType.SLIDE,
+    HazardType.SNOW_AVALANCHE,
+    HazardType.TECH_DISASTER,
+    HazardType.TORNADO,
+    HazardType.VIOLENT_WIND,
+    HazardType.WAVE_SURGE,
 ]
 
 
 @shared_task
-def _ext_and_transform_glide_latest_data(hazard_type, hazard_type_str):
+def _ext_and_transform_glide_latest_data(hazard_type: HazardType):
     ext_object = (
         ExtractionData.objects.filter(
             source=ExtractionData.Source.GLIDE,
@@ -48,31 +48,55 @@ def _ext_and_transform_glide_latest_data(hazard_type, hazard_type_str):
         .order_by("-created_at")
         .first()
     )
+
     if ext_object:
         from_date = ext_object.created_at.date()
     else:
         from_date = datetime.strptime(settings.GLIDE_START_DATE, "%Y-%m-%d").date()
 
     to_date = datetime.today().date()
-    url = f"{settings.GLIDE_URL}/glide/jsonglideset.jsp?fromyear={from_date.year}&frommonth={from_date.month}&fromday={from_date.day}&toyear={to_date.year}&tomonth={to_date.month}&today={to_date.day}&events={hazard_type}"  # noqa: E501
 
-    chain(GlideExtraction.task.s(url), GlideTransformHandler.task.s()).apply_async()
+    # FIXME: Check if the date filters are inclusive
+    url = f"{settings.GLIDE_URL}/glide/jsonglideset.jsp"
+    variables: GlideQueryVars = {
+        "fromyear": from_date.year,
+        "frommonth": from_date.month,
+        "fromday": from_date.day,
+        "toyear": to_date.year,
+        "tomonth": to_date.month,
+        "today": to_date.day,
+        "events": hazard_type.value,
+    }
+
+    chain(GlideExtraction.task.s(url, variables), GlideTransformHandler.task.s()).apply_async()
 
 
 @shared_task
-def _ext_and_transform_glide_historical_data(hazard_type, hazard_type_str):
+def _ext_and_transform_glide_historical_data(hazard_type: HazardType):
     to_date = datetime.today().date()
-    url = f"{settings.GLIDE_URL}/glide/jsonglideset.jsp?toyear={to_date.year}&tomonth={to_date.month}&to_date={to_date.day}&events={hazard_type}"  # noqa: E501
-    chain(GlideExtraction.task.s(url), GlideTransformHandler.task.s()).apply_async()
+
+    # FIXME: Check if the date filters are inclusive
+    url = f"{settings.GLIDE_URL}/glide/jsonglideset.jsp"
+    variables: GlideQueryVars = {
+        "fromyear": None,
+        "frommonth": None,
+        "fromday": None,
+        "toyear": to_date.year,
+        "tomonth": to_date.month,
+        "today": to_date.day,
+        "events": hazard_type.value,
+    }
+
+    chain(GlideExtraction.task.s(url, variables), GlideTransformHandler.task.s()).apply_async()
 
 
 @shared_task
 def ext_and_transform_glide_latest_data():
-    for hazard_type, hazard_type_str in GLIDE_HAZARDS:
-        _ext_and_transform_glide_latest_data(hazard_type, hazard_type_str)
+    for hazard_type in GLIDE_HAZARDS:
+        _ext_and_transform_glide_latest_data(hazard_type)
 
 
 @shared_task
 def ext_and_transform_glide_historical_data():
-    for hazard_type, hazard_type_str in GLIDE_HAZARDS:
-        _ext_and_transform_glide_historical_data(hazard_type, hazard_type_str)
+    for hazard_type in GLIDE_HAZARDS:
+        _ext_and_transform_glide_historical_data(hazard_type)
diff --git a/apps/etl/extraction/sources/base/handler.py b/apps/etl/extraction/sources/base/handler.py
@@ -26,7 +26,7 @@ def store_extraction_data(
         validate_source_func: Callable[[Any], None],
         source: int,
         response: dict,
-        instance_id: int = None,
+        instance_id: int | None = None,
     ):
         """
         Save extracted data into data base. Checks for duplicate conent using hashing.
@@ -78,7 +78,7 @@ def _create_extraction_instance(cls, url: str, source: int) -> ExtractionData:
 
     @classmethod
     def _update_instance_status(
-        cls, instance: ExtractionData, status: int, validation_status: int = None, update_validation: bool = False
+        cls, instance: ExtractionData, status: int, validation_status: int | None = None, update_validation: bool = False
     ) -> None:
         """
         Update the status of the extraction instance.
@@ -115,7 +115,7 @@ def _save_response_data(cls, instance: ExtractionData, response: requests.Respon
         return json.loads(response.content)
 
     @classmethod
-    def handle_extraction(cls, url: str, params: dict, headers: dict, source: int) -> dict:
+    def handle_extraction(cls, url: str, params: dict | None, headers: dict, source: int) -> dict:
         """
         Process data extraction.
         Returns:
diff --git a/apps/etl/extraction/sources/desinventar/extract.py b/apps/etl/extraction/sources/desinventar/extract.py
@@ -25,7 +25,7 @@ def store_extraction_data(
         instance_id: int = None,
     ):
         """
-        Save extracted data into database. Checks for duplicate content using hashing.
+        Save extracted data into database.
         """
         file_name = f"{source}.zip"
         resp_data = response
@@ -40,6 +40,7 @@ def store_extraction_data(
             # manage duplicate file content.
             manage_duplicate_file_content(
                 source=extraction_instance.source,
+                # FIXME: We need to calculate has for zip file
                 hash_content=None,
                 instance=extraction_instance,
                 response_data=resp_data.content,
diff --git a/apps/etl/transform/sources/handler.py b/apps/etl/transform/sources/handler.py
@@ -105,7 +105,7 @@ def load_stac_item_to_queue(cls, transform_items, transform_obj_id):
 
     @staticmethod
     @app.task
-    def task(extraction_id):
+    def task(extraction_id: int):
         """
         Not NotImplemented due to celery limitation with classmethod
         Eg: return XYZTransformHandler.handle_transformation(extraction_id)
