Sync staging prod may 1

ckan-backend-dev/.env.example

@@ -101,7 +101,7 @@ CKANEXT__S3FILESTORE__HOST_NAME=http://minio:9000
 
 # scheming
 CKAN___SCHEMING__DATASET_SCHEMAS=ckanext.wri.schema:ckan_dataset.yaml
-CKAN___SCHEMING__ORGANIZATION_SCHEMAS=ckanext.scheming:custom_org_with_address.json
+CKAN___SCHEMING__ORGANIZATION_SCHEMAS=ckanext.wri.schema:custom_org_with_address.json
 CKAN___SCHEMING__GROUP_SCHEMAS=ckanext.wri.schema:wri_application.json
 CKAN___SCHEMING__PRESETS=ckanext.wri.schema:presets.json
 

ckan-backend-dev/ckan/Dockerfile.dev

@@ -137,6 +137,14 @@ COPY setup/supervisord.conf /etc/supervisord.conf
 COPY setup/supervisor.worker.conf /etc/supervisord.d/worker.conf
 COPY setup/supervisor.harvest.conf /etc/supervisord.d/harvest.conf
 
+RUN SUPERVISOR_USER="root"; \
+    SUPERVISOR_HOME="/tmp"; \
+    sed -i "s|SUPERVISOR_USER|${SUPERVISOR_USER}|g" /etc/supervisord.d/harvest.conf && \
+    sed -i "s|SUPERVISOR_HOME|${SUPERVISOR_HOME}|g" /etc/supervisord.d/harvest.conf && \
+    sed -i "s|SUPERVISOR_USER|${SUPERVISOR_USER}|g" /etc/supervisord.d/worker.conf && \
+    sed -i "s|SUPERVISOR_HOME|${SUPERVISOR_HOME}|g" /etc/supervisord.d/worker.conf && \
+    sed -i "s|SUPERVISOR_HOME|${SUPERVISOR_HOME}|g" /etc/supervisord.conf
+
 RUN chown -R ckan:ckan /etc/supervisord.d
 RUN chown -R ckan:ckan /etc/supervisord.conf
 

ckan-backend-dev/src/ckanext-wri/ckanext/wri/logic/action/create.py

@@ -905,4 +905,4 @@ def organization_create(context, data_dict):
 
 
     result = old_organization_create(context, data_dict)
-    return result
+    return result

ckan-backend-dev/src/ckanext-wri/ckanext/wri/logic/action/datapusher_download.py

@@ -6,7 +6,7 @@
 import ckanext.s3filestore.uploader as uploader
 from ckan.lib.mailer import mail_recipient
 from ckan.common import config
-from .datapusher_download_zip import  fetch_dataset_name
+from .datapusher_download_zip import  fetch_dataset_name, get_admin_emails_for_dataset
 
 import datetime
 import requests
@@ -75,8 +75,21 @@ def download_request(context: Context, data_dict: dict[str, Any]):
         "error": "{}",
     }
 
+    dataset_id = fetch_dataset_name({
+        "entity_id": res_id,
+        "entity_type": "resource"
+    })
+    admin_email= get_admin_emails_for_dataset(dataset_id)
+
+    value = {}
+    if admin_email:
+        value["admin_emails"] = admin_email
+
     if email:
-        task["value"] = json.dumps({"emails": [email]})
+        value["emails"] = [email]
+
+    task["value"] = json.dumps(value)
+
 
     try:
         existing_task = p.toolkit.get_action("task_status_show")(
@@ -105,6 +118,7 @@ def download_request(context: Context, data_dict: dict[str, Any]):
 
                 if update_emails:
                     existing_task_values['emails'] = existing_task_emails
+                    existing_task_values["admin_emails"] = admin_email
                     existing_task['value'] = json.dumps(existing_task_values)
                     p.toolkit.get_action("task_status_update")({ "ignore_auth": True }, existing_task)
 
@@ -182,7 +196,7 @@ def download_request(context: Context, data_dict: dict[str, Any]):
             "entity_id": res_id,
             "entity_type": "resource"
         })
-        send_error([email], resource_title, dataset_name)
+        send_error([email]+admin_email, resource_title, dataset_name)
         raise p.toolkit.ValidationError(error)
 
     try:
@@ -208,11 +222,11 @@ def download_request(context: Context, data_dict: dict[str, Any]):
             "entity_id": res_id,
             "entity_type": "resource"
         })
-        send_error([email], resource_title, dataset_name)
+        send_error([email]+admin_email, resource_title, dataset_name)
         raise p.toolkit.ValidationError(error)
 
     value = {"job_id": r.json()["id"]}
-
+    value["admin_emails"] = admin_email
     if email:
         value["emails"] = [email]
 
@@ -251,6 +265,7 @@ def download_callback(context: Context, data_dict: dict[str, Any]):
 
     value = json.loads(task["value"])
     emails = value.get("emails", [])
+    admin_email = value.get("admin_emails", [])
     download_filename = value.get("download_filename")
 
     log.info("Preparing to send email...")
@@ -263,7 +278,7 @@ def download_callback(context: Context, data_dict: dict[str, Any]):
             "entity_id": entity_id,
             "entity_type": "resource"
         })
-        send_error(emails, download_filename, dataset_name)
+        send_error(emails+admin_email, download_filename, dataset_name)
         log.error(error)
 
 

ckan-backend-dev/src/ckanext-wri/ckanext/wri/logic/action/datapusher_download_subset.py

@@ -6,7 +6,7 @@
 import ckanext.s3filestore.uploader as uploader
 from ckan.lib.mailer import mail_recipient
 from ckan.common import config
-from .datapusher_download_zip import fetch_dataset_name
+from .datapusher_download_zip import fetch_dataset_name, get_admin_emails_for_dataset
 
 import datetime
 import requests
@@ -143,8 +143,15 @@ def subset_download_request(context: Context, data_dict: dict[str, Any]):
         "error": "{}",
     }
 
+    admin_email= get_admin_emails_for_dataset(dataset_id)
+    value = {}
+    if admin_email:
+        value["admin_emails"] = admin_email
+
     if email:
-        task["value"] = json.dumps({"emails": [email]})
+        value["emails"] = [email]
+
+    task["value"] = json.dumps(value)
 
     try:
         existing_task = p.toolkit.get_action("task_status_show")(
@@ -176,6 +183,7 @@ def subset_download_request(context: Context, data_dict: dict[str, Any]):
 
                 if update_emails:
                     existing_task_values["emails"] = existing_task_emails
+                    existing_task_values["admin_emails"] = admin_email
                     existing_task["value"] = json.dumps(existing_task_values)
                     p.toolkit.get_action("task_status_update")(
                         {"ignore_auth": True}, existing_task
@@ -260,7 +268,7 @@ def subset_download_request(context: Context, data_dict: dict[str, Any]):
             "entity_id": id if provider == "datastore" else dataset_id,
             "entity_type": "resource" if provider == "datastore" else "dataset"
         })
-        send_error([email], "Subset of data", dataset_name)
+        send_error([email]+admin_email, "Subset of data", dataset_name)
         raise p.toolkit.ValidationError(error)
 
     try:
@@ -286,11 +294,11 @@ def subset_download_request(context: Context, data_dict: dict[str, Any]):
             "entity_id": id if provider == "datastore" else dataset_id,
             "entity_type": "resource" if provider == "datastore" else "dataset"
         })
-        send_error([email], "Subset of data", dataset_name)
+        send_error([email]+admin_email, "Subset of data", dataset_name)
         raise p.toolkit.ValidationError(error)
 
     value = {"job_id": r.json()["id"]}
-
+    value["admin_emails"] = admin_email
     if email:
         value["emails"] = [email]
 
@@ -325,6 +333,7 @@ def subset_download_callback(context: Context, data_dict: dict[str, Any]):
 
     value = json.loads(task["value"])
     emails = value.get("emails", [])
+    admin_emails = value.get("admin_emails", [])
     download_filename = value.get("download_filename")
 
 
@@ -336,7 +345,7 @@ def subset_download_callback(context: Context, data_dict: dict[str, Any]):
             "entity_id": entity_id,
             "entity_type": data_dict.get("entity_type", "resource")
         })
-        send_error(emails, download_filename, dataset_name)
+        send_error(emails + admin_emails, download_filename, dataset_name)
         log.error(error)
 
 

ckan-backend-dev/src/ckanext-wri/ckanext/wri/logic/action/datapusher_download_zip.py

@@ -66,6 +66,31 @@ def build_download_filename(dataset_id: str, context) -> str:
         return False
 
 
+
+def get_admin_emails_for_dataset(dataset_id: str) -> list[str]:
+    package = p.toolkit.get_action("package_show")(
+        {"ignore_auth": True}, {"id": dataset_id}
+    )
+    organization = package.get("organization", None)
+    admin_email = []
+    if organization:
+        organization = organization.get("name")
+        org = p.toolkit.get_action("organization_show")(
+            {"ignore_auth": True}, {"id": organization, "include_users": True}
+        )
+        users = org.get("users", [])
+        if users:
+            for user in users:
+                if user.get("capacity") == "admin":
+                    users_obj = p.toolkit.get_action("user_show")(
+                        {"ignore_auth": True}, {"id": user.get("id")}
+                    )
+                    if users_obj.get("email", False):
+                        admin_email.append(users_obj.get("email"))
+    return admin_email
+
+
+
 def zipped_download_request(context: Context, data_dict: dict[str, Any]):
     prefect_url: str = config.get("ckanext.wri.prefect_url")
     deployment_name: str = config.get("ckanext.wri.datapusher_deployment_name")
@@ -93,11 +118,21 @@ def zipped_download_request(context: Context, data_dict: dict[str, Any]):
         "state": "submitting",
         "key": filename,
         "value": "{}",
-        "error": "{}",
+        "error": "{}", 
     }
 
+    admin_email= get_admin_emails_for_dataset(dataset_id)
+    log.error(f"admin_email: {admin_email}")
+
+
+    value = {}
+    if admin_email:
+        value["admin_emails"] = admin_email
+
     if email:
-        task["value"] = json.dumps({"emails": [email]})
+        value["emails"] = [email]
+
+    task["value"] = json.dumps(value)
 
     try:
         existing_task = p.toolkit.get_action("task_status_show")(
@@ -129,6 +164,8 @@ def zipped_download_request(context: Context, data_dict: dict[str, Any]):
 
                 if update_emails:
                     existing_task_values["emails"] = existing_task_emails
+                    if admin_email:
+                        existing_task_values["admin_emails"] = admin_email
                     existing_task["value"] = json.dumps(existing_task_values)
                     p.toolkit.get_action("task_status_update")(
                         {"ignore_auth": True}, existing_task
@@ -204,7 +241,7 @@ def zipped_download_request(context: Context, data_dict: dict[str, Any]):
         task["state"] = "error"
         task["last_updated"] = (str(datetime.datetime.utcnow()),)
         p.toolkit.get_action("task_status_update")(context, task)
-        send_error([email], "Zipped data")
+        send_error([email]+admin_email, "Zipped data")
         raise p.toolkit.ValidationError(error)
 
     try:
@@ -226,14 +263,17 @@ def zipped_download_request(context: Context, data_dict: dict[str, Any]):
         task["state"] = "error"
         task["last_updated"] = (str(datetime.datetime.utcnow()),)
         p.toolkit.get_action("task_status_update")(context, task)
-        send_error([email], "Zipped data")
+        send_error([email] + admin_email, "Zipped data")
         raise p.toolkit.ValidationError(error)
 
     value = {"job_id": r.json()["id"]}
 
     if email:
         value["emails"] = [email]
 
+    if admin_email:
+        value["admin_emails"] = admin_email
+
     value["download_filename"] = download_filename
 
     task["value"] = json.dumps(value)
@@ -265,28 +305,32 @@ def zipped_download_callback(context: Context, data_dict: dict[str, Any]):
 
     value = json.loads(task["value"])
     emails = value.get("emails", [])
+    admin_email = value.get("admin_emails", [])
     download_filename = value.get("download_filename")
 
     if state == "complete":
         url = data_dict.get("url")
         send_email(emails, url, download_filename)
     else:
-        send_error(emails, download_filename)
+        send_error(emails+admin_email, download_filename)
         log.error(error)
 
 def send_error_callback(context: Context, data_dict: dict[str, Any]):
     entity_id = data_dict.get("entity_id")
+    task_type = data_dict.get("task_type")
     key = data_dict.get("key")
     task = p.toolkit.get_action("task_status_show")(
         context,
-        {"entity_id": entity_id, "task_type": "download_zipped", "key": key},
+        {"entity_id": entity_id, "task_type": task_type, "key": key},
     )
 
     if not task:
         raise logic.NotFound("Task not found")
 
     value = json.loads(task["value"])
     emails = value.get("emails", [])
+    admin_emails = value.get("admin_emails", [])
+    emails += admin_emails
     download_filename = value.get("download_filename")
     dataset_name = fetch_dataset_name({
         "entity_id": entity_id,

ckan-backend-dev/src/ckanext-wri/ckanext/wri/logic/action/get.py

@@ -1848,7 +1848,6 @@ def organization_show(context, data_dict):
     data_dict = old_organization_show(context, data_dict)
     user = context.get("user")
 
-
     if not authz.is_sysadmin(user):
         is_authorized = get_action("organization_list")(context, {"q": data_dict.get("name")})
         if not is_authorized:

ckan-backend-dev/src/ckanext-wri/ckanext/wri/plugin.py

@@ -275,6 +275,7 @@ def get_actions(self):
             "download_event_create": download_event_create,
             "download_event_list": get_download_events,
             "organization_create": organization_create,
+
             "prefect_send_error_callback": send_error_callback,
         }
 

datapusher/main.py

@@ -131,7 +131,6 @@ def convert_store_to_file(
     try:
         logger = get_run_logger()
         ckan_url = config.get("CKAN_URL")
-
         logger.info("Fetching data...")
         data = query_datastore(
             api_key, ckan_url, sql, provider, rw_id, carto_account, format
@@ -169,7 +168,8 @@ def convert_store_to_file(
             "prefect_send_error_callback",
             {
                 "task_id": task_id,
-                "url": url,
+                "url": "",
+                "task_type": "download",
                 "state": "complete",
                 "entity_id": resource_id,
                 "entity_type": "resource",
@@ -195,7 +195,6 @@ async def download_subset_of_data(
     try:
         logger = get_run_logger()
         ckan_url = config.get("CKAN_URL")
-
         logger.info("Fetching data...")
         data = []
         if provider == "datastore":
@@ -237,6 +236,7 @@ async def download_subset_of_data(
             "prefect_send_error_callback",
             {
                 "task_id": task_id,
+                "task_type": "download_subset",
                 "url": "",
                 "state": "failed",
                 "entity_id": id if provider == "datastore" else dataset_id,
@@ -260,7 +260,6 @@ async def download_resources_zipped(
     try:
         logger = get_run_logger()
         ckan_url = config.get("CKAN_URL")
-        print("Filename", filename)
         with tempfile.TemporaryDirectory() as temp_dir:
             tasks = await download_keys(keys, filename, temp_dir)
             data = list(tasks)
@@ -295,6 +294,7 @@ async def download_resources_zipped(
             {
                 "task_id": task_id,
                 "url": "",
+                "task_type": "download_zipped",
                 "state": "failed",
                 "entity_id": dataset_id,
                 "entity_type": "dataset",

deployment/ckan/Dockerfile

@@ -32,6 +32,7 @@ RUN chmod 777 ${APP_DIR}/src/ckanext-s3filestore
 # Required by shapely dependency
 RUN apk --update add build-base libxslt-dev python3-dev 
 RUN apk update
+RUN apk add git=2.40.4-r0
 RUN apk add postgresql15-client=15.11-r0
 
 RUN pip install --force-reinstall -v "Pillow==11.0.0"
@@ -96,10 +97,6 @@ RUN ckan config-tool ${CKAN_INI} "ckan.plugins = ${CKAN__PLUGINS}"
 RUN if [ "$GITHUB_ACTIONS" = "true" ]; then ckan config-tool ${CKAN_INI} "ckan.cors.origin_allow_all = True"; fi
 
 
-RUN if [ "$GITHUB_ACTIONS" = "true" ]; then \
-      ckan config-tool ${CKAN_INI} "ckan.cors.origin_allow_all = True"; \
-    fi
-
 USER root
 
 RUN chown -R ckan:ckan ${APP_DIR}/src/