from arcgis.gis import GIS import sys import logging from functions import get_local_date, set_groups_data, get_edit_dates, extract_user_props # create a list of dictionaries containing survey properties def get_survey_props(url, username, password, platform): try: # Connect to GIS Portal gis = GIS(url, username, password) # Search all Survey123 forms surveys = gis.content.search(query='type:"Form"', max_items=5000) # Container for results data = [] # add message for log file logging.info(f'generated list of surveys for {platform}') # control date created field based upon platform if platform == 'Portal': date_field = "created_date" else: date_field = "CreationDate" # iterate over Survey123 items for survey in surveys: # groups the survey is shared shared with groups = survey.sharing.groups.list() # if no grop sharing, skip survey if len(groups) == 0: continue else: record_count = None records_last_added = None field_schema_last_modified = None service_found = False try: # Try related_items approach first related_services = survey.related_items('Survey2Service', 'forward') for service in related_services: if service.type == "Feature Service": try: if service.layers and len(service.layers) > 0: layer = service.layers[0] record_count = layer.query(return_count_only=True) dates = get_edit_dates(layer.properties) # query most recent date that a record was added query_result = layer.query( where="1=1", out_fields=date_field, order_by_fields=f"{date_field} DESC", return_geometry=False, result_record_count=1 ) if query_result.features: recent_date = query_result.features[0].attributes[f'{date_field}'] records_last_added = get_local_date(recent_date) #records_last_added = get_local_date(dates['dataLastEditDate']) field_schema_last_modified = get_local_date(dates['schemaLastEditDate']) service_found = True break except Exception as e: # add message for log file logging.info(f"Error querying layer from related service: {e}") except Exception as e: # add message for log file logging.info(f"Error getting related items for {survey.title}: {e}") # Fallback: Use serviceItemId (more reliable in ArcGIS Enterprise) if not service_found: try: service_item_id = survey.properties.get("serviceItemId") if service_item_id: service_item = gis.content.get(service_item_id) if service_item and service_item.type == "Feature Service": if service_item.layers and len(service_item.layers) > 0: layer = service_item.layers[0] record_count = layer.query(return_count_only=True) dates = get_edit_dates(layer.properties) #records_last_added = get_local_date(dates['dataLastEditDate']) query_result = layer.query( where="1=1", out_fields=date_field, order_by_fields=f"{date_field} DESC", return_geometry=False, result_record_count=1 ) if query_result.features: recent_date = query_result.features[0].attributes[f'{date_field}'] records_last_added = get_local_date(recent_date) field_schema_last_modified = get_local_date(dates['schemaLastEditDate']) service_found = True except Exception as e: # add message for log file logging.info(f"Fallback using serviceItemId failed for {survey.title}: {e}") # process group membership; start of loop tests that survey is shared with a group formatted_group_members = None # groups the survey is shared shared with #groups = survey.sharing.groups.list() # number of people in the groups group_members_count = 0 # case where survey is not shared with groups if len(groups) == 0: group_names = 'None' group_members = 'None' # case where survey is shared with at least 1 group else: # get names of groups group_names = [group.title for group in groups] # container for group memembers group_members = [] for group in groups: # get list of members # {'owner': 'user', 'admins': ['user'], 'users': ['user']} members = group.get_members() # add owner group_members.append(members['owner']) # add admins for member in members['admins']: group_members.append(member) # add users for member in members['users']: group_members.append(member) # remove duplicates group_members = sorted(set(group_members)) group_members_count += len(group_members) formatted_group_members = extract_user_props(group_members) # standard survey properties # remove whitespace survey_id = str(survey.id).lstrip().rstrip() if survey.id is not None else survey.id survey_title = str(survey.title).lstrip().rstrip() if survey.title is not None else survey.title survey_owner = str(survey.owner).lstrip().rstrip() if survey.owner is not None else survey.owner survey_summary = str(survey.snippet).lstrip().rstrip() if survey.snippet is not None else survey.snippet survey_desc = str(survey.description).lstrip().rstrip() if survey.description is not None else survey.description survey_access = str(survey.access).lstrip().rstrip() if survey.access is not None else survey.access # add entry to data container data.append({ "Survey_ID": survey_id, "Title": survey_title, "Owner": survey_owner, "Survey_Created": get_local_date(survey.created), "Survey_Last_Modified_By_Owner": get_local_date(survey.modified), "Summary": survey_summary, "Description": survey_desc, "Sharing_Access": survey_access, "Record_Count": record_count, "Records_Last_Added": records_last_added, "Survey_Table_Schema_Last_Modified_By_Owner": field_schema_last_modified, "Shared_with_Groups": set_groups_data(group_names), "Group_Members": set_groups_data(formatted_group_members) if formatted_group_members else set_groups_data(group_members), "Count_Group_Members": group_members_count, "Platform": platform }) # end iteration over surveys except (Exception, EnvironmentError) as e: # information about the error tbE = sys.exc_info()[2] # write the line number the error occured on logging.info(f"Failed at Line {tbE.tb_lineno}") # write error message logging.info(f"Error: {e}") finally: # return however much data has been produced # TODO: check for entries in data # how to process if empty if data is not None: return data else: logging.info("no data to return from function call") return None