> 持续流程改进 > 客户服务 > Zendesk Support > 数据模板

您的客户服务数据模板

Zendesk Support

← 返回改进指南

客户服务属性 (18) 客户服务活动 (14) 提取指南 (4)

您的客户服务数据模板

此模板帮助您收集分析客户服务流程所需的数据，列出需要采集的关键数据属性和跟踪的核心活动，并提供实用的信息提取指导。借助本指南，您可以构建可靠的事件日志，发现流程洞察并改善服务运营。

建议采集的属性
客户服务流程中需要跟踪的关键活动
Zendesk支持数据提取指南

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

下载模板

客户服务属性

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

5 必需 6 建议 7 可选

	名称	说明
	开始时间 StartTime	表示活动或事件开始时间的时间戳。
说明开始时间或事件时间戳记录具体活动发生的准确日期和时间。例如，记录客户评论添加、客服人员分配或工单状态变为“Resolved”的时间。该时间戳是流程挖掘的基础，用于确定事件的时间顺序、计算活动间隔时长、衡量SLA达成情况并了解服务流程的时间动态。为什么重要该时间戳对于排列事件顺序、计算时长和分析服务请求流程时间线至关重要。获取位置 Zendesk Ticket Audits API中审计轨迹每个事件的`created_at`字段。示例 2023-04-15T10:00:00Z2023-04-15T10:05:14Z2023-04-16T14:30:00Z
	服务请求 ServiceRequest	每个客户服务请求的唯一标识，也称为工单或案例。
说明服务请求是主要案例标识，用于关联从客户咨询或问题创建到最终解决的所有相关活动。每次互动、更新或内部操作都与此唯一ID关联。在流程挖掘中，按服务请求对事件分组进行分析，可以完整呈现客户服务旅程。它是计算总解决时间、识别流程偏差和了解每个客户问题生命周期的基础。为什么重要这是连接所有流程步骤的关键案例ID，可用于重建和分析每个客户服务旅程。获取位置 Zendesk Tickets API中的`id`字段。示例 102451287415332
	最近数据更新时间 LastDataUpdate	源系统最近一次刷新或提取数据的时间戳。
说明该属性表示数据集最近一次从Zendesk Support更新的时间，帮助说明当前分析数据的新鲜度。了解最近更新时间对于分析人员和业务用户判断所查看的流程信息是否为最新内容至关重要。它有助于管理对数据时效性的预期，也是报告和监控的重要依据。为什么重要明确数据的时效性，帮助用户了解当前流程分析所基于的数据有多新。获取位置这是在数据提取过程中生成并存储的元数据字段，用于记录提取任务的时间戳。示例 2023-10-27T02:00:00Z
	活动名称 ActivityName	服务请求生命周期中发生的具体事件或任务的名称。
说明 Activity Name用于描述客户服务流程中的单个步骤或里程碑，例如“服务请求已创建”“请求已分配给工作人员”或“服务请求已解决”。这些事件带有时间戳，构成每个服务请求的操作顺序。该属性对于可视化流程顺序、发现流程变体以及分析事件的频率和顺序至关重要。它可以帮助分析人员了解正在执行的操作，并识别常见路径、瓶颈以及偏离标准过程的情况。为什么重要该属性定义流程中的步骤，用于可视化流程图，并分析流程顺序和变体。获取位置源自记录变更和操作的Zendesk工单审计日志或事件流。示例创建服务请求请求分配给客服人员发送客服人员首次公开回复服务请求已解决
	源系统 SourceSystem	提取数据的记录系统。
说明该属性指定服务请求数据的来源系统，本例中为Zendesk Support。在整合多个系统的数据时，它有助于数据治理和数据血缘管理。在分析中，它确保数据正确归属到来源系统，这对于维护数据完整性和理解流程背景至关重要，尤其是在多系统环境中。为什么重要标识数据来源，对于数据治理以及在集成环境中区分流程数据至关重要。获取位置这是在数据提取过程中添加的静态值，用于标记数据来源。示例 Zendesk Support
	SLA目标解决时间 SlaTargetResolutionTime	根据SLA策略，预期完成服务请求解决的目标时长。
说明该属性定义解决工单的服务级别协议（SLA）目标。目标通常是动态的，取决于请求优先级、类型或客户服务计划等因素。这是“SLA合规绩效”仪表板的基础属性，用作衡量实际解决时间的基准。通过分析相对于该目标的绩效，可以量化服务交付质量，并确保履行对客户的合同义务。为什么重要定义对客户的服务承诺，作为衡量按时完成情况和SLA合规性的基准。获取位置根据应用于工单的SLA策略推导得出。相关信息可通过Zendesk Ticket Metrics API获取。示例 144002880086400
	优先级 Priority	分配给服务请求的优先级，例如“Low”“Normal”“High”或“Urgent”。
说明优先级表示服务请求的紧急程度，通常会影响其在队列中的位置和目标解决时间。它帮助客服人员优先处理最关键的问题。该属性对于绩效和SLA分析至关重要。“服务请求解决时间分析”仪表板使用优先级细分数据，帮助判断高优先级请求是否比低优先级请求处理得更快，以及资源是否根据业务需求得到有效配置。为什么重要表示请求的紧急程度，对于分析SLA合规性并确保关键问题得到及时处理至关重要。获取位置 Zendesk Tickets API中的`priority`字段。示例低正常高紧急
	是否违反SLA IsSlaBreached	用于标识服务请求的解决时间是否超过SLA目标的布尔标记。
说明此计算属性是一个简单的真或假标记，用于显示服务请求是否未在规定的解决时间内满足其服务级别协议。该属性通过比较实际解决时长与计划SLA目标得出。此标记简化了SLA合规分析，是“SLA合规绩效”仪表板和“SLA合规率”KPI的核心数据点，可快速汇总并可视化达到服务目标与未达到目标的请求数量。为什么重要为每个案例提供清晰的二元SLA绩效结果，简化合规监控和报告。获取位置在数据转换过程中，通过比较总解决时间与SlaTargetResolutionTime计算得出。示例 truefalse
	服务请求类型 ServiceRequestType	服务请求的分类，例如“Question”“Incident”“Problem”或“Task”。
说明该属性根据服务请求的性质对其进行分类。通常，分类会在请求创建或分诊时设置，用于确定适当的工作流和优先级。在分析中，按服务请求类型进行细分十分重要。这样可以比较不同问题类型的解决时间、升级率和流程顺序，如“服务请求解决时间分析”和“内部升级率及原因”等仪表板所示。这有助于识别某些请求类型是否更难处理或效率更低。为什么重要对请求进行分类，以便比较和分析不同问题类型的绩效，这对于有针对性地改进流程至关重要。获取位置 Zendesk Tickets API中的`type`字段。示例问题事件故障任务
	沟通渠道 CommunicationChannel	提交服务请求或进行沟通所使用的渠道。
说明该属性标识所使用的沟通方式，例如电子邮件、Web表单、聊天或电话，反映客户与服务台互动的方式。了解渠道使用情况对于资源规划和改善客户体验非常重要。“沟通渠道使用概览”仪表板通过分析这些数据，了解最受欢迎的渠道，以及某些渠道是否与更长的解决时间或不同的流程路径相关。这有助于确定服务改进或自动化的投入方向。为什么重要展示客户与客服人员的互动方式，支持分析渠道效率及其对流程和客户体验的影响。获取位置 Zendesk Tickets API中的`via.channel`字段。示例 Web电子邮件API聊天
	负责客服人员 AssignedAgent	负责处理服务请求的客户服务人员姓名或ID。
说明该属性标识特定时间点负责某项活动或服务请求的客服人员。如果请求被重新分配，负责人可能在生命周期内发生变化。按负责客服人员分析对于了解人员工作量、绩效和效率至关重要。它支持“客服人员工作量与效率”仪表板，可比较不同客服人员的处理时长和案例量，帮助发现辅导机会并平衡工作量。为什么重要记录执行操作的客服人员，从而支持个人绩效、工作量分配和资源配置分析。获取位置 Zendesk Tickets API中的`assignee_id`字段。用户详情可通过Users API获取。示例 John SmithJane DoeSupportBot
	产品服务类别 ProductServiceCategory	客户请求所涉及的具体产品、服务或功能。
说明该属性按产品或服务领域对服务请求进行分类，提供更详细的背景信息。通常由客服人员手动设置，或根据请求内容自动设置。该分类对于“请求分类准确性”和“调查周期时间拆解”仪表板至关重要。它支持深入分析哪些产品产生的支持请求最多、哪些产品最难解决，以及初始分类是否与最终解决方案一致，从而帮助改进路由和客服培训。为什么重要将请求关联至具体业务领域、产品或服务，支持针对问题区域及其对流程影响的重点分析。获取位置通常是Zendesk中的自定义工单字段。具体字段名称取决于Zendesk的配置。示例移动应用订阅管理API集成硬件
	信息请求次数 InformationRequestCount	针对单个服务请求向客户请求信息的总次数。
说明此计算指标统计每个服务请求中“向客户请求信息”活动的发生次数。次数较多通常表明客服人员未能在开始处理时收集全部必要信息。此属性用于“重复信息请求分析”仪表板。跟踪该次数有助于识别流程低效环节和客服培训需求。减少信息请求次数可以显著缩短解决时间并改善客户体验。为什么重要量化与客户之间的反复沟通，突出延长解决时间并降低客户体验的低效环节。获取位置通过统计每个服务请求中ActivityName为“Information Requested From Customer”的事件数量计算得出。示例 013
	客服团队 AgentGroup	负责服务请求的支持组或团队。
说明该属性表示负责服务请求的客服团队。请求通常会根据技能、产品领域或语言路由至特定团队。按客服团队分析有助于了解团队级绩效、工作量分配和团队间的升级模式。相比个人客服人员分析，它提供更高层级的视角，并有助于识别特定部门或职能中的系统性问题。为什么重要记录团队责任，支持分析不同支持层级或专业团队的团队绩效、跨团队交接和资源配置。获取位置 Zendesk Tickets API中的`group_id`字段。团队详情可通过Groups API获取。示例一级支持技术支持账单
	是否重新打开 IsReopened	用于标识服务请求在标记为已解决后是否重新打开的布尔标记。
说明当服务请求在此前被设为已解决或已关闭后重新转为打开状态时，此属性将标记为真。它表明初次解决并不充分。此标记对于跟踪返工和首次联系解决失败至关重要。它直接支持“重新打开的服务请求趋势”仪表板和“服务请求重新打开率”KPI，便于统计和分析需要进一步处理的案例，而这通常意味着存在更深层次的问题。为什么重要识别返工和解决失败情况，帮助衡量所提供解决方案的质量和有效性。获取位置在数据转换过程中，通过检查工单状态是否从“resolved”或“closed”变回“open”计算得出。示例 truefalse
	满意度评分 SatisfactionRating	客户在服务请求解决后提供的满意度评分。
说明此属性记录客户对服务体验的反馈，通常在工单解决后通过调查收集。常见评分包括“好”“差”或数值评分。这是衡量客户情绪的直接指标，也是关键结果指标。该属性用于计算“客户情绪评分”KPI。将满意度评分与流程数据结合分析，例如解决时间或客服人员处理次数，可以发现哪些流程行为能够带来更好的客户结果。为什么重要直接衡量客户对所提供服务的反馈，将流程绩效与客户结果关联起来。获取位置 Zendesk Tickets API，`satisfaction_rating.score`字段。示例良好不佳已提供
	结束时间 EndTime	表示活动或事件完成时间的时间戳。
说明结束时间表示活动完成的时间。在许多事件日志结构中，下一个活动的开始时间可以作为当前活动的结束时间。对于“客服人员调查问题”等基于状态的活动，结束时间表示该状态结束的时点。该属性对于精确计算活动时长至关重要，是绩效分析的核心组成部分。它有助于识别最耗时的步骤，并为详细的瓶颈分析和资源效率计算提供依据。为什么重要支持计算活动时长，对于识别瓶颈和衡量绩效至关重要。获取位置通常通过获取同一服务请求事件序列中后续事件的StartTime推导得出。示例 2023-04-15T10:05:14Z2023-04-15T11:20:30Z2023-04-16T15:00:00Z
	解决代码 ResolutionCode	用于表示请求最终解决或关闭原因的代码或类别。
说明解决代码以结构化方式说明服务请求的结果。例如“客服人员解决”“重复请求”“无需采取行动”或“已知问题”。相比简单的“Closed”状态，它能提供更多背景信息。该属性对于根因分析尤其有价值。在“重新打开的服务请求趋势”仪表板中，按解决代码分析重新打开率，可以发现某些解决方案是否效果较差，导致客户需要再次联系支持团队。为什么重要帮助了解服务请求的处理结果，对于根因分析和理解请求重新打开的原因至关重要。获取位置通常是Zendesk中的自定义工单字段。具体字段名称取决于Zendesk的配置。示例首次联系解决已升级至二级支持等待客户回复产品缺陷

必需建议可选

客户服务活动

以下是需要记录在事件日志中的关键流程步骤和里程碑，用于准确发现客户服务流程。

6 建议 8 可选

	活动	说明
	创建服务请求	当客户通过电子邮件、Web表单或聊天等任意渠道提交请求，并在Zendesk中生成新工单时，客户服务流程便从此活动开始。系统会在创建工单时记录这一事件，并生成唯一工单ID和时间戳。
为什么重要作为主要开始事件，该活动对于计算完整工单时长和分析一段时间内的请求量至关重要。它也是衡量首次响应时间和总解决时间等关键绩效指标的基准。获取位置这是Zendesk Ticket Audits API中明确记录的事件，对应工单的“Create”事件，并提供初始创建时间戳。采集从Ticket Audits日志中的工单创建事件获取。事件类型 explicit
	向客户请求信息	当客服人员需要客户提供更多信息才能继续处理，并将工单状态改为“pending”时，该活动便会发生。此状态变更明确表示流程正在等待外部参与方。
为什么重要该活动体现了流程对客户的依赖，并会暂停内部SLA计时。同一工单频繁或重复出现此活动，可能表明初始信息收集不完整，并导致解决时间延长。获取位置这是Zendesk Ticket Audits API中记录的明确状态变更事件，在“status”字段变为“pending”时捕获。采集在Ticket Audits中识别将工单状态设置为“pending”的“Change”事件。事件类型 explicit
	服务请求已关闭	这是最后一项活动，标志着服务请求永久关闭。通常在工单标记为“solved”后经过一段设定时间，且客户未再回复时自动发生。
为什么重要作为最终结束事件，该活动会完成工单生命周期。从“solved”到“closed”的时间代表可能重新打开工单的窗口，而“closed”事件则确认该请求已完成解决。获取位置这是Zendesk Ticket Audits API中记录的明确状态变更事件，在“status”字段设置为“closed”时产生。采集在Ticket Audits中识别将工单状态设置为“closed”的“Change”事件。事件类型 explicit
	服务请求已解决	客服人员向客户提供解决方案后，会将服务请求标记为“solved”。这是一个临时状态，在工单永久关闭前，客户仍可通过回复重新打开工单。
为什么重要这是衡量解决时间和客服人员效率的主要里程碑。它表示客服人员认为工作已完成，也为分析工单重新打开后的返工提供依据。获取位置这是Zendesk Ticket Audits API中记录的明确状态变更事件，在“status”字段设置为“solved”时产生。采集在Ticket Audits中识别将工单状态设置为“solved”的“Change”事件。事件类型 explicit
	服务请求重新打开	当客户回复处于“solved”状态的工单时，该活动便会发生。Zendesk会自动将状态改回“open”，表示问题尚未完全解决。
为什么重要工单重新打开是首次联系解决失败和解决方案质量不佳的重要指标。分析重新打开工单的频率和原因，有助于确定客服培训和解决流程的改进方向。获取位置这是Ticket Audits API中的明确状态变更，在“status”从“solved”改回“open”时捕获。采集跟踪从“solved”到“open”的状态“Change”事件。事件类型 explicit
	请求分配给客服人员	该事件表示服务请求已分配给特定客服人员处理。分配可能依据路由规则自动完成，也可能由团队负责人或客服人员手动完成。
为什么重要分配是明确责任和管理工作量的重要里程碑。分析分配耗时和重新分配模式，有助于发现分诊与分发流程中的瓶颈。获取位置这是Zendesk Ticket Audits API中的明确“Change”事件，每当“assignee_id”字段被填充或发生变更时记录。采集跟踪Ticket Audits日志中“assignee_id”字段的变更。事件类型 explicit
	发生SLA违约	该活动标志着服务请求未能在预定义的服务级别协议内完成，例如首次回复时间或解决时间。系统会将工单活动时间戳与SLA策略进行比较后计算该事件。
为什么重要 SLA违约会直接影响客户满意度和合同合规性。分析违约发生的时间和原因，对于识别系统性延迟、资源不足或不切实际的绩效目标至关重要。获取位置可从Zendesk Ticket Metrics API获取，该接口会存储SLA违约时间戳（“breached_at”）。也可以将工单事件时间戳与定义的SLA规则进行比较后计算。采集使用Ticket Metrics API中的“breached_at”时间戳，或将解决时间与SLA策略规定的时间进行比较后计算。事件类型 calculated
	发送客服人员首次公开回复	该活动表示客服人员首次向客户发送公开评论，而不是自动确认消息。这是衡量客服人员首次介入客户问题的重要事件。
为什么重要这是衡量“First Reply Time”SLA的关键里程碑，也是服务响应能力的重要指标。它将自动通信与人工主导的主动调查和支持区分开来。获取位置在Ticket Comments流中查找首条由人工客服人员而非自动化系统用户发布的公开评论即可识别。采集扫描工单评论，筛选客服人员发布的公开评论，并选择时间戳最早的一条。事件类型 inferred
	发送满意度调查	表示自动向客户发送客户满意度（CSAT）调查的时点。通常在工单标记为“solved”后不久发送。
为什么重要该活动启动客户反馈闭环。了解调查何时发送以及是否发送，对于正确解读满意度评分和衡量反馈计划的有效性非常重要。获取位置可通过自动化日志或工单新增的特定标签推断。Ticket Audits API中的“satisfaction_rating”部分也会记录调查发出的时间。采集查找“csat_sent”等标签，或使用提供满意度评分时的时间戳。事件类型 inferred
	发送首次确认	表示向客户发送的自动首次回复，用于确认已收到客户请求。通常由Zendesk触发器在工单创建后立即发送模板化电子邮件通知。
为什么重要跟踪该活动对于衡量初始响应速度和管理客户预期至关重要。从请求创建到发送确认之间的时间，是衡量客户满意度的关键指标。获取位置如果工单首条公开评论的作者是自动化用户，或评论在工单创建后几秒内产生，则可据此推断。可通过分析Ticket Comments流识别该活动。采集识别工单创建后由自动化程序或触发器立即创建的首条公开评论。事件类型 inferred
	客户已回复	当客户回复工单时触发该事件，通常发生在工单处于“pending”状态时。Zendesk会自动将工单状态从“pending”切换回“open”，表示客服人员可以继续处理。
为什么重要该活动标志着等待时间结束，并触发流程继续执行。分析客户响应所需时间，有助于了解客服人员请求的清晰程度。获取位置该事件对应终端用户发布的新公开评论，并会在Ticket Audits API中触发从“pending”到“open”的明确状态变更。采集跟踪从“pending”到“open”的状态“Change”事件，或识别终端用户发布的新公开评论。事件类型 explicit
	收到满意度评分	当客户提交满意度调查回复并给出“Good”或“Bad”等评分时，该事件便会发生。评分及相关评论会记录在工单中。
为什么重要直接的客户反馈对于衡量服务质量和客户情绪极具价值。在流程顺序的背景下分析这些评分，有助于将具体活动或工作人员与结果建立关联。获取位置当“satisfaction_rating”字段填入客户评分和评论时，作为Ticket Audits API中的“Change”事件捕获。采集筛选Ticket Audit日志，查找“satisfaction_rating”字段的变更。事件类型 explicit
	触发内部升级	表示将服务请求转交给其他内部团队或更高支持层级。通常可通过工单所属组发生变更来推断。
为什么重要跟踪升级对于识别流程薄弱环节、一线支持的知识缺口和复杂请求类型至关重要。较高的升级率可能表明需要加强培训或完善流程文档。获取位置根据Ticket Audits API中“group_id”字段发生修改的“Change”事件推断。组别变更表示团队之间发生了交接。采集监控Ticket Audits日志中“group_id”字段的变更。事件类型 inferred
	请求完成分类和优先级设置	当客服人员或自动化程序设置或更新工单类型、类别和优先级等字段时，该活动便会发生。此步骤会作为变更事件记录在工单历史中。
为什么重要准确的分类和优先级设置对于高效路由和资源分配至关重要。分析该活动有助于评估初始分诊的准确性及其对解决时间的影响。获取位置从Zendesk Ticket Audits API中的“Change”事件获取。查找“priority”“type”或与分类相关的自定义字段的首次更新即可识别。采集筛选Ticket Audit日志，查找创建后关键分类字段的首个“Change”事件。事件类型 explicit

建议可选

提取指南

如何从Zendesk支持获取数据

步骤

进入Zendesk Explore：登录您的Zendesk实例，点击顶部栏中的产品启动器图标，选择Explore，打开该产品。
创建新查询：在Explore中点击左侧边栏的Queries图标，然后点击右上角的“New query”按钮。
选择数据集：在“Choose a dataset”窗口中选择Support: Tickets数据集。该数据集包含主要工单属性和关键事件时间戳。
添加指标：在查询构建器中进入“Metrics”面板并点击“Add”，选择COUNT(Tickets)，用于验证数据量。
将属性添加到行：进入“Rows”面板并点击“Add”，添加以下属性。这些属性将构成宽格式导出的基础列。
- Ticket ID（这是您的案例标识符，ServiceRequest）
- Assignee name（AssignedAgent）
- Ticket type（ServiceRequestType）
- Ticket priority（Priority）
- Ticket channel（CommunicationChannel）
- SLA resolution time（SlaTargetResolutionTime）
- SLA breach status（IsSlaBreached）
- Full resolution time (min)（ServiceRequestResolutionTime）
将活动时间戳添加到行：将以下时间戳属性添加到“Rows”面板。每个属性都代表一个潜在活动。部分活动需要创建自定义计算属性。
- Ticket created - Timestamp：用于Service Request Created和Initial Acknowledgment Sent。
- Ticket assigned - Timestamp：用于Request Assigned To Agent。
- First reply time - Timestamp：用于First Agent Public Reply Sent。
- Ticket solved - Timestamp：用于Service Request Resolved和Satisfaction Survey Sent。
- Ticket reopened - Timestamp：用于Service Request Reopened。
- Satisfaction rating received - Timestamp：用于Satisfaction Rating Received。
- Ticket closed - Timestamp：用于Service Request Closed。
为其他活动创建自定义计算属性：部分活动基于特定字段变更，需要创建自定义属性来记录其时间戳。进入Calculations菜单（计算器图标），选择“Standard calculated attribute”。为以下每项活动创建一个属性。您需要在Zendesk实例中确认字段和值的准确名称。
- 名称：Timestamp - Categorized and Prioritized。公式：IF ([Changes - Field name] = "priority" OR [Changes - Field name] = "type") THEN [Update - Timestamp] ENDIF。创建后，将此新属性添加到“Rows”面板，并将聚合器改为MIN。
- 名称：Timestamp - Info Requested。公式：IF ([Changes - Field name] = "status" AND [Changes - New value] = "pending") THEN [Update - Timestamp] ENDIF。将其添加到“Rows”并将聚合器设为MIN。
- 名称：Timestamp - Customer Responded。公式：IF ([Changes - Field name] = "status" AND [Changes - Previous value] = "pending" AND [Changes - New value] = "open") THEN [Update - Timestamp] ENDIF。将其添加到“Rows”并将聚合器设为MIN。
- 名称：Timestamp - Escalated。公式：IF ([Changes - Field name] = "group") THEN [Update - Timestamp] ENDIF。将其添加到“Rows”并将聚合器设为MIN。
- 名称：Timestamp - SLA Breach。公式：IF ([SLA breach status]=TRUE) THEN [SLA update - Timestamp] ENDIF。将其添加到“Rows”并将聚合器设为MIN。
筛选数据：进入“Filters”面板并点击“Add”，为Ticket created - Date添加筛选条件，选择合适的日期范围，例如最近3至6个月，以控制数据集规模。
运行并导出报告：点击“Save”运行查询。表格显示数据后，点击“Export”按钮并选择所需格式，例如CSV或Excel。
将数据转换为事件日志格式：导出文件为“宽”格式，每行代表一个工单，每列代表一个事件时间戳。您必须将其转换为适用于流程挖掘的“长”格式。可以使用Microsoft Excel（通过Power Query的取消透视功能）、Python或其他数据准备工具。
- 对每个工单行，为每个非空时间戳列创建一行新记录。
- 最终表格应包含ServiceRequest、ActivityName、StartTime及其他属性，这些属性将针对同一服务请求的每个活动重复。

配置

数据集：主要使用的数据集是Support: Tickets，可在一次查询中访问大多数工单属性和关键时间戳。
权限：您至少需要在Zendesk Explore中拥有Editor或Admin角色，才能创建和保存查询。
日期范围：必须为Ticket created - Date设置日期范围筛选。建议使用3至6个月的数据，在洞察深度与性能之间取得平衡。导出数百万张工单可能速度较慢，甚至失败。
查询筛选条件：可以根据特定工单属性添加筛选条件，例如Ticket brand、Ticket form或Ticket group，以聚焦某项服务或某个团队。
计算属性：此方法高度依赖标准计算属性，以捕获默认属性中没有的事件时间戳，例如状态变更或升级。请确保您拥有创建这些属性的权限。

a 示例查询 config

This extraction is configured through the Zendesk Explore graphical user interface. There is no single script to execute. The configuration below outlines the components to add in the query builder.

**Dataset**: `Support: Tickets`

**Metrics**:
1.  `COUNT(Tickets)`

**Rows (Attributes)**:
1.  `Ticket ID`
2.  `Assignee name`
3.  `Ticket type`
4.  `Ticket priority`
5.  `Ticket channel`
6.  `SLA resolution time`
7.  `SLA breach status`
8.  `Full resolution time (min)`
9.  `MIN(Ticket created - Timestamp)`
10. `MIN(Ticket assigned - Timestamp)`
11. `MIN(First reply time - Timestamp)`
12. `MIN(Ticket solved - Timestamp)`
13. `MIN(Ticket reopened - Timestamp)`
14. `MIN(Satisfaction rating received - Timestamp)`
15. `MIN(Ticket closed - Timestamp)`
16. `MIN(Timestamp - Categorized and Prioritized)` (Custom calculated attribute)
17. `MIN(Timestamp - Info Requested)` (Custom calculated attribute)
18. `MIN(Timestamp - Customer Responded)` (Custom calculated attribute)
19. `MIN(Timestamp - Escalated)` (Custom calculated attribute)
20. `MIN(Timestamp - SLA Breach)` (Custom calculated attribute)

**Filters**:
1.  `Ticket created - Date`: Select a date range (e.g., last 180 days)

**Standard Calculated Attribute Formulas**:
*   `Timestamp - Categorized and Prioritized`: `IF ([Changes - Field name] = "priority" OR [Changes - Field name] = "type") THEN [Update - Timestamp] ENDIF`
*   `Timestamp - Info Requested`: `IF ([Changes - Field name] = "status" AND [Changes - New value] = "pending") THEN [Update - Timestamp] ENDIF`
*   `Timestamp - Customer Responded`: `IF ([Changes - Field name] = "status" AND [Changes - Previous value] = "pending" AND [Changes - New value] = "open") THEN [Update - Timestamp] ENDIF`
*   `Timestamp - Escalated`: `IF ([Changes - Field name] = "group") THEN [Update - Timestamp] ENDIF`
*   `Timestamp - SLA Breach`: `IF ([SLA breach status]=TRUE) THEN [SLA update - Timestamp] ENDIF`

步骤

准备Zendesk API访问权限：以管理员身份登录Zendesk。进入管理中心，然后依次进入Apps and integrations > APIs > Zendesk API。在Settings选项卡中，确保已启用Token Access。点击“Add API token”创建新令牌。请安全复制生成的令牌，因为之后将无法再次显示。
设置Python环境：确保已安装Python 3.6或更高版本。安装用于向Zendesk API发送HTTP请求的requests库。您可以通过命令行安装：pip install requests。
配置Python脚本：将提供的Python脚本保存为文件，例如extract_zendesk_log.py。在文本编辑器中打开文件，使用您的Zendesk实例信息和凭据更新顶部的占位变量：ZENDESK_SUBDOMAIN、ZENDESK_EMAIL和API_TOKEN。
定义提取周期：在脚本中找到START_TIMESTAMP变量。该值决定数据提取的历史起始日期，必须是Unix纪元时间戳。首次分析建议使用对应最近3至6个月的值。
执行脚本：打开终端或命令提示符，进入保存脚本的目录，然后运行命令：python extract_zendesk_log.py。
监控执行过程：脚本会在控制台输出进度更新，显示已处理的工单数量。总执行时间取决于提取数据量以及Zendesk实施的API速率限制。脚本包含处理速率限制响应的基本逻辑。
获取事件日志：成功完成后，脚本会在同一目录中创建名为zendesk_event_log.csv的文件。该文件包含可直接用于分析的结构化事件日志。
准备上传：打开zendesk_event_log.csv文件，确认生成无误。检查列是否符合所需属性：ServiceRequest、ActivityName、StartTime、SourceSystem和LastDataUpdate。现在即可将文件上传到流程挖掘工具。

配置

API凭据：脚本需要有效的Zendesk子域名、客服人员或管理员用户的电子邮件地址，以及有效的API令牌。凭据关联用户必须拥有读取工单、工单审计、用户和工单指标的权限。
日期范围：提取起始日期由脚本中的START_TIMESTAMP变量控制，该变量必须是Unix纪元时间戳。建议先使用较短周期，例如3个月，以验证脚本输出和性能，再尝试提取更长的历史数据。
增量导出API：此方案使用Zendesk的Incremental Export API。这是获取大量工单最高效且推荐的方法，专为批量数据导出设计，可避免常见的分页问题。
性能与速率限制：Zendesk会根据订阅计划实施不同的API速率限制。脚本内置逻辑，会根据API返回的Retry-After标头在触发速率限制时暂停执行。对于超大型实例，提取可能需要数小时。
前提条件：此提取方法需要可正常运行的Python 3.6+环境和requests外部库。您需要拥有目标Zendesk实例的管理员访问权限，才能生成所需的API令牌。

a 示例查询 python

import requests
import json
import time
import datetime
import csv

class=class="hl-string">"hl-comment"># --- Configuration ---
class=class="hl-string">"hl-comment"># Replace with your Zendesk details
ZENDESK_SUBDOMAIN = class="hl-string">'[your_zendesk_subdomain]'
ZENDESK_EMAIL = class="hl-string">'[your_zendesk_login_email]'
API_TOKEN = class="hl-string">'[your_zendesk_api_token]'

class=class="hl-string">"hl-comment"># Set the start time for the export (Unix epoch timestamp)
class=class="hl-string">"hl-comment"># Example: 1672531200 is Jan 1, 2023 00:00:00 GMT
START_TIMESTAMP = 1672531200

class=class="hl-string">"hl-comment"># --- Script --- 

BASE_URL = fclass="hl-string">'https://{ZENDESK_SUBDOMAIN}.zendesk.com/api/v2'
AUTH = (fclass="hl-string">'{ZENDESK_EMAIL}/token', API_TOKEN)
SOURCE_SYSTEM = class="hl-string">'Zendesk Support'
LAST_DATA_UPDATE = datetime.datetime.utcnow().isoformat() + class="hl-string">'Z'

CSV_FILENAME = class="hl-string">'zendesk_event_log.csv'
CSV_HEADERS = [
    class="hl-string">'ServiceRequest', class="hl-string">'ActivityName', class="hl-string">'StartTime', class="hl-string">'SourceSystem', class="hl-string">'LastDataUpdate',
    class="hl-string">'AssignedAgent', class="hl-string">'ServiceRequestType', class="hl-string">'Priority', class="hl-string">'CommunicationChannel',
    class="hl-string">'SlaTargetResolutionTime', class="hl-string">'IsSlaBreached', class="hl-string">'ServiceRequestResolutionTime'
]

def get_user_name(user_id, users_cache, session):
    if user_id in users_cache:
        return users_cache[user_id]
    if user_id is None:
        return None
    try:
        response = session.get(fclass="hl-string">'{BASE_URL}/users/{user_id}.json')
        response.raise_for_status()
        user_name = response.json().get(class="hl-string">'user', {}).get(class="hl-string">'name')
        users_cache[user_id] = user_name
        return user_name
    except requests.exceptions.RequestException as e:
        print(fclass="hl-string">'Warning: Could not fetch user {user_id}. Error: {e}')
        users_cache[user_id] = str(user_id)
        return str(user_id)

def format_duration(seconds):
    if seconds is None:
        return None
    return str(datetime.timedelta(seconds=seconds))

def main():
    session = requests.Session()
    session.auth = AUTH
    users_cache = {}
    event_log = []

    class=class="hl-string">"hl-comment"># Fetch tickets using the incremental export API
    url = fclass="hl-string">'{BASE_URL}/incremental/tickets/cursor.json?start_time={START_TIMESTAMP}'
    ticket_count = 0

    while url:
        print(fclass="hl-string">'Fetching tickets from: {url}')
        try:
            response = session.get(url)

            if response.status_code == 429:
                retry_after = int(response.headers.get(class="hl-string">'Retry-After', 60))
                print(fclass="hl-string">'Rate limit hit. Retrying after {retry_after} seconds.')
                time.sleep(retry_after)
                continue

            response.raise_for_status()
            data = response.json()
            tickets = data.get(class="hl-string">'tickets', [])

            for ticket in tickets:
                ticket_count += 1
                ticket_id = ticket[class="hl-string">'id']
                print(fclass="hl-string">'Processing ticket {ticket_id} ({ticket_count} total)...')

                class=class="hl-string">"hl-comment"># Fetch ticket metrics for SLA and resolution time
                metrics_response = session.get(fclass="hl-string">'{BASE_URL}/tickets/{ticket_id}/metrics.json')
                ticket_metrics = metrics_response.json().get(class="hl-string">'ticket_metric', {}) if metrics_response.ok else {}

                resolution_time_seconds = ticket_metrics.get(class="hl-string">'full_resolution_time_in_minutes', {}).get(class="hl-string">'business')
                if resolution_time_seconds:
                    resolution_time_seconds *= 60

                class=class="hl-string">"hl-comment"># Fetch ticket audits for events
                audits_url = fclass="hl-string">'{BASE_URL}/tickets/{ticket_id}/audits.json'
                audits_response = session.get(audits_url)
                audits = audits_response.json().get(class="hl-string">'audits', []) if audits_response.ok else []

                first_agent_reply_sent = False

                for audit in audits:
                    audit_created_at = audit[class="hl-string">'created_at']
                    base_event = {
                        class="hl-string">'ServiceRequest': ticket_id,
                        class="hl-string">'StartTime': audit_created_at,
                        class="hl-string">'SourceSystem': SOURCE_SYSTEM,
                        class="hl-string">'LastDataUpdate': LAST_DATA_UPDATE,
                        class="hl-string">'AssignedAgent': get_user_name(ticket.get(class="hl-string">'assignee_id'), users_cache, session),
                        class="hl-string">'ServiceRequestType': ticket.get(class="hl-string">'type'),
                        class="hl-string">'Priority': ticket.get(class="hl-string">'priority'),
                        class="hl-string">'CommunicationChannel': ticket.get(class="hl-string">'via', {}).get(class="hl-string">'channel'),
                        class="hl-string">'SlaTargetResolutionTime': None, class=class="hl-string">"hl-comment"># Requires parsing SLA policies, complex
                        class="hl-string">'IsSlaBreached': ticket_metrics.get(class="hl-string">'sla_breach_at') is not None,
                        class="hl-string">'ServiceRequestResolutionTime': format_duration(resolution_time_seconds)
                    }

                    for event in audit.get(class="hl-string">'events', []):
                        event_type = event.get(class="hl-string">'type')
                        class=class="hl-string">"hl-comment"># Service Request Created
                        if event_type == class="hl-string">'Create' and event.get(class="hl-string">'field_name') == class="hl-string">'id':
                            new_event = base_event.copy()
                            new_event[class="hl-string">'ActivityName'] = class="hl-string">'Service Request Created'
                            event_log.append(new_event)

                        class=class="hl-string">"hl-comment"># Initial Acknowledgment Sent (Proxy by Notification event)
                        if event_type == class="hl-string">'Notification':
                            new_event = base_event.copy()
                            new_event[class="hl-string">'ActivityName'] = class="hl-string">'Initial Acknowledgment Sent'
                            event_log.append(new_event)

                        class=class="hl-string">"hl-comment"># Request Categorized And Prioritized, Assigned, Status Changes
                        if event_type == class="hl-string">'Change':
                            field = event.get(class="hl-string">'field_name')
                            if field in [class="hl-string">'type', class="hl-string">'priority', class="hl-string">'custom_field_[your_category_field_id]']:
                                new_event = base_event.copy()
                                new_event[class="hl-string">'ActivityName'] = class="hl-string">'Request Categorized And Prioritized'
                                event_log.append(new_event)
                            
                            if field == class="hl-string">'assignee_id':
                                new_event = base_event.copy()
                                new_event[class="hl-string">'ActivityName'] = class="hl-string">'Request Assigned To Agent'
                                new_event[class="hl-string">'AssignedAgent'] = get_user_name(event.get(class="hl-string">'value'), users_cache, session)
                                event_log.append(new_event)

                            if field == class="hl-string">'group_id':
                                new_event = base_event.copy()
                                new_event[class="hl-string">'ActivityName'] = class="hl-string">'Internal Escalation Triggered'
                                event_log.append(new_event)

                            if field == class="hl-string">'status':
                                new_status = event.get(class="hl-string">'value')
                                if new_status == class="hl-string">'pending':
                                    new_event = base_event.copy()
                                    new_event[class="hl-string">'ActivityName'] = class="hl-string">'Information Requested From Customer'
                                    event_log.append(new_event)
                                elif new_status == class="hl-string">'solved':
                                    new_event = base_event.copy()
                                    new_event[class="hl-string">'ActivityName'] = class="hl-string">'Service Request Resolved'
                                    event_log.append(new_event)
                                    class=class="hl-string">"hl-comment"># Satisfaction Survey Sent (Proxy: occurs after solve)
                                    solve_time = datetime.datetime.fromisoformat(audit_created_at.replace(class="hl-string">'Z', class="hl-string">'+00:00'))
                                    survey_time = solve_time + datetime.timedelta(hours=1)
                                    survey_event = new_event.copy()
                                    survey_event[class="hl-string">'ActivityName'] = class="hl-string">'Satisfaction Survey Sent'
                                    survey_event[class="hl-string">'StartTime'] = survey_time.isoformat().replace(class="hl-string">'+00:00', class="hl-string">'Z')
                                    event_log.append(survey_event)
                                elif new_status == class="hl-string">'closed':
                                    new_event = base_event.copy()
                                    new_event[class="hl-string">'ActivityName'] = class="hl-string">'Service Request Closed'
                                    event_log.append(new_event)
                                elif new_status == class="hl-string">'open' and event.get(class="hl-string">'previous_value') in [class="hl-string">'solved', class="hl-string">'pending']:
                                    activity = class="hl-string">'Service Request Reopened' if event.get(class="hl-string">'previous_value') == class="hl-string">'solved' else class="hl-string">'Customer Responded'
                                    new_event = base_event.copy()
                                    new_event[class="hl-string">'ActivityName'] = activity
                                    event_log.append(new_event)
                            
                            if field == class="hl-string">'satisfaction_score':
                                new_event = base_event.copy()
                                new_event[class="hl-string">'ActivityName'] = class="hl-string">'Satisfaction Rating Received'
                                event_log.append(new_event)

                        class=class="hl-string">"hl-comment"># First Agent Public Reply Sent & Customer Responded
                        if event_type == class="hl-string">'Comment':
                            author_id = event.get(class="hl-string">'author_id')
                            author_details_response = session.get(fclass="hl-string">'{BASE_URL}/users/{author_id}.json')
                            author_role = author_details_response.json().get(class="hl-string">'user',{}).get(class="hl-string">'role', class="hl-string">'end-user') if author_details_response.ok else class="hl-string">'end-user'

                            if author_role in [class="hl-string">'agent', class="hl-string">'admin'] and event.get(class="hl-string">'public') and not first_agent_reply_sent:
                                new_event = base_event.copy()
                                new_event[class="hl-string">'ActivityName'] = class="hl-string">'First Agent Public Reply Sent'
                                event_log.append(new_event)
                                first_agent_reply_sent = True

                class=class="hl-string">"hl-comment"># SLA Breach Occurred
                if ticket_metrics.get(class="hl-string">'sla_breach_at'):
                    sla_breach_time = ticket_metrics.get(class="hl-string">'sla_breach_at')
                    new_event = base_event.copy()
                    new_event[class="hl-string">'ActivityName'] = class="hl-string">'SLA Breach Occurred'
                    new_event[class="hl-string">'StartTime'] = sla_breach_time
                    event_log.append(new_event)

            url = data.get(class="hl-string">'after_url')
            if not data.get(class="hl-string">'end_of_stream', False):
                time.sleep(1) class=class="hl-string">"hl-comment"># Be a good API citizen
            else:
                url = None

        except requests.exceptions.RequestException as e:
            print(fclass="hl-string">'An error occurred: {e}')
            break

    class=class="hl-string">"hl-comment"># Write to CSV
    print(fclass="hl-string">'Writing {len(event_log)} events to {CSV_FILENAME}...')
    with open(CSV_FILENAME, class="hl-string">'w', newline=class="hl-string">'', encoding=class="hl-string">'utf-8') as csvfile:
        writer = csv.DictWriter(csvfile, fieldnames=CSV_HEADERS)
        writer.writeheader()
        class=class="hl-string">"hl-comment"># Sort events by ticket and then by time for chronological order
        sorted_log = sorted(event_log, key=lambda x: (x[class="hl-string">'ServiceRequest'], x[class="hl-string">'StartTime']))
        for row in sorted_log:
            writer.writerow(row)
    print(class="hl-string">'Extraction complete.')

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

步骤

确认您的Zendesk Support账户允许访问Incremental Exports API，并确认集成身份有权读取工单、工单事件、用户、组、评论、满意度评分，以及账户已启用功能对应的SLA数据。根据组织批准的配置，确认账户子域名和身份验证方式。
选择覆盖所需分析周期的提取起始时间。增量工单导出使用Unix时间戳，并返回从该时间点起创建或更新的记录。若要完整加载历史数据，请将[Configure based on your system]作为初始起始时间，然后保存返回的结束时间，用于后续增量运行。日常分析可在运营历史数据充足时使用3至6个月的窗口。
配置Zendesk子域名、身份验证凭据、起始时间戳和可选的公司特定筛选条件，然后运行Python提取脚本。脚本会请求增量工单数据、处理分页，并在转换前保存原始工单记录和可用的事件详情。
当增量工单响应未包含足够的字段变更、分配、评论或状态详情时，获取每张工单的审计或事件历史。仅使用账户中已启用且获授权的Zendesk API资源。对于返回的字段，保留源工单ID、审计ID、评论ID、事件时间戳、操作者、字段名称、原值和新值。
将时间戳统一为UTC，并将Unix时间戳或ISO 8601值转换为一致的日期时间格式。按ServiceRequest、事件时间戳和源事件标识符对源记录排序。使用可用的最具体稳定源标识符去重，例如审计或评论标识符。
为全部14项必需活动创建明确的事件日志行。不要依赖ProcessMind从状态变更或字段值推断活动。根据以下规则生成活动：从工单创建生成Service Request Created；从已验证的自动首次回复生成Initial Acknowledgment Sent；从类别、类型或优先级变更生成Request Categorized And Prioritized；从受派人变更生成Request Assigned To Agent；从第一条符合条件的客服人员公开评论生成First Agent Public Reply Sent；从与信息请求相关的pending状态变更生成Information Requested From Customer；从符合条件的客户回复生成Customer Responded；从受派组变更生成Internal Escalation Triggered；从已验证的SLA违反时间戳或明确计算的违反结果生成SLA Breach Occurred；从solved状态变更生成Service Request Resolved；从客户回复导致的solved到open状态变更生成Service Request Reopened；从已验证的调查发送事件生成Satisfaction Survey Sent；从已提交的评分生成Satisfaction Rating Received；从closed状态变更或已验证的永久关闭事件生成Service Request Closed。
填充必需列ServiceRequest、ActivityName、StartTime、SourceSystem和LastDataUpdate。当源数据提供可靠值时，填充推荐列AssignedAgent、ServiceRequestType、Priority、CommunicationChannel、SlaTargetResolutionTime和IsSlaBreached。对于不可用的可选属性使用null，不要自行编造值。
在提取完成且导出前，应用已批准的公司特定筛选条件。在内部暂存副本中保留原始源标识符和源时间戳，以便审计和重新运行事件判定。
验证行数、活动覆盖率、时间戳顺序、重复记录处理和属性填充情况。确认每一行输出都包含ServiceRequest、ActivityName、StartTime、SourceSystem和LastDataUpdate。
将转换后的事件日志导出为UTF-8 CSV或其他ProcessMind支持的表格格式。确保列名与所需属性名称完全一致，每行保留一个活动发生记录，然后通过已配置的ProcessMind导入流程上传文件。

配置

提取方式：使用Zendesk Incremental Exports API提取大量工单。脚本应从已保存的Unix起始时间戳请求数据，处理分页，遵守速率限制，并保存返回的结束时间戳供下次运行使用。
身份验证：根据组织安全策略配置已批准的Zendesk身份验证方式、账户子域名、凭据和API权限。不要将凭据直接放入源代码管理系统。
日期范围：日常流程分析使用3至6个月的数据。首次加载时，应选择早于所需流程周期的起始时间，因为较早创建的工单可能在分析窗口内发生后续活动。增量加载时，使用上一次成功提取的水位线，并保留较小的重叠区间，以防遗漏延迟到达的更新，然后根据源标识符去重。
筛选条件：仅使用账户可用Zendesk资源支持的筛选条件，例如已批准的组织、组、品牌、工单类型、状态或自定义字段筛选。公司特定筛选条件必须根据您的系统配置。不要假设Zendesk中存在Company Code或Document Type。
必需输出列：ServiceRequest、ActivityName、StartTime、SourceSystem和LastDataUpdate。
推荐输出列：AssignedAgent、ServiceRequestType、Priority、CommunicationChannel、SlaTargetResolutionTime和IsSlaBreached。
事件详情：仅凭增量工单记录可能无法获取所有历史状态变更。请根据需要获取已获授权的工单审计、评论、满意度记录和SLA事件。将每项必需活动明确输出为一行。
SLA配置：在可用时，使用账户返回的实际SLA策略和违反时间戳。如果计算违反状态，请记录所使用的策略、工作时间、暂停规则和时区。当适用工作时间规则时，不要仅根据自然时间流逝计算SLA违反。
性能：增量处理分页结果，以流式或批处理方式处理记录，缓存工单详情，使用受控并发，遵守Zendesk速率限制响应，并在分页成功后保存检查点。避免反复请求完整历史数据。
数据留存：保留原始API响应、提取水位线、请求元数据和转换日志，以便对账和重新运行。
前提条件：Zendesk API访问权限、读取相关工单和事件资源的权限、按需启用的工单审计和评论功能、执行CSAT活动所需的满意度数据访问权限，以及执行SLA违反活动所需的SLA数据访问权限。具体功能取决于Zendesk订阅计划和账户配置。

a 示例查询 python

import csv
import json
import os
import sys
import time
from datetime import datetime, timezone
from typing import Any, Dict, Iterable, List, Optional, Tuple

import requests

ZENDESK_SUBDOMAIN = os.environ.get(class="hl-string">"ZENDESK_SUBDOMAIN", class="hl-string">"[Configure based on your system]")
ZENDESK_EMAIL = os.environ.get(class="hl-string">"ZENDESK_EMAIL", class="hl-string">"[Configure based on your system]")
ZENDESK_API_TOKEN = os.environ.get(class="hl-string">"ZENDESK_API_TOKEN", class="hl-string">"[Configure based on your system]")
START_TIME = os.environ.get(class="hl-string">"START_TIME", class="hl-string">"[Configure based on your system]")
OUTPUT_FILE = os.environ.get(class="hl-string">"OUTPUT_FILE", class="hl-string">"zendesk_processmind_event_log.csv")
SOURCE_SYSTEM = class="hl-string">"Zendesk Support"

REQUIRED_COLUMNS = [
    class="hl-string">"ServiceRequest",
    class="hl-string">"ActivityName",
    class="hl-string">"StartTime",
    class="hl-string">"SourceSystem",
    class="hl-string">"LastDataUpdate",
]
RECOMMENDED_COLUMNS = [
    class="hl-string">"AssignedAgent",
    class="hl-string">"ServiceRequestType",
    class="hl-string">"Priority",
    class="hl-string">"CommunicationChannel",
    class="hl-string">"SlaTargetResolutionTime",
    class="hl-string">"IsSlaBreached",
]
ALL_COLUMNS = REQUIRED_COLUMNS + RECOMMENDED_COLUMNS


def fail(message: str) -> None:
    raise RuntimeError(message)


def parse_start_time(value: str) -> int:
    if value.isdigit():
        return int(value)
    try:
        return int(datetime.fromisoformat(value.replace(class="hl-string">"Z", class="hl-string">"+00:00")).timestamp())
    except ValueError as exc:
        raise RuntimeError(class="hl-string">"START_TIME must be a Unix timestamp or ISO 8601 timestamp") from exc


def iso_timestamp(value: Any) -> Optional[str]:
    if value is None or value == class="hl-string">"":
        return None
    if isinstance(value, (int, float)):
        return datetime.fromtimestamp(value, tz=timezone.utc).isoformat().replace(class="hl-string">"+00:00", class="hl-string">"Z")
    text = str(value)
    if text.isdigit():
        return datetime.fromtimestamp(int(text), tz=timezone.utc).isoformat().replace(class="hl-string">"+00:00", class="hl-string">"Z")
    return text


def now_utc() -> str:
    return datetime.now(timezone.utc).isoformat().replace(class="hl-string">"+00:00", class="hl-string">"Z")


def api_get(session: requests.Session, path: str, params: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
    url = fclass="hl-string">"https://{ZENDESK_SUBDOMAIN}.zendesk.com{path}"
    while True:
        response = session.get(url, params=params, timeout=60)
        if response.status_code == 429:
            retry_after = int(response.headers.get(class="hl-string">"Retry-After", class="hl-string">"5"))
            time.sleep(max(retry_after, 1))
            continue
        response.raise_for_status()
        return response.json()


def paged_get(session: requests.Session, path: str, params: Optional[Dict[str, Any]] = None) -> Iterable[Dict[str, Any]]:
    next_url: Optional[str] = fclass="hl-string">"https://{ZENDESK_SUBDOMAIN}.zendesk.com{path}"
    next_params = params
    while next_url:
        while True:
            response = session.get(next_url, params=next_params, timeout=60)
            if response.status_code == 429:
                retry_after = int(response.headers.get(class="hl-string">"Retry-After", class="hl-string">"5"))
                time.sleep(max(retry_after, 1))
                continue
            response.raise_for_status()
            payload = response.json()
            break
        for item in payload.get(class="hl-string">"tickets", payload.get(class="hl-string">"audits", payload.get(class="hl-string">"comments", payload.get(class="hl-string">"results", [])))):
            yield item
        next_url = payload.get(class="hl-string">"next_page")
        next_params = None


def get_user_name(session: requests.Session, user_id: Any, cache: Dict[str, str]) -> Optional[str]:
    if not user_id:
        return None
    key = str(user_id)
    if key in cache:
        return cache[key]
    try:
        user = api_get(session, fclass="hl-string">"/api/v2/users/{key}.json").get(class="hl-string">"user", {})
        name = user.get(class="hl-string">"name") or user.get(class="hl-string">"email") or key
        cache[key] = str(name)
        return cache[key]
    except requests.HTTPError:
        cache[key] = key
        return key


def get_group_name(session: requests.Session, group_id: Any, cache: Dict[str, str]) -> Optional[str]:
    if not group_id:
        return None
    key = str(group_id)
    if key in cache:
        return cache[key]
    try:
        group = api_get(session, fclass="hl-string">"/api/v2/groups/{key}.json").get(class="hl-string">"group", {})
        name = group.get(class="hl-string">"name") or key
        cache[key] = str(name)
        return cache[key]
    except requests.HTTPError:
        cache[key] = key
        return key


def ticket_field(ticket: Dict[str, Any], names: List[str]) -> Any:
    for name in names:
        if ticket.get(name) is not None:
            return ticket.get(name)
    return None


def base_row(ticket: Dict[str, Any], activity: str, event_time: Any, last_update: str, user_cache: Dict[str, str]) -> Dict[str, Any]:
    return {
        class="hl-string">"ServiceRequest": str(ticket.get(class="hl-string">"id")),
        class="hl-string">"ActivityName": activity,
        class="hl-string">"StartTime": iso_timestamp(event_time),
        class="hl-string">"SourceSystem": SOURCE_SYSTEM,
        class="hl-string">"LastDataUpdate": last_update,
        class="hl-string">"AssignedAgent": get_user_name(session, ticket.get(class="hl-string">"assignee_id"), user_cache),
        class="hl-string">"ServiceRequestType": ticket.get(class="hl-string">"type"),
        class="hl-string">"Priority": ticket.get(class="hl-string">"priority"),
        class="hl-string">"CommunicationChannel": ticket.get(class="hl-string">"via", {}).get(class="hl-string">"channel") if isinstance(ticket.get(class="hl-string">"via"), dict) else ticket.get(class="hl-string">"via"),
        class="hl-string">"SlaTargetResolutionTime": ticket.get(class="hl-string">"sla_target_resolution_time"),
        class="hl-string">"IsSlaBreached": ticket.get(class="hl-string">"is_sla_breached"),
    }


def add_event(events: List[Dict[str, Any]], seen: set, ticket: Dict[str, Any], activity: str, event_time: Any, last_update: str, user_cache: Dict[str, str], source_key: str) -> None:
    if event_time is None:
        return
    key = (str(ticket.get(class="hl-string">"id")), activity, source_key, str(event_time))
    if key in seen:
        return
    seen.add(key)
    events.append(base_row(ticket, activity, event_time, last_update, user_cache))


def audit_events(session: requests.Session, ticket_id: Any) -> List[Dict[str, Any]]:
    return list(paged_get(session, fclass="hl-string">"/api/v2/tickets/{ticket_id}/audits.json"))


def audit_events_for_field(audit: Dict[str, Any], field_names: List[str]) -> List[Tuple[str, Any, Any]]:
    output: List[Tuple[str, Any, Any]] = []
    for event in audit.get(class="hl-string">"events", []):
        if event.get(class="hl-string">"type") != class="hl-string">"Change":
            continue
        field = str(event.get(class="hl-string">"field_name") or event.get(class="hl-string">"field") or class="hl-string">"")
        if field in field_names:
            output.append((field, event.get(class="hl-string">"previous_value"), event.get(class="hl-string">"value")))
    return output


def is_customer_actor(audit: Dict[str, Any], ticket: Dict[str, Any]) -> bool:
    author_id = audit.get(class="hl-string">"author_id")
    requester_id = ticket.get(class="hl-string">"requester_id")
    return author_id is not None and requester_id is not None and str(author_id) == str(requester_id)


def is_public_comment(event: Dict[str, Any]) -> bool:
    return event.get(class="hl-string">"type") in {class="hl-string">"Comment", class="hl-string">"VoiceComment"} and event.get(class="hl-string">"public") is not False


def is_automated_acknowledgment(event: Dict[str, Any]) -> bool:
    body = str(event.get(class="hl-string">"body") or event.get(class="hl-string">"html_body") or class="hl-string">"").lower()
    marker = str(event.get(class="hl-string">"via", {}).get(class="hl-string">"channel") if isinstance(event.get(class="hl-string">"via"), dict) else event.get(class="hl-string">"via") or class="hl-string">"").lower()
    return class="hl-string">"acknowledg" in body or class="hl-string">"received your request" in body or marker in {class="hl-string">"trigger", class="hl-string">"automation", class="hl-string">"system"}


def extract_ticket_events(session: requests.Session, ticket: Dict[str, Any], last_update: str, user_cache: Dict[str, str], group_cache: Dict[str, str], seen: set) -> List[Dict[str, Any]]:
    events: List[Dict[str, Any]] = []
    ticket_id = ticket.get(class="hl-string">"id")
    audits = audit_events(session, ticket_id)
    audits.sort(key=lambda item: (str(item.get(class="hl-string">"created_at") or class="hl-string">""), str(item.get(class="hl-string">"id") or class="hl-string">"")))
    add_event(events, seen, ticket, class="hl-string">"Service Request Created", ticket.get(class="hl-string">"created_at"), last_update, user_cache, class="hl-string">"ticket_created")
    first_public_agent_reply = False
    first_ack = False
    for audit in audits:
        audit_time = audit.get(class="hl-string">"created_at")
        audit_id = str(audit.get(class="hl-string">"id") or audit_time)
        for event in audit.get(class="hl-string">"events", []):
            event_type = event.get(class="hl-string">"type")
            if event_type in {class="hl-string">"Comment", class="hl-string">"VoiceComment"} and is_public_comment(event):
                if not first_ack and is_automated_acknowledgment(event):
                    add_event(events, seen, ticket, class="hl-string">"Initial Acknowledgment Sent", audit_time, last_update, user_cache, audit_id + class="hl-string">":ack")
                    first_ack = True
                if not is_customer_actor(audit, ticket) and not first_public_agent_reply and not is_automated_acknowledgment(event):
                    add_event(events, seen, ticket, class="hl-string">"First Agent Public Reply Sent", audit_time, last_update, user_cache, audit_id + class="hl-string">":first_agent_reply")
                    first_public_agent_reply = True
                if is_customer_actor(audit, ticket):
                    add_event(events, seen, ticket, class="hl-string">"Customer Responded", audit_time, last_update, user_cache, audit_id + class="hl-string">":customer_reply")
            if event_type == class="hl-string">"Change":
                field = str(event.get(class="hl-string">"field_name") or event.get(class="hl-string">"field") or class="hl-string">"")
                previous = event.get(class="hl-string">"previous_value")
                value = event.get(class="hl-string">"value")
                if field in {class="hl-string">"type", class="hl-string">"priority", class="hl-string">"custom_fields"}:
                    add_event(events, seen, ticket, class="hl-string">"Request Categorized And Prioritized", audit_time, last_update, user_cache, audit_id + class="hl-string">":category_priority:" + field)
                if field in {class="hl-string">"assignee_id", class="hl-string">"assignee"} and value not in (None, class="hl-string">"", 0, class="hl-string">"0"):
                    add_event(events, seen, ticket, class="hl-string">"Request Assigned To Agent", audit_time, last_update, user_cache, audit_id + class="hl-string">":assignee")
                if field in {class="hl-string">"group_id", class="hl-string">"group"} and previous not in (None, class="hl-string">"", 0, class="hl-string">"0") and value not in (None, class="hl-string">"", 0, class="hl-string">"0") and str(previous) != str(value):
                    get_group_name(session, value, group_cache)
                    add_event(events, seen, ticket, class="hl-string">"Internal Escalation Triggered", audit_time, last_update, user_cache, audit_id + class="hl-string">":group")
                if field == class="hl-string">"status":
                    old_status = str(previous or class="hl-string">"").lower()
                    new_status = str(value or class="hl-string">"").lower()
                    if new_status == class="hl-string">"pending":
                        add_event(events, seen, ticket, class="hl-string">"Information Requested From Customer", audit_time, last_update, user_cache, audit_id + class="hl-string">":pending")
                    if new_status == class="hl-string">"solved":
                        add_event(events, seen, ticket, class="hl-string">"Service Request Resolved", audit_time, last_update, user_cache, audit_id + class="hl-string">":solved")
                    if old_status == class="hl-string">"solved" and new_status == class="hl-string">"open":
                        add_event(events, seen, ticket, class="hl-string">"Service Request Reopened", audit_time, last_update, user_cache, audit_id + class="hl-string">":reopened")
                    if new_status == class="hl-string">"closed":
                        add_event(events, seen, ticket, class="hl-string">"Service Request Closed", audit_time, last_update, user_cache, audit_id + class="hl-string">":closed")
                if field in {class="hl-string">"satisfaction_survey_sent", class="hl-string">"csat_survey_sent"} and str(value).lower() in {class="hl-string">"true", class="hl-string">"1", class="hl-string">"yes", class="hl-string">"sent"}:
                    add_event(events, seen, ticket, class="hl-string">"Satisfaction Survey Sent", audit_time, last_update, user_cache, audit_id + class="hl-string">":survey_sent")
                if field in {class="hl-string">"sla_breached", class="hl-string">"is_sla_breached"} and str(value).lower() in {class="hl-string">"true", class="hl-string">"1", class="hl-string">"yes"}:
                    row = base_row(ticket, class="hl-string">"SLA Breach Occurred", audit_time, last_update, user_cache)
                    row[class="hl-string">"IsSlaBreached"] = True
                    key = (str(ticket_id), class="hl-string">"SLA Breach Occurred", audit_id, str(audit_time))
                    if key not in seen:
                        seen.add(key)
                        events.append(row)
    try:
        satisfaction = api_get(session, fclass="hl-string">"/api/v2/tickets/{ticket_id}/satisfaction_rating.json").get(class="hl-string">"satisfaction_rating")
    except requests.HTTPError:
        satisfaction = None
    if satisfaction:
        add_event(events, seen, ticket, class="hl-string">"Satisfaction Rating Received", satisfaction.get(class="hl-string">"created_at") or satisfaction.get(class="hl-string">"updated_at"), last_update, user_cache, str(satisfaction.get(class="hl-string">"id") or class="hl-string">"rating"))
    return events


def validate_rows(rows: List[Dict[str, Any]]) -> None:
    activities = {
        class="hl-string">"Service Request Created",
        class="hl-string">"Initial Acknowledgment Sent",
        class="hl-string">"Request Categorized And Prioritized",
        class="hl-string">"Request Assigned To Agent",
        class="hl-string">"First Agent Public Reply Sent",
        class="hl-string">"Information Requested From Customer",
        class="hl-string">"Customer Responded",
        class="hl-string">"Internal Escalation Triggered",
        class="hl-string">"SLA Breach Occurred",
        class="hl-string">"Service Request Resolved",
        class="hl-string">"Service Request Reopened",
        class="hl-string">"Satisfaction Survey Sent",
        class="hl-string">"Satisfaction Rating Received",
        class="hl-string">"Service Request Closed",
    }
    missing_columns = [column for column in ALL_COLUMNS if column not in rows[0]] if rows else ALL_COLUMNS
    if missing_columns:
        fail(class="hl-string">"Missing output columns: " + class="hl-string">", ".join(missing_columns))
    invalid = [row for row in rows if not row.get(class="hl-string">"ServiceRequest") or not row.get(class="hl-string">"ActivityName") or not row.get(class="hl-string">"StartTime") or not row.get(class="hl-string">"SourceSystem") or not row.get(class="hl-string">"LastDataUpdate")]
    if invalid:
        fail(fclass="hl-string">"{len(invalid)} rows are missing required values")
    unexpected = sorted({row[class="hl-string">"ActivityName"] for row in rows} - activities)
    if unexpected:
        fail(class="hl-string">"Unexpected activity names: " + class="hl-string">", ".join(unexpected))


def main() -> None:
    global session
    if ZENDESK_SUBDOMAIN.startswith(class="hl-string">"[") or ZENDESK_EMAIL.startswith(class="hl-string">"[") or ZENDESK_API_TOKEN.startswith(class="hl-string">"[") or START_TIME.startswith(class="hl-string">"["):
        fail(class="hl-string">"Configure ZENDESK_SUBDOMAIN, ZENDESK_EMAIL, ZENDESK_API_TOKEN, and START_TIME")
    start_time = parse_start_time(START_TIME)
    session = requests.Session()
    session.auth = (fclass="hl-string">"{ZENDESK_EMAIL}/token", ZENDESK_API_TOKEN)
    session.headers.update({class="hl-string">"Accept": class="hl-string">"application/json"})
    last_update = now_utc()
    user_cache: Dict[str, str] = {}
    group_cache: Dict[str, str] = {}
    seen = set()
    rows: List[Dict[str, Any]] = []
    tickets = paged_get(session, class="hl-string">"/api/v2/incremental/tickets/cursor.json", {class="hl-string">"start_time": start_time})
    for ticket in tickets:
        rows.extend(extract_ticket_events(session, ticket, last_update, user_cache, group_cache, seen))
    rows.sort(key=lambda row: (row[class="hl-string">"ServiceRequest"], row[class="hl-string">"StartTime"], row[class="hl-string">"ActivityName"]))
    validate_rows(rows)
    with open(OUTPUT_FILE, class="hl-string">"w", newline=class="hl-string">"", encoding=class="hl-string">"utf-8") as output:
        writer = csv.DictWriter(output, fieldnames=ALL_COLUMNS, extrasaction=class="hl-string">"ignore")
        writer.writeheader()
        writer.writerows(rows)
    print(json.dumps({class="hl-string">"output_file": OUTPUT_FILE, class="hl-string">"rows": len(rows), class="hl-string">"last_data_update": last_update}))


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

步骤

选择支持Zendesk Support连接器的商业ETL或ELT平台，并确认该连接器能够复制您Zendesk账户中的工单、工单事件或审计记录、评论、用户、组、满意度评分、SLA数据以及触发器或自动化结果。
使用您所在组织和所选连接器支持的身份验证方式创建Zendesk API集成。仅授予工单、审计记录、评论、用户、组、满意度和SLA数据所需的读取权限。将连接字符串和凭据存储在工具的安全凭据库中。
配置连接器，将所需的Zendesk对象复制到目标数据仓库。使用连接器生成的源对象，不要假定固定的表名或列名。如果连接器使用账户专属名称，请在项目文档中将实际对象和字段映射记录为[您的表名]和[您的列名]。
使用源系统支持的更新时间戳、游标或变更跟踪机制配置增量提取。纳入所选分析周期的历史数据，通常为三至六个月，并保留源事件或审计标识符，以便对重复更新进行去重。
构建标准化暂存模型，将工单记录与工单历史、审计或事件记录、评论、用户、组、满意度评分、SLA记录以及连接器专属的触发器或自动化记录合并。只要数据可用，就保留原始源时间戳、源记录标识符、工单标识符、执行者标识符、事件类型、字段名、原值和新值。
创建事件转换逻辑，为每个明确观测到的活动生成一行记录。将工单创建映射为Service Request Created，将符合条件的自动首次响应映射为Initial Acknowledgment Sent，将类型、类别或优先级变更映射为Request Categorized And Prioritized，将受理人变更映射为Request Assigned To Agent，将首次符合条件的客服公开回复映射为First Agent Public Reply Sent，将与信息请求相关的待处理状态转换映射为Information Requested From Customer，将客户回复映射为Customer Responded，将组变更映射为Internal Escalation Triggered，将观测到的SLA违约记录或违约事件映射为SLA Breach Occurred，将已解决状态转换映射为Service Request Resolved，将由客户回复导致的已解决到开放状态转换映射为Service Request Reopened，将观测到的调查发送映射为Satisfaction Survey Sent，将已提交的评分映射为Satisfaction Rating Received，并将永久关闭状态转换映射为Service Request Closed。
不要仅根据当前工单状态或经过的时间创建推断记录。如果源系统未明确提供调查发送、SLA违约、自动确认或永久关闭事件，请保留该活动为空，或使用单独记录且已明确记录这些事件的源数据流。ProcessMind按原样读取事件日志，不会推导缺失事件。
在可用时使用源事件或审计标识符对事件去重。如果不存在稳定标识符，请使用有文档记录的复合键，其中包含ServiceRequest、ActivityName、StartTime、执行者和源记录标识符。保留Customer Responded、Information Requested From Customer、Internal Escalation Triggered和Service Request Reopened等活动的多次合法发生记录。
填充必需的输出列ServiceRequest、ActivityName、StartTime、SourceSystem和LastDataUpdate。当源系统提供时，添加AssignedAgent、ServiceRequestType、Priority、CommunicationChannel、SlaTargetResolutionTime和IsSlaBreached。统一使用UTC时间戳标准，并保留源时区转换规则。
将最终事件日志导出为兼容ProcessMind的CSV，或加载到已配置的ProcessMind数据接入位置。确保每次活动发生对应一行记录，ServiceRequest为案例标识符，ActivityName包含准确的活动标签，StartTime为有效时间戳，并在技术上可用时确保每行输出都包含必需的属性。

配置

**连接器和源范围：**为Zendesk Support配置所选商业ETL或ELT工具，并选择能够提供工单、工单审计记录或事件、评论、用户、组、满意度评分、SLA信息以及自动化或触发器结果的源对象。对象名称因连接器和账户而异，因此应使用连接器的实际目录，不要假定表名。
**身份验证和授权：**使用您所在组织批准的Zendesk身份验证选项。为所有选定的源对象授予读取权限，包括历史工单事件和客户满意度数据。确认API访问权限、所需的Zendesk套餐功能以及任何连接器许可证均已启用。
**日期范围：**先加载三至六个月的历史数据。只有在验证API配额、数据仓库容量、事件量以及历史审计、SLA和满意度记录的可用性后，才扩大范围。
**增量加载：**优先使用连接器原生游标、变更跟踪或源更新时间戳。安排一次初始历史加载，随后频繁执行增量加载。为延迟到达的评论、评分、SLA结果和工单更新设置回溯窗口。
**活动筛选：**将源记录筛选至所选分析周期和相关Zendesk Support实例。不要仅按当前工单状态筛选，因为历史活动可能属于当前状态已发生变化的工单。
**公司或账户筛选：**如果连接器提供多个Zendesk实例、品牌、组织或业务单元，请使用连接器中的实际字段应用已批准的公司或账户筛选条件。不要假定存在Company Code字段。
**事件映射：**维护明确的映射表，将源事件类型、字段名、状态转换、执行者类型和自动化标识映射到14个必需的ActivityName值。每一条生成的记录都必须有明确的源证据。
**确认识别：**仅当源系统识别出自动首次响应或等效的触发器生成通知时，才将其映射为Initial Acknowledgment Sent。没有源证据时，不要将客服回复归类为自动确认。
**SLA处理：**使用明确的SLA违约记录或连接器提供的SLA结果。将目标时长存储在SlaTargetResolutionTime中，将违约结果存储在IsSlaBreached中。除非SLA日历、政策、工作时间和目标均可用且计算过程有文档记录，否则不要根据经过的时间计算违约。
**性能：**按日期限制初始加载范围，仅投影所需列，使用增量复制；在支持的情况下，按ServiceRequest和StartTime启用数据仓库分区或聚簇，并在原始数据接入后物化标准化事件模型。避免反复扫描全部工单历史。
**数据质量：**保留源标识符和时间戳，将时间戳标准化为UTC，统一布尔值，并在源系统未提供某项属性时保留空值。不要为缺失的客服、类别、优先级、SLA或渠道数据填充虚构值。
**刷新时间戳：**将LastDataUpdate设置为提取或转换任务完成时的时间戳，并在整个事件日志中统一使用同一时区和格式。

a 示例查询 config

WITH source_data AS (
    SELECT
        [Your ticket identifier column] AS service_request,
        [Your source event identifier column] AS source_event_id,
        [Your event type column] AS source_event_type,
        [Your field name column] AS source_field_name,
        [Your previous value column] AS previous_value,
        [Your new value column] AS new_value,
        [Your event timestamp column] AS event_time,
        [Your actor type column] AS actor_type,
        [Your actor identifier column] AS actor_id,
        [Your assigned agent column] AS assigned_agent,
        [Your request type column] AS service_request_type,
        [Your priority column] AS priority,
        [Your communication channel column] AS communication_channel,
        [Your SLA target resolution time column] AS sla_target_resolution_time,
        [Your SLA breached flag column] AS is_sla_breached,
        [Your automated response flag column] AS is_automated_response,
        [Your public comment flag column] AS is_public_comment,
        [Your customer reply flag column] AS is_customer_reply,
        [Your survey sent flag column] AS is_survey_sent,
        [Your satisfaction rating column] AS satisfaction_rating,
        [Your source system value] AS source_system,
        [Your extraction timestamp column] AS last_data_update
    FROM [Your normalized Zendesk event source]
    WHERE [Your event timestamp column] >= [Your start date parameter]
      AND [Your event timestamp column] < [Your end date parameter]
),
activity_events AS (
    SELECT service_request, 'Service Request Created' AS activity_name, event_time AS start_time, source_event_id, assigned_agent, service_request_type, priority, communication_channel, sla_target_resolution_time, is_sla_breached, source_system, last_data_update
    FROM source_data
    WHERE source_event_type = 'ticket_created'
    UNION ALL
    SELECT service_request, 'Initial Acknowledgment Sent', event_time, source_event_id, assigned_agent, service_request_type, priority, communication_channel, sla_target_resolution_time, is_sla_breached, source_system, last_data_update
    FROM source_data
    WHERE source_event_type = 'comment_added' AND is_automated_response = TRUE AND is_public_comment = TRUE
    UNION ALL
    SELECT service_request, 'Request Categorized And Prioritized', event_time, source_event_id, assigned_agent, service_request_type, priority, communication_channel, sla_target_resolution_time, is_sla_breached, source_system, last_data_update
    FROM source_data
    WHERE source_event_type = 'ticket_field_changed' AND source_field_name IN ('type', 'category', 'priority')
    UNION ALL
    SELECT service_request, 'Request Assigned To Agent', event_time, source_event_id, assigned_agent, service_request_type, priority, communication_channel, sla_target_resolution_time, is_sla_breached, source_system, last_data_update
    FROM source_data
    WHERE source_event_type = 'ticket_field_changed' AND source_field_name IN ('assignee', 'assigned_agent')
    UNION ALL
    SELECT service_request, 'First Agent Public Reply Sent', event_time, source_event_id, assigned_agent, service_request_type, priority, communication_channel, sla_target_resolution_time, is_sla_breached, source_system, last_data_update
    FROM source_data
    WHERE source_event_type = 'comment_added' AND is_public_comment = TRUE AND actor_type = 'agent' AND is_automated_response = FALSE
    UNION ALL
    SELECT service_request, 'Information Requested From Customer', event_time, source_event_id, assigned_agent, service_request_type, priority, communication_channel, sla_target_resolution_time, is_sla_breached, source_system, last_data_update
    FROM source_data
    WHERE source_event_type = 'ticket_status_changed' AND new_value = 'pending'
    UNION ALL
    SELECT service_request, 'Customer Responded', event_time, source_event_id, assigned_agent, service_request_type, priority, communication_channel, sla_target_resolution_time, is_sla_breached, source_system, last_data_update
    FROM source_data
    WHERE source_event_type = 'comment_added' AND is_public_comment = TRUE AND is_customer_reply = TRUE
    UNION ALL
    SELECT service_request, 'Internal Escalation Triggered', event_time, source_event_id, assigned_agent, service_request_type, priority, communication_channel, sla_target_resolution_time, is_sla_breached, source_system, last_data_update
    FROM source_data
    WHERE source_event_type = 'ticket_field_changed' AND source_field_name IN ('group', 'assigned_group')
    UNION ALL
    SELECT service_request, 'SLA Breach Occurred', event_time, source_event_id, assigned_agent, service_request_type, priority, communication_channel, sla_target_resolution_time, TRUE, source_system, last_data_update
    FROM source_data
    WHERE source_event_type IN ('sla_breach', 'sla_breached') OR is_sla_breached = TRUE
    UNION ALL
    SELECT service_request, 'Service Request Resolved', event_time, source_event_id, assigned_agent, service_request_type, priority, communication_channel, sla_target_resolution_time, is_sla_breached, source_system, last_data_update
    FROM source_data
    WHERE source_event_type = 'ticket_status_changed' AND new_value = 'solved'
    UNION ALL
    SELECT service_request, 'Service Request Reopened', event_time, source_event_id, assigned_agent, service_request_type, priority, communication_channel, sla_target_resolution_time, is_sla_breached, source_system, last_data_update
    FROM source_data
    WHERE source_event_type = 'ticket_status_changed' AND previous_value = 'solved' AND new_value = 'open' AND is_customer_reply = TRUE
    UNION ALL
    SELECT service_request, 'Satisfaction Survey Sent', event_time, source_event_id, assigned_agent, service_request_type, priority, communication_channel, sla_target_resolution_time, is_sla_breached, source_system, last_data_update
    FROM source_data
    WHERE source_event_type IN ('satisfaction_survey_sent', 'csat_survey_sent') OR is_survey_sent = TRUE
    UNION ALL
    SELECT service_request, 'Satisfaction Rating Received', event_time, source_event_id, assigned_agent, service_request_type, priority, communication_channel, sla_target_resolution_time, is_sla_breached, source_system, last_data_update
    FROM source_data
    WHERE source_event_type IN ('satisfaction_rating_received', 'csat_rating_received') OR satisfaction_rating IS NOT NULL
    UNION ALL
    SELECT service_request, 'Service Request Closed', event_time, source_event_id, assigned_agent, service_request_type, priority, communication_channel, sla_target_resolution_time, is_sla_breached, source_system, last_data_update
    FROM source_data
    WHERE source_event_type = 'ticket_status_changed' AND new_value = 'closed'
),
deduplicated_events AS (
    SELECT
        service_request,
        activity_name,
        start_time,
        assigned_agent,
        service_request_type,
        priority,
        communication_channel,
        sla_target_resolution_time,
        is_sla_breached,
        source_system,
        last_data_update,
        ROW_NUMBER() OVER (PARTITION BY service_request, activity_name, source_event_id ORDER BY start_time) AS event_rank
    FROM activity_events
)
SELECT
    service_request AS ServiceRequest,
    activity_name AS ActivityName,
    start_time AS StartTime,
    source_system AS SourceSystem,
    last_data_update AS LastDataUpdate,
    assigned_agent AS AssignedAgent,
    service_request_type AS ServiceRequestType,
    priority AS Priority,
    communication_channel AS CommunicationChannel,
    sla_target_resolution_time AS SlaTargetResolutionTime,
    is_sla_breached AS IsSlaBreached
FROM deduplicated_events
WHERE event_rank = 1
ORDER BY service_request, start_time, activity_name;

准备好开始了吗？

立即使用此模板准备数据，开始优化客户服务。发现有价值的洞察，提升Zendesk支持运营效率。

返回改进指南下载模板

消除客户服务瓶颈，立即提升CSAT

减少重复联系，实现80%的首次联系解决率，提升客户满意度。

开始免费试用

无需信用卡

您的客户服务数据模板

您的客户服务数据模板

客户服务属性

客户服务活动

提取指南

Zendesk Explore报告与数据导出的用户界面

步骤

配置

a 示例查询 config

使用Zendesk Support REST API的Python脚本

步骤

配置

a 示例查询 python

用于批量数据的Zendesk Incremental Exports API

步骤

配置

a 示例查询 python

Zendesk第三方ETL工具连接器

步骤

配置

a 示例查询 config

准备好开始了吗？

消除客户服务瓶颈，立即提升CSAT

您的客户服务数据模板

您的客户服务数据模板

客户服务属性

客户服务活动

提取指南

Zendesk Explore报告与数据导出的用户界面

步骤

配置

a 示例查询 config

使用Zendesk Support REST API的Python脚本

步骤

配置

a 示例查询 python

用于批量数据的Zendesk Incremental Exports API

步骤

配置

a 示例查询 python

Zendesk第三方ETL工具连接器

步骤

配置

a 示例查询 config

准备好开始了吗？

消除客户服务瓶颈，立即提升CSAT

我们重视您的隐私