Files
OSIT-AE-API-FastAPI/app/routers/e_impexium.py
2023-02-03 13:53:19 -05:00

972 lines
51 KiB
Python

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, get_individual_profile
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_individual_profile: bool = True,
inc_individual_profile_custom_fields: bool = True, # NOTE: Do not set True if inc_individual_profile is also set 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
account_id_random = commons.x_account_id_random
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,
)
if event_registrant_li:
log.info(f'Total record count: {len(event_registrant_li)}')
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
# BEGIN: Loop through the registrants pulled from Impexium
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'BEGIN: **** *** ** * Impexium registrant import loop #{loop_count} * ** *** ****')
# ISHLT 2022 used event_registrant.get('registrantTypeCode') for the registration type
# ISHLT 2023 used a custom field for their main registration.
custom_fields = [] # NOTE: These are custom fields for registration
if custom_fields := event_registrant.get('itemizedCustomFields'):
if isinstance(custom_fields, list): pass
else:
log.warning('Found custom fields section, but no list was found.')
custom_fields = []
else:
log.warning('The custom fields were not found at all. Are the "details" being included?')
pass
event_badge_data = {}
if account_id_random == 'd8TqXqf1EOg':
if e_impexium_event_id == '2023_Annual Meeting':
log.info('Processing ISHLT 2023 Annual Meeting Registration...')
event_badge_data['event_badge_template_id'] = 10
log.info('Found list of custom fields for a registrant. Searching for "registration_selection"')
for field in custom_fields:
if field.get('name') == 'registration_selection':
event_badge_data['badge_type_code'] = field.get('value') # This is probably not exactly right
event_badge_data['badge_type'] = field.get('value') # This is probably not exactly right
event_badge_data['registration_type_code'] = field.get('value')
event_badge_data['registration_type'] = field.get('value')
break
else:
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'
elif e_impexium_event_id == 'ISHLT 43rd Annual Meeting - Exhib Booth Staff Only':
log.info('Processing ISHLT 2023 Annual Meeting Exhibit Only...')
event_badge_data['event_badge_template_id'] = 10
event_badge_data['badge_type_code'] = 'EXO'
event_badge_data['badge_type'] = 'Exhibitor Booth Staff'
event_badge_data['registration_type_code'] = 'EXO'
event_badge_data['registration_type'] = 'Exhibitor Booth Staff'
elif e_impexium_event_id in ['2023 Lung Transplantation Morning Master Class', '2023 Lung Transplantation Afternoon Master Class', '2023 Nursing and Allied Health Master Class', '2023 Pediatric MCS Master Class', '2023 Pulmonary Hypertension Master Class']:
log.info('Processing ISHLT 2023 Annual Meeting Academies...')
event_badge_data['event_badge_template_id'] = 11
if e_impexium_event_id == '2023 Lung Transplantation Morning Master Class':
event_badge_data['badge_type_code'] = 'lung_morning'
event_badge_data['badge_type'] = 'Lung Transplantation Morning'
elif e_impexium_event_id == '2023 Lung Transplantation Afternoon Master Class':
event_badge_data['badge_type_code'] = 'lung_afternoon'
event_badge_data['badge_type'] = 'Lung Transplantation Afternoon'
elif e_impexium_event_id == '2023 Nursing and Allied Health Master Class':
event_badge_data['badge_type_code'] = 'nurse_allied_health'
event_badge_data['badge_type'] = 'Nursing and Allied Health'
elif e_impexium_event_id == '2023 Pediatric MCS Master Class':
event_badge_data['badge_type_code'] = 'pediatric'
event_badge_data['badge_type'] = 'Pediatric'
elif e_impexium_event_id == '2023 Pulmonary Hypertension Master Class':
event_badge_data['badge_type_code'] = 'pulmonary'
event_badge_data['badge_type'] = 'Pulmonary Hypertension'
else:
log.warning('Setting generic badge type. Event ID not fully setup?')
event_badge_data['badge_type_code'] = event_registrant.get('registrantTypeCode') # MBR19, NonM
event_badge_data['badge_type'] = event_registrant.get('registrantTypeName') # Member, Non-Member
event_badge_data['registration_type_code'] = event_registrant.get('registrantTypeCode')
event_badge_data['registration_type'] = event_registrant.get('registrantTypeName')
else:
log.error(f'This is an unknown meeting ID ({e_impexium_event_id}) for account ID ({account_id_random})!')
return mk_resp(data=len(event_registrant_li), status_code=501, response=commons.response) # Not Implemented
else:
log.error(f'This account ID ({account_id_random}) is not currently configured!')
return mk_resp(data=len(event_registrant_li), status_code=501, response=commons.response) # Not Implemented
# event_registrant.get('id') is the Impexium UUID for an individual person and their registrations
# event_registrant.get('recordNumber') is the Impexium 5 digit number for an individual person and their registrations
log.info(f"ID: {event_registrant.get('id')}, Record Num: {event_registrant.get('recordNumber')}, Registration Num: {event_registrant.get('registrationNumber')}, Registrant Type Code: {event_badge_data.get('registration_type_code')}")
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_v1 = 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_person_id = event_registrant.get('id') # Impexium UUID for person
external_registration_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_person_id'] = external_person_id
event_person_data['external_registration_id'] = external_registration_id
# event_person_data['external_id_old'] = external_id_old
email = None
address_line_1 = None
address_line_2 = None
address_line_3 = None
city = None
country_subdivision_code = None
state_province = None
state_province_abb = None
postal_code = 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'):
address_line_1 = address.get('line1')
address_line_2 = address.get('line2')
address_line_3 = address.get('line3')
city = address.get('city')
country_subdivision_code = address.get('stateISOCode')
state_province = address.get('state')
state_province_abb = address.get('stateAbbreviation')
postal_code = address.get('zipcode')
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
membership_code = None
membership_type = None
membership_effective_datetime = None
membership_expire_datetime = None
membership_status = None
if inc_individual_profile: # This also includes membership information if details is True
# NOTE: Reminder that the ID under a registration is also the person's profile ID.
log.info(f'Getting individual profile for ID {event_registrant.get("id")}')
individual_profile_result = get_individual_profile(
individual_id = event_registrant.get('id'),
details = details,
# page = page,
)
log.debug(individual_profile_result)
if individual_profile_result:
individual_profile_custom_field_li = []
if individual_profile_custom_field_li := individual_profile_result.get('customFields'):
if isinstance(individual_profile_custom_field_li, list):
for individual_profile_custom_field in individual_profile_custom_field_li:
if individual_profile_custom_field.get('name') == 'degree':
degrees = individual_profile_custom_field.get('value')
# NOTE: Use only one of the organization names with a preference for the exhibiting name. (new for ISHLT 2023)
if individual_profile_custom_field.get('name') == 'exhibiting_organization_name':
organization_name = individual_profile_custom_field.get('value')
elif individual_profile_custom_field.get('name') == 'organization_name':
organization_name = individual_profile_custom_field.get('value')
log.debug(f'Degrees: {degrees}; Organization Name: {organization_name}')
else:
log.warning('Found individual profile custom fields section, but no list was found.')
individual_profile_custom_field_li = []
else:
log.warning('The individual profile custom fields were not found at all. Are the "details" being included?')
pass
individual_profile_membership_li = []
membership_code = None
membership_type = None
membership_effective_datetime = None
membership_expire_datetime = None
if individual_profile_membership_li := individual_profile_result.get('memberships'):
if isinstance(individual_profile_membership_li, list):
for individual_profile_membership in individual_profile_membership_li:
if membership_code := individual_profile_membership.get('code'): pass
if membership_type := individual_profile_membership.get('membershipType'): pass
if membership_effective_datetime := individual_profile_membership.get('effectiveDate'): pass
if membership_expire_datetime := individual_profile_membership.get('expireDate'): pass
log.debug(f'Membership - Code: {membership_code}; Type: {membership_type}; Effective Datetime: {membership_effective_datetime}; Expire Datetime: {membership_expire_datetime}')
else:
log.warning('Found individual profile membership section, but no list was found.')
individual_profile_membership_li = []
else:
log.warning('The individual profile membership section was not found at all. Are the "details" being included?')
pass
else:
log.warning('Tried to get individual profile, but no results!')
pass
# if inc_individual_profile_custom_fields:
# log.info(f'Getting individual custom fields list for Event Registrant ID {event_registrant.get("id")}')
# individual_custom_fields_li = get_individual_custom_fields(individual_id=event_registrant.get('id'))
# log.debug(individual_custom_fields_li)
# 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 = {} # Initialized above ^
event_badge_data['enable'] = True
event_badge_data['event_id'] = event_id
event_badge_data['external_id'] = external_id_v3
event_badge_data['external_person_id'] = external_person_id
event_badge_data['external_registration_id'] = external_registration_id
# event_badge_data['external_id_old'] = external_id_old
event_badge_data['allow_tracking'] = False
event_person_data['allow_tracking'] = False
for field in custom_fields:
if field.get('name') == 'EventSponsorOptin' and field.get('value') == 'True':
log.info('Allow Tracking')
event_badge_data['allow_tracking'] = True
event_person_data['allow_tracking'] = True
elif field.get('name') == 'EventSponsorOptin' and field.get('value') == 'False':
log.info('Tracking Not Allowed')
event_badge_data['allow_tracking'] = False
event_person_data['allow_tracking'] = False
# 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. Will update below if possible
event_badge_data['member_type'] = None # Using this as the member_type. Will update below if possible
if membership_code: event_badge_data['member_type_code'] = membership_code
else: event_badge_data['member_type_code'] = event_registrant.get('customerType')
if membership_type: event_badge_data['member_type'] = membership_type
if membership_expire_datetime:
current_datetime_utc = datetime.datetime.utcnow()
# membership_expire_datetime_obj = datetime.datetime.strptime(membership_expire_datetime, '%Y-%m-%d')
membership_expire_datetime_obj = datetime.datetime.strptime(membership_expire_datetime, '%Y-%m-%dT%H:%M:%S')
# log.debug(current_datetime_utc)
# log.debug(membership_expire_datetime_obj)
# log.debug(membership_expire_datetime)
if membership_expire_datetime_obj > current_datetime_utc:
log.info('Current membership; not expired')
# member_status = 'current' # Active member; not expired
event_badge_data['member_status'] = 'current' # Active member; not expired
else:
log.info('Inactive membership; expired')
# member_status = 'inactive' # Inactive member; expired
event_badge_data['member_status'] = 'inactive' # Inactive member; expired
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
if address_line_1: event_badge_data['address_line_1'] = address_line_1
if address_line_2: event_badge_data['address_line_2'] = address_line_2
if address_line_3: event_badge_data['address_line_3'] = address_line_3
event_badge_data['city'] = event_registrant.get('badgeCity')
event_badge_data['state_province'] = event_registrant.get('badgeState')
if not event_badge_data['city'] and city:
event_badge_data['city'] = city
if country_subdivision_code:
event_badge_data['country_subdivision_code'] = country_subdivision_code
if not event_badge_data['state_province'] and state_province:
event_badge_data['state_province'] = state_province
if state_province_abb:
event_badge_data['state_province_abb'] = state_province_abb
if postal_code:
event_badge_data['postal_code'] = postal_code
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}'
elif event_badge_data['city']:
city = event_badge_data['city']
location = f'{city}'
elif country:
location = f'{country}'
if location: event_badge_data['location'] = location
event_person_data['event_badge'] = {}
event_person_data['event_badge'] = event_badge_data
# log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(event_person_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 ### event_check_individual() ###
# Updated 2022-04-33
# @router.get('/event/{e_impexium_event_id}/{e_impexium_individual_id}/import_individual', response_model=Resp_Body_Base)
# @router.get('/event/{event_badge_id}/check_individual', response_model=Resp_Body_Base)
@router.get('/event/{e_impexium_individual_id}/check_individual', response_model=Resp_Body_Base)
async def event_check_individual(
e_impexium_individual_id: str = Query(..., min_length=50, max_length=55),
# event_badge_id: str = Query(..., min_length=11, max_length=22),
details: bool = False,
inc_membership_fields: bool = True,
page: int = 1, # 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
refresh_record: bool = False,
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
individual_profile_result = get_individual_profile(
individual_id = e_impexium_individual_id,
details = details,
page = page,
)
log.debug(individual_profile_result)
if individual_profile_result: pass
else: return mk_resp(data=None, status_code=404, status_message=f'Checked for individual profile in Impexium. No Impexium individual found.', response=commons.response)
individual_profile = individual_profile_result
event_person_summary_data = {}
event_person_summary_data['external_person_id'] = e_impexium_individual_id
event_person_data = {}
event_person_data['external_person_id'] = e_impexium_individual_id # The Impexium individual ID
email = None
city = None
country_subdivision_code = None
state_province = None
state_province_abb = None
country_alpha_2_code = None
country = None
if details:
if emails := individual_profile.get('emails'):
if isinstance(emails, list) and len(emails):
email = emails[0].get('address')
event_person_summary_data['email'] = email
if addresses := individual_profile.get('addresses'):
if isinstance(addresses, list) and len(addresses):
for address in addresses:
if primary_address := address.get('primary'):
city = address.get('city')
country_subdivision_code = address.get('stateISOCode')
state_province = address.get('state')
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=individual_profile.get('id'))
individual_custom_fields_li = individual_profile.get('customFields')
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'] = individual_profile.get('gender')
event_person_profile_data['informal_name'] = individual_profile.get('preferredFirstName')
event_person_profile_data['title_names'] = individual_profile.get('prefix')
event_person_profile_data['given_name'] = individual_profile.get('firstName')
event_person_profile_data['middle_name'] = individual_profile.get('middleName')
event_person_profile_data['family_name'] = individual_profile.get('lastName')
event_person_profile_data['designations'] = individual_profile.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 # individual_profile.get('title') # Should this be None if no degrees found above???
# event_person_profile_data['full_name'] = individual_profile.get('badgeName')
# event_person_summary_data['full_name'] = individual_profile.get('badgeName')
event_person_summary_data['full_name'] = individual_profile.get('firstName') + ' ' + individual_profile.get('lastName')
if organization_name:
event_person_profile_data['affiliations'] = organization_name # Ideally should be organization_name for ISHLT
else:
# event_person_profile_data['affiliations'] = individual_profile.get('badgeOrganization') # Should this be None if no organization_name found above???
pass
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_person_id'] = external_person_id
# event_badge_data['external_registration_id'] = external_registration_id
event_badge_data['pronouns'] = individual_profile.get('gender')
event_badge_data['informal_name'] = individual_profile.get('preferredFirstName')
event_badge_data['title_names'] = individual_profile.get('prefix')
event_badge_data['given_name'] = individual_profile.get('firstName')
event_badge_data['middle_name'] = individual_profile.get('middleName')
event_badge_data['family_name'] = individual_profile.get('lastName')
event_badge_data['designations'] = individual_profile.get('suffix')
if degrees:
event_badge_data['professional_title'] = degrees # Ideally should be degrees for ISHLT
else:
event_badge_data['professional_title'] = None # individual_profile.get('title') # Should this be None if no degrees found above???
# event_badge_data['full_name'] = individual_profile.get('badgeName')
if organization_name:
event_badge_data['affiliations'] = organization_name # Ideally should be organization_name for ISHLT
else:
# event_badge_data['affiliations'] = individual_profile.get('badgeOrganization') # Should this be None if no organization_name found above???
pass
if email: event_badge_data['email'] = email
# event_badge_data['city'] = individual_profile.get('badgeCity')
# event_badge_data['state_province'] = individual_profile.get('badgeState')
# if not event_badge_data['city'] and city:
# event_badge_data['city'] = city
if country_subdivision_code:
event_badge_data['country_subdivision_code'] = country_subdivision_code
# if not event_badge_data['state_province'] and state_province:
# event_badge_data['state_province'] = state_province
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}'
# elif event_badge_data['city']:
# city = event_badge_data['city']
# location = f'{city}'
# elif country:
# location = f'{country}'
# if location: event_badge_data['location'] = location
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, external_person_id AS event_person_external_person_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_person_id = :external_person_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)}')
# 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)
# 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)
# return False
if return_detail:
return mk_resp(data=individual_profile, status_message=f'Checked for individual profile in Impexium. Found. Returning details.', response=commons.response)
else:
return mk_resp(data=event_person_summary_data, status_message=f'Checked for individual profile in Impexium. Found. Returning summary.', response=commons.response)
return mk_resp(data=individual_profile, status_code=200, response=commons.response)
# ### END ### API Impexium ### event_check_individual() ###
# ### 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() ###