This is an old revision of the document!
ECE Interactive API
Retrieve Activities for Date X
note - this is a draft version - and currently doesn't loop through the pages / pagination. So this needs to be done!
# Author Gerry O'Rourke # Date: 29/03/2023 # Version: 0.10 import datetime import requests # used for http requests import sys # used for sys.exit import urllib3 # urllib3.disable_warnings() #Surpress cert warnings import json import os #check / creating folders import argparse eceUsername = 'eceapi_agent_tst' ecePassword = '12345' eceHost = 'ecewebserver.mydomain.com' rootFolder = 'C:/ECE-Export-Activities/LAB/' def getArguments(): global myDate parser = argparse.ArgumentParser() parser.add_argument('-d', '--date', type=lambda d: datetime.datetime.strptime(d, '%Y-%m-%d').date(), help='Set a start date') parser.parse_args(["-d", "2023-01-01"]) args = parser.parse_args() if args.date: myDate = (args.date) print('From Date set using arguments. From Date set to: ' + myDate.strftime('%Y-%m-%d')) else: myDate = datetime.date.today() - datetime.timedelta(days=1) print('No date argument set. From Date set to yesterday: ' + myDate.strftime('%Y-%m-%d')) def loginEce(): #Attempt to Login loginUrl = 'https://' + eceHost + '/system/ws/v12/authentication/user/login?forceLogin=yes' loginUser='{"userName":"' + eceUsername + '","password":"' + ecePassword + '"}' loginHeaders={"Content-Type":"application/json"} print('Attemping to login...') response = requests.post(url=loginUrl,verify=False, headers=loginHeaders, data=loginUser) #Store Response Values responseHeaders = (response.headers) responseStatusCode = (response.status_code) responseBody = (response.text) if (responseStatusCode == 204): #Get the X-egain-session Header value returned after successful (204) login. responseToken = (responseHeaders['X-egain-session']) #ECE defaults to returning XML unless you set the Accept header to application/json #Here we are also setting the X-egain-session header (which we got when we logged in), which is needed for all requests. #Note the user we logged in - must have the correct roles for the specific API calls #we create a global variable here - so we can use it in other functions. global eceHeaders eceHeaders={"Accept":"application/json","X-egain-session":responseToken} print ('Login Successful!, X-egain-session value is: ',responseToken) else: print ('Login Failed!, Status Code: ',responseStatusCode) sys.exit(1) def logoutEce(): logoutUrl = 'https://' + eceHost + '/system/ws/v12/authentication/user/logout' response = requests.delete(url=logoutUrl,verify=False, headers=eceHeaders) #Store Response Values responseStatusCode = (response.status_code) responseText = response.text if (responseStatusCode == 204): print ('Logged out successfully.') else: print('Failed to logout successfully. This is what was returned: ' + response.text) def listActivities(fromDate): toDate = fromDate + datetime.timedelta(days=1) fromDateString = (fromDate.strftime('%Y-%m-%d')) toDateString = (toDate.strftime('%Y-%m-%d')) listActivitiesUrl = 'https://' + eceHost + '/system/ws/v12/interaction/activity?lastModifiedDate=[' + fromDateString + 'T00:00:00.000Z,' + toDateString + 'T00:00:00.000Z]' #Attempt to get Activity Id response = requests.get(url=listActivitiesUrl,verify=False, headers=eceHeaders) #Store Response Values responseStatusCode = (response.status_code) responseText = response.text if (responseStatusCode == 200): try: print('successful Retrieved Json File') activityDict = json.loads(responseText) #print (activityDict.keys()) #activityId_0 = str(activityDict['activity'][0]['id']) #Get activity id[0] #print(activityId_0) myFolder = rootFolder + fromDate.strftime('%Y%m%d') # Check whether the specified path exists or not isExist = os.path.exists(myFolder) if not isExist: # Create a new directory because it does not exist os.makedirs(myFolder) print('The new directory: "' + myFolder + '" is created.') fileName = 'activities_' + fromDate.strftime('%Y%m%d') + '.json' filePath = myFolder + '/' + fileName #Save the activity as a text file text_file = open(filePath, "w", encoding="utf-8") text_file.write(responseText) text_file.close() print('File: ' + filePath + ' written.') #Loop through the file and get each Activity id for id in activityDict['activity']: #print(id ['id']) activityId = str(id ['id']) getActivity(activityId,fromDate) print('All activies completed.') except Exception as e: print('Error in listActivities Function. error:' + e) logoutEce() sys.exit(2) else: print('Failed to list Activities for the specified Date.') def getActivity(id,date): print('Attempting to retrieve Activity Id: ' + id + '...') activityUrl = 'https://' + eceHost + '/system/ws/v12/interaction/activity/' + id + '?$attribute=all' #Attempt to get Activity Id response = requests.get(url=activityUrl,verify=False, headers=eceHeaders) #Store Response Values responseStatusCode = (response.status_code) responseText = response.text #print (activityDict.keys()) #print (activityDict['activity'][0]['case']['id']) if (responseStatusCode == 200): try: activityDict = json.loads(responseText) caseId = str(activityDict['activity'][0]['case']['id']) #Get the Case Id fileName = 'activityId-' + id + '-caseId-' + caseId + '_' + date.strftime('%Y%m%d') + '.json' myFolder = rootFolder + date.strftime('%Y%m%d') filePath = myFolder + '/' + fileName #Check whether the specified path exists or not isExist = os.path.exists(myFolder) if not isExist: # Create a new directory because it does not exist os.makedirs(myFolder) print('The new directory: "' + myFolder + '" is created.') #Save the activity as a text file text_file = open(filePath, "w", encoding="utf-8") text_file.write(responseText) text_file.close() print ('File: ' + filePath + ' written.') except Exception as e: print('Error in getActivity function. error: ' + e) logoutEce() sys.exit(2) else: print('Failed to retrieve Activity') #get the Command line parameters getArguments() #Login loginEce() #list and retrieve activities for the specific date listActivities(myDate) #Logout logoutEce()
Masking an activity or deleting attachments from an activity
Permissions for Masking an activity
All of the following are required:
- User must have 'Edit' action on 'Activity' resource or 'Manage Utilities' action on 'Utilities' resource.
- The activity must either belong to the user's home department, or to a department in which the user is a foreign user.
In PCCE - the above permissions are added to the specific user the following: ECE → User → Relationships:
- Manage Completed Activity
Example Code to Mask or delete an attachment in an ECE Activity
# Author Gerry O'Rourke # Date: 06/09/2023 # Version: 2.0 import requests # used for http requests import sys # used for sys.exit import urllib3 # urllib3.disable_warnings() # Surpress cert warnings import json import os # check / creating folders import time eceUsername = 'eceapi_agent_tst' ecePassword = 'BaNaNa5!' eceHost = 'ece-webserver.mydomain.com' def loginEce(): # Attempt to Login loginUrl = 'https://' + eceHost + '/system/ws/v12/authentication/user/login?forceLogin=yes' loginUser = '{"userName":"' + eceUsername + '","password":"' + ecePassword + '"}' loginHeaders = {"Content-Type": "application/json"} print('Attemping to login...') response = requests.post(url=loginUrl, verify=False, headers=loginHeaders, data=loginUser) # Store Response Values responseHeaders = (response.headers) responseStatusCode = (response.status_code) responseBody = (response.text) if (responseStatusCode == 204): # Get the X-egain-session Header value returned after successful (204) login. global responseToken responseToken = (responseHeaders['X-egain-session']) # ECE defaults to returning XML unless you set the Accept header to application/json # Here we are also setting the X-egain-session header (which we got when we logged in), which is needed for all requests. # Note the user we logged in - must have the correct roles for the specific API calls # we create a global variable here - so we can use it in other functions. global eceHeaders eceHeaders = {"Accept": "application/json", "X-egain-session": responseToken} print('Login Successful!, X-egain-session value is: ', responseToken) else: print('Login Failed!, Status Code: ', responseStatusCode) sys.exit(1) def logoutEce(): logoutUrl = 'https://' + eceHost + '/system/ws/v12/authentication/user/logout' response = requests.delete(url=logoutUrl, verify=False, headers=eceHeaders) # Store Response Values responseStatusCode = (response.status_code) responseText = response.text if (responseStatusCode == 204): print('Logged out successfully.') else: print('Failed to logout successfully. This is what was returned: ' + response.text) def getActivity(id): global caseId global lastModifiedDate global activityAttachmentsCount print('Attempting to retrieve Activity Id: ' + id + '.') activityUrl = 'https://' + eceHost + '/system/ws/v12/interaction/activity/' + id # Attempt to get Activity Id response = requests.get(url=activityUrl, verify=False, headers=eceHeaders) # Store Response Values responseStatusCode = (response.status_code) responseText = response.text if (responseStatusCode == 200): try: activityDict = json.loads(responseText) caseId = str(activityDict['activity'][0]['case']['id']) # Get the Case Id # print('CaseID: ' + caseId) lastModifiedDate = str(activityDict['activity'][0]['lastModified']['date']) # Get the lastModified date # print('Last Modified Date: ' + lastModifiedDate) activityAttachmentsCount = str(activityDict['activity'][0]['attachments']['count']) # Get the Case Id # print('Activity Attachments Count: ' + activityAttachmentsCount) #maskActivity(id, lastModifiedDate) except Exception as e: print('Error in getActivity function. error: ' + e) logoutEce() sys.exit(2) else: print('Failed to retrieve Activity') def maskActivity(id, date): maskActivityUrl = 'https://' + eceHost + '/system/ws/v12/interaction/activity' maskActivityHeaders = {"Content-Type": "application/xml", "X-egain-session": responseToken} maskActivityBody = '<?xml version="1.0" encoding="UTF-8"?><activities xmlns="http://bindings.egain.com/ws/model/v12/gen/interaction/activity" xmlns:ns2="http://bindings.egain.com/ws/model/v12/gen/common"><activity id="' + id + '"><lastModified date="' + date + '" /><payload><email><contents><content><ns2:type>html</ns2:type><ns2:value><b>Contents are removed.</b></ns2:value></content></contents></email></payload></activity></activities>' print('Attempting to Mask Activity Id: ' + id + '.') response = requests.put(url=maskActivityUrl, verify=False, headers=maskActivityHeaders, data=maskActivityBody) # Store Response Values responseHeaders = (response.headers) responseStatusCode = (response.status_code) responseBody = (response.text) if (responseStatusCode == 204): print('Successfully Masked Activity ID: ' + id) confirm = input('\nDo you want to Review another Activity ID?. Enter "y" to confirm? ') if (confirm == 'y' or confirm == 'Y'): enterActivityId() else: print('Exiting now...') logoutEce() else: print('----- ERROR, Unable to Mask Activity -----') print('Status Code: ' + str(responseStatusCode)) print('Response Body: ' + responseBody) logoutEce() def deleteAttachments(id): deleteAttachmentsUrl = 'https://' + eceHost + '/system/ws/v12/interaction/activity/' +id + '/attachment' deleteAttachmentsHeaders = {"X-egain-session": responseToken} print('Attempting to delete ALL attachments from Activity Id: ' + id + '.') response = requests.delete(url=deleteAttachmentsUrl, verify=False, headers=deleteAttachmentsHeaders) # Store Response Values responseHeaders = (response.headers) responseStatusCode = (response.status_code) responseBody = (response.text) if (responseStatusCode == 204): print('Successfully Deleted Attachments from Activity ID: ' + id) confirm = input('\nDo you want to review another Activity ID?. Enter "y" to confirm? ') if (confirm == 'y' or confirm == 'Y'): enterActivityId() else: print('Exiting now...') logoutEce() else: print('----- ERROR, Unable to Delete Attachments -----') print('Status Code: ' + str(responseStatusCode)) print('Response Body: ' + responseBody) logoutEce() def enterActivityId(): global activityId activityId = input('\nEnter your Activity Id or enter "exit": ') if (activityId == 'exit'): print('exiting...') logoutEce() elif (activityId.isnumeric()): getActivity(str(activityId)) print('Retrieved the Activity ID successfully.') print('\nActivity ID: ' + activityId) print('Case ID:' + caseId) print('Attachment Count:' + activityAttachmentsCount) if (int(activityAttachmentsCount)>0): print('This activity has one or more attachments!') print('----------------------') print('1. Mask Activity') print('2. Delete Attachments') print('9. Exit') print('----------------------') confirm = input('\nEnter Menu option [1], [2] or [9]?') if (confirm == '1'): maskActivity(activityId, lastModifiedDate) elif (confirm == '2'): deleteAttachments(activityId) else: logoutEce() else: print('This activity has NO attachments!') print('----------------------') print('1. Mask Activity') print('9. Exit') print('----------------------') confirm = input('\nEnter Menu option [1] or [9]?') if (confirm == '1'): maskActivity(activityId, lastModifiedDate) else: logoutEce() else: print('Entered Activity ID is not a number or the word "exit".') print('Please try again.') time.sleep(2) enterActivityId() os.system('cls') print('****** WARNING - Masking an activity is not reversable! ******') print('Proceed with caution !!!') print('\n') # Login loginEce() # Enter the Activity Id to Mask enterActivityId()
Sample output of the above script
Backup (draft version) of script to purge attachments
# Author Gerry O'Rourke # Date: 07/09/2023 # Version: 0.10 import datetime import requests # used for http requests import sys # used for sys.exit import urllib3 # urllib3.disable_warnings() #Surpress cert warnings import json import os #check / creating folders import argparse eceUsername = 'eirsupport1' ecePassword = 'Pa$$w0rd eceHost = 'ece-webserver.mydomain.com' rootFolder = 'C:/eirScripts/ECE-Purge-Activities-Attachments-by-Queue/LAB/' activityIdsSearched = [] activityIdsWithAttachments = [] def getArguments(): global endDate global startDate global silentMode global queueIds parser = argparse.ArgumentParser() parser.add_argument('-d', '--days', type=int, help='sets the end date, e.g. -d 180, sets the end dat as 180 days ago') parser.add_argument('-r', '--dayrange', type=int, default=1, help='by default we only purge a single day, the if we ' 'want to purge more, enter day range') parser.add_argument('-s', '--silent', action='store_true') parser.add_argument('-q', '--queue', action='append', help='Queue IDs, for multiple queue ID enter -q before each one') args = parser.parse_args() if args.days: #days = (args.days) endDate = datetime.date.today() - datetime.timedelta(days=args.days) print('End Date set to: ' + endDate.strftime('%Y-%m-%d')) elif args.days == 0: endDate = datetime.date.today() print('End Date set to: ' + endDate.strftime('%Y-%m-%d')) else: print('No date argument set. Enter the number of days from today, e.g. "-d 180", would purge data from 180 days ' 'ago.') exit(1) if args.dayrange: startDate = endDate - datetime.timedelta(days=args.dayrange) print('Start Date set to: ' + startDate.strftime('%Y-%m-%d')) if args.silent: print('Silent Mode set - the application will not prompt for input to continue') silentMode = true else: print('Silent Mode NOT set - do NOT use this mode if running as a scheduled task.') if args.queue: queueIds = args.queue print('Queue ID list: ' + str(queueIds)) else: print('No queue data set. This app could be used to purge all activities for a date range but that is too risky! ' 'Hence you must enter at least 1 queue you want to purge using the argument "-q".') exit(1) def loginEce(): #Attempt to Login loginUrl = 'https://' + eceHost + '/system/ws/v12/authentication/user/login?forceLogin=yes' loginUser='{"userName":"' + eceUsername + '","password":"' + ecePassword + '"}' loginHeaders={"Content-Type":"application/json"} print('Attemping to login...') response = requests.post(url=loginUrl,verify=False, headers=loginHeaders, data=loginUser) #Store Response Values responseHeaders = (response.headers) responseStatusCode = (response.status_code) responseBody = (response.text) if (responseStatusCode == 204): #Get the X-egain-session Header value returned after successful (204) login. responseToken = (responseHeaders['X-egain-session']) global eceHeaders eceHeaders={"Accept":"application/json","X-egain-session":responseToken} print ('Login Successful!, X-egain-session value is: ',responseToken) else: print ('Login Failed!, Status Code: ' + str(responseStatusCode)) sys.exit(1) def logoutEce(): logoutUrl = 'https://' + eceHost + '/system/ws/v12/authentication/user/logout' response = requests.delete(url=logoutUrl,verify=False, headers=eceHeaders) #Store Response Values responseStatusCode = (response.status_code) responseText = response.text if (responseStatusCode == 204): print ('Logged out successfully.') else: print('Logout Failed!, Status Code: ' + str(responseStatusCode)) print('Error Response: ' + response.text) def listActivities(startDate,endDate,pageNum): global activityIdsSearched global activityIdsWithAttachments global activityCount global currentPageNum global pageSize startDateString = (startDate.strftime('%Y-%m-%d')) endDateString = (endDate.strftime('%Y-%m-%d')) currentPageNum = pageNum listActivitiesUrl = 'https://' + eceHost + '/system/ws/v12/interaction/activity?$pagenum=' + str(currentPageNum) + '&lastModifiedDate=[' + startDateString + 'T00:00:00.000Z,' + endDateString + 'T00:00:00.000Z]' print('listActivitiesUrl: ' + listActivitiesUrl) #Attempt to Retrieve List of Activities response = requests.get(url=listActivitiesUrl,verify=False, headers=eceHeaders) responseStatusCode = (response.status_code) responseText = response.text if (responseStatusCode == 200): try: activityDict = json.loads(responseText) activityCount = activityDict['paginationInfo']['count'] #currentPageNum = activityDict['paginationInfo']['pagenum'] pageSize = activityDict['paginationInfo']['pagesize'] #Loop through the file and get each Activity id for id in activityDict['activity']: activityId = str(id ['id']) #get the Activity ID activityIdsSearched.append(activityId) activityAttachmentsCount = activityDict['activity'][0]['attachments']['count'] # Get the Attachments Count if (activityAttachmentsCount > 0): activityIdsWithAttachments.append(activityId) except Exception as e: print('Error in listActivities Function. error:' + e) logoutEce() sys.exit(2) elif (currentPageNum == 1 and responseStatusCode == 204): print('No Activities Content for that date range') elif (responseStatusCode == 204): print('completed search.') else: print('Error - Failed to list Activities, response code: ' + str(responseStatusCode)) logoutEce() sys.exit(2) #get the Command line parameters getArguments() loginEce() listActivities(startDate, endDate,1) print('Activity Count: ' + str(activityCount)) print('PageSize: ' + str(pageSize)) print('Current Page Num: ', currentPageNum) while ((currentPageNum * pageSize) < activityCount): currentPageNum += 1 print('Current Page Num: ', currentPageNum) listActivities(startDate, endDate, currentPageNum) print('Number of Activities Scanned: ', len(activityIdsSearched)) print('Number of Activity IDs with attachments: ', len(activityIdsWithAttachments)) #print('Activity Ids Searched: ' + str(activityIdsSearched)) #print('Activity Ids with attachments: ' + str(activityIdsWithAttachments)) #Login #loginEce() #list and retrieve activities for the specific date #Logout logoutEce()
ECE Purge Attachments by Queue ID and day range
# Author Gerry O'Rourke # Date: 14/09/2023 # Version: 1.0 #import datetime from datetime import datetime from datetime import date from datetime import timedelta import requests # used for http requests import sys # used for sys.exit and standard output import urllib3 # urllib3.disable_warnings() #Surpress cert warnings import json import os #check / creating folders import argparse import logging from logging.handlers import RotatingFileHandler import smtplib from email.mime.text import MIMEText #loglevel = logging.INFO loglevel = logging.DEBUG logging.basicConfig(handlers=[logging.StreamHandler(sys.stdout), RotatingFileHandler('ece-purge-attachments.log', maxBytes=100000, backupCount=10)], style='{', format='{asctime} - {levelname:8s} - {message}', level=loglevel) eceUsername = 'eceapi_user' ecePassword = 'Pa$$w0rd' eceHost = 'ece-webserver.mydomain.com' environment = 'LAB' # Do not have any spaces - set to 'LAB' or 'PROD' rootFolder = 'C:/eirScripts/ECE-Purge-Activities-Attachments-by-Queue/LAB/' activityIdsSearched = [] activityIdsWithAttachments = [] deletedActivityIds = [] errorCounter = 0 errorMessage = '' startTime = datetime.now() message = ('---------- Starting! ----------') logging.info(message) message = ('startTime: ' + startTime.strftime("%d/%m/%Y %H:%M:%S")) logging.info(message) message = ('Arguments supplied: ' + str(sys.argv)) logging.info(message) def getArguments(): global endDate global startDate global silentMode global queueIds global writeAccess global email global errorCounter global errorMessage parser = argparse.ArgumentParser() parser.add_argument('-d', '--days', type=int, help='sets the end date, e.g. -d 180, sets the end dat as 180 days ago') parser.add_argument('-r', '--dayrange', type=int, default=1, help='by default we only purge a single day, the if we ' 'want to purge more, enter day range') parser.add_argument('-s', '--silent', action='store_true') parser.add_argument('-q', '--queue', action='append', help='Queue IDs, for multiple queue ID enter -q before each one') parser.add_argument('-w', '--write', action='store_true', help='Write Access. Write access must be set if you want to delete attachments') parser.add_argument('-e', '--email', action='store_true', help='Send an email at the end of the script to a summary') args = parser.parse_args() if args.days: endDate = date.today() - timedelta(days=args.days) message = ('End Date set to: ' + endDate.strftime('%Y-%m-%d') + 'T23:59:59') logging.info(message) elif args.days == 0: endDate = date.today() message = ('End Date set to: ' + endDate.strftime('%Y-%m-%d')+ 'T23:59:59') logging.info(message) else: message = ('No date argument set. Enter the number of days from today, e.g. "-d 180", would purge data from 180 ' 'days ago.') logging.error(message) errorCounter = errorCounter + 1 errorMessage = 'Arguments not set correctly - please check log file for more information.' sendEmail() exit(1) if args.dayrange: startDate = endDate - timedelta(days=(args.dayrange-1)) message = ('Start Date set to: ' + startDate.strftime('%Y-%m-%d') + 'T00:00:00') logging.info(message) if args.silent: message = ('Silent Mode set - the application will not prompt for input to continue') logging.info(message) silentMode = True else: message = ('Silent Mode NOT set - do NOT use this mode if running as a scheduled task.') logging.info(message) silentMode = False if args.write: message = ('Write mode enabled - the application will delete found attachements.') logging.info(message) writeAccess = True else: message = ('Read Only mode set - the application will NOT delete attachments.') logging.info(message) writeAccess = False if args.email: message = ('email mode enabled - the application should send an email when completed.') logging.info(message) email = True else: message = ('email mode is NOT enabled - the application will NOT send an email when completed.') logging.info(message) email = False if args.queue: queueIds = args.queue message = ('Queue ID list: ' + str(queueIds)) logging.info(message) else: message = ('No queue data set. This app COULD (if we wanted to!) be used to purge all activities for a date range but that is too risky! ' 'Hence you must enter at least 1 queue you want to purge using the argument "-q".') logging.info(message) errorCounter = errorCounter + 1 errorMessage = 'Arguments not set correctly - please check log file for more information.' exit(1) def loginEce(): global errorCounter global errorMessage loginUrl = 'https://' + eceHost + '/system/ws/v12/authentication/user/login?forceLogin=yes' loginUser='{"userName":"' + eceUsername + '","password":"' + ecePassword + '"}' loginHeaders={"Content-Type":"application/json"} message = ('Attemping to login...') logging.info(message) responseStatusCode = 0 try: response = requests.post(url=loginUrl,verify=False, headers=loginHeaders, data=loginUser, timeout=5) responseHeaders = (response.headers) responseStatusCode = (response.status_code) except: message = ('Issue Calling Login URL.') logging.error(message) if (responseStatusCode == 204): #Get the X-egain-session Header value returned after successful (204) login. global responseToken responseToken = (responseHeaders['X-egain-session']) global eceHeaders eceHeaders={"Accept":"application/json","X-egain-session":responseToken} message = ('Login Successful!, X-egain-session value is: ' + str(responseToken)) logging.info(message) else: message = ('Login Failed!, Status Code: ' + str(responseStatusCode)) logging.error(message) errorCounter = errorCounter + 1 errorMessage = 'Failed to log into ECE successfully. Please check log file for more information.' sendEmail() sys.exit(1) def logoutEce(): logoutUrl = 'https://' + eceHost + '/system/ws/v12/authentication/user/logout' responseStatusCode = 0 try: response = requests.delete(url=logoutUrl,verify=False, headers=eceHeaders) responseStatusCode = (response.status_code) responseText = response.text except: message = ('Issue Calling Logout URL.') logging.error(message) if (responseStatusCode == 204): message = ('Logged out successfully.') logging.info(message) else: message = ('Logout Failed!, Status Code: ' + str(responseStatusCode) + '\nError Response: ' + response.text) logging.error(message) def listActivities(startDate,endDate,pageNum,queueId): global activityIdsSearched global activityIdsWithAttachments global activityCount global currentPageNum global pageSize global retrievedData global errorCounter startDateString = (startDate.strftime('%Y-%m-%d')) endDateString = (endDate.strftime('%Y-%m-%d')) currentPageNum = pageNum retrievedData = False listActivitiesUrl = 'https://' + eceHost + '/system/ws/v12/interaction/activity?type=email&queue=' + str(queueId) + '&$pagenum=' + str(currentPageNum) + '&lastModifiedDate=[' + startDateString + 'T00:00:00.000Z,' + endDateString + 'T23:59:59.999Z]' message = ('listActivitiesUrl: ' + listActivitiesUrl) logging.debug(message) #Attempt to Retrieve List of Activities responseStatusCode = 0 try: response = requests.get(url=listActivitiesUrl, verify=False, headers=eceHeaders, timeout=5) responseStatusCode = (response.status_code) responseText = response.text except: message = ('Issue Calling List Activity URL.') logging.error(message) if (responseStatusCode == 200): try: activityDict = json.loads(responseText) activityCount = activityDict['paginationInfo']['count'] #currentPageNum = activityDict['paginationInfo']['pagenum'] pageSize = activityDict['paginationInfo']['pagesize'] #Loop through the file and get each Activity id for id in activityDict['activity']: activityId = str(id ['id']) #get the Activity ID #print('activityId: ',activityId) activityIdsSearched.append(activityId) activityAttachmentsCount = id['attachments']['count'] # Get the Attachments Count #print('activityAttachmentsCount: ', activityAttachmentsCount) if (activityAttachmentsCount > 0): activityIdsWithAttachments.append(activityId) retrievedData = True except Exception as e: message = ('Error in listActivities Function. error:' + str(e)) logging.error(message) errorCounter = errorCounter + 1 elif (currentPageNum == 1 and responseStatusCode == 204): message = ('No Activities Content for that date range') logging.info(message) elif (responseStatusCode == 204): message = ('Completed search.') logging.info(message) else: message = ('Failed to list Activities, statusCode: ' + str(responseStatusCode)) logging.error(message) errorCounter = errorCounter + 1 def deleteAttachments(id): deleteAttachmentsUrl = 'https://' + eceHost + '/system/ws/v12/interaction/activity/' + id + '/attachment' deleteAttachmentsHeaders = {"X-egain-session": responseToken} message=('Attempting to delete ALL attachments from Activity Id: ' + id + '.') logging.info(message) responseStatusCode = 0 try: response = requests.delete(url=deleteAttachmentsUrl, verify=False, headers=deleteAttachmentsHeaders) responseStatusCode = (response.status_code) responseBody = (response.text) except: message = ('Issue Calling Delete attachment URL.') logging.error(message) errorCounter = errorCounter + 1 if (responseStatusCode == 204): deletedActivityIds.append(activityId) message('Successfully Deleted Attachments from Activity ID: ' + id) logging.info(message) else: message = ('Unable to Delete Attachments, Status Code: ' + str(responseStatusCode) + ', Response Body: ' + responseBody) logging.error(message) errorCounter = errorCounter + 1 def sendEmail(): global activityIdsWithAttachments global activityIdsSearched global deletedActivityIds global errorCounter global errorMessage global environment global startTime global email global writeAccess global queueIds global endDate global startDate if writeAccess==True: mode = 'Write Mode Enabled - will attempt to delete attachments' else: mode = 'Write Mode Disabled - will NOT attempt to delete attachments' if (email==False): message = ('Not sending email as email argument not enmabled') logging.info(message) return #don't continue if email is not enabled. endTime = datetime.now() message = ('Sending email now...') logging.info(message) s = smtplib.SMTP('exchsmtphost2013.cld1.tld.int') sender = environment + '[email protected]' recipients = ['[email protected]', '[email protected]'] msg = MIMEText('\n\n--------------------------------------------------------------------' '\nEnvironment: ' + environment + '' '\nMode: ' + mode + '' '\nScript Start Time: ' + startTime.strftime("%d/%m/%Y %H:%M:%S") + '' '\nScript End Time: ' + endTime.strftime("%d/%m/%Y %H:%M:%S") + '' '\nError Count: ' + str(errorCounter) + '' '\nError Message: ' + errorMessage + '' '\n--------------------------------------------------------------------' '\n\n--------------------------------------------------------------------' '\nQueue ID list: ' + str(queueIds) + '' '\nStart Date: ' + startDate.strftime('%Y-%m-%d') + 'T00:00:00' '\nEnd Date: ' + endDate.strftime('%Y-%m-%d') + 'T23:59:59' '\nNumber of Activities Scanned: ' + str(len(activityIdsSearched)) + '' '\nActivity Id Attachments Count: ' + str(len(activityIdsWithAttachments)) + '' '\nDeleted Activity Id Count: ' + str(len(deletedActivityIds)) + '' '\n--------------------------------------------------------------------\n\n\n\n') msg['Subject'] = environment + ' - ECE - Delete Attachment Script' msg['From'] = sender msg['To'] = ", ".join(recipients) s.sendmail(sender, recipients, msg.as_string()) s.quit() #get the Command line parameters getArguments() #Login loginEce() #Loop through queues provided for queueId in queueIds: message = ('QueueId: ' + str(queueId)) logging.info(message) listActivities(startDate, endDate,1,queueId) if (retrievedData): message = ('Activity Count: ' + str(activityCount)) logging.info(message) message = ('PageSize: ' + str(pageSize)) logging.info(message) message = ('Current Page Num: ' + str(currentPageNum)) logging.info(message) while ((currentPageNum * pageSize) < activityCount): currentPageNum += 1 message = ('Current Page Num: ' + str(currentPageNum)) logging.info(message) listActivities(startDate, endDate, currentPageNum,queueId) #print('Activity Ids Searched: ' + str(activityIdsSearched)) message = ('Activity Ids with attachments: ' + str(activityIdsWithAttachments)) logging.info(message) message = ('Number of Activities Scanned: ' + str(len(activityIdsSearched))) logging.info(message) message = ('Number of Activity IDs with attachments: ' + str(len(activityIdsWithAttachments))) logging.info(message) if (len(activityIdsWithAttachments)>0 and writeAccess and (silentMode==False)): confirm = input('\nAre you sure you want to delete these attachments?') if (confirm == 'y' or confirm == 'Y'): for activityIdWithAttachment in activityIdsWithAttachments: deleteAttachments(activityIdWithAttachment) if (len(activityIdsWithAttachments)>0 and writeAccess and (silentMode==True)): for activityIdWithAttachment in activityIdsWithAttachments: deleteAttachments(activityIdWithAttachment) message = ('Activity IDs Deleted: ' + str(deletedActivityIds)) logging.info(message) message = ('Number of Activity IDs Deleted: ' + str(len(deletedActivityIds))) logging.info(message) #Send Email sendEmail() #Logout logoutEce() endTime = datetime.now() message = ('endTime: ' + endTime.strftime("%d/%m/%Y %H:%M:%S")) logging.info(message) message = ('---------- Finished! ----------') logging.info(message)
===== Retrieve Activity and Case ID info for Activities in specific Date Range and only for certain Queue IDs
# Author Gerry O'Rourke # Date: 05/06/2024 # Version: 2.02 # Example1 # 'ECE-Get-Activities_V2.0.py -q 1015 -r 10' # This will get Activities for yesterday (-d defaults to 1 and -r default to only get 1 days worth) # # Example2 # 'ECE-Get-Activities_V2.0.py -d 0 -q 1015 -r 10' # Get Activities for Queue ID 1015 from today (-d 0), and go back 10 days. i.e. the last 10 days including today. # Example3 'ECE-Get-Activities_V2.0.py -d 1 -q 1015 -1 1023 -r 3' # Get Activities for yesterday, the day before yesterday and the day before that, for queues 1015 AND 1023 # End date is "-d 1" - i.e. 1 day ago and range (-r) is 3 days (-d 0) # # Note this script should be improved - as the Case ID for each Activity can be retrieved from the Activity Search Response (and there is no need to get the Activity to find the Case ID) import time # used when pausing script for X milliseconds import json import os # check / creating folders import argparse import sys # used for sys.exit and standard output import requests # used for http requests import urllib3 # from datetime import datetime from datetime import date from datetime import timedelta import logging from logging.handlers import RotatingFileHandler urllib3.disable_warnings() # Suppress cert warnings # Initial Variables loglevel = logging.DEBUG # DEBUG, INFO logging.basicConfig(handlers=[logging.StreamHandler(sys.stdout), RotatingFileHandler('logs/ece-get-activities.log', maxBytes=10 * 1024 * 1024, backupCount=10)], style='{', format='{asctime} - {levelname:8s} - {message}', level=loglevel) ece_username = 'my_username' ece_password = 'my_pa$$w0rd' ece_host = 'ece-webserver.mydomain.com' environment = 'PRD' # Do not have any spaces - set to 'LAB' or 'PROD' delay = 333 / 1000 # Time to pause when retrieving HTTP Request loops - to reduce load on server, e.g. 200/1000 -> 4 # API Requests per second, 333/1000 -> 3 API requests per second, 500/100 => 2 API calls per second. rootFolder = 'C:/eceScripts/ECE-Export-Activities/' + environment + '/' activity_ids_searched = [] activity_ids_with_attachments = [] case_ids = [] case_ids_closed = [] error_counter = 0 error_message = '' startTime = datetime.now() message = ('--------------- Starting! ---------------') logging.info(message) message = ('startTime: ' + startTime.strftime("%d/%m/%Y %H:%M:%S")) logging.info(message) message = ('Arguments supplied: ' + str(sys.argv)) logging.info(message) def getArguments(): global endDate global startDate global queueIds global retrieveActivities global writeActivitiesToFile global error_counter global error_message parser = argparse.ArgumentParser() parser.add_argument('-d', '--days', type=int, help='sets the end date, e.g. -d 180, sets the end dat as 180 days ago') parser.add_argument('-r', '--dayrange', type=int, default=1, help='by default only a single day') parser.add_argument('-q', '--queue', action='append', help='Queue IDs, for multiple queue ID enter -q before each one') parser.add_argument('-a', '--activities', action='store_true', help='Retrieve the Activities found (need to use this if we want to get Case info)') parser.add_argument('-w', '--write', action='store_true', help='Write Activities out to file') args = parser.parse_args() if args.days: endDate = date.today() - timedelta(days=args.days) message = ('End Date set to: ' + endDate.strftime('%Y-%m-%d') + 'T23:59:59') logging.info(message) elif args.days == 0: endDate = date.today() message = ('End Date set to: ' + endDate.strftime('%Y-%m-%d') + 'T23:59:59') logging.info(message) else: message = ('No date argument set. Enter the number of days from today, e.g. "-d 1", would be 1 day ago.') logging.error(message) error_counter = error_counter + 1 error_message = 'Arguments not set correctly - please check log file for more information.' exit(1) if args.dayrange: startDate = endDate - timedelta(days=(args.dayrange - 1)) message = ('Start Date set to: ' + startDate.strftime('%Y-%m-%d') + 'T00:00:00') logging.info(message) if args.queue: queueIds = args.queue message = ('Queue ID list: ' + str(queueIds)) logging.info(message) else: message = ( 'No queue ID data set. This app COULD (if we wanted to!) be used to retrieve all activities for a date ' 'range but that is not what we want' 'Hence you must enter at least 1 queue with the "-q" parameter. For multiple queues add a "-q" for in ' 'front of each queue ID.') logging.info(message) error_counter = error_counter + 1 error_message = 'Arguments not set correctly - please check log file for more information.' exit(1) if args.activities: retrieveActivities = True else: retrieveActivities = False if args.write: writeActivitiesToFile = True else: writeActivitiesToFile = False def loginEce(): global error_counter global error_message login_url = 'https://' + ece_host + '/system/ws/v12/authentication/user/login?forceLogin=yes' loginUser = '{"userName":"' + ece_username + '","password":"' + ece_password + '"}' loginHeaders = {"Content-Type": "application/json"} message = ('Attempting to login...') logging.info(message) responseStatusCode = 0 try: response = requests.post(url=login_url, verify=False, headers=loginHeaders, data=loginUser, timeout=5) responseHeaders = (response.headers) responseStatusCode = (response.status_code) except: message = ('Issue Calling Login URL.') logging.error(message) if (responseStatusCode == 204): # Get the X-egain-session Header value returned after successful (204) login. global responseToken responseToken = (responseHeaders['X-egain-session']) global eceHeaders eceHeaders = {"Accept": "application/json", "X-egain-session": responseToken} message = ('Login Successful!, X-egain-session value is: ' + str(responseToken)) logging.info(message) else: message = ('Login Failed!, Status Code: ' + str(responseStatusCode)) logging.error(message) error_counter = error_counter + 1 error_message = 'Failed to log into ECE successfully. Please check log file for more information.' sys.exit(1) def logoutEce(): logoutUrl = 'https://' + ece_host + '/system/ws/v12/authentication/user/logout' responseStatusCode = 0 try: response = requests.delete(url=logoutUrl, verify=False, headers=eceHeaders) responseStatusCode = (response.status_code) response_text = response.text except: message = ('Issue Calling Logout URL.') logging.error(message) if (responseStatusCode == 204): message = ('Logged out successfully.') logging.info(message) else: message = ('Logout Failed!, Status Code: ' + str(responseStatusCode) + '\nError Response: ' + response_text) logging.error(message) def listActivities(startDate, endDate, pageNum, queueId): global activity_ids_searched global activity_ids_with_attachments global activityCount global currentPageNum global pageSize global retrievedData global error_counter startDateString = (startDate.strftime('%Y-%m-%d')) endDateString = (endDate.strftime('%Y-%m-%d')) currentPageNum = pageNum retrievedData = False listActivitiesUrl = 'https://' + ece_host + '/system/ws/v12/interaction/activity?type=email&queue=' + str( queueId) + '&$pagenum=' + str( currentPageNum) + '&lastModifiedDate=[' + startDateString + 'T00:00:00.000Z,' + endDateString + 'T23:59:59.999Z]' message = ('listActivitiesUrl: ' + listActivitiesUrl) logging.debug(message) # Attempt to Retrieve List of Activities responseStatusCode = 0 try: response = requests.get(url=listActivitiesUrl, verify=False, headers=eceHeaders, timeout=5) responseStatusCode = (response.status_code) responseText = response.text except: message = ('Issue Calling List Activity URL.') logging.error(message) if (responseStatusCode == 200): try: activityDict = json.loads(responseText) activityCount = activityDict['paginationInfo']['count'] # currentPageNum = activityDict['paginationInfo']['pagenum'] pageSize = activityDict['paginationInfo']['pagesize'] # Loop through the file and get each Activity id for id in activityDict['activity']: activityId = str(id['id']) # get the Activity ID # print('activityId: ',activityId) activity_ids_searched.append(activityId) activityAttachmentsCount = id['attachments']['count'] # Get the Attachments Count # print('activityAttachmentsCount: ', activityAttachmentsCount) if (activityAttachmentsCount > 0): activity_ids_with_attachments.append(activityId) retrievedData = True except Exception as e: message = ('Error in listActivities Function. error:' + str(e)) logging.error(message) error_counter = error_counter + 1 elif (currentPageNum == 1 and responseStatusCode == 204): message = ('No Activities Content for that date range') logging.info(message) elif (responseStatusCode == 204): message = ('Completed search.') logging.info(message) else: message = ('Failed to list Activities, statusCode: ' + str(responseStatusCode)) logging.error(message) error_counter = error_counter + 1 def getActivity(id, folder): message = ('Attempting to retrieve Activity Id: ' + id + '...') logging.debug(message) activityUrl = 'https://' + ece_host + '/system/ws/v12/interaction/activity/' + id + '?$attribute=all' # Attempt to get Activity Id response = requests.get(url=activityUrl, verify=False, headers=eceHeaders) # Store Response Values responseStatusCode = (response.status_code) responseText = response.text # print (activityDict.keys()) # print (activityDict['activity'][0]['case']['id']) if (responseStatusCode == 200): try: activityDict = json.loads(responseText) caseId = str(activityDict['activity'][0]['case']['id']) # Get the Case Id if caseId not in case_ids: case_ids.append(caseId) if (writeActivitiesToFile): fileName = 'activityId-' + id + '-caseId-' + caseId + '.json' myFolder = rootFolder + folder filePath = myFolder + '/' + fileName # Check whether the specified path exists or not isExist = os.path.exists(myFolder) if not isExist: # Create a new directory because it does not exist os.makedirs(myFolder) message = ('New directory: "' + myFolder + '" is created.') logging.info(message) # Save the activity as a text file text_file = open(filePath, "w", encoding="utf-8") text_file.write(responseText) text_file.close() message = ('Saved Activity Id : ' + id + 'to file.') logging.debug(message) except Exception as e: message = ('Error in getActivity function. error: ' + str(e)) logging.error(message) logoutEce() sys.exit(2) else: print('Failed to retrieve Activity') def getCase(id): message = ('Attempting to retrieve Case Id: ' + id + '...') logging.debug(message) caseUrl = 'https://' + ece_host + '/system/ws/v12/interaction/case/' + id + '?$attribute=all' # Attempt to get Activity Id response = requests.get(url=caseUrl, verify=False, headers=eceHeaders) # Store Response Values responseStatusCode = (response.status_code) responseText = response.text if (responseStatusCode == 200): try: caseDict = json.loads(responseText) caseStatus = str(caseDict['case'][0]['status']['value']) # Get the Case Id if (caseStatus == 'closed'): case_ids_closed.append(caseId) if (writeActivitiesToFile): fileName = 'caseId-' + id + '-' + caseStatus + '.json' myFolder = rootFolder + 'Cases' filePath = myFolder + '/' + fileName # Check whether the specified path exists or not isExist = os.path.exists(myFolder) if not isExist: # Create a new directory because it does not exist os.makedirs(myFolder) message = ('New directory: "' + myFolder + '" is created.') logging.info(message) # Save the activity as a text file text_file = open(filePath, "w", encoding="utf-8") text_file.write(responseText) text_file.close() message = ('Saved Case Id : ' + id + ' to file.') logging.debug(message) except Exception as e: message = ('Error in getCase function. error: ' + str(e)) logging.error(message) logoutEce() sys.exit(2) else: print('Failed to retrieve Case') # get the Command line parameters getArguments() # Login loginEce() # Loop through queues provided for queueId in queueIds: message = ('QueueId: ' + str(queueId)) logging.info(message) listActivities(startDate, endDate, 1, queueId) if (retrievedData): message = ('Activity Count: ' + str(activityCount)) logging.info(message) message = ('PageSize: ' + str(pageSize)) logging.info(message) message = ('Current Page Num: ' + str(currentPageNum)) logging.debug(message) while ((currentPageNum * pageSize) < activityCount): currentPageNum += 1 message = ('Current Page Num: ' + str(currentPageNum)) logging.debug(message) time.sleep(delay) listActivities(startDate, endDate, currentPageNum, queueId) message = ('Number of Activities Searched: ' + str(len(activity_ids_searched))) logging.debug(message) message = ('Number of Activity IDs with attachments: ' + str(len(activity_ids_with_attachments))) logging.debug(message) message = ('Activity Ids Searched: ' + str(activity_ids_searched)) logging.debug(message) message = ('Activity Ids with attachments: ' + str(activity_ids_with_attachments)) logging.debug(message) if (retrieveActivities): message = ('Getting activityIds. This might take some time...') logging.info(message) for activityId in activity_ids_searched: message = ('Getting activityId: ' + str(activityId)) logging.debug(message) time.sleep(delay) getActivity(activityId, 'Activities-All') message = ('Number of case_ids: ' + str(len(case_ids))) logging.debug(message) message = ('case_ids: ' + str(case_ids)) logging.debug(message) message = ('Getting case_ids. This might take some time...') logging.info(message) for caseId in case_ids: message = ('Getting caseId: ' + str(caseId)) logging.debug(message) time.sleep(delay) getCase(caseId) message = ('Number of Closed case_ids: ' + str(len(case_ids_closed))) logging.debug(message) message = ('Closed case_ids: ' + str(case_ids_closed)) logging.debug(message) # Logout logoutEce() # Print Summary At the end message = ('#########################################') logging.info(message) message = ('Filter StartDate: ' + startDate.strftime('%Y-%m-%d') + ' 00:00:00') logging.info(message) message = ('Filter EndDate: ' + endDate.strftime('%Y-%m-%d') + ' 23:59:59') logging.info(message) message = ('Number of Activities Searched: ' + str(len(activity_ids_searched))) logging.info(message) message = ('Number of Activity IDs with attachments: ' + str(len(activity_ids_with_attachments))) logging.info(message) if (retrieveActivities): message = ('Number of case_ids: ' + str(len(case_ids))) logging.info(message) message = ('Number of Closed case_ids: ' + str(len(case_ids_closed))) logging.info(message) message = ('Activity Ids Searched: ' + str(activity_ids_searched)) logging.info(message) message = ('Activity Ids with attachments: ' + str(activity_ids_with_attachments)) logging.info(message) if (retrieveActivities): message = ('case_ids: ' + str(case_ids)) logging.info(message) message = ('Closed case_ids: ' + str(case_ids_closed)) logging.info(message) message = ('startTime: ' + startTime.strftime("%d/%m/%Y %H:%M:%S")) logging.info(message) endTime = datetime.now() message = ('endTime: ' + endTime.strftime("%d/%m/%Y %H:%M:%S")) logging.info(message) duration = endTime - startTime message = ('duration: ' + str(duration)) logging.info(message) message = ('#########################################') logging.info(message) message = ('--------------- Finished! ---------------') logging.info(message)