Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
Examine each Salesforce report definition, and generate a csv file of those with relevant filters
#!/usr/local/bin/python3
# https://developer.salesforce.com/docs/atlas.en-us.api_analytics.meta/api_analytics/sforce_analytics_rest_api_getbasic_reportmetadata.htm
import pdb
import json
from collections import OrderedDict
import config
from progressbar import ProgressBar
import simple_salesforce
class SFDC(simple_salesforce.Salesforce):
def __init__(self, username=None, password=None, security_token=None,
sandbox=False, version=config.DEFAULT_API_VERSION):
super(SFDC, self).__init__(username=username, password=password, security_token=security_token,
sandbox=sandbox, version=version)
def describeReport(self, id):
"""Describes report with given id
"""
url = self.base_url + "analytics/reports/{}/describe".format(id)
result = self._call_salesforce('GET', url)
if result.status_code != 200:
raise SalesforceGeneralError(url,
'describe',
result.status_code,
result.content)
json_result = result.json(object_pairs_hook=OrderedDict)
# json_result = result.json()
if len(json_result) == 0:
return None
else:
return json_result
def init():
salesforce_credentials = json.loads(open(config.SFDC_CONFIGFILE).read())
username = salesforce_credentials['user']
if salesforce_credentials['sandbox']:
username += '.' + salesforce_credentials['sandbox']
config.SFDC_URL = 'https://canonical--staging--c.cs87.visual.force.com/'
else:
config.SFDC_URL = 'https://eu1.salesforce.com/';
sfdc = SFDC(username=username,
password=salesforce_credentials['password'],
security_token=salesforce_credentials['token'],
sandbox=salesforce_credentials['sandbox'],
version=config.DEFAULT_API_VERSION)
return sfdc
reportFields = [
'Id'
,'Name'
# ,'FolderName'
,'DeveloperName'
,'CreatedDate'
,'CreatedBy.Name'
,'LastRunDate'
,'LastModifiedBy.Name'
,'LastModifiedDate'
,'LastViewedDate'
,'Description'
]
def dumpHeader():
print('\t'.join([
'FolderName'
,'Name'
,'CreatedDate'
,'LastModifiedDate'
,'LastRunDate'
,'CreatedBy.Name'
,'LastModifiedBy.Name'
,'Note'
,'Url'
]))
def quotify(s):
if s == None:
return '';
s = str(s)
if len(s) == 0:
return '';
return '"' + s.replace('\n', '\\n').replace('"', '\"') + '"'
def get(dict, keys):
if not keys:
return dict
if not isinstance(keys, list):
return dict[keys]
return get(dict[keys.pop(0)], keys)
def dumpReport(folderName, reportJson, reportRecord, sfdcUrl, note):
print('\t'.join([quotify(cell) for cell in
[folderName] +
[reportJson['reportMetadata']['name']] +
[get(reportRecord, keys) for keys in [
'CreatedDate'
, 'LastModifiedDate'
, 'LastRunDate'
, ['CreatedBy', 'Name']
, ['LastModifiedBy', 'Name']
]] +
[note] +
[sfdcUrl + reportJson['reportMetadata']['id']]
]))
pass
# queryWhereClause = "where id = '00OD00000072LjH'"
queryWhereClause = ""
relevantStageNames = ['Value Proposition', 'Negotiation/Review', 'Proposal']
def relevant(reportJson):
stages = []
if reportJson and 'reportMetadata' in reportJson:
for filter in reportJson['reportMetadata']['reportFilters']:
if filter['column'] == 'Renewal__c':
return 'Renewal__c'
if filter['column'] == 'FORECAST_CATEGORY':
return 'Forecast Category'
elif filter['column'] == 'STAGE_NAME':
values = filter['value'].split(',')
for value in relevantStageNames:
if value in values:
return 'Stage'
return None
def getFolderName(reportJson, folderNamesById):
folderId = reportJson['reportMetadata']['folderId']
folderName = ''
if folderId in folderNamesById:
folderName = folderNamesById[folderId]
elif folderId.startswith('005'):
folderName = 'My Personal Custom Reports'
elif folderId.startswith('00D'):
folderName = 'Unfiled Custom Reports'
else:
folderName = folderId
return folderName
def main():
sfdc = init()
sfdcUrl = 'https://{instance}/'.format(instance=sfdc.sf_instance)
folderNamesById = {}
for folder in sfdc.query_all("select Id, Name from folder")['records']:
folderNamesById[folder['Id']] = folder['Name']
result = sfdc.query_all("select {} from report {} ".format(','.join(reportFields), queryWhereClause))
dumpHeader()
progressbar = ProgressBar()
for reportRecord in progressbar(result['records']):
reportJson = None
try:
reportJson = sfdc.describeReport(reportRecord['Id'])
except Exception as e:
if e.status == 501:
pass
else:
pdb.set_trace()
print(reportRecord['Id'], str(e))
relevance = relevant(reportJson)
if relevance:
folderName = getFolderName(reportJson, folderNamesById)
try:
dumpReport(folderName, reportJson, reportRecord, sfdcUrl, relevance)
except Exception as e:
pdb.set_trace()
print(reportRecord['Id'], str(e))
continue
if __name__ == "__main__":
main()
pass
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.