使用 Python 为 Tableau 获取数据

本教程介绍如何使用 Python 3 将 Communications Mining 平台中的数据提取为适合导入到 Tableau 或类似分析应用程序中的格式。

重要提示：

使用本教程所需的权限

• 查看来源
• 查看标签
• 流管理员
• 使用流

本教程中的数据

本教程中的数据

本教程将演示如何获取以下数据：

Communications Mining 字段：

• 分类中每个标签的标签预测（如果未预测标签，则为0如果为0.0到1.0之间的值，则为
• 匹配的常规字段
• 服务质量分数（如果为数据集启用了服务质量）
• 语气分数（如果为数据集启用了语气）

注释数据:

• ID
  • 注释 ID
  • 来源 ID
  • 电子邮件消息 ID（由 Exchange 提供的唯一 ID）
  • 线程 ID
• 电子邮件主题
• 电子邮件正文
• 来自发件人的电子邮件
• 电子邮件收件人列表
• 电子邮件抄送收件人列表
• 电子邮件密送收件人列表
• 邮箱文件夹（同步时电子邮件所在的位置）
• 附件数量
• 附件名称列表
• 电子邮件的“发送日期”时间戳

Python 示例

Python 示例

本节演示如何从 Communications Mining 平台获取注释，并将其转换为适合 Tableau 或类似分析应用程序的格式。 请根据您的特定要求调整此示例。

请确保您使用的是 Python 3 并已安装以下库:

• urllib3和requests （用于向 Communications Mining API 发出请求）
• pandas （用于在本教程的最后一步中将数据转换为数据框）

设置

确定以下步骤所需的资源。

• 您的API 端点
  • 对于通过 UiPath 加入的租户： https://cloud.uipath.com/<my_uipath_organisation>/<my_uipath_tenant>/reinfer_/api/v1
  • 对于独立租户： https://<my_domain>.reinfer.io/api/v1
• 您的API 令牌
• 要从中获取注释的数据集的名称

用于获取注释的流的名称。 您应该创建一个新流或使用现有流。

API_ENDPOINT = "YOUR API ENDPOINT"
API_TOKEN = "YOUR API TOKEN"
DATASET_NAME = "project-name/dataset-name"
STREAM_NAME = "stream-name"

创建将用于所有 API 请求的requests会话。 建议您将其配置为重试失败的请求（请参阅示例）。

from requests import Session
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
from http import HTTPStatus

RETRY_STRATEGY = Retry(
    total=5,
    status_forcelist=[
        HTTPStatus.TOO_MANY_REQUESTS,
        HTTPStatus.BAD_GATEWAY,
        HTTPStatus.GATEWAY_TIMEOUT,
        HTTPStatus.INTERNAL_SERVER_ERROR,
        HTTPStatus.REQUEST_TIMEOUT,
        HTTPStatus.SERVICE_UNAVAILABLE,
],
    allowed_methods=["GET", "POST"],
    backoff_factor=1,
)

adapter = HTTPAdapter(max_retries=RETRY_STRATEGY)
session = Session()
session.mount("https://", adapter)
session.mount("http://", adapter)

session.headers.update({"Authorization": "Bearer " + API_TOKEN})

# If you need to use a proxy to connect to the internet, see# https://requests.readthedocs.io/en/latest/user/advanced/#proxies# on how to configure a proxy for your `requests` session.

如果出现错误，以下步骤将引发DownloadError 。

class DownloadError(Exception):
    pass

标签分类和常规字段

获取标签分类 (即 所有标签的名称）的数据集，以获取流使用的模型版本。 这是下一步所必需的。

# get model version used by the stream
stream_response = session.get(
    f"{API_ENDPOINT}/datasets/{DATASET_NAME}/streams/{STREAM_NAME}",
)
stream_response_json = stream_response.json()
if not stream_response.ok:
    raise DownloadError(stream_response_json)
model_version = stream_response_json["stream"]["model"]["version"]

# get label taxonomy
model_stats_response = session.get(
    f"{API_ENDPOINT}/datasets/{DATASET_NAME}/labellers/{model_version}/validation",
)
model_stats_response_json = model_stats_response.json()
if not model_stats_response.ok:
    raise DownloadError(model_stats_response_json)
label_taxonomy = [
    label["name"] for label in model_stats_response_json["validation"]["labels"]
]
entities = [
    entity["name"] for entity in model_stats_response_json["validation"]["entities"]
]

# sort for use in next steps
label_taxonomy.sort()
entities.sort()

注释转换

Communications Mining 将注释返回为嵌套 JSON 对象。 为了在 Tableau 或类似应用程序中使用，需要将嵌套的 JSON 对象转换为更合适的表格格式。

定义实用程序函数，以将 Communications Mining comment对象转换为表格格式。

请注意，由于注释可以具有多个相同的常规字段，因此在此示例中，相同类型的所有匹配实体都将连接并放置在同一列中。

def comment_to_dict(comment, sorted_taxonomy, sorted_entities):

    message = comment["comment"]["messages"][0]  # email fields
    userprops = comment["comment"]["user_properties"]  # comment metadata
    labelprops = {
        prop["property_name"]:prop["value"]
        for prop in comment.get("label_properties", [])
    }  # QOS and Tone scores (if enabled in dataset)
    predictions = {
        " > ".join(prediction["name"]):prediction["probability"]
        for prediction in comment.get("labels", [])
    }
    entities = comment.get("entities", [])
    attachments = comment["comment"].get("attachments", [])
    comment_dict = {
        # comment
        "comment_id": comment["comment"]['id'],
        "comment_uid": comment["comment"]['uid'],
        "source_id": comment["comment"]['source_id'],
        "comment_timestamp": comment["comment"]['timestamp'],
        # email fields
        "email_subject": message.get("subject", {}).get("text"),
        "email_message": message.get("body", {}).get("text"),
        "email_from": message.get("from"),
        "email_to": message.get("to", []),
        "email_cc": message.get("cc", []),
        "email_bcc": message.get("bcc", []),
        "email_sent_at": message.get("sent_at"),
        "email_message_id": userprops.get("string:Message ID"),
        "email_folder": userprops.get("string:Folder"),
        "email_num_attachments": len(attachments),
        "email_attachments": attachments,
        "has_attachments": len(attachments) > 0,
        "total_attachment_size_bytes": sum([item["size"] for item in attachments]),
        "attachment_names": [item["name"] for item in attachments],
        "attachment_types": [item["content_type"] for item in attachments],
        "thread_id": comment["comment"].get('thread_id'),
        # QOS and Tone scores
        "qos_score": labelprops.get("quality_of_service"),
        "tone_score": labelprops.get("tone"),
    }
    for label in sorted_taxonomy:
        comment_dict[label] = predictions.get(label, 0)
    for entity in sorted_entities:
        comment_dict[entity] = ", ".join([
            item["formatted_value"]
            for item in entities if item["name"] == entity])
    return comment_dict

设置流起点

默认情况下，流将返回晚于其创建时间的注释。 在开发过程中，通常需要将流重置为从特定时间点开始。

STARTING_TIME = "2023-01-03T16:05:00"  # change to required starting time

stream_reset_response = session.post(
    f"{API_ENDPOINT}/datasets/{DATASET_NAME}/streams/{STREAM_NAME}/reset",
    json={
        "to_comment_created_at": STARTING_TIME
    },
)
stream_reset_response_json = stream_reset_response.json()
if not stream_reset_response.ok:
    raise DownloadError(stream_reset_response_json)

获取-高级循环

流可批量提供注释，并会跟踪上次提取的注释。 使用流获取路由获取注释，并使用流高级路由确认批次。 如果批次未获确认，则流将不会继续提供下一个批次。 因此，从 Communications Mining 获取注释的过程称为“高级获取循环”。

定义一个实用程序函数，该函数通过重复“获取-推进”循环直至获取所有注释来获取注释。 出于演示目的，此函数将提取的所有注释存储在内存中。 在生产场景或任何包含大量数据的场景中，应将每批注释推送到数据存储或附加到文件。

由于流会跟踪上次提取的注释，因此可以安全地停止和继续此流程。

import pandas as pd

def fetch_comments_from_stream(api_endpoint, dataset_name, stream_name, batch_size, label_taxonomy):
    """Fetch comments until no more comments are available"""

    comment_dicts = []

    while True:
        # fetch BATCH_SIZE comments from stream
        fetch_response = session.post(
            f"{api_endpoint}/datasets/{dataset_name}/streams/{stream_name}/fetch",
            json={
                "size": batch_size,
            },
        )

        # get comments from response
        fetch_response_json = fetch_response.json()
        if not fetch_response.ok:
            raise DownloadError(fetch_response_json)
        comments = fetch_response_json["results"]
        if len(comments) == 0:
            break

        # process comments
        for comment in comments:
            comment_dicts.append(comment_to_dict(comment, label_taxonomy, entities))

        # advance stream using the `sequence_id` from response
        advance_response = session.post(
            f"{api_endpoint}/datasets/{dataset_name}/streams/{stream_name}/advance",
            json={
                "sequence_id": fetch_response_json["sequence_id"],
            },
        )
        advance_response_json = advance_response.json()
        if not advance_response.ok:
            raise DownloadError(advance_response_json)

    return comment_dicts

BATCH_SIZE = 100  # number of comments to fetch in each `fetch` request. max value is 1024.

comment_dicts = fetch_comments_from_stream(
    API_ENDPOINT, DATASET_NAME, STREAM_NAME, BATCH_SIZE, label_taxonomy
)

df = pd.DataFrame.from_records(comment_dicts)

# do something with `df`

此时，您可以根据要求继续处理或存储数据。

如果您需要再次获取相同的数据（出于测试目的），则需要重置流。