pandas - 使用 Beatbox 解析 Pandas DataFrame 中的 Salesforce 报告

Question

有人试过使用 Beatbox 将 SalesForce 报告解析为 Pandas DataFrame 吗？ SO 上有几个示例，但没有一个提供全面的解决方案，或者至少我认为没有。

#!/usr/bin/env python3

import beatbox
import pandas as pd


sf = beatbox._tPartnerNS
service = beatbox.Client()
service.serverUrl = 'https://login.salesforce.com/services/Soap/u/38.0'
service.login('my-username', 'my-password')

report_id = '00myreport4G3V'

query = "SELECT Name FROM Report where id = '{}'".format(report_id)
query_result = service.query(query)

这只是选择名称，但理想情况下我想将报告的内容加载到 DataFrame 中。有什么帮助吗？

Answer 1

报告数据可以通过Salesforce Reports and Dashboards REST API检索.自 Summer'15(版本 34.0)以来，这在 Salesforce 中有效。

我用包 Simple-salesforce 写了一个例子，由于 REST API。 (然而，可以在没有 simple-salesforce 的情况下重写它并使用来自 Beatbox 的 api session ，编写至少大约 10 行额外的代码并仅安装 requests 包。)

通用代码

from collections import OrderedDict
from simple_salesforce import Salesforce
import pandas as pd
import json

class SfReportsApi(Salesforce):
    def __init__(self, *args, **kwargs):
        super(SfReportsApi, self).__init__(*args, **kwargs)

    def describe_report(self, report_id):
        return self._call_report(report_id, command='/describe')

    def to_pandas_dataframe(self, report_id, metadata=None):
        """SF report details exported to DataFrame, can be modified by metadata"""
        resp = self._call_report(report_id, metadata=metadata)
        if not resp['allData']:
            print("Detailed data have been truncated to the usual report limit (2000).")
        columns = []
        converters = []
        get_label = lambda x: x['label']
        sf_pandas_map = {
            'boolean': lambda x: x['value'],
            'currency': lambda x: x['value'].get('amount'),
            'date': lambda x: pd.Timestamp(x['value']),
            'datetime': lambda x: pd.Timestamp(x['value']),
            'double': lambda x: x['value'],
            'picklist': get_label,
            'string': get_label,
            'textarea': get_label,
        }
        for col in resp['reportExtendedMetadata']['detailColumnInfo'].values():
            columns.append(col['label'])
            converters.append(sf_pandas_map.get(col['dataType'], get_label))
        data = [[conv(cell) for conv, cell in zip(converters, row['dataCells'])]
                for sect_key, section in resp['factMap'].items()
                    if 'rows' in section
                        for row in section['rows']
                ]
        df = pd.DataFrame(data, columns=columns)
        return df

    def _call_report(self, report_id, metadata=None, command=None):
        url = '{}analytics/reports/{}{}'.format(self.base_url, report_id, command or '')
        data = json.dumps({'reportMetadata': metadata}) if metadata else None
        resp = self._call_salesforce('POST' if metadata else 'GET', url, data=data)
        return resp.json(object_pairs_hook=OrderedDict)

使用示例

report_id = '00O24000004qtI4EAI'
# set Salesforce session_id some way (by login or reused from other app)
sf = SfReportsApi(username='me@example.com', password='password', security_token='token')
# sf = SfReportsApi(instance_url='https://na1.salesforce.com', session_id='')
# get report metadata if useful
metadata = sf.describe_report(report_id)['reportMetadata']
# modify them or write only the modified keys, e.g. change filters or remove subtotals etc.
metadata = {
    'orderBy': ['ACCOUNT.NAME'],
    'reportFilters': [{'value': 'W', 'column': 'ACCOUNT.NAME', 'operator': greaterOrEqual'}]
}
# or you can omit `metadata` parameter and use the report as is without changing anything
df = sf.to_pandas_dataframe(report_id, metadata)

可以动态添加列、过滤器、排序等(关于 report Execute synchronous 的文档)。 to_pandas_dataframe 方法用于包含详细信息的普通表格报告，并且可以选择包含一个总计且不超过一级小计。可以从更复杂的报告中检索数据(请参阅有关 Decode the Fact Map 或 cheatsheet 的文档)，但它没有实现，因为在运行之前通过元数据参数更容易即时删除它们。

只能报告2000行详细数据。可以使用多个带有过滤器的请求来查看所有数据。

固定代码 新代码适用于任何包含原始行的报告。它不适用于没有原始行的摘要报告。 (旧代码仅适用于包含行和小计的报告。请原谅)