> 持续流程改进 > 客户服务 > Genesys Cloud CX > 数据模板

您的客户服务数据模板

Genesys Cloud CX

← 返回改进指南

客户服务属性(19) 客户服务活动(13) 数据提取指南(3)

您的客户服务数据模板

此模板为收集客户服务流程分析与优化所需的关键数据提供了清晰指引，列出建议采集的属性、需要跟踪的关键活动，以及从源系统提取这些信息的实用方法。您可以据此构建事件日志，支持流程挖掘工作。

建议采集的属性
需要跟踪的关键活动
数据提取指南

刚接触事件日志？了解如何创建流程挖掘事件日志.

下载模板

客户服务属性

以下是建议纳入事件日志的数据字段，可用于全面分析客户服务。

5 必填 6 建议填写 8 选填

	名称	说明
	开始时间 EventTime	活动或事件开始时的时间戳。
说明此属性记录每项活动发生的准确日期和时间，对于正确排列事件顺序及开展所有基于时间的分析至关重要。开始时间用于按时间顺序构建流程路径，也是计算不同活动之间持续时间、等待时间和周期时间的依据。准确的时间戳是可靠开展流程分析和绩效监控的关键。重要性此时间戳按时间顺序排列所有活动，从而准确分析流程路径、持续时间和瓶颈。获取途径可在Genesys Cloud CX的事件日志或交互详情中找到，通常与每条已记录的系统事件或用户事件相关联。示例 2024-05-21T10:00:15Z2024-05-21T10:02:30Z2024-05-21T10:15:00Z
	服务请求 ServiceRequest	客户服务交互的主要标识符，用于关联所有相关活动。
说明服务请求通常也称为工单或案例，是单个客户咨询或问题的主要标识符。它将从首次联系到最终解决的所有相关事件归入同一个流程实例。按服务请求进行分析，可以完整了解客户旅程和内部处理过程。这也是计算总解决时间和首次联系解决率等关键指标的基础。重要性这是关键的案例ID，用于关联所有流程步骤，从头到尾全面分析每次客户交互。获取途径这是概念上的案例标识符。在Genesys中，它通常对应对话ID，但具体取决于实施方式，也可能是自定义标识符。示例 SR-20240521-00123SR-20240521-00124SR-20240522-00001
	活动名称 ActivityName	服务请求生命周期中某个时间点发生的具体事件或任务名称。
说明此属性描述客户服务流程中的具体步骤或状态变更。每项活动都代表一个独立事件，例如“坐席接受交互”或“分配结束代码”。分析这些活动的顺序和频率是流程挖掘的基础，有助于呈现流程的实际流转情况、识别偏差，并找出瓶颈或低效步骤。重要性活动构成流程图的基础，可用于呈现和分析实际流程与设计流程之间的差异。获取途径通常通过将Genesys Cloud CX中的系统事件、坐席状态或特定审计跟踪记录映射为标准化活动名称得出。示例交互开始坐席接受交互通话后处理结束服务请求重新打开
	最近数据更新时间 LastDataUpdate	最近一次数据刷新的时间戳。
说明此属性表示数据集最近一次从源系统更新的时间，可帮助您判断所分析数据的新鲜度，从而及时作出相关业务决策。仪表板和报告应醒目显示此信息，让用户了解数据的时效性，并评估分析结果是否有效。重要性显示数据的新鲜度，确保分析和决策基于最新信息。获取途径此值在提取数据或将数据加载到流程挖掘工具时生成，并写入数据集。示例 2024-05-23T04:00:00Z2024-05-24T04:00:00Z
	源系统 SourceSystem	数据提取自哪个系统。
说明此属性标识事件数据的来源，本例中为Genesys Cloud CX。在集成多个系统的环境中，此字段对于追踪数据来源、排查问题和了解数据背景至关重要。它有助于确保分析使用正确的数据源；当多个系统共同构成一个流程视图时，也可用于筛选或细分数据。重要性标识数据来源，对于数据治理、验证以及合并多个系统的数据至关重要。获取途径通常在数据提取和转换过程中添加固定值，用于标记记录来源。示例 Genesys Cloud CXGenesysCloudCX_US1Genesys
	坐席ID AgentId	处理交互或活动的坐席唯一标识符。
说明坐席ID是每位客服代表的唯一标识。分析坐席绩效、工作量分配和效率时，离不开此属性。您可以按坐席或团队筛选流程图，了解他们如何处理请求，并比较解决时间、返工率和标准流程遵循情况等指标。这也是“坐席绩效与效率”仪表板的核心数据。重要性此属性将流程活动关联到具体员工，便于分析个人绩效、工作量和团队效率。获取途径可在Genesys Cloud CX对话详情记录中找到，与处理交互特定片段的用户相关联。示例 a1b2c3d4-e5f6-7890-1234-567890abcdeff0e9d8c7-b6a5-4321-fedc-ba0987654321
	坐席姓名 AgentName	处理交互或活动的坐席全名。
说明坐席姓名是与坐席ID对应的易读标识，便于管理者和团队负责人查看仪表板和报告。坐席分析会广泛使用此属性来评估绩效、发现培训需求并公平分配工作量，无需再查找系统ID。重要性显示坐席姓名，便于您分析绩效和交流发现，无需使用技术ID。获取途径通过查询AgentId，从Genesys Cloud CX的Users或Directory服务中获取。示例约翰·史密斯简·多伊彼得·琼斯
	沟通渠道 MediaType	交互使用的沟通渠道，例如语音、聊天或电子邮件。
说明此属性指定客户与坐席沟通所用的渠道，常见渠道包括语音、聊天、电子邮件和社交媒体。像“沟通渠道绩效”仪表板一样，按渠道分析绩效有助于企业了解各渠道的交互量、解决时间和客户满意度。这些信息对优化渠道策略和资源分配至关重要。重要性按沟通渠道细分流程，有助于了解各渠道的绩效、客户行为和资源需求。获取途径 Genesys Cloud CX对话详情记录中的标准字段。示例语音聊天电子邮件消息
	结束代码 WrapUpCode	坐席在交互结束时分配的代码，用于归类交互结果或主题。
说明结束代码是坐席选择的标签，用于归类客户交互的性质或处理结果。这些代码以结构化数据记录客户联系支持团队的原因。分析结束代码有助于识别常见问题类型、跟踪处理结果并统计特定请求的频率。这些数据有助于开展根因分析并了解服务需求模式。重要性归类交互结果，为分析常见问题、解决效果和联系原因提供结构化数据。获取途径可在Genesys Cloud CX对话详情记录中找到，具体位于坐席参与者的会话详情中。示例密码重置账单争议已解决产品信息咨询已升级至二级支持
	结束时间 EventEndTime	表示活动或事件结束时间的时间戳。
说明此属性记录活动结束的准确时间。结合开始时间，可精确计算每项活动的持续时间。分析活动耗时有助于找出流程中最耗时的步骤，发现低效环节和优化机会。例如，可识别“交互置于保持”耗时过长或“通话后处理”时间过长的情况。重要性可精确计算各项活动的持续时间，帮助识别耗时步骤和绩效瓶颈。获取途径可在Genesys Cloud CX的事件日志或交互详情中找到，也可根据后续事件的开始时间推算。示例 2024-05-21T10:02:30Z2024-05-21T10:15:00Z2024-05-21T10:18:45Z
	队列名称 QueueName	交互被路由到的队列名称。
说明此属性用于识别交互在分配给坐席前等待的队列。按队列名称分析数据，是“服务队列瓶颈检测”仪表板的关键。管理者可据此了解不同技能组或服务线的工作量分布、衡量各队列的等待时间，并找出长期人手不足或负荷过重的队列。这些信息有助于优化资源分配、缩短客户等待时间。重要性显示服务请求等待分配的位置，帮助您发现瓶颈并分析工作量分布。获取途径可在Genesys Cloud CX对话详情记录中找到。每次交互可能经过一个或多个队列。示例一级支持-语音账单咨询-聊天技术支持-电子邮件
	CSAT评分 CustomerSatisfactionScore	客户在交互结束后填写的满意度调查评分。
说明客户满意度评分（CSAT）直接反映客户对所获服务的评价。通常会在交互结束后通过调查收集。结合流程数据分析CSAT评分，可以了解流程差异、解决时间或特定坐席如何影响客户满意度。这是评估整体流程成效的重要结果指标。重要性直接衡量客户满意度，便于分析流程绩效与客户结果之间的关联。获取途径此数据通常来自Genesys Cloud CX质量管理模块，或与Genesys集成的第三方调查工具。示例 5413
	SLA目标解决时间 SlaTargetResolutionTime	服务请求解决时间的合同约定目标。
说明此属性根据服务级别协议（SLA）规定服务请求的最长解决时间，并作为衡量实际解决时间的基准。它是“SLA合规概览”仪表板和“SLA合规率”KPI的重要数据，可帮助组织对照客户承诺监控绩效，并在可能违约前发现风险。重要性为衡量SLA合规情况提供基准，这是服务绩效和合同义务的重要指标。获取途径此值可能存储为对话的自定义属性，也可能根据队列、客户类型或请求类型等规则推算。示例 86400144003600
	对话ID ConversationId	Genesys为整个对话分配的唯一标识符。
说明对话ID是Genesys Cloud CX中的主要技术键，用于汇总单次客户对话中的所有相关交互、片段和参与者。概念上的“服务请求”用作案例ID，而对话ID则是通过Genesys API获取所有相关数据的底层键。它对于提取数据、关联不同数据集以及验证流程数据至关重要。重要性这是Genesys中的主要技术键，对于提取数据、排查问题和关联回源系统至关重要。获取途径这是Genesys Cloud CX所有分析和对话API中的主要字段。示例 d8a7c6b5-e4f3-2109-8765-fedcba098765c7b6a5d4-f3e2-1098-7654-edcbaf987654
	是否符合SLA IsSlaCompliant	标记服务请求是否在SLA目标时间内解决。
说明此布尔属性表示服务请求是否达到解决时间目标，通过比较“ServiceResolutionTime”和“SlaTargetResolutionTime”计算得出。此标记简化了“SLA合规概览”仪表板中的分析和可视化，也是计算“SLA合规率”KPI的依据。您可以快速筛选并细分符合或未符合SLA的案例，识别SLA违约的常见模式。重要性清晰标记每个案例是否符合SLA或已违约，简化SLA绩效分析，并支持对未达标情况进行根因分析。获取途径计算字段：如果ServiceResolutionTime <= SlaTargetResolutionTime，则为True，否则为False。示例 truefalse
	是否重新打开 IsReopened	标记已解决的服务请求之后是否重新打开。
说明此布尔属性用于标记状态已设为已解决或已关闭、之后又重新变为活跃的服务请求。重新打开的案例通常表示初次解决方案无效或不完整，导致客户不满并增加工作量。此属性支持“服务请求重新打开趋势”仪表板和“服务请求重新打开率”KPI。分析这些案例有助于找出解决无效的根本原因。重要性揭示解决流程中的问题，指出解决方案质量缺陷，并可能导致返工和糟糕的客户体验。获取途径计算字段：检测“服务请求已解决”活动发生后重新激活案例的活动，并据此推导。示例 truefalse
	是否首次联系解决 IsFirstContactResolution	标记服务请求是否在一次交互内解决。
说明此布尔属性用于识别在首次交互中解决的案例，无需客户后续联系或内部转接。值为“true”表示问题得到理想且高效的解决。此属性是“首次联系解决率”KPI及其对应仪表板的计算依据。分析首次联系未能解决的案例，有助于发现坐席培训、知识库改进或流程调整的机会。重要性直接衡量效率和客户满意度，因为首次解决问题是良好服务体验的重要因素。获取途径计算字段，根据案例的事件顺序推导。如果案例解决前未发生转接或重新打开等特定活动，则视为首次联系解决（FCR）。示例 truefalse
	服务解决时间 ServiceResolutionTime	从首次客户交互开始到最终解决所经过的总时间。
说明此指标衡量服务请求从客户首次联系到问题标记为已解决的总时长，是衡量整体流程效率和客户体验的重要绩效指标。此计算属性是“服务请求解决时间”仪表板和“平均服务解决时间”KPI的核心。分析其分布有助于发现处理时间过长的案例和系统性延误。重要性这是衡量整体流程效率及其对客户体验影响的重要KPI。获取途径计算字段：最终解决活动的时间戳减去首次客户联系活动的时间戳。示例 90010800172800
	服务请求类型 ServiceRequestType	服务请求的分类，例如“咨询”“投诉”或“技术问题”。
说明此属性根据服务请求的性质或目的进行分类，便于细分分析，了解不同类型请求的处理方式。例如，“投诉”与“一般咨询”的流程和解决时间可能大不相同。在“内部升级分析”等仪表板中，此属性有助于判断某些请求类型是否更常被升级。重要性可按请求类型细分流程，比较不同请求的处理方式，并发现特定类型的瓶颈或低效环节。获取途径此信息可能来自IVR选项、客户在网页表单中的选择，也可能由坐席指定。在Genesys中，可将其存储为参与者属性或结束代码。示例账单咨询技术支持账户管理产品投诉

必填建议填写选填

客户服务活动

以下是建议记录在事件日志中的关键流程步骤和里程碑，可用于准确发现流程。

5 建议填写 8 选填

	活动	说明
	交互已转接	表示坐席将交互转接至其他队列或坐席。转接可以是坐席立即断开的盲转，也可以是先与接收方沟通的咨询转接。
重要性此活动是内部升级率这一关键绩效指标的重要依据。较高的转接率可能意味着需要加强培训、调整路由或简化流程。获取途径如果初始坐席参与后，对话详情记录中新增了ACD或坐席参与者，通常即可识别此事件；该变更往往由“transfer”事件触发。采集根据参与者会话数据中的转接事件识别。事件类型 explicit
	交互开始	此活动标记客户服务交互的开始，例如来电、聊天或电子邮件。Genesys Cloud CX会在系统创建新的对话对象时明确记录此事件。
重要性这是服务请求流程的主要开始事件，对于计算总解决时间和了解客户需求模式至关重要。获取途径此事件记录在对话详情中，对应对话对象本身的开始时间，通常可在conversationStart时间戳中找到。采集系统发起新对话时记录。事件类型 explicit
	交互断开	所有参与者断开连接时，对话结束。此活动标记对话结束，通常也代表服务请求在技术层面已关闭。
重要性这是交互本身明确的结束事件，对于计算交互总时长至关重要，也常用作请求关闭的替代指标。获取途径此信息来自对话详情记录，对应对话对象的结束时间，通常可在conversationEnd时间戳中找到。采集所有参与者断开连接、对话结束时记录。事件类型 explicit
	分配结束代码	坐席为交互分配预设的结束代码或处理结果代码，明确标记服务请求的结果，例如“已解决”或“已升级”。
重要性结束代码是判断请求业务结果的主要依据，对于计算解决率和划分案例以供分析至关重要。获取途径这是对话详情记录中的明确事件，通常位于坐席参与者的会话数据中，并记录结束代码及时间戳。采集坐席为交互选择结束代码时记录。事件类型 explicit
	坐席接受交互	标记坐席接受已分配的交互并与客户接通的时刻。这是服务请求开始由坐席直接处理的重要节点。
重要性此活动对于衡量首次联系解决率和坐席处理时间至关重要，也标志着坐席开始处理请求。获取途径可在对话详情记录中找到。坐席参与者的状态变为“connected”时会记录相应时间戳。采集坐席参与者的状态变为“connected”时记录。事件类型 explicit
	交互中开始保持	坐席在交互过程中将客户置于保持状态时，系统会记录此活动。这是对话中坐席参与者状态的明确变更。
重要性分析保持的频率和时长，有助于发现流程低效之处，例如坐席经常需要查找信息或向他人咨询。获取途径此信息来自对话详情记录中的坐席参与者会话数据。状态变为“held”时会记录时间戳。采集坐席参与者的状态变为“held”时记录。事件类型 explicit
	交互结束保持	坐席取消客户的保持状态并恢复对话时会发生此事件。它标记保持时段结束，并作为状态变更记录。
重要性结合“交互中开始保持”活动，可准确计算总保持时间。这是总处理时间和客户体验的重要组成部分。获取途径根据坐席参与者会话数据推断，状态从“held”变回“connected”时即为此事件。采集坐席参与者的状态从“held”变为“connected”时记录。事件类型 explicit
	向坐席发出交互	系统向特定坐席发出交互时会发生此事件。系统等待坐席接受或拒绝对话期间，坐席状态会变为“alerting”。
重要性此活动有助于分析坐席响应速度和路由算法的效果。此后若出现延迟，可能说明坐席未能及时接单。获取途径可在对话详情记录中坐席的参与者数据内找到。参与者会话中将显示“alerting”状态及对应时间戳。采集路由引擎向坐席发出新交互提醒时记录。事件类型 explicit
	对话进入队列	表示新的交互进入特定队列、等待坐席接听的时刻。这是Genesys路由引擎（ACD）明确记录的事件。
重要性跟踪此活动对于衡量队列等待时间和识别路由过程中的瓶颈至关重要。从进入队列到分配坐席之间的间隔较长，可能表明人员配置或路由逻辑存在问题。获取途径此信息来自对话详情记录，具体查看对话中ACD参与者的“purpose”和“state”。指标数组中队列的“enterTime”表示此事件。采集 ACD路由引擎将对话放入队列时记录。事件类型 explicit
	已发送客户满意度调查	交互关闭后发送客户满意度（CSAT）调查时会发生此活动。通常，系统会根据预设规则自动触发调查。
重要性此活动是衡量CSAT调查发送及时性这一关键绩效指标所必需的。及时发送调查有助于在客户仍清楚记得服务体验时收集准确反馈。获取途径此信息通常可从Genesys Cloud CX中的调查或质量管理数据获取，并可关联回原始对话ID。采集调查模块针对特定对话发送调查时记录。事件类型 explicit
	服务请求重新打开	客户在问题被视为已解决后不久，再次就同一问题联系服务中心的情况。此事件不会被明确记录，而是根据业务逻辑计算得出。
重要性跟踪重新打开的请求是衡量服务请求重新打开率的关键。这表明初次解决未能奏效，影响客户满意度和运营效率。获取途径如果某个客户（客户ID）在上一次交互解决后的预设时间范围内，就相关问题（例如服务请求类型）再次触发“交互开始”事件，即可推断此活动。采集将同一客户针对同一问题的新交互与近期关闭的交互关联后计算得出。事件类型 calculated
	通话后处理开始	此活动标记通话后处理（ACW）时段的开始。坐席已与客户断开连接，接下来会处理记录备注或更新系统等任务。
重要性衡量ACW时长有助于了解坐席效率和总处理时间。ACW时间过长可能表明交互后的处理流程效率低下。获取途径此信息来自坐席参与者会话数据。客户参与者断开连接后，坐席状态变为“acw”时会记录此事件。采集坐席参与者的状态变为“acw”时记录。事件类型 explicit
	通话后处理结束	此事件标记坐席通话后处理时段的结束。此时，坐席可以开始处理另一项交互。
重要性结合“通话后处理开始”活动，可准确计算结束处理阶段的时长，帮助分析坐席效率和流程开销。获取途径根据坐席状态从“acw”变为可用状态（例如“idle”）推断，并使用该状态变更的时间戳。采集根据坐席状态从“acw”变为可用状态推断。事件类型 inferred

建议填写选填

数据提取指南

如何从Genesys Cloud CX获取数据

步骤

进入Analytics工作空间：登录Genesys Cloud CX。打开菜单，选择Admin，然后在Analytics下选择Workspace。
创建新视图：点击加号添加新标签页，然后选择“创建新视图”。将视图类型设为“交互”，打开交互视图编辑器。
设置日期范围：在左侧筛选面板中，设置要提取数据的日期和时间范围。建议先选择较短的范围（例如一周）测试流程。
选择导出列：这是数据准备中最关键的一步。点击“列”按钮，从可用字段中选择并添加以下列。这些字段是构建事件日志所需的数据：
- Conversation ID
- Conversation Start
- Conversation End
- Media Type
- Participants（此复杂字段包含嵌套的客服人员、队列和客户数据）
- Metrics.tAnswered
- Metrics.tACW
- Metrics.tHeld
- Surveys
应用其他筛选条件：如有需要，可使用筛选面板缩小数据范围。您可以按Media Type（例如voice、chat）、队列、Direction（inbound/outbound）或其他交互属性筛选，聚焦于特定的服务请求流程。
保存并运行视图：为视图设置描述性名称，例如“服务请求流程挖掘数据导出”。视图随后会加载符合条件的数据。
开始导出：视图加载数据后，点击“导出”按钮（通常显示为下载图标），选择“导出为CSV”或“导出为JSON”。初步处理时通常使用CSV更方便。
下载导出文件：导出任务将在后台运行。您可以前往Performance > My Exports查看进度。任务完成后，将文件下载到本地系统。
转换原始数据：导出文件每行对应一次交互，参与者和分段数据则嵌套在单列中。您需要运行转换脚本（例如Query部分提供的Python脚本）展开这些数据。脚本会解析原始导出文件并生成新文件，每行代表一项活动，例如“客服人员接受交互”或“交互进入保持状态”。
实现自定义逻辑：在转换过程中加入适用于您业务的逻辑。例如，“服务请求重新打开”活动需要根据指定时间范围内同一客户后续的交互进行计算。
生成最终事件日志：转换脚本应输出CSV文件，列名符合ProcessMind所需格式，包括ServiceRequest、ActivityName、EventTime及其他属性。请确保所有时间戳格式一致。
上传到ProcessMind：转换后的最终CSV文件现已准备就绪，可上传到您的流程挖掘工具。

配置

权限：您的Genesys Cloud CX用户角色必须具有Analytics > Conversation > View权限，才能访问Analytics工作空间并创建交互视图。
日期范围：首次提取建议使用具有代表性的3至6个月数据。对于数据量很大的联络中心，建议按周或按月分批导出，以避免触及系统限制并控制文件大小。
关键筛选条件：
- Media Type：筛选与客户服务流程相关的渠道，例如voice、chat或email。
- 队列：如果流程由特定团队处理，请按相关队列ID筛选，确保只提取目标流程的数据。
- Direction：通常建议筛选inbound交互，以捕获由客户发起的服务请求。
导出列：Participants列至关重要，其中包含客服人员和队列的详细事件时间线。缺少此列将无法提取单项活动。请同时确保包含Conversation ID、Conversation Start和Conversation End。
性能注意事项：导出大量列或很长日期范围的数据可能会显著延长导出任务的耗时。如果任务失败或超时，请缩短日期范围后重试。Genesys Cloud每个导出任务最多约可导出100万次交互。

查询示例 python

import pandas as pd
import json
from datetime import datetime, timezone

class=class="hl-string">"hl-comment"># This script transforms a Genesys Cloud Conversation Detail Export into a process mining event log.
class=class="hl-string">"hl-comment"># It assumes the raw data is in a CSV file named class="hl-string">'genesys_export.csv'.

def transform_genesys_export(file_path):
    class=class="hl-string">"hl-comment"># Load the raw export file
    try:
        df = pd.read_csv(file_path)
    except FileNotFoundError:
        print(fclass="hl-string">"Error: The file {file_path} was not found.")
        return None

    event_log = []
    source_system = class="hl-string">'Genesys Cloud CX'
    last_data_update = datetime.now(timezone.utc).isoformat()

    class=class="hl-string">"hl-comment"># Rename columns for easier access
    df.rename(columns={
        class="hl-string">'Conversation ID': class="hl-string">'conversationId',
        class="hl-string">'Conversation Start': class="hl-string">'conversationStart',
        class="hl-string">'Conversation End': class="hl-string">'conversationEnd',
        class="hl-string">'Media Type': class="hl-string">'mediaType'
    }, inplace=True)

    for index, row in df.iterrows():
        service_request_id = row[class="hl-string">'conversationId']
        media_type = row[class="hl-string">'mediaType']
        
        class=class="hl-string">"hl-comment"># Activity: Interaction Started
        event_log.append({
            class="hl-string">'ServiceRequest': service_request_id,
            class="hl-string">'ActivityName': class="hl-string">'Interaction Started',
            class="hl-string">'EventTime': row[class="hl-string">'conversationStart'],
            class="hl-string">'EventEndTime': None,
            class="hl-string">'AgentId': None, class="hl-string">'AgentName': None, class="hl-string">'QueueName': None, class="hl-string">'WrapUpCode': None,
            class="hl-string">'MediaType': media_type, class="hl-string">'SourceSystem': source_system, class="hl-string">'LastDataUpdate': last_data_update
        })

        class=class="hl-string">"hl-comment"># The class="hl-string">'Participants' column contains a JSON string that needs to be parsed
        participants = json.loads(row[class="hl-string">'Participants'])

        for p in participants:
            participant_id = p.get(class="hl-string">'participantId')
            user_id = p.get(class="hl-string">'userId')
            queue_name = p.get(class="hl-string">'queueName')
            purpose = p.get(class="hl-string">'purpose')

            for s in p.get(class="hl-string">'sessions', []):
                wrap_up_code = s.get(class="hl-string">'wrapUpCode')
                for seg in s.get(class="hl-string">'segments', []):
                    seg_start = seg.get(class="hl-string">'segmentStart')
                    seg_end = seg.get(class="hl-string">'segmentEnd')
                    seg_type = seg.get(class="hl-string">'segmentType')

                    activity_name = None
                    if purpose == class="hl-string">'acd' and seg_type == class="hl-string">'interact':
                        activity_name = class="hl-string">'Conversation Routed to Queue'
                    elif purpose == class="hl-string">'agent' and seg_type == class="hl-string">'alert':
                        activity_name = class="hl-string">'Agent Offered Interaction'
                    elif purpose == class="hl-string">'agent' and seg_type == class="hl-string">'interacting':
                        activity_name = class="hl-string">'Agent Accepted Interaction'
                    elif purpose == class="hl-string">'agent' and seg_type == class="hl-string">'hold':
                        activity_name = class="hl-string">'Hold Placed on Interaction'
                        class=class="hl-string">"hl-comment"># Also create the Hold Removed event
                        event_log.append({
                            class="hl-string">'ServiceRequest': service_request_id,
                            class="hl-string">'ActivityName': class="hl-string">'Hold Removed from Interaction',
                            class="hl-string">'EventTime': seg_end,
                            class="hl-string">'EventEndTime': None,
                            class="hl-string">'AgentId': user_id, class="hl-string">'AgentName': class="hl-string">'[Configure based on your system]', class="hl-string">'QueueName': queue_name, class="hl-string">'WrapUpCode': None,
                            class="hl-string">'MediaType': media_type, class="hl-string">'SourceSystem': source_system, class="hl-string">'LastDataUpdate': last_data_update
                        })
                    elif purpose == class="hl-string">'agent' and seg_type == class="hl-string">'acw':
                        activity_name = class="hl-string">'After-Call Work Started'
                        class=class="hl-string">"hl-comment"># Also create ACW Ended and Wrap-Up Assigned events
                        if wrap_up_code:
                            event_log.append({
                                class="hl-string">'ServiceRequest': service_request_id,
                                class="hl-string">'ActivityName': class="hl-string">'Wrap-Up Code Assigned',
                                class="hl-string">'EventTime': seg_end, class=class="hl-string">"hl-comment"># Assume code assigned at end of ACW
                                class="hl-string">'EventEndTime': None,
                                class="hl-string">'AgentId': user_id, class="hl-string">'AgentName': class="hl-string">'[Configure based on your system]', class="hl-string">'QueueName': queue_name, class="hl-string">'WrapUpCode': wrap_up_code,
                                class="hl-string">'MediaType': media_type, class="hl-string">'SourceSystem': source_system, class="hl-string">'LastDataUpdate': last_data_update
                            })
                        event_log.append({
                            class="hl-string">'ServiceRequest': service_request_id,
                            class="hl-string">'ActivityName': class="hl-string">'After-Call Work Ended',
                            class="hl-string">'EventTime': seg_end,
                            class="hl-string">'EventEndTime': None,
                            class="hl-string">'AgentId': user_id, class="hl-string">'AgentName': class="hl-string">'[Configure based on your system]', class="hl-string">'QueueName': queue_name, class="hl-string">'WrapUpCode': wrap_up_code,
                            class="hl-string">'MediaType': media_type, class="hl-string">'SourceSystem': source_system, class="hl-string">'LastDataUpdate': last_data_update
                        })
                    
                    if seg.get(class="hl-string">'disconnectType') == class="hl-string">'transfer':
                        activity_name = class="hl-string">'Interaction Transferred'

                    if activity_name:
                        event_log.append({
                            class="hl-string">'ServiceRequest': service_request_id,
                            class="hl-string">'ActivityName': activity_name,
                            class="hl-string">'EventTime': seg_start,
                            class="hl-string">'EventEndTime': seg_end,
                            class="hl-string">'AgentId': user_id, class="hl-string">'AgentName': class="hl-string">'[Configure based on your system]', class="hl-string">'QueueName': queue_name, class="hl-string">'WrapUpCode': None,
                            class="hl-string">'MediaType': media_type, class="hl-string">'SourceSystem': source_system, class="hl-string">'LastDataUpdate': last_data_update
                        })
        
        class=class="hl-string">"hl-comment"># Activity: Interaction Disconnected
        event_log.append({
            class="hl-string">'ServiceRequest': service_request_id,
            class="hl-string">'ActivityName': class="hl-string">'Interaction Disconnected',
            class="hl-string">'EventTime': row[class="hl-string">'conversationEnd'],
            class="hl-string">'EventEndTime': None,
            class="hl-string">'AgentId': None, class="hl-string">'AgentName': None, class="hl-string">'QueueName': None, class="hl-string">'WrapUpCode': None,
            class="hl-string">'MediaType': media_type, class="hl-string">'SourceSystem': source_system, class="hl-string">'LastDataUpdate': last_data_update
        })

        class=class="hl-string">"hl-comment"># Activity: Customer Satisfaction Survey Sent (from surveys column)
        if pd.notna(row.get(class="hl-string">'Surveys')):
            surveys = json.loads(row[class="hl-string">'Surveys'])
            for survey in surveys:
                if survey.get(class="hl-string">'surveyStatus') == class="hl-string">'Sent':
                    class=class="hl-string">"hl-comment"># Survey sent time is not always directly available, using conversation end as an approximation
                    event_log.append({
                        class="hl-string">'ServiceRequest': service_request_id,
                        class="hl-string">'ActivityName': class="hl-string">'Customer Satisfaction Survey Sent',
                        class="hl-string">'EventTime': survey.get(class="hl-string">'surveySentDate', row[class="hl-string">'conversationEnd']),
                        class="hl-string">'EventEndTime': None,
                        class="hl-string">'AgentId': None, class="hl-string">'AgentName': None, class="hl-string">'QueueName': None, class="hl-string">'WrapUpCode': None,
                        class="hl-string">'MediaType': media_type, class="hl-string">'SourceSystem': source_system, class="hl-string">'LastDataUpdate': last_data_update
                    })

    class=class="hl-string">"hl-comment"># NOTE: class="hl-string">'Service Request Re-Opened' requires custom cross-conversation logic.
    class=class="hl-string">"hl-comment"># This logic would need to be implemented here, analyzing the entire dataset
    class=class="hl-string">"hl-comment"># to find subsequent interactions from the same customer within a specific timeframe.

    final_df = pd.DataFrame(event_log)
    class=class="hl-string">"hl-comment"># Sort by case and event time to create a chronological log
    final_df.sort_values(by=[class="hl-string">'ServiceRequest', class="hl-string">'EventTime'], inplace=True)
    
    return final_df

if __name__ == class="hl-string">'__main__':
    class=class="hl-string">"hl-comment"># Specify the path to your downloaded Genesys export CSV file
    input_file = class="hl-string">'genesys_export.csv'
    output_file = class="hl-string">'genesys_event_log.csv'
    
    event_log_df = transform_genesys_export(input_file)
    
    if event_log_df is not None:
        event_log_df.to_csv(output_file, index=False)
        print(fclass="hl-string">"Transformation complete. Event log saved to {output_file}")

步骤

准备工作：确保已安装Python 3.8+以及pandas和purecloud-platform-client-v2库。可使用pip安装：pip install pandas purecloud-platform-client-v2。
创建Genesys Cloud OAuth客户端：登录Genesys Cloud CX管理员账户。前往Integrations > OAuth，创建一个新的客户端，并将授权类型设为'Client Credentials'。分配包含analytics:conversationDetail:view权限的角色。
配置脚本凭据：保存上一步获取的Client ID和Client Secret。在下方提供的Python脚本中，将CLIENT_ID、CLIENT_SECRET和GENESYS_CLOUD_REGION的占位值替换为您的凭据。
设置提取时间范围：在脚本中找到query_body变量，修改interval值以指定要提取数据的日期和时间范围。建议先用较短范围（例如一天）进行测试。
运行Python脚本：在终端中运行python your_script_name.py。脚本将通过Genesys Cloud API进行身份验证，提交异步分析任务，并输出任务ID。
监控任务运行情况：脚本每15秒自动轮询API以检查任务状态，状态更新会显示在控制台中。根据请求的数据量，此步骤可能需要数分钟或更长时间。
处理和转换数据：任务完成后，脚本会下载生成的JSON数据，并逐条处理交互记录，解析参与者、会话和分段数据，以识别并整理关键活动和属性。
处理计算得出的事件：脚本包含一个后处理步骤，用于识别“服务请求重新打开”事件。您可以自定义此逻辑，它会检查上一次交互解决后指定时间范围内，同一客户是否发起了新交互。
生成事件日志：转换后的数据会汇总到pandas DataFrame中，其结构与流程挖掘事件日志一致。
导出最终CSV文件：脚本会将最终DataFrame保存为genesys_event_log.csv。此文件格式可直接上传到ProcessMind平台。

配置

API凭据：身份验证需要OAuth Client ID和Client Secret。必须在Genesys Cloud CX管理界面中创建这些凭据。
API权限：关联的OAuth客户端角色必须具有analytics:conversationDetail:view权限，才能访问所需的交互数据。
提取时间范围：API查询正文中的interval参数用于控制提取日期范围。对于大型环境，建议按较小时间段依次分批提取（例如每次一周），以避免API超时并控制内存使用。
数据筛选条件：您可以修改API查询正文，加入conversationFilters、segmentFilters或evaluationFilters以缩小提取范围。例如，按特定的queueId或mediaType筛选，聚焦于客户服务流程的特定部分。
重新打开逻辑：脚本中可配置用于识别“服务请求重新打开”事件的时间范围。默认值为72小时，您应根据组织的业务规则进行调整。
环境：运行脚本需要本地Python环境及所需的SDK和库。无需直接访问Genesys Cloud CX服务器或数据库。

查询示例 python

import PureCloudPlatformClientV2
import pandas as pd
import requests
import json
import time
import os
from datetime import datetime, timedelta

class=class="hl-string">"hl-comment"># --- Configuration ---
class=class="hl-string">"hl-comment"># Replace with your Genesys Cloud credentials and region
CLIENT_ID = class="hl-string">"[Your_Client_Id]"
CLIENT_SECRET = class="hl-string">"[Your_Client_Secret]"
GENESYS_CLOUD_REGION = class="hl-string">"[Your_Genesys_Cloud_Region]" class=class="hl-string">"hl-comment"># e.g., mypurecloud.com, mypurecloud.de

class=class="hl-string">"hl-comment"># Set the API environment
PureCloudPlatformClientV2.configuration.host = fclass="hl-string">'https://api.{GENESYS_CLOUD_REGION}'

class=class="hl-string">"hl-comment"># --- Authentication ---
api_client = PureCloudPlatformClientV2.api_client.ApiClient().get_client_credentials_token(CLIENT_ID, CLIENT_SECRET)
analytics_api = PureCloudPlatformClientV2.AnalyticsApi(api_client)


def parse_conversation(conv_detail):
    class="hl-string">"""Parses a single conversation detail object into multiple event log rows."""
    events = []
    conversation_id = conv_detail.get(class="hl-string">'conversationId')
    source_system = class="hl-string">'Genesys Cloud CX'
    last_data_update = datetime.utcnow().isoformat() + class="hl-string">'Z'

    class=class="hl-string">"hl-comment"># Event: Interaction Started
    events.append({
        class="hl-string">'ServiceRequest': conversation_id,
        class="hl-string">'ActivityName': class="hl-string">'Interaction Started',
        class="hl-string">'EventTime': conv_detail.get(class="hl-string">'conversationStart'),
        class="hl-string">'EventEndTime': None,
        class="hl-string">'AgentId': None, class="hl-string">'AgentName': None, class="hl-string">'QueueName': None, class="hl-string">'WrapUpCode': None,
        class="hl-string">'MediaType': next((p.get(class="hl-string">'sessions')[0].get(class="hl-string">'mediaType') for p in conv_detail.get(class="hl-string">'participants', []) if p.get(class="hl-string">'purpose') == class="hl-string">'customer'), None),
        class="hl-string">'SourceSystem': source_system, class="hl-string">'LastDataUpdate': last_data_update
    })

    for p in conv_detail.get(class="hl-string">'participants', []):
        agent_id = p.get(class="hl-string">'userId')
        agent_name = p.get(class="hl-string">'participantName')

        for s in p.get(class="hl-string">'sessions', []):
            media_type = s.get(class="hl-string">'mediaType')

            for seg in s.get(class="hl-string">'segments', []):
                seg_start = seg.get(class="hl-string">'segmentStart')
                seg_end = seg.get(class="hl-string">'segmentEnd')

                if seg.get(class="hl-string">'segmentType') == class="hl-string">'alert':
                    events.append({
                        class="hl-string">'ServiceRequest': conversation_id, class="hl-string">'ActivityName': class="hl-string">'Agent Offered Interaction', class="hl-string">'EventTime': seg_start, class="hl-string">'EventEndTime': seg_end,
                        class="hl-string">'AgentId': agent_id, class="hl-string">'AgentName': agent_name, class="hl-string">'QueueName': s.get(class="hl-string">'queueName'), class="hl-string">'WrapUpCode': None, class="hl-string">'MediaType': media_type,
                        class="hl-string">'SourceSystem': source_system, class="hl-string">'LastDataUpdate': last_data_update
                    })

                if seg.get(class="hl-string">'segmentType') == class="hl-string">'interact' and p.get(class="hl-string">'purpose') == class="hl-string">'agent':
                    events.append({
                        class="hl-string">'ServiceRequest': conversation_id, class="hl-string">'ActivityName': class="hl-string">'Agent Accepted Interaction', class="hl-string">'EventTime': seg_start, class="hl-string">'EventEndTime': seg_end,
                        class="hl-string">'AgentId': agent_id, class="hl-string">'AgentName': agent_name, class="hl-string">'QueueName': s.get(class="hl-string">'queueName'), class="hl-string">'WrapUpCode': None, class="hl-string">'MediaType': media_type,
                        class="hl-string">'SourceSystem': source_system, class="hl-string">'LastDataUpdate': last_data_update
                    })

                if seg.get(class="hl-string">'segmentType') == class="hl-string">'hold':
                    events.append({
                        class="hl-string">'ServiceRequest': conversation_id, class="hl-string">'ActivityName': class="hl-string">'Hold Placed on Interaction', class="hl-string">'EventTime': seg_start, class="hl-string">'EventEndTime': None,
                        class="hl-string">'AgentId': agent_id, class="hl-string">'AgentName': agent_name, class="hl-string">'QueueName': s.get(class="hl-string">'queueName'), class="hl-string">'WrapUpCode': None, class="hl-string">'MediaType': media_type,
                        class="hl-string">'SourceSystem': source_system, class="hl-string">'LastDataUpdate': last_data_update
                    })
                    if seg_end:
                        events.append({
                            class="hl-string">'ServiceRequest': conversation_id, class="hl-string">'ActivityName': class="hl-string">'Hold Removed from Interaction', class="hl-string">'EventTime': seg_end, class="hl-string">'EventEndTime': None,
                            class="hl-string">'AgentId': agent_id, class="hl-string">'AgentName': agent_name, class="hl-string">'QueueName': s.get(class="hl-string">'queueName'), class="hl-string">'WrapUpCode': None, class="hl-string">'MediaType': media_type,
                            class="hl-string">'SourceSystem': source_system, class="hl-string">'LastDataUpdate': last_data_update
                        })

                if seg.get(class="hl-string">'segmentType') == class="hl-string">'acw':
                    events.append({
                        class="hl-string">'ServiceRequest': conversation_id, class="hl-string">'ActivityName': class="hl-string">'After-Call Work Started', class="hl-string">'EventTime': seg_start, class="hl-string">'EventEndTime': seg_end,
                        class="hl-string">'AgentId': agent_id, class="hl-string">'AgentName': agent_name, class="hl-string">'QueueName': s.get(class="hl-string">'queueName'), class="hl-string">'WrapUpCode': p.get(class="hl-string">'wrapUp', {}).get(class="hl-string">'code'), class="hl-string">'MediaType': media_type,
                        class="hl-string">'SourceSystem': source_system, class="hl-string">'LastDataUpdate': last_data_update
                    })
                    if seg_end:
                         events.append({
                            class="hl-string">'ServiceRequest': conversation_id, class="hl-string">'ActivityName': class="hl-string">'After-Call Work Ended', class="hl-string">'EventTime': seg_end, class="hl-string">'EventEndTime': None,
                            class="hl-string">'AgentId': agent_id, class="hl-string">'AgentName': agent_name, class="hl-string">'QueueName': s.get(class="hl-string">'queueName'), class="hl-string">'WrapUpCode': p.get(class="hl-string">'wrapUp', {}).get(class="hl-string">'code'), class="hl-string">'MediaType': media_type,
                            class="hl-string">'SourceSystem': source_system, class="hl-string">'LastDataUpdate': last_data_update
                        })
                
                if seg.get(class="hl-string">'segmentType') == class="hl-string">'transmitting' and seg.get(class="hl-string">'disconnectType') == class="hl-string">'transfer':
                    events.append({
                        class="hl-string">'ServiceRequest': conversation_id, class="hl-string">'ActivityName': class="hl-string">'Interaction Transferred', class="hl-string">'EventTime': seg_start, class="hl-string">'EventEndTime': None,
                        class="hl-string">'AgentId': agent_id, class="hl-string">'AgentName': agent_name, class="hl-string">'QueueName': s.get(class="hl-string">'queueName'), class="hl-string">'WrapUpCode': None, class="hl-string">'MediaType': media_type,
                        class="hl-string">'SourceSystem': source_system, class="hl-string">'LastDataUpdate': last_data_update
                    })
            
            if s.get(class="hl-string">'flow', {}).get(class="hl-string">'outcomeValue') == class="hl-string">'SURVEY_OFFERED':
                events.append({
                    class="hl-string">'ServiceRequest': conversation_id, class="hl-string">'ActivityName': class="hl-string">'Customer Satisfaction Survey Sent', class="hl-string">'EventTime': s.get(class="hl-string">'flow', {}).get(class="hl-string">'flowEnd'), class="hl-string">'EventEndTime': None,
                    class="hl-string">'AgentId': None, class="hl-string">'AgentName': None, class="hl-string">'QueueName': None, class="hl-string">'WrapUpCode': None, class="hl-string">'MediaType': media_type,
                    class="hl-string">'SourceSystem': source_system, class="hl-string">'LastDataUpdate': last_data_update
                })

        if p.get(class="hl-string">'purpose') == class="hl-string">'acd':
             for s in p.get(class="hl-string">'sessions', []):
                for seg in s.get(class="hl-string">'segments', []):
                    if seg.get(class="hl-string">'segmentType') == class="hl-string">'interact':
                         events.append({
                            class="hl-string">'ServiceRequest': conversation_id, class="hl-string">'ActivityName': class="hl-string">'Conversation Routed to Queue', class="hl-string">'EventTime': seg.get(class="hl-string">'segmentStart'), class="hl-string">'EventEndTime': seg.get(class="hl-string">'segmentEnd'),
                            class="hl-string">'AgentId': None, class="hl-string">'AgentName': None, class="hl-string">'QueueName': s.get(class="hl-string">'queueName'), class="hl-string">'WrapUpCode': None, class="hl-string">'MediaType': s.get(class="hl-string">'mediaType'),
                            class="hl-string">'SourceSystem': source_system, class="hl-string">'LastDataUpdate': last_data_update
                        })

        if p.get(class="hl-string">'wrapUp'):
            events.append({
                class="hl-string">'ServiceRequest': conversation_id, class="hl-string">'ActivityName': class="hl-string">'Wrap-Up Code Assigned', class="hl-string">'EventTime': p[class="hl-string">'wrapUp'].get(class="hl-string">'wrapUpEnd'), class="hl-string">'EventEndTime': None,
                class="hl-string">'AgentId': agent_id, class="hl-string">'AgentName': agent_name, class="hl-string">'QueueName': s.get(class="hl-string">'queueName'), class="hl-string">'WrapUpCode': p[class="hl-string">'wrapUp'].get(class="hl-string">'code'), class="hl-string">'MediaType': s.get(class="hl-string">'mediaType'),
                class="hl-string">'SourceSystem': source_system, class="hl-string">'LastDataUpdate': last_data_update
            })

    class=class="hl-string">"hl-comment"># Event: Interaction Disconnected
    events.append({
        class="hl-string">'ServiceRequest': conversation_id, class="hl-string">'ActivityName': class="hl-string">'Interaction Disconnected', class="hl-string">'EventTime': conv_detail.get(class="hl-string">'conversationEnd'), class="hl-string">'EventEndTime': None,
        class="hl-string">'AgentId': None, class="hl-string">'AgentName': None, class="hl-string">'QueueName': None, class="hl-string">'WrapUpCode': None,
        class="hl-string">'MediaType': next((p.get(class="hl-string">'sessions')[0].get(class="hl-string">'mediaType') for p in conv_detail.get(class="hl-string">'participants', []) if p.get(class="hl-string">'purpose') == class="hl-string">'customer'), None),
        class="hl-string">'SourceSystem': source_system, class="hl-string">'LastDataUpdate': last_data_update
    })
    
    return events


def main():
    class=class="hl-string">"hl-comment"># --- Define Analytics Query ---
    class=class="hl-string">"hl-comment"># Define your time interval here. Format: YYYY-MM-DDTHH:MM:SS.sssZ/YYYY-MM-DDTHH:MM:SS.sssZ
    interval = class="hl-string">"2023-10-01T00:00:00.000Z/2023-10-02T00:00:00.000Z"
    
    query_body = {
        class="hl-string">"interval": interval,
        class="hl-string">"paging": {
            class="hl-string">"pageSize": 100,
            class="hl-string">"pageNumber": 1
        }
    }

    class=class="hl-string">"hl-comment"># --- Submit and Monitor Job ---
    try:
        print(fclass="hl-string">"Submitting analytics job for interval: {interval}")
        api_response = analytics_api.post_analytics_conversations_details_jobs(query_body)
        job_id = api_response.job_id
        print(fclass="hl-string">"Job submitted successfully. Job ID: {job_id}")

        while True:
            job_status = analytics_api.get_analytics_conversations_details_job(job_id)
            print(fclass="hl-string">"Job status: {job_status.state}. Progress: {job_status.percent_complete}%")
            if job_status.state == class="hl-string">'FULFILLED':
                print(class="hl-string">"Job completed successfully.")
                break
            elif job_status.state in [class="hl-string">'FAILED', class="hl-string">'CANCELLED']:
                print(fclass="hl-string">"Job failed or was cancelled. State: {job_status.state}, Error: {job_status.error}")
                return
            time.sleep(15)

        class=class="hl-string">"hl-comment"># --- Fetch and Process Results ---
        all_events = []
        results_response = analytics_api.get_analytics_conversations_details_job_results(job_id)
        
        if not results_response.conversations:
            print(class="hl-string">"No conversations found for the given interval.")
            return
            
        print(fclass="hl-string">"Processing {len(results_response.conversations)} conversations...")
        for conv in results_response.conversations:
            all_events.extend(parse_conversation(conv))

        df = pd.DataFrame(all_events)
        df[class="hl-string">'EventTime'] = pd.to_datetime(df[class="hl-string">'EventTime'], errors=class="hl-string">'coerce')
        df = df.dropna(subset=[class="hl-string">'EventTime'])
        df = df.sort_values(by=[class="hl-string">'ServiceRequest', class="hl-string">'EventTime']).reset_index(drop=True)

        class=class="hl-string">"hl-comment"># --- Post-Processing for class="hl-string">'Service Request Re-Opened' ---
        class=class="hl-string">"hl-comment"># This logic is an example. Adjust the window and criteria to fit your business rules.
        print(class="hl-string">"Applying post-processing for 'Re-Opened' events...")
        reopen_events = []
        df_resolved = df[df[class="hl-string">'WrapUpCode'].str.contains(class="hl-string">'Resolved', na=False)].copy()
        df_resolved = df_resolved.sort_values(class="hl-string">'EventTime').drop_duplicates(class="hl-string">'ServiceRequest', keep=class="hl-string">'last')
        reopen_window = timedelta(hours=72)

        for index, row in df_resolved.iterrows():
            customer_id = next((p.get(class="hl-string">'participantName') for c in results_response.conversations if c.get(class="hl-string">'conversationId') == row[class="hl-string">'ServiceRequest'] for p in c.get(class="hl-string">'participants', []) if p.get(class="hl-string">'purpose') == class="hl-string">'customer'), None)
            if not customer_id: continue
            
            resolved_time = row[class="hl-string">'EventTime']
            future_interactions = df[(df[class="hl-string">'EventTime'] > resolved_time) & (df[class="hl-string">'EventTime'] <= resolved_time + reopen_window)]
            
            for _, future_row in future_interactions.iterrows():
                future_customer_id = next((p.get(class="hl-string">'participantName') for c in results_response.conversations if c.get(class="hl-string">'conversationId') == future_row[class="hl-string">'ServiceRequest'] for p in c.get(class="hl-string">'participants', []) if p.get(class="hl-string">'purpose') == class="hl-string">'customer'), None)
                if customer_id == future_customer_id and future_row[class="hl-string">'ActivityName'] == class="hl-string">'Interaction Started':
                    reopen_events.append({
                        class="hl-string">'ServiceRequest': row[class="hl-string">'ServiceRequest'], class="hl-string">'ActivityName': class="hl-string">'Service Request Re-Opened', class="hl-string">'EventTime': future_row[class="hl-string">'EventTime'],
                        class="hl-string">'SourceSystem': class="hl-string">'Genesys Cloud CX', class="hl-string">'LastDataUpdate': datetime.utcnow().isoformat() + class="hl-string">'Z'
                    })
                    break

        if reopen_events:
            df_reopen = pd.DataFrame(reopen_events)
            df = pd.concat([df, df_reopen], ignore_index=True)
            df = df.sort_values(by=[class="hl-string">'ServiceRequest', class="hl-string">'EventTime']).reset_index(drop=True)
        
        class=class="hl-string">"hl-comment"># --- Save to CSV ---
        output_filename = class="hl-string">'genesys_event_log.csv'
        df.to_csv(output_filename, index=False, date_format=class="hl-string">'%Y-%m-%dT%H:%M:%S.%fZ')
        print(fclass="hl-string">"Extraction complete. Event log saved to {output_filename}")

    except PureCloudPlatformClientV2.ApiException as e:
        print(fclass="hl-string">"Error calling Genesys Cloud API: {e}")
    except Exception as e:
        print(fclass="hl-string">"An unexpected error occurred: {e}")

if __name__ == class="hl-string">'__main__':
    main()

步骤

确认您的Genesys Cloud CX组织具备所需的EventBridge集成能力、相应管理员权限和交互事件访问权限，并且AWS账户有权接收事件。与系统负责人确认目标AWS Region、EventBridge事件总线，以及组织、部门、队列或媒体筛选条件。
在Genesys Cloud CX中，根据当前管理界面和组织的集成文档配置AWS EventBridge集成。选择目标AWS账户、Region、事件总线，以及详细监控交互生命周期所需的交互事件类别。Genesys Cloud菜单名称和可用事件架构可能因组织和版本而异；组织专属的操作路径或设置请使用[根据您的系统配置]。
配置AWS EventBridge，将Genesys Cloud事件接收到所选事件总线。适用时，设置最小权限、加密、保留策略、监控和死信目标。创建规则，将所需事件转发到数据湖、数据仓库、流处理器或对象存储接收区所使用的使用方。
捕获原始EventBridge负载，不要更改原始时间戳或标识符。在[您的原始事件存储]中保存负载、接收时间戳、事件类型、来源、详细类型和传递元数据。保留原始负载，以便Genesys Cloud架构变更时重新验证映射。
使用Python构建标准化步骤，读取原始负载并识别交互标识符、参与者和会话详情、队列信息、客服人员信息、媒体类型、结束后处理数据及事件时间戳，然后为每项明确记录的活动输出一行。请使用您组织收到的事件样本中的实际字段路径。不要推断未由明确事件或状态转换表示的活动。
将明确的Genesys Cloud交互和参与者状态变更映射到所需的ActivityName值。存在交互创建事件时输出Interaction Started；存在路由事件时输出Conversation Routed to Queue；客服人员状态为响铃时输出Agent Offered Interaction；客服人员接受或接通时输出Agent Accepted Interaction；发生相应保持状态转换时输出Hold Placed on Interaction和Hold Removed from Interaction；存在明确转接事件时输出Interaction Transferred；存在明确ACW转换时输出After-Call Work Started和After-Call Work Ended；存在结束后处理代码分配时输出Wrap-Up Code Assigned；交互或相关的最终参与者会话断开时输出Interaction Disconnected。
分别处理Service Request Re-Opened和Customer Satisfaction Survey Sent。EventBridge交互事件不一定会直接提供这两项活动。如果组织有明确的调查事件来源，仅在收到该事件时纳入。Service Request Re-Opened只能通过单独且有文档记录的业务规则流程计算，然后将结果行追加到事件日志。ProcessMind不会计算或推断此活动。
将业务映射选定的稳定服务请求或交互标识符填入ServiceRequest。ActivityName填写且仅填写一个所需活动标签，EventTime填写活动开始时间戳；如有EventEndTime，则一并填写；SourceSystem填写Genesys Cloud CX和AWS EventBridge；LastDataUpdate填写提取刷新时间戳。存在可选属性时填写，否则留空（NULL）。
如有提供方事件标识符，请据此去重；否则使用有文档记录的复合键，其中包含ServiceRequest、ActivityName、EventTime、参与者或客服人员标识符，以及来源事件标识符。处理EventBridge重试和乱序传递。不要删除不同的重复保持、转接、呼叫邀请或其他有效重复活动。
根据原始负载数量、事件时间戳、交互标识符、队列和客服人员详情以及完整活动列表，验证标准化输出。导出平面文件或加载包含必需列ServiceRequest、ActivityName、EventTime、SourceSystem和LastDataUpdate的表，并在可用时加入推荐列。确保时间戳采用ProcessMind接受的一致时区和格式，然后将事件日志作为服务请求流程的输入上传。

配置

集成和目标位置：为目标AWS账户、Region和事件总线配置Genesys Cloud CX AWS EventBridge集成。对于可用事件样本未定义的组织专属集成设置，请使用[根据您的系统配置]。
事件范围：订阅Genesys Cloud组织中可用的详细交互、参与者、会话、路由、状态、转接、断开、结束后处理和ACW事件。正式部署前，先通过受控测试交互确认确切的事件类型和负载路径。
案例标识符：将业务映射选定的稳定交互或服务请求标识符用作ServiceRequest。如果其他系统中存在单独的服务请求标识符，请维护该标识符与Genesys Cloud交互标识符之间有文档记录的交叉引用。
日期范围：流程分析可从3至6个月的历史数据开始。近实时运行时，使用较短的轮询或流处理检查点间隔，并保留可重放的原始数据接收层。发生中断后恢复时使用重叠时间范围，再按提供方事件标识符去重。
筛选条件：仅应用已确认的组织专属筛选条件，例如部门、队列、媒体类型、业务单元或交互日期。除非集成数据源提供了Company Code或Document Type概念，且已在您的环境中明确映射，否则不要按这些条件筛选。
时间戳处理：EventTime保留来源事件时间戳。接收或刷新时间单独存入LastDataUpdate。将所有时间戳标准化为有文档记录的时区，建议使用UTC；如审计需要，则保留原始时区偏移。
可选属性：AgentId和AgentName可从参与者或用户详情中获取，QueueName可从路由数据中获取，WrapUpCode可从结束后处理数据中获取，MediaType可从交互或会话渠道中获取，EventEndTime可根据明确的结束状态转换或来源时长填写。
性能：按事件日期对原始数据和标准化数据分区；如支持，也可按组织或部门分区。使用EventBridge检查点增量处理。分批写入数据仓库或数据湖，监控延迟，避免反复扫描全部历史数据。
可靠性：配置重试处理、死信处理、重放流程、重复检测，并监控传递失败、架构变更和事件延迟。
前提条件：需要Genesys Cloud管理员或集成权限、AWS EventBridge权限、所选事件总线和下游存储的访问权限、读取交互详情的授权，以及组织事件集成所需的Genesys Cloud功能或许可证。存储交互数据前，请确认数据保留和隐私要求。

查询示例 python

import json
import os
from datetime import datetime, timezone
from typing import Any, Dict, Iterable, List, Optional

import boto3


SOURCE_SYSTEM = class="hl-string">"Genesys Cloud CX via AWS EventBridge"
ACTIVITY_NAMES = {
    class="hl-string">"interaction_started": class="hl-string">"Interaction Started",
    class="hl-string">"conversation_routed_to_queue": class="hl-string">"Conversation Routed to Queue",
    class="hl-string">"agent_offered_interaction": class="hl-string">"Agent Offered Interaction",
    class="hl-string">"agent_accepted_interaction": class="hl-string">"Agent Accepted Interaction",
    class="hl-string">"hold_placed_on_interaction": class="hl-string">"Hold Placed on Interaction",
    class="hl-string">"hold_removed_from_interaction": class="hl-string">"Hold Removed from Interaction",
    class="hl-string">"interaction_transferred": class="hl-string">"Interaction Transferred",
    class="hl-string">"after_call_work_started": class="hl-string">"After-Call Work Started",
    class="hl-string">"wrap_up_code_assigned": class="hl-string">"Wrap-Up Code Assigned",
    class="hl-string">"after_call_work_ended": class="hl-string">"After-Call Work Ended",
    class="hl-string">"interaction_disconnected": class="hl-string">"Interaction Disconnected",
    class="hl-string">"service_request_re_opened": class="hl-string">"Service Request Re-Opened",
    class="hl-string">"customer_satisfaction_survey_sent": class="hl-string">"Customer Satisfaction Survey Sent",
}


def first_value(data: Dict[str, Any], paths: Iterable[str]) -> Optional[Any]:
    for path in paths:
        value: Any = data
        found = True
        for part in path.split(class="hl-string">"."):
            if isinstance(value, dict) and part in value:
                value = value[part]
            else:
                found = False
                break
        if found and value not in (None, class="hl-string">""):
            return value
    return None


def parse_time(value: Any) -> Optional[str]:
    if value is None:
        return None
    if isinstance(value, (int, float)):
        return datetime.fromtimestamp(value / 1000 if value > 100000000000 else value, tz=timezone.utc).isoformat()
    text = str(value).strip()
    if not text:
        return None
    return text


def event_detail(event: Dict[str, Any]) -> Dict[str, Any]:
    detail = event.get(class="hl-string">"detail")
    if isinstance(detail, dict):
        return detail
    if isinstance(detail, str):
        try:
            parsed = json.loads(detail)
            return parsed if isinstance(parsed, dict) else {}
        except json.JSONDecodeError:
            return {}
    return {}


def event_type(event: Dict[str, Any], detail: Dict[str, Any]) -> str:
    value = first_value(detail, [
        class="hl-string">"eventType", class="hl-string">"type", class="hl-string">"event.type", class="hl-string">"topicName", class="hl-string">"name", class="hl-string">"state", class="hl-string">"participant.state", class="hl-string">"session.state"
    ])
    return str(value or event.get(class="hl-string">"detail-type") or event.get(class="hl-string">"source") or class="hl-string">"").lower().replace(class="hl-string">"_", class="hl-string">" ")


def base_row(detail: Dict[str, Any], activity: str, event: Dict[str, Any], refresh_time: str) -> Dict[str, Any]:
    service_request = first_value(detail, [
        class="hl-string">"conversationId", class="hl-string">"conversation.id", class="hl-string">"conversation.conversationId", class="hl-string">"serviceRequest", class="hl-string">"serviceRequestId", class="hl-string">"caseId"
    ])
    event_time = parse_time(first_value(detail, [
        class="hl-string">"eventTime", class="hl-string">"timestamp", class="hl-string">"event.timestamp", class="hl-string">"conversation.startTime", class="hl-string">"participant.startTime", class="hl-string">"session.startTime", class="hl-string">"startTime"
    ]))
    event_end_time = parse_time(first_value(detail, [
        class="hl-string">"eventEndTime", class="hl-string">"endTime", class="hl-string">"event.endTime", class="hl-string">"conversation.endTime", class="hl-string">"participant.endTime", class="hl-string">"session.endTime"
    ]))
    agent_id = first_value(detail, [class="hl-string">"agentId", class="hl-string">"userId", class="hl-string">"agent.id", class="hl-string">"user.id", class="hl-string">"participant.userId", class="hl-string">"participant.user.id"])
    agent_name = first_value(detail, [class="hl-string">"agentName", class="hl-string">"userName", class="hl-string">"agent.name", class="hl-string">"user.name", class="hl-string">"participant.userName", class="hl-string">"participant.user.name"])
    queue_name = first_value(detail, [class="hl-string">"queueName", class="hl-string">"queue.name", class="hl-string">"routing.queueName", class="hl-string">"participant.queueName"])
    wrap_up_code = first_value(detail, [class="hl-string">"wrapUpCode", class="hl-string">"wrapupCode", class="hl-string">"wrapUp.code", class="hl-string">"wrapup.code", class="hl-string">"dispositionCode"])
    media_type = first_value(detail, [class="hl-string">"mediaType", class="hl-string">"media", class="hl-string">"channel", class="hl-string">"conversation.mediaType", class="hl-string">"session.mediaType"])
    return {
        class="hl-string">"ServiceRequest": service_request,
        class="hl-string">"ActivityName": activity,
        class="hl-string">"EventTime": event_time,
        class="hl-string">"EventEndTime": event_end_time,
        class="hl-string">"SourceSystem": SOURCE_SYSTEM,
        class="hl-string">"LastDataUpdate": refresh_time,
        class="hl-string">"AgentId": agent_id,
        class="hl-string">"AgentName": agent_name,
        class="hl-string">"QueueName": queue_name,
        class="hl-string">"WrapUpCode": wrap_up_code,
        class="hl-string">"MediaType": media_type,
        class="hl-string">"SourceEventId": event.get(class="hl-string">"id") or first_value(detail, [class="hl-string">"eventId", class="hl-string">"id"]),
    }


def classify(detail: Dict[str, Any], event: Dict[str, Any]) -> List[str]:
    text = event_type(event, detail)
    values = [str(first_value(detail, paths) or class="hl-string">"").lower() for paths in [
        [class="hl-string">"eventType", class="hl-string">"type"], [class="hl-string">"state"], [class="hl-string">"participant.state"], [class="hl-string">"session.state"], [class="hl-string">"event.name"], [class="hl-string">"event.type"]
    ]]
    text = class="hl-string">" ".join([text] + values)
    activities: List[str] = []
    if any(token in text for token in [class="hl-string">"conversation created", class="hl-string">"interaction started", class="hl-string">"conversation started", class="hl-string">"conversation.created"]):
        activities.append(ACTIVITY_NAMES[class="hl-string">"interaction_started"])
    if any(token in text for token in [class="hl-string">"routed", class="hl-string">"route to queue", class="hl-string">"queue joined", class="hl-string">"acd queue"]):
        activities.append(ACTIVITY_NAMES[class="hl-string">"conversation_routed_to_queue"])
    if any(token in text for token in [class="hl-string">"alerting", class="hl-string">"offered", class="hl-string">"agent offered"]):
        activities.append(ACTIVITY_NAMES[class="hl-string">"agent_offered_interaction"])
    if any(token in text for token in [class="hl-string">"connected", class="hl-string">"accepted", class="hl-string">"agent accepted", class="hl-string">"interacting"]):
        activities.append(ACTIVITY_NAMES[class="hl-string">"agent_accepted_interaction"])
    if any(token in text for token in [class="hl-string">"hold", class="hl-string">"held"]) and not any(token in text for token in [class="hl-string">"resume", class="hl-string">"unhold", class="hl-string">"off hold"]):
        activities.append(ACTIVITY_NAMES[class="hl-string">"hold_placed_on_interaction"])
    if any(token in text for token in [class="hl-string">"resume", class="hl-string">"unhold", class="hl-string">"off hold", class="hl-string">"hold removed"]):
        activities.append(ACTIVITY_NAMES[class="hl-string">"hold_removed_from_interaction"])
    if any(token in text for token in [class="hl-string">"transfer", class="hl-string">"transferred"]):
        activities.append(ACTIVITY_NAMES[class="hl-string">"interaction_transferred"])
    if any(token in text for token in [class="hl-string">"acw start", class="hl-string">"after call work started", class="hl-string">"wrapup start", class="hl-string">"wrap-up start"]):
        activities.append(ACTIVITY_NAMES[class="hl-string">"after_call_work_started"])
    if any(token in text for token in [class="hl-string">"wrap-up code", class="hl-string">"wrap up code", class="hl-string">"wrapup code", class="hl-string">"disposition"]):
        activities.append(ACTIVITY_NAMES[class="hl-string">"wrap_up_code_assigned"])
    if any(token in text for token in [class="hl-string">"acw end", class="hl-string">"after call work ended", class="hl-string">"wrapup end", class="hl-string">"wrap-up end"]):
        activities.append(ACTIVITY_NAMES[class="hl-string">"after_call_work_ended"])
    if any(token in text for token in [class="hl-string">"disconnected", class="hl-string">"disconnect", class="hl-string">"conversation ended", class="hl-string">"conversation completed"]):
        activities.append(ACTIVITY_NAMES[class="hl-string">"interaction_disconnected"])
    if any(token in text for token in [class="hl-string">"re-opened", class="hl-string">"reopened", class="hl-string">"service request reopened"]):
        activities.append(ACTIVITY_NAMES[class="hl-string">"service_request_re_opened"])
    if any(token in text for token in [class="hl-string">"csat sent", class="hl-string">"customer satisfaction survey sent", class="hl-string">"survey sent"]):
        activities.append(ACTIVITY_NAMES[class="hl-string">"customer_satisfaction_survey_sent"])
    return list(dict.fromkeys(activities))


def normalize_event(event: Dict[str, Any], refresh_time: str) -> List[Dict[str, Any]]:
    detail = event_detail(event)
    rows: List[Dict[str, Any]] = []
    for activity in classify(detail, event):
        rows.append(base_row(detail, activity, event, refresh_time))
    return rows


def extract_events(event_bus_name: str, start_time: datetime, end_time: datetime) -> List[Dict[str, Any]]:
    client = boto3.client(class="hl-string">"events", region_name=os.environ.get(class="hl-string">"AWS_REGION", class="hl-string">"[Your AWS Region]"))
    rows: List[Dict[str, Any]] = []
    next_token: Optional[str] = None
    refresh_time = datetime.now(timezone.utc).isoformat()
    while True:
        request: Dict[str, Any] = {
            class="hl-string">"EventBusName": event_bus_name,
            class="hl-string">"StartTime": start_time,
            class="hl-string">"EndTime": end_time,
            class="hl-string">"Limit": 100,
        }
        if next_token:
            request[class="hl-string">"NextToken"] = next_token
        response = client.list_events(**request)
        for event in response.get(class="hl-string">"Events", []):
            rows.extend(normalize_event(event, refresh_time))
        next_token = response.get(class="hl-string">"NextToken")
        if not next_token:
            break
    return rows


def append_reopened_and_survey_rows(rows: List[Dict[str, Any]], explicit_business_events: Iterable[Dict[str, Any]]) -> None:
    refresh_time = datetime.now(timezone.utc).isoformat()
    for event in explicit_business_events:
        detail = event_detail(event)
        for activity in [
            ACTIVITY_NAMES[class="hl-string">"service_request_re_opened"],
            ACTIVITY_NAMES[class="hl-string">"customer_satisfaction_survey_sent"],
        ]:
            if activity in classify(detail, event):
                rows.append(base_row(detail, activity, event, refresh_time))


def deduplicate(rows: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
    result: List[Dict[str, Any]] = []
    seen = set()
    for row in rows:
        key = (
            row.get(class="hl-string">"SourceEventId"), row.get(class="hl-string">"ServiceRequest"), row.get(class="hl-string">"ActivityName"),
            row.get(class="hl-string">"EventTime"), row.get(class="hl-string">"AgentId"), row.get(class="hl-string">"QueueName")
        )
        if key in seen:
            continue
        seen.add(key)
        result.append(row)
    return result


if __name__ == class="hl-string">"__main__":
    event_bus_name = os.environ.get(class="hl-string">"EVENT_BUS_NAME", class="hl-string">"[Your EventBridge event bus]")
    start_time = datetime.fromisoformat(os.environ.get(class="hl-string">"START_TIME", class="hl-string">"[Start date and time in ISO 8601 format]").replace(class="hl-string">"Z", class="hl-string">"+00:00"))
    end_time = datetime.fromisoformat(os.environ.get(class="hl-string">"END_TIME", class="hl-string">"[End date and time in ISO 8601 format]").replace(class="hl-string">"Z", class="hl-string">"+00:00"))
    normalized_rows = extract_events(event_bus_name, start_time, end_time)
    normalized_rows = deduplicate(normalized_rows)
    output_columns = [
        class="hl-string">"ServiceRequest", class="hl-string">"ActivityName", class="hl-string">"EventTime", class="hl-string">"EventEndTime", class="hl-string">"SourceSystem",
        class="hl-string">"LastDataUpdate", class="hl-string">"AgentId", class="hl-string">"AgentName", class="hl-string">"QueueName", class="hl-string">"WrapUpCode", class="hl-string">"MediaType", class="hl-string">"SourceEventId"
    ]
    output_path = os.environ.get(class="hl-string">"OUTPUT_PATH", class="hl-string">"[Your output file path]")
    import csv
    with open(output_path, class="hl-string">"w", newline=class="hl-string">"", encoding=class="hl-string">"utf-8") as output_file:
        writer = csv.DictWriter(output_file, fieldnames=output_columns)
        writer.writeheader()
        writer.writerows(normalized_rows)

准备好开始了吗？

深入了解客户服务运营。按照此模板操作，您即可逐步发现洞察并提升效率。

返回改进指南

下载模板

减少重复联系，立即提升客户服务！

轻松提升CSAT，将首次联系解决率提高至80%。

开始免费试用

无需信用卡，几分钟即可激活试用。

您的客户服务数据模板

您的客户服务数据模板

客户服务属性

客户服务活动

数据提取指南

Genesys Cloud Analytics交互详情导出

步骤

配置

查询示例 python

使用Genesys Cloud Platform API的Python脚本

步骤

配置

查询示例 python

通过AWS EventBridge集成进行事件流传输

步骤

配置

查询示例 python

准备好开始了吗？

减少重复联系，立即提升客户服务！