import datetime, json, pytz, secrets, time import pandas, xlrd # qrcode from fastapi import APIRouter, Body, Depends, Header, HTTPException, Query, Response, status from pydantic import BaseModel, EmailStr, Field from typing import Dict, List, Optional, Set, Union from app.lib_general import log, logging, secure_hash_string, common_route_params, Common_Route_Params from app.config import settings from app.db_sql import sql_insert, sql_update, sql_insert_or_update, sql_select, sql_delete, redis_lookup_id_random from app.routers.api_crud import delete_obj_template, get_obj_template, get_obj_li_template, patch_obj_template, post_obj_template from app.methods.e_impexium_methods import get_custom_fields, get_events, get_event_registrants, get_individual_custom_fields from app.methods.event_person_methods import create_event_person_obj, create_update_event_person_obj_v4, get_event_person_rec_list, load_event_person_obj, update_event_person_obj, update_event_person_obj_v3 from app.models.event_badge_models import Event_Badge_Base from app.models.response_models import Resp_Body_Base, mk_resp router = APIRouter() # ### BEGIN ### API Impexium ### event_import_reg() ### # Updated 2022-04-15 @router.get('/event/{e_impexium_event_id}/import_reg', response_model=Resp_Body_Base) async def event_import_reg( e_impexium_event_id: str = Query(..., min_length=11, max_length=22), # For ISHLT: 42_AM (2022-04); EX22_AM (2022-04); 41V_2 (2021-04) registered_since: datetime.datetime = None, # datetime.datetime.now() + datetime.timedelta(seconds=120) details: bool = True, inc_membership_fields: bool = True, page: int = 0, # 250 per page from Impexium event_id: str = Query(..., min_length=11, max_length=22), # For ISHLT: ZDzTBlevhZs (2022-04) # Account ID For ISHLT: d8TqXqf1EOg return_detail: bool = False, commons: Common_Route_Params = Depends(common_route_params), ): log.setLevel(logging.INFO) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL log.debug(locals()) log.info('Starting Impexium event registration import...') account_id = commons.x_account_id if event_id := redis_lookup_id_random(record_id_random=event_id, table_name='event'): pass else: return mk_resp(data=None, status_code=404, status_message=f'The Event ID was not found. Event ID: {event_id}', response=commons.response) event_registrant_li = get_event_registrants( event_code = e_impexium_event_id, registered_since = registered_since, details = details, page = page ) log.debug(type(event_registrant_li)) if event_registrant_li: pass else: return mk_resp(data=None, status_code=404, status_message=f'Checked for registrations in Impexium. No Impexium registrations found.', response=commons.response) # return mk_resp(data=len(event_registrant_li), status_code=501, response=commons.response) # Not Implemented loop_count = 0 loop_limit = 2500 # 200 event_person_li = [] event_person_summary_li = [] for event_registrant in event_registrant_li: if loop_count > loop_limit: break log.info(f"Record Num: {event_registrant.get('recordNumber')}, Registration Num: {event_registrant.get('registrationNumber')}, Registrant Type Code: {event_registrant.get('registrantTypeCode')}") # log.debug(event_registrant) # Creating an external ID from the Impexium record number and registration number. Very explicitly converting each number to a string before joining, just in case. external_id_old = str(event_registrant.get('recordNumber'))+':'+str(event_registrant.get('registrationNumber')) external_id_v2 = str(e_impexium_event_id)+':'+str(event_registrant.get('recordNumber'))+':'+str(event_registrant.get('registrationNumber')) external_id_v3 = str(e_impexium_event_id)+':'+str(event_registrant.get('recordNumber')) external_sys_id = event_registrant.get('id') # Impexium UUID for person external_reg_id = event_registrant.get('recordNumber') # Impexium record number for a registration (guest should match) event_person_summary_data = {} event_person_summary_data['external_id'] = external_id_v3 event_person_data = {} event_person_data['enable'] = True event_person_data['account_id'] = account_id event_person_data['event_id'] = event_id event_person_data['external_id'] = external_id_v3 event_person_data['external_sys_id'] = external_sys_id event_person_data['external_reg_id'] = external_reg_id # event_person_data['external_id_old'] = external_id_old email = None country_subdivision_code = None state_province_abb = None country_alpha_2_code = None country = None if details: if emails := event_registrant.get('emails'): if isinstance(emails, list) and len(emails): email = emails[0].get('address') event_person_summary_data['email'] = email if addresses := event_registrant.get('addresses'): if isinstance(addresses, list) and len(addresses): for address in addresses: if primary_address := address.get('primary'): country_subdivision_code = address.get('stateISOCode') state_province_abb = address.get('stateAbbreviation') if country_data := address.get('countryData'): country_alpha_3_code = country_data.get('threeLetterIsoCode') country_alpha_2_code = country_data.get('twoLetterIsoCode') country = address.get('country') degrees = None organization_name = None if inc_membership_fields: individual_custom_fields_li = get_individual_custom_fields(individual_id=event_registrant.get('id')) for individual_custom_field in individual_custom_fields_li: if individual_custom_field.get('name') == 'degree': degrees = individual_custom_field.get('value') if individual_custom_field.get('name') == 'organization_name': organization_name = individual_custom_field.get('value') log.debug(f'Degrees: {degrees}; Organization Name: {organization_name}') event_person_profile_data = {} event_person_profile_data['enable'] = True event_person_profile_data['pronouns'] = event_registrant.get('gender') event_person_profile_data['informal_name'] = event_registrant.get('preferredFirstName') event_person_profile_data['title_names'] = event_registrant.get('prefix') event_person_profile_data['given_name'] = event_registrant.get('firstName') event_person_profile_data['middle_name'] = event_registrant.get('middleName') event_person_profile_data['family_name'] = event_registrant.get('lastName') event_person_profile_data['designations'] = event_registrant.get('suffix') if degrees: event_person_profile_data['professional_title'] = degrees # Ideally should be degrees for ISHLT else: event_person_profile_data['professional_title'] = None # event_registrant.get('title') # Should this be None if no degrees found above??? event_person_profile_data['full_name'] = event_registrant.get('badgeName') event_person_summary_data['full_name'] = event_registrant.get('badgeName') if organization_name: event_person_profile_data['affiliations'] = organization_name # Ideally should be organization_name for ISHLT else: event_person_profile_data['affiliations'] = event_registrant.get('badgeOrganization') # Should this be None if no organization_name found above??? if email: event_person_profile_data['email'] = email event_person_data['event_person_profile'] = {} event_person_data['event_person_profile'] = event_person_profile_data event_badge_data = {} event_badge_data['enable'] = True event_badge_data['event_id'] = event_id event_badge_data['external_id'] = external_id_v3 event_badge_data['external_sys_id'] = external_sys_id event_badge_data['external_reg_id'] = external_reg_id # event_badge_data['external_id_old'] = external_id_old if reg_type_code := event_registrant.get('registrantTypeCode'): event_badge_data['badge_type_code'] = event_registrant.get('registrantTypeCode') # Using this as the badge_type ishlt_m = ('AHMB', 'IMBR', 'MBR') ishlt_nm = ('ANHM', 'INMB', 'NMBR') ishlt_student_m = ('SMBR') ishlt_student_nm = ('SNMB') ishlt_ex_all = ('EX') ishlt_ex_booth = ('EX') ishlt_guest = ('GUEST') if reg_type_code in ishlt_m: event_badge_data['badge_type'] = 'Member' elif reg_type_code in ishlt_nm: event_badge_data['badge_type'] = 'Non-Member' elif reg_type_code in ishlt_student_m: event_badge_data['badge_type'] = 'Student/Trainee Member' elif reg_type_code in ishlt_student_nm: event_badge_data['badge_type'] = 'Student/Trainee Non-Member' elif reg_type_code in ishlt_ex_all: event_badge_data['badge_type'] = 'Exhibitor All Access' elif reg_type_code in ishlt_ex_booth: event_badge_data['badge_type'] = 'Exhibitor Booth Staff' elif reg_type_code in ishlt_guest: event_badge_data['badge_type'] = 'Guest' else: event_badge_data['badge_type'] = event_registrant.get('registrantTypeName') if custom_fields := event_registrant.get('itemizedCustomFields'): if isinstance(custom_fields, list): log.info('Found list of custom fields for an exhibitor. Searching for "exhibitor_reg_type') for field in custom_fields: if field.get('name') == 'registration_type': event_badge_data['registration_type_code'] = field.get('value') event_badge_data['registration_type'] = field.get('value') break else: log.info('Found custom fields section, but no list was found.') else: log.warning('The custom fields were not found at all. Are the "details" being included?') event_badge_data['event_badge_template_id'] = 3 elif e_impexium_event_id == '42_AM': if not reg_type_code and event_registrant.get('guestOfRecordNumber'): event_badge_data['badge_type_code'] = 'GUEST' event_badge_data['badge_type'] = 'Guest' event_badge_data['registration_type_code'] = 'GUEST' event_badge_data['registration_type'] = 'Guest' else: event_badge_data['badge_type_code'] = 'UNKNOWN' event_badge_data['badge_type'] = 'Unknown Type' event_badge_data['registration_type_code'] = 'UNKNOWN' event_badge_data['registration_type'] = 'Unknown Type' event_badge_data['event_badge_template_id'] = 3 elif e_impexium_event_id == 'EX22_AM': event_badge_data['badge_type_code'] = 'EX' event_badge_data['badge_type'] = 'Exhibit Staff' if custom_fields := event_registrant.get('itemizedCustomFields'): if isinstance(custom_fields, list): log.info('Found list of custom fields for an exhibitor. Searching for "exhibitor_reg_type') for field in custom_fields: if field.get('name') == 'exhibitor_reg_type': if field.get('value') == 'Exhibitor Booth Staff ': event_badge_data['badge_type_code'] = 'EXO' event_badge_data['badge_type'] = 'Exhibitor Booth Staff' elif field.get('value') == 'Exhibitor All Access ': event_badge_data['badge_type_code'] = 'EXALL' event_badge_data['badge_type'] = 'Exhibitor All Access' event_badge_data['registration_type_code'] = field.get('value') event_badge_data['registration_type'] = field.get('value') break else: log.info('Found custom fields section, but no list was found.') else: log.warning('The custom fields were not found at all. Are the "details" being included?') event_badge_data['event_badge_template_id'] = 3 elif e_impexium_event_id == '2022HEART': if not reg_type_code and event_registrant.get('guestOfRecordNumber'): event_badge_data['badge_type_code'] = 'GUEST' event_badge_data['badge_type'] = 'Guest' event_badge_data['registration_type_code'] = 'GUEST' event_badge_data['registration_type'] = 'Guest' else: event_badge_data['badge_type_code'] = 'HEART' event_badge_data['badge_type'] = 'HFTX Core' event_badge_data['registration_type_code'] = 'HEART' event_badge_data['registration_type'] = 'HFTX Core' event_badge_data['event_badge_template_id'] = 5 # event_badge_data['registration_type_code'] = event_registrant.get('registrantTypeCode') # event_badge_data['registration_type'] = event_registrant.get('registrantTypeName') elif e_impexium_event_id == '2022LUNG': if not reg_type_code and event_registrant.get('guestOfRecordNumber'): event_badge_data['badge_type_code'] = 'GUEST' event_badge_data['badge_type'] = 'Guest' event_badge_data['badge_type_code'] = 'GUEST' event_badge_data['badge_type'] = 'Guest' else: event_badge_data['badge_type_code'] = 'LUNG' event_badge_data['badge_type'] = 'LTX Core' event_badge_data['registration_type_code'] = 'LUNG' event_badge_data['registration_type'] = 'LTX Core' event_badge_data['event_badge_template_id'] = 5 # event_badge_data['registration_type_code'] = event_registrant.get('registrantTypeCode') # event_badge_data['registration_type'] = event_registrant.get('registrantTypeName') else: event_badge_data['badge_type_code'] = 'UNKNOWN' event_badge_data['badge_type'] = 'Unknown Type' event_badge_data['event_badge_template_id'] = 3 event_badge_data['registration_type_code'] = event_registrant.get('registrantTypeCode') event_badge_data['registration_type'] = event_registrant.get('registrantTypeName') event_badge_data['member_type_code'] = event_registrant.get('customerType') # Using this as the member_type event_badge_data['pronouns'] = event_registrant.get('gender') event_badge_data['informal_name'] = event_registrant.get('preferredFirstName') event_badge_data['title_names'] = event_registrant.get('prefix') event_badge_data['given_name'] = event_registrant.get('firstName') event_badge_data['middle_name'] = event_registrant.get('middleName') event_badge_data['family_name'] = event_registrant.get('lastName') event_badge_data['designations'] = event_registrant.get('suffix') if degrees: event_badge_data['professional_title'] = degrees # Ideally should be degrees for ISHLT else: event_badge_data['professional_title'] = None # event_registrant.get('title') # Should this be None if no degrees found above??? event_badge_data['full_name'] = event_registrant.get('badgeName') if organization_name: event_badge_data['affiliations'] = organization_name # Ideally should be organization_name for ISHLT else: event_badge_data['affiliations'] = event_registrant.get('badgeOrganization') # Should this be None if no organization_name found above??? if email: event_badge_data['email'] = email event_badge_data['city'] = event_registrant.get('badgeCity') event_badge_data['state_province'] = event_registrant.get('badgeState') if country_subdivision_code: event_badge_data['country_subdivision_code'] = country_subdivision_code if state_province_abb: event_badge_data['state_province_abb'] = state_province_abb if country_alpha_2_code: event_badge_data['country_alpha_2_code'] = country_alpha_2_code if country: event_badge_data['country'] = country location = None if event_badge_data['city'] and state_province_abb and country_alpha_2_code and country_alpha_2_code == 'US': city = event_badge_data['city'] location = f'{city}, {state_province_abb} {country_alpha_2_code}' elif event_badge_data['city'] and state_province_abb and country and country == 'United States': city = event_badge_data['city'] location = f'{city}, {state_province_abb} {country}' elif event_badge_data['city'] and event_badge_data['state_province'] and country: city = event_badge_data['city'] state_province = event_badge_data['state_province'] location = f'{city}, {state_province} {country}' elif event_badge_data['city'] and country: city = event_badge_data['city'] location = f'{city}, {country}' if location: event_badge_data['location'] = location event_person_data['event_badge'] = {} event_person_data['event_badge'] = event_badge_data sql_select_event_person = f""" SELECT id AS event_person_id, id_random AS event_person_id_random, external_id AS event_person_external_id, event_badge_id AS event_badge_id, event_person_profile_id AS event_person_profile_id FROM `event_person` AS `event_person` WHERE event_person.event_id = :event_id AND event_person.external_id = :external_id /*LIMIT 1*/; """ if event_person_result := sql_select(sql=sql_select_event_person, data=event_person_data): if isinstance(event_person_result, list): log.error(f'Found more than one Event Person with the same External ID. Count: {len(event_person_result)}') break # return False else: event_person_id = event_person_result.get('event_person_id') event_badge_id = event_person_result.get('event_badge_id') event_person_profile_id = event_person_result.get('event_person_profile_id') log.info(f'Found Event Person. Updating existing... Event Person ID: {event_person_id}') if create_event_person_obj_result := create_update_event_person_obj_v4( event_person_dict_obj = event_person_data, event_person_id = event_person_id, account_id = account_id, event_id = event_id, event_badge_id = event_badge_id, event_person_profile_id = event_person_profile_id, # create_sub_obj = create_sub_obj, # fail_any = fail_any, # return_outline = False, ): event_person_id = create_event_person_obj_result log.warning(f'Event Person updated. Event Person ID: {event_person_id}') else: log.warning(f'Event Person not updated. Event Person ID: {event_person_id}') log.debug(event_badge_obj_in_result) break # return False else: log.info('No Event Person found. Creating new...') if create_event_person_obj_result := create_update_event_person_obj_v4( event_person_dict_obj = event_person_data, account_id = account_id, event_id = event_id, # create_sub_obj = create_sub_obj, # fail_any = fail_any, # return_outline = False, ): event_person_id = create_event_person_obj_result log.warning(f'Event Person created. Event Person ID: {event_person_id}') else: log.warning(f'Event Person not created.') log.debug(event_badge_obj_in_result) break # return False # event_badge_data = {} # external_id = event_registrant.get('recordNumber')+':'+event_registrant.get('registrationNumber') # event_badge_data['external_id'] = external_id # # event_badge_data['reg_num'] = event_registrant.get('registrationNumber') # event_badge_data['badge_type_code'] = event_registrant.get('registrantTypeCode') # Using this as the badge_type # event_badge_data['badge_type'] = event_registrant.get('registrantTypeName') # event_badge_data['member_type_code'] = event_registrant.get('customerType') # Using this as the member_type # event_badge_data['registration_type_code'] = event_registrant.get('registrantTypeCode') # event_badge_data['registration_type'] = event_registrant.get('registrantTypeName') # event_badge_data['pronouns'] = event_registrant.get('gender') # event_badge_data['informal_name'] = event_registrant.get('preferredFirstName') # event_badge_data['title_names'] = event_registrant.get('prefix') # event_badge_data['given_name'] = event_registrant.get('firstName') # event_badge_data['middle_name'] = event_registrant.get('middleName') # event_badge_data['family_name'] = event_registrant.get('lastName') # # event_badge_data['not_sure'] = event_registrant.get('secondLastName') # event_badge_data['designations'] = event_registrant.get('suffix') # event_badge_data['professional_title'] = event_registrant.get('title') # event_badge_data['full_name'] = event_registrant.get('badgeName') # event_badge_data['affiliations'] = event_registrant.get('badgeOrganization') # event_badge_data['city'] = event_registrant.get('badgeCity') # event_badge_data['state_province'] = event_registrant.get('badgeState') # id, attendedDate, sessions, itemizedCustomFields, guestOfRecordNumber, registrantTypeName, boughtTogetherWith, primaryOrganization, customerType, showInDirectory, addresses, emails, phones event_person_li.append(event_person_data) event_person_summary_li.append(event_person_summary_data) loop_count = loop_count + 1 if return_detail: return mk_resp(data=event_person_li, status_message=f'Checked for registrations in Impexium. Found {len(event_person_li)} Impexium registrations.', response=commons.response) else: return mk_resp(data=event_person_summary_li, status_message=f'Checked for registrations in Impexium. Found {len(event_person_summary_li)} Impexium registrations.', response=commons.response) # ### END ### API Impexium ### event_import_reg() ### # ### BEGIN ### API Impexium ### testing() ### # Updated 2022-03-22 @router.get('/event/{e_impexium_event_id}/testing', response_model=Resp_Body_Base) async def testing( e_impexium_event_id: str = Query(..., min_length=11, max_length=22), # For ISHLT: 42_AM (2022-04); EX22_AM (2022-04); 41V_2 (2021-04) registered_since: datetime.datetime = None, # datetime.datetime.now() + datetime.timedelta(seconds=120) details: bool = True, page: int = 0, # 250 per page from Impexium event_id: str = Query(..., min_length=11, max_length=22), # For ISHLT: ZDzTBlevhZs (2022-04) # Account ID For ISHLT: d8TqXqf1EOg commons: Common_Route_Params = Depends(common_route_params), ): log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL log.debug(locals()) log.info('Starting Impexium event registration import...') account_id = commons.x_account_id event_id = 'ZDzTBlevhZs' # ISHLT 2022-04 Boston if event_id := redis_lookup_id_random(record_id_random=event_id, table_name='event'): pass else: return mk_resp(data=None, status_code=404, status_message=f'The Event ID was not found. Event ID: {event_id}', response=commons.response) event_registrant_li = get_event_registrants( event_code = e_impexium_event_id, registered_since = registered_since, details = details, page = page ) log.debug(type(event_registrant_li)) for event_registrant in event_registrant_li: log.debug(event_registrant) log.info(f"ID: {event_registrant.get('id')}, Registration Num: {event_registrant.get('registrationNumber')}, Registrant Name: {event_registrant.get('badgeName')}") individual_custom_fields_li = get_individual_custom_fields(individual_id=event_registrant.get('id')) # log.debug(individual_custom_fields_li) degrees = None organization_name = None for individual_custom_field in individual_custom_fields_li: if individual_custom_field.get('name') == 'degree': degrees = individual_custom_field.get('value') if individual_custom_field.get('organization_name') == 'organization_name': organization_name = individual_custom_field.get('value') log.debug(f'Degree: {degrees}; Organization Name: {organization_name}') if event_registrant_li: return mk_resp(data=event_registrant_li, status_code=200, response=commons.response) # return mk_resp(data=len(event_registrant_li), status_code=200, response=commons.response) else: return mk_resp(data=event_registrant_li, status_code=400, response=commons.response, status_message='Possible bad request???') # Bad Request # ### END ### API Impexium ### testing() ###