feat/msg type cancel

.github/docker-compose.yml

@@ -0,0 +1,12 @@
+services:
+  web:
+    image: $DOCKER_IMAGE_BACKEND
+    environment:
+      CI: "true"
+    volumes:
+      - ./:/code
+      - ./ci-share/:/ci-share/
+      # - ./coverage/:/code/coverage/
+
+  worker: !reset null
+  worker-beat: !reset null

.github/workflows/ci.yml

@@ -23,6 +23,10 @@ permissions:
   contents: read
   packages: write
 
+env:
+  COMPOSE_FILE: docker-compose.yml:./.github/docker-compose.yml
+  DJANGO_SECRET_KEY: INSECURE_DJANGO_SECRET_KEY
+
 jobs:
   pre_commit_checks:
     name: 🚴 Pre-Commit checks 🚴
@@ -121,11 +125,21 @@ jobs:
           cache-from: type=local,src=/tmp/.buildx-cache
           cache-to: type=local,dest=/tmp/.buildx-cache-new
 
+      - name: Start app resources
+        timeout-minutes: 1
+        env:
+          DOCKER_IMAGE: ${{ steps.prep.outputs.tagged_image }}
+        run: |
+          timeout 60s docker compose run --rm web bash -c './manage.py wait_for_resources --all' || {
+            echo 'Failed to wait for resources'
+            exit 1
+          }
+
       - name: 🕮 Validate if there are no pending django migrations.
         env:
           DOCKER_IMAGE_BACKEND: ${{ steps.prep.outputs.tagged_image }}
         run: |
-          docker compose -f ./gh-docker-compose.yml run --rm web bash -c 'wait-for-it db:5432 && ./manage.py makemigrations --check --dry-run' || {
+          docker compose run --rm web bash -c './manage.py makemigrations --check --dry-run' || {
             echo 'There are some changes to be reflected in the migration. Make sure to run makemigrations';
             exit 1;
           }
@@ -134,7 +148,7 @@ jobs:
         env:
           DOCKER_IMAGE_BACKEND: ${{ steps.prep.outputs.tagged_image }}
         run: |
-          docker compose -f ./gh-docker-compose.yml run --rm web bash -c 'wait-for-it db:5432 && ./manage.py graphql_schema --out /ci-share/schema-latest.graphql' &&
+          docker compose run --rm web bash -c './manage.py graphql_schema --out /ci-share/schema-latest.graphql' &&
           cmp --silent schema.graphql ./ci-share/schema-latest.graphql || {
             echo 'The schema.graphql is not up to date with the latest changes. Please update and push latest';
             diff schema.graphql ./ci-share/schema-latest.graphql;
@@ -144,7 +158,7 @@ jobs:
       - name: 🤞 Run Test 🧪 & Publish coverage to code climate
         env:
           DOCKER_IMAGE_BACKEND: ${{ steps.prep.outputs.tagged_image }}
-        run: docker compose -f gh-docker-compose.yml run --rm web /code/scripts/run_tests.sh
+        run: docker compose run --rm web /code/scripts/run_tests.sh
 
       - name: 🐳 Docker push
         if: ${{ inputs.push_docker_image }}
@@ -174,8 +188,8 @@ jobs:
 
       - name: 🐳 Helm dependency
         run: |
-          yq --indent 0 '.dependencies | map(["helm", "repo", "add", .name, .repository] | join(" ")) | .[]' ./helm/Chart.lock  | sh --
-          helm dependency build ./helm
+          yq --indent 0 '.dependencies | map(select(.repository | test("^oci:") | not)) | map(["helm", "repo", "add", .name, .repository] | join(" ")) | .[]' ./helm/Chart.lock | sh --
+          helm dependency build ./helm/
 
       - name: 🐳 Helm lint
         run: helm lint ./helm --values ./helm/values-test.yaml

.github/workflows/helm-publish.yml

@@ -39,8 +39,8 @@ jobs:
 
     - name: 🐳 Helm dependency
       run: |
-        yq --indent 0 '.dependencies | map(["helm", "repo", "add", .name, .repository] | join(" ")) | .[]' ./helm/Chart.lock  | sh --
-        helm dependency build ./helm
+        yq --indent 0 '.dependencies | map(select(.repository | test("^oci:") | not)) | map(["helm", "repo", "add", .name, .repository] | join(" ")) | .[]' ./helm/Chart.lock | sh --
+        helm dependency build ./helm/
 
     - name: Tag docker image in Helm Chart values.yaml
       env:

Dockerfile

@@ -22,8 +22,6 @@ RUN --mount=type=cache,target=/root/.cache/uv \
         gcc libc-dev gdal-bin libproj-dev \
         # Django translation
         gettext \
-        # Helper packages
-        wait-for-it \
     && uv lock --locked --offline \
         && uv sync --frozen --no-install-project --all-groups \
     # Clean-up

apps/cap_feed/data_injector/geo.py

@@ -4,7 +4,7 @@
 import re
 from collections import defaultdict
 
-import requests
+import httpx
 from django.conf import settings
 from django.contrib.gis.gdal import DataSource
 from django.contrib.gis.geos import GEOSGeometry
@@ -136,7 +136,7 @@ def handle_pagination(self, url, **requests_kwargs):
         self.log_info(f'Fetching data from IFRC-GO: ({_lang})', _url)
         # TODO: Add some check to avoid infinite run
         while True:
-            resp = requests.get(_url, **requests_kwargs).json()
+            resp = httpx.get(_url, **requests_kwargs).json()
             for item in resp['results']:
                 yield item
             if resp['next'] is not None:

apps/cap_feed/formats/atom.py

@@ -1,7 +1,7 @@
 import logging
 import xml.etree.ElementTree as ET
 
-import requests
+import httpx
 import validators
 
 from apps.cap_feed.models import Alert, Feed, ProcessedAlert
@@ -21,13 +21,13 @@ def get_alerts_atom(feed, ns):
 
     # navigate list of alerts
     try:
-        response = requests.get(
+        response = httpx.get(
             feed.url,
             headers=COMMON_REQUESTS_HEADERS,
             timeout=Feed.MAX_REQUEST_TIMEOUT,
         )
         response.raise_for_status()
-    except requests.exceptions.RequestException:
+    except httpx.HTTPError:
         logger.error(
             '[ATOM] Failed to fetch feed alerts',
             exc_info=True,

apps/cap_feed/formats/cap_xml.py

@@ -26,9 +26,11 @@
     AlertInfoParameter,
     Feed,
     ProcessedAlert,
+    alert_info_default_expire,
 )
 from apps.cap_feed.utils import distance_to_decimal_degrees
 from main.managers import BulkCreateManager
+from utils.common import logger_log_extra
 
 logger = logging.getLogger(__name__)
 
@@ -97,7 +99,7 @@ def create_alert_info(
         instruction=find_element(alert_info_entry, ns, 'cap:instruction'),
         web=find_element(alert_info_entry, ns, 'cap:web'),
         contact=find_element(alert_info_entry, ns, 'cap:contact'),
-        expires=expire_time,
+        expires=expire_time or alert_info_default_expire(),
     )
 
 
@@ -203,6 +205,61 @@ def process_geo_codes(
     return list(set(possible_admin1_ids))
 
 
+def process_related_alerts(alert: Alert):
+    # Currenlty only processing "Cancel" update to pre-expire existing active alerts
+
+    """
+    Source: https://docs.oasis-open.org/emergency/cap/v1.2/CAP-v1.2-os.html
+
+    # `msg_type`
+    Code Values:
+    "Alert" - Initial information requiring attention by targeted recipients
+    "Update" - Updates and supercedes the earlier message(s) identified in <references>
+    "Cancel" - Cancels the earlier message(s) identified in <references>
+    "Ack" - Acknowledges receipt and acceptance of the message(s) identified in <references>
+    "Error" - Indicates rejection of the message(s) identified in <references>; explanation SHOULD appear in <note>
+
+    # `references`
+    (1) The extended message identifier(s) (in the form sender,identifier,sent) of an earlier CAP message or messages referenced by this one.
+    (2) If multiple messages are referenced, they SHALL be separated by whitespace.
+    """  # noqa: E501
+
+    if alert.msg_type.lower() != "cancel":
+        return
+
+    # Auto expire this alert
+    alert.is_expired = True
+    alert.save(update_fields=("is_expired",))
+
+    # Now also expire existing alerts
+    references_raw = (alert.references or "").strip(" ")
+
+    if not references_raw:
+        return
+
+    existing_alert_identifiers = set()
+
+    try:
+        for reference in references_raw.split(" "):
+            reference_split = reference.split(",") or []
+            if len(reference_split) > 1:
+                existing_alert_identifiers.add(reference_split[1])
+    except Exception:
+        logger.error(
+            "Failed to extract existing alert identifiers from references",
+            exc_info=True,
+            extra=logger_log_extra({"alert_id": alert.pk}),
+        )
+
+    if existing_alert_identifiers:
+        updated_qs_resp = Alert.objects.filter(
+            feed=alert.feed,
+            is_expired=False,
+            identifier__in=existing_alert_identifiers,
+        ).update(is_expired=True)
+        logger.info("Expired existing alerts due to cancel msg_type: %s", updated_qs_resp)
+
+
 def process_alert(
     url: str,
     alert_root: XmlElement,
@@ -296,6 +353,7 @@ def process_alert(
     mgr.done()
     if mrg_summary := mgr.summary(ignore_empty=True):
         logger.debug(f"DB ops summary for alert: {alert.pk}: {str(mrg_summary)}")
+    process_related_alerts(alert)
     return alert
 
 

apps/cap_feed/formats/format_handler.py

@@ -1,6 +1,5 @@
 import logging
 
-from apps.cap_feed.models import Alert
 from main.sentry import SentryTag
 
 from .atom import get_alerts_atom
@@ -37,12 +36,16 @@ def get_alerts(feed, all_alert_urls=set()):
         logger.error(f'Error getting alerts from {feed.url}', exc_info=True)
     else:
         if valid_poll:
+            all_alert_urls.update(alert_urls)
+            # XXX: Remove this later...
+            # for now not deleting the alerts missing from alerting authority..
+            # just using expire date
+            # ---------------------------------------------------------------------------------
             # alerts that are in the database and have not expired but are no longer available -
             # - on the feed must have been deleted by the alerting authority
             # remove these alerts from the database
-            all_alert_urls.update(alert_urls)
-            deleted_alerts = Alert.objects.filter(feed=feed).exclude(url__in=all_alert_urls)
-            # Tag expired alerts
-            deleted_alerts.filter(is_expired=False).update(is_expired=True)
+            # deleted_alerts = Alert.objects.filter(feed=feed).exclude(url__in=all_alert_urls)
+            # # Tag expired alerts
+            # deleted_alerts.filter(is_expired=False).update(is_expired=True)
 
     return polled_alerts_count

apps/cap_feed/formats/nws_us.py

@@ -1,7 +1,7 @@
 import logging
 import xml.etree.ElementTree as ET
 
-import requests
+import httpx
 import validators
 
 from apps.cap_feed.formats.cap_xml import get_alert
@@ -21,7 +21,7 @@ def get_alerts_nws_us(feed, ns):
 
     # navigate list of alerts
     try:
-        response = requests.get(
+        response = httpx.get(
             feed.url,
             headers={
                 **COMMON_REQUESTS_HEADERS,
@@ -30,7 +30,7 @@ def get_alerts_nws_us(feed, ns):
             timeout=Feed.MAX_REQUEST_TIMEOUT,
         )
         response.raise_for_status()
-    except requests.exceptions.RequestException:
+    except httpx.HTTPError:
         logger.error(
             '[NWS_US] Failed to fetch feed alerts',
             exc_info=True,

apps/cap_feed/formats/rss.py

@@ -1,7 +1,7 @@
 import logging
 import xml.etree.ElementTree as ET
 
-import requests
+import httpx
 import validators
 
 from apps.cap_feed.formats.cap_xml import get_alert
@@ -21,13 +21,13 @@ def get_alerts_rss(feed, ns):
 
     # navigate list of alerts
     try:
-        response = requests.get(
+        response = httpx.get(
             feed.url,
             headers=COMMON_REQUESTS_HEADERS,
             timeout=Feed.MAX_REQUEST_TIMEOUT,
         )
         response.raise_for_status()
-    except requests.exceptions.RequestException:
+    except httpx.HTTPError:
         logger.error(
             '[RSS] Failed to fetch feed alerts',
             exc_info=True,

apps/cap_feed/formats/utils.py

@@ -3,8 +3,8 @@
 import xml.etree.ElementTree as ET
 from datetime import datetime
 
+import httpx
 import pytz
-import requests
 
 from apps.cap_feed.models import FeedLog
 
@@ -25,7 +25,7 @@ def convert_datetime(original_datetime):
 
 def fetch_alert_using_url(url) -> tuple[typing.Literal[False], None] | tuple[typing.Literal[True], ET.Element]:
     # navigate alert
-    alert_response = requests.get(url, headers=COMMON_REQUESTS_HEADERS)
+    alert_response = httpx.get(url, headers=COMMON_REQUESTS_HEADERS)
     alert_response_content = alert_response.content
     if alert_response.status_code != 200:
         logger.warning(f'Skipping for url {url}: Invalid status_code {alert_response.status_code}')

apps/cap_feed/management/commands/regenerate_feed_periodic_tasks.py

@@ -1,33 +1,25 @@
 from django.core.management.base import BaseCommand
-from django_celery_beat.models import PeriodicTask
 
 from apps.cap_feed.models import Feed
 from apps.cap_feed.utils import FeedTaskManager
 
 
 class Command(BaseCommand):
 
-    def force_delete_all_tasks(self):
-        # TODO: Use INTERNAL_CELERY_TASK_NAME_PREFIX to remove legacy tasks
-        deleted = PeriodicTask.objects.filter(
-            name__in=[
-                # Legacy
-                'remove_expired_alerts',
-                # Name changed
-                'tag_expired_alerts',
-                'remove_expired_alert_records',
-            ]
-        ).delete()
-        self.stdout.write(self.style.WARNING(f'Removed {deleted}'))
-        deleted = PeriodicTask.objects.filter(name__startswith='poll_feed').delete()
-        self.stdout.write(self.style.WARNING(f'Removed {deleted}'))
-
     def handle(self, *_, **options):
-        self.force_delete_all_tasks()
         deleted_count = FeedTaskManager.force_delete_all_tasks()
-        self.stdout.write(self.style.WARNING(f'Removing existing feed tasks {deleted_count}'))
+
+        self.stdout.write(self.style.WARNING(f'Removing existing feed tasks count: {deleted_count}'))
+
         proccessed_feeds = 0
         for feed in Feed.objects.all():
-            FeedTaskManager.add_task(feed)
+            if not feed.is_archived:
+                FeedTaskManager.add_task(feed)
+            else:
+                FeedTaskManager.remove_task(feed)
             proccessed_feeds += 1
+
         self.stdout.write(self.style.SUCCESS(f'Create tasks for {proccessed_feeds} feeds'))
+        self.stdout.write(f'Total feeds: {Feed.objects.count()}')
+        self.stdout.write(f'Active feeds: {Feed.objects.filter(is_archived=False).count()}')
+        self.stdout.write(f'Archived feeds: {Feed.objects.filter(is_archived=True).count()}')