feat(webportal): Add export to Web Portal function

specifyweb/backend/stored_queries/execution.py

@@ -3,23 +3,31 @@
 import logging
 import os
 import re
+import traceback
+import uuid
+from io import StringIO
+from xml.sax.saxutils import escape
+from zipfile import ZIP_DEFLATED, ZipFile
 
-from typing import Literal, NamedTuple
+from typing import Any, Literal, NamedTuple
 import xml.dom.minidom
 from collections import namedtuple, defaultdict
 from functools import reduce
 
 from django.conf import settings
+from django.apps import apps
 from django.db import transaction
 from django.utils import timezone
 from specifyweb.backend.inheritance.api import cog_inheritance_post_query_processing, parent_inheritance_post_query_processing
 from specifyweb.backend.inheritance.utils import get_cat_num_inheritance_setting, get_parent_cat_num_inheritance_setting
+from specifyweb.backend.context.schema_localization import get_schema_localization
 from specifyweb.backend.stored_queries.utils import log_sqlalchemy_query
 from specifyweb.specify.utils.field_change_info import FieldChangeInfo
+from specifyweb.specify.utils.uiformatters import CNNField, get_catalognumber_format, get_uiformatter
 from sqlalchemy import sql, orm, func, text
 from sqlalchemy.sql.expression import asc, desc, insert, literal
 
-from specifyweb.specify.models_utils.models_by_table_id import get_table_id_by_model_name
+from specifyweb.specify.models_utils.models_by_table_id import get_model_by_table_id, get_table_id_by_model_name
 from specifyweb.backend.stored_queries.group_concat import group_by_displayed_fields
 from specifyweb.backend.trees.utils import get_search_filters
 
@@ -28,13 +36,15 @@
 from .query_construct import QueryConstruct
 from .relative_date_utils import apply_absolute_date
 from .field_spec_maps import apply_specify_user_name
+from .web_portal_export import query_to_web_portal_zip as _query_to_web_portal_zip, _portal_attachment_map
 from specifyweb.backend.notifications.models import Message
 from specifyweb.backend.permissions.permissions import check_table_permissions
 from specifyweb.specify.models import Loan, Loanpreparation, Loanreturnpreparation, Taxontreedef
 from specifyweb.backend.workbench.upload.auditlog import auditlog
 from specifyweb.backend.stored_queries.group_concat import group_by_displayed_fields
-from specifyweb.backend.stored_queries.queryfield import fields_from_json, QUREYFIELD_SORT_T
+from specifyweb.backend.stored_queries.queryfield import QueryField, fields_from_json, QUREYFIELD_SORT_T
 from specifyweb.backend.stored_queries.synonomy import synonymize_tree_query
+
 from specifyweb.specify.datamodel import datamodel, is_tree_table
 
 logger = logging.getLogger(__name__)
@@ -258,22 +268,84 @@ def do_export(spquery, collection, user, filename, exporttype, host):
     message_type = "query-export-to-csv-complete"
 
     with models.session_context() as session:
-        field_specs = fields_from_json(spquery['fields'])
-        if exporttype == 'csv':
-            query_to_csv(session, collection, user, tableid, field_specs, path,
-                         recordsetid=recordsetid, 
-                         captions=spquery['captions'], strip_id=True,
-                         distinct=spquery['selectdistinct'], delimiter=spquery['delimiter'], bom=spquery['bom'])
-        elif exporttype == 'kml':
-            query_to_kml(session, collection, user, tableid, field_specs, path, spquery['captions'], host,
-                         recordsetid=recordsetid, strip_id=False, selected_rows=spquery.get('selectedrows', None))
-            message_type = 'query-export-to-kml-complete'
+        try:
+            field_specs = fields_from_json(spquery['fields'])
+            if exporttype == 'csv':
+                query_to_csv(session, collection, user, tableid, field_specs, path,
+                             recordsetid=recordsetid, 
+                             captions=spquery['captions'], strip_id=True,
+                             distinct=spquery['selectdistinct'], delimiter=spquery['delimiter'], bom=spquery['bom'])
+                message_type = 'query-export-to-csv-complete'
+            elif exporttype == 'kml':
+                query_to_kml(session, collection, user, tableid, field_specs, path, spquery['captions'], host,
+                             recordsetid=recordsetid, strip_id=False, selected_rows=spquery.get('selectedrows', None))
+                message_type = 'query-export-to-kml-complete'
+            elif exporttype == 'webportal':
+                query_to_web_portal_zip(
+                    session,
+                    collection,
+                    user,
+                    tableid,
+                    field_specs,
+                    path,
+                    spquery['captions'],
+                    recordsetid=recordsetid,
+                    distinct=spquery['selectdistinct'],
+                )
+                message_type = 'query-export-to-webportal-complete'
+            else:
+                # This should never happen because the export type is controlled by the backend, but just in case.
+                raise ValueError(f"Unsupported export type: {exporttype}")
+        except Exception as e:
+            logger.exception(
+                "Export failed for %s: collection %s, file %s, type %s",
+                user, collection, filename, exporttype,
+            )
+            tb = traceback.format_exc()
+            error_details = {'error': str(e)}
+            if tb:
+                error_details['traceback'] = tb
+            message_type = f'query-export-to-{exporttype}-failed'
+            Message.objects.create(user=user, content=json.dumps({
+                'type': message_type,
+                'file': filename,
+                'error': error_details,
+            }))
+            raise
 
     Message.objects.create(user=user, content=json.dumps({
         'type': message_type,
         'file': filename,
     }))
 
+
+def query_to_web_portal_zip(
+    session,
+    collection,
+    user,
+    tableid,
+    field_specs,
+    path,
+    captions,
+    recordsetid=None,
+    distinct=False,
+):
+    return _query_to_web_portal_zip(
+        session,
+        collection,
+        user,
+        tableid,
+        field_specs,
+        path,
+        captions,
+        build_query_fn=build_query,
+        build_query_props_cls=BuildQueryProps,
+        apply_special_post_query_processing_fn=apply_special_post_query_processing,
+        set_group_concat_max_len_fn=set_group_concat_max_len,
+        recordsetid=recordsetid,
+        distinct=distinct,
+    )
+
 # def stored_query_to_csv(query_id, collection, user, path):
 #     """Executes a query from the Spquery table with the given id and send
 #     the results to a CSV file at path.

specifyweb/backend/stored_queries/queryfieldspec.py

@@ -18,6 +18,7 @@
 from . import models
 from .query_ops import QueryOps
 from specifyweb.specify.models_utils.load_datamodel import Table, Field, Relationship
+from specifyweb.specify.datamodel import is_tree_table
 
 logger = logging.getLogger(__name__)
 
@@ -259,7 +260,19 @@ def from_stringid(cls, stringid: str, is_relation: bool):
         field = node.get_field(extracted_fieldname, strict=False)
 
         tree_rank_name = None
-        if field is None:  # try finding tree
+        if (
+            field is None
+            and is_relation
+            and not is_tree_table(node)
+            and extracted_fieldname.lower() == table_name.lower() == node.name.lower()
+        ):
+            # Legacy relation stringids like "locality.locality" serialize the current related table as a formatted
+            # step, not as an actual field on that table.
+            # Preserve that sentinel so nested formatted relations keep the same row plan shape, without treating
+            # arbitrary unknown fields on non-tree tables as tree ranks.
+            tree_rank_name = extracted_fieldname
+            join_path.append(TreeRankQuery.create(tree_rank_name, node.name))
+        elif field is None and is_tree_table(node):  # try finding tree only on tree tables
             tree_rank_name, field = find_tree_and_field(node, extracted_fieldname)
             if tree_rank_name:
                 tree_rank = TreeRankQuery.create(

specifyweb/backend/stored_queries/tests/test_execution/test_field_specs_from_json.py

@@ -108,3 +108,24 @@ def test_static_field_specs(self): # pragma: no cover
         # generate_fields_test_str(query_fields, "static_simple_field_spec")
 
         self.assertEqual(static_simple_field_spec, query_fields)
+
+    def test_non_tree_table_does_not_parse_tree_rank(self):
+        table = datamodel.get_table_strict("CollectionObject")
+        stringid = f"{table.tableId}.collectionobject.NotARealField"
+
+        fieldspec = QueryFieldSpec.from_stringid(stringid, False)
+
+        self.assertFalse(fieldspec.contains_tree_rank())
+        self.assertIsNone(fieldspec.tree_rank)
+        self.assertIsNone(fieldspec.get_field())
+
+    def test_nested_formatted_relation_keeps_legacy_sentinel(self):
+        fieldspec = QueryFieldSpec.from_stringid("1,10,2.locality.locality", True)
+
+        self.assertTrue(fieldspec.contains_tree_rank())
+        self.assertEqual(fieldspec.tree_rank, "locality")
+        self.assertEqual(
+            [node.name for node in fieldspec.join_path],
+            ["collectingEvent", "locality", "locality"],
+        )
+        self.assertIsInstance(fieldspec.get_field(), TreeRankQuery)

specifyweb/backend/stored_queries/tests/test_views/test_export_web_portal.py

@@ -0,0 +1,70 @@
+from unittest.mock import MagicMock, Mock, patch
+
+from django.test import Client
+
+from specifyweb.backend.stored_queries.tests.tests import SQLAlchemySetup
+
+from .raw_query import get_simple_query
+
+
+class TestExportWebPortal(SQLAlchemySetup):
+    @patch("specifyweb.backend.stored_queries.views.Thread")
+    def test_export(self, thread: Mock):
+        c = Client()
+        c.force_login(self.specifyuser)
+
+        response = c.post(
+            "/stored_query/exportwebportal/",
+            get_simple_query(self.specifyuser),
+            content_type="application/json",
+        )
+
+        self._assertStatusCodeEqual(response, 200)
+        thread.assert_called_once()
+        self.assertTrue(thread.return_value.daemon)
+        thread.return_value.start.assert_called_once()
+        self._assertContentEqual(response, "OK")
+
+    def test_portal_attachment_map(self):
+        from specifyweb.backend.stored_queries import execution
+
+        class FakeAttachment:
+            id = 5291
+            attachmentlocation = "sp6896513492722436219.att.JPG"
+            origfilename = "29432.JPG"
+            title = "Figure 1"
+
+        class FakeJoinRecord:
+            collectionobject_id = 123
+            attachment = FakeAttachment()
+
+        class FakeJoinQuery:
+            def select_related(self, *_args, **_kwargs):
+                return [FakeJoinRecord()]
+
+        class FakeJoinManager:
+            def __init__(self):
+                self.filter_kwargs = None
+
+            def filter(self, **kwargs):
+                self.filter_kwargs = kwargs
+                return FakeJoinQuery()
+
+        fake_join_manager = FakeJoinManager()
+        fake_base_model = type("Collectionobject", (), {"_meta": MagicMock(app_label="specifyweb")})
+        fake_table = MagicMock()
+        fake_table.attachments_field = MagicMock()
+
+        with patch.object(execution.datamodel, "get_table_by_id", return_value=fake_table), patch.object(
+            execution, "get_model_by_table_id", return_value=fake_base_model
+        ), patch.object(execution.apps, "get_model", return_value=type("Collectionobjectattachment", (), {"objects": fake_join_manager})):
+            result = execution._portal_attachment_map(1, [123])
+
+        self.assertEqual(
+            fake_join_manager.filter_kwargs,
+            {"collectionobject_id__in": [123], "attachment__ispublic": True},
+        )
+        self.assertEqual(
+            result["123"],
+            '[{AttachmentID:5291,AttachmentLocation:"sp6896513492722436219.att.JPG",Title:"Figure 1"}]',
+        )

specifyweb/backend/stored_queries/urls.py

@@ -7,6 +7,7 @@
     path('ephemeral/', views.ephemeral),
     path('exportcsv/', views.export_csv),
     path('exportkml/', views.export_kml),
+    path('exportwebportal/', views.export_to_web_portal),
     path('make_recordset/', views.make_recordset),
     path('merge_recordsets/', views.merge_recordsets),
     path('return_loan_preps/', views.return_loan_preps),

specifyweb/backend/stored_queries/views.py

@@ -34,6 +34,7 @@ class QueryBuilderPt(PermissionTarget):
     execute = PermissionTargetAction()
     export_csv = PermissionTargetAction()
     export_kml = PermissionTargetAction()
+    export_to_web_portal = PermissionTargetAction()
     create_recordset = PermissionTargetAction()
 
 def value_from_request(field, get):
@@ -150,8 +151,6 @@ def batch_edit(request):
 @never_cache
 def export_csv(request):
     """Executes and return as CSV the results of the query provided as JSON in the POST body."""
-    check_permission_targets(request.specify_collection.id, request.specify_user.id, [
-        QueryBuilderPt.execute, QueryBuilderPt.export_csv])
     try:
         spquery = json.load(request)
     except ValueError as e:
@@ -164,6 +163,9 @@ def export_csv(request):
         logger.debug('forcing collection to %s', collection.collectionname)
     else:
         collection = request.specify_collection
+
+    check_permission_targets(collection.id, request.specify_user.id, [
+        QueryBuilderPt.execute, QueryBuilderPt.export_csv])
 
     file_name = format_export_file_name(spquery, "csv")
 
@@ -177,8 +179,6 @@ def export_csv(request):
 @never_cache
 def export_kml(request):
     """Executes and return as KML the results of the query provided as JSON in the POST body."""
-    check_permission_targets(request.specify_collection.id, request.specify_user.id, [
-        QueryBuilderPt.execute, QueryBuilderPt.export_kml])
     try:
         spquery = json.load(request)
     except ValueError as e:
@@ -195,13 +195,50 @@ def export_kml(request):
     else:
         collection = request.specify_collection
 
+    check_permission_targets(collection.id, request.specify_user.id, [
+        QueryBuilderPt.execute, QueryBuilderPt.export_kml])
+
     file_name = format_export_file_name(spquery, "kml")
 
     thread = Thread(target=do_export, args=(spquery, collection, request.specify_user, file_name, 'kml', the_host))
     thread.daemon = True
     thread.start()
     return HttpResponse('OK', content_type='text/plain')
 
+
+@require_POST
+@login_maybe_required
+@never_cache
+def export_to_web_portal(request):
+    """Executes and returns as ZIP the web portal export package for the query provided as JSON in the POST body."""
+    try:
+        spquery = json.load(request)
+    except ValueError as e:
+        return HttpResponseBadRequest(e)
+
+    logger.info('export web portal query: %s', spquery)
+
+    if 'collectionid' in spquery:
+        collection = Collection.objects.get(pk=spquery['collectionid'])
+        logger.debug('forcing collection to %s', collection.collectionname)
+    else:
+        collection = request.specify_collection
+
+    check_permission_targets(collection.id, request.specify_user.id, [
+        QueryBuilderPt.execute,
+        QueryBuilderPt.export_to_web_portal,
+    ])
+
+    file_name = format_export_file_name(spquery, 'zip')
+
+    thread = Thread(
+        target=do_export,
+        args=(spquery, collection, request.specify_user, file_name, 'webportal', None),
+    )
+    thread.daemon = True
+    thread.start()
+    return HttpResponse('OK', content_type='text/plain')
+
 @require_POST
 @login_maybe_required
 @never_cache