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

您的客户服务数据模板

Zendesk Support

← 返回改进指南

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

您的客户服务数据模板

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

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

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

下载模板

客户服务属性

建议将这些数据字段纳入事件日志，以便全面分析客户服务。

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	服务请求生命周期中发生的具体事件或任务名称。
说明活动名称用于描述客户服务流程中的单个步骤或节点，例如“服务请求已创建”“请求已分配给客服人员”或“服务请求已解决”。这些事件均带有时间戳，共同构成每个服务请求的操作顺序。此属性对于可视化流程、发现流程变体以及分析事件的频率和顺序至关重要。它能帮助分析人员了解实际执行的操作，并识别常见路径、瓶颈以及偏离标准过程的情况。重要性此属性定义流程步骤，可用于绘制流程图并分析流程路径和变体。获取途径从记录变更和操作的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`字段。示例低普通高紧急
	已分配的客服人员 AssignedAgent	负责处理服务请求的客服人员姓名或ID。
说明此属性用于标识特定时间点负责某项活动或服务请求的客服人员。如果请求被重新分配，负责人可能会在生命周期内发生变化。按已分配的客服人员进行分析，有助于了解工作量、绩效和效率。此属性支持“客服人员工作量与效率”仪表板，可比较不同客服人员的处理时长和案例数量，帮助发现辅导机会并平衡工作量。重要性跟踪执行操作的客服人员，以便分析个人绩效、工作量分配和资源配置。获取途径 Zendesk Tickets API的`assignee_id`字段。可通过Users API获取用户详情。示例 John SmithJane DoeSupportBot
	是否违反SLA IsSlaBreached	用于标记服务请求的解决时间是否超过SLA目标的布尔值。
说明此计算属性是一个简单的真/假标记，用于表示服务请求是否未能满足规定的解决时限。该标记通过比较实际解决时长与SLA目标时限得出。此标记简化了SLA合规分析，是“SLA合规绩效”仪表板和“SLA合规率”KPI的核心数据点，可快速汇总并直观展示达到或未达到服务目标的请求数量。重要性为每个案例提供明确的SLA绩效结果，简化合规监控与报告。获取途径在数据转换期间，将总解决时间与SlaTargetResolutionTime进行比较后计算得出。示例 truefalse
	服务请求类型 ServiceRequestType	服务请求的分类，例如“Question”“Incident”“Problem”或“Task”。
说明此属性根据服务请求的性质对其进行分类。通常在创建请求或初步分流时设置，有助于确定适当的工作流和优先级。按服务请求类型细分数据是分析的基础。您可以比较不同问题类型的解决时间、升级率和流程路径，例如查看“服务请求解决时间分析”和“内部升级率及原因”仪表板。这有助于发现某些类型的请求是否更难处理或处理效率更低。重要性对请求进行分类，以便比较和分析不同问题类型的绩效，为有针对性的流程改进提供依据。获取途径 Zendesk Tickets API的`type`字段。示例问题事件问题Task
	沟通渠道 CommunicationChannel	提交服务请求或进行沟通的渠道。
说明此属性用于标识所使用的沟通方式，例如电子邮件、网页表单、聊天或电话，反映客户与服务台的互动方式。了解各渠道的使用情况有助于规划资源并改善客户体验。“沟通渠道使用概览”仪表板会分析这些数据，查看哪些渠道最常用，以及某些渠道是否与更长的解决时间或不同的流程路径相关。这些信息有助于确定应在哪些方面投入资源，以改进服务或实现自动化。重要性展示客户与客服人员的互动情况，便于分析各渠道的效率及其对流程和客户体验的影响。获取途径 Zendesk Tickets API中的`via.channel`字段。示例网页电子邮件API聊天
	产品或服务类别 ProductServiceCategory	客户请求所涉及的具体产品、服务或功能。
说明此属性根据产品或服务领域对服务请求进行分类，提供更详细的背景信息。分类通常由客服人员手动设置，也可以根据请求内容自动设置。此分类对“请求分类准确率”和“调查周期时间细分”仪表板至关重要。它有助于深入分析哪些产品引发的支持请求最多、哪些请求最难解决，以及初始分类是否与最终解决方案相符，从而改进请求分派和客服培训。重要性将请求关联到具体业务领域、产品或服务，便于针对问题领域及其对流程的影响开展分析。获取途径通常是Zendesk中的自定义工单字段。具体字段名称取决于Zendesk的配置。示例移动应用订阅管理API集成硬件
	信息请求次数 InformationRequestCount	针对单个服务请求向客户索取信息的总次数。
说明此计算指标统计每个服务请求中“向客户索取信息”活动的发生次数。次数较多可能说明客服人员没有在一开始收集所有必要信息。此属性用于“重复索取信息分析”仪表板。跟踪该次数有助于发现流程低效之处和需要加强客服培训的领域。减少索取信息的次数，可以显著缩短解决时间并改善客户体验。重要性量化与客户反复沟通的情况，揭示导致解决时间延长、客户体验下降的低效环节。获取途径统计每个服务请求中ActivityName为“Information Requested From Customer”的事件数量后计算得出。示例 013
	客服团队 AgentGroup	负责处理服务请求的支持组或团队。
说明此属性表示负责处理服务请求的客服团队。请求通常会根据技能、产品领域或语言分配给特定团队。按客服团队分析，有助于了解团队层面的绩效、工作量分布以及团队间的升级模式。与分析单个客服人员相比，这种分析视角更全面，也有助于发现特定部门或职能中的系统性问题。重要性记录团队责任归属，便于分析团队绩效、团队间交接情况，以及不同支持层级或专业领域的资源分配。获取途径 Zendesk Tickets API中的`group_id`字段。可通过Groups API获取团队详情。示例一线支持技术支持账单
	是否重新打开 IsReopened	用于标记服务请求在标记为已解决后是否重新打开的布尔值。
说明如果服务请求在此前已标记为已解决或已关闭，之后又转回打开状态，此属性就会标记为真。这表明最初的解决方案未能解决问题。此标记对跟踪返工和首次联系解决失败至关重要。它直接支持“服务请求重新打开趋势”仪表板和“服务请求重新打开率”KPI，便于统计和分析需要进一步处理的案例，而这些案例往往反映了更深层的问题。重要性识别返工和未能解决的问题，帮助衡量所提供解决方案的质量与效果。获取途径在数据转换期间，检查工单状态是否从“已解决”或“已关闭”变为“打开”后计算得出。示例 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	用于说明请求最终解决或关闭原因的代码或类别。
说明解决代码以结构化方式记录服务请求的处理结果，例如“客服人员已解决”“重复请求”“无需处理”或“已知问题”。相比简单的“已关闭”状态，它能提供更多背景信息。此属性对根因分析尤其有价值。在“服务请求重新打开趋势”仪表板中，按解决代码分析重新打开率，可以发现某些解决方案是否效果不佳，导致客户再次联系支持团队。重要性帮助了解服务请求的处理结果，为根因分析以及查明请求重新打开的原因提供重要信息。获取途径通常是Zendesk中的自定义工单字段。具体字段名称取决于Zendesk的配置。示例首次联系即解决升级至二线支持等待客户回复产品缺陷

必填建议填写选填

客户服务活动

建议在事件日志中记录这些关键流程步骤和里程碑，以准确发现客户服务流程。

6 建议填写 8 选填

	活动	说明
	已向客户索取信息	客服人员需要客户提供更多信息才能继续处理，并将工单状态改为“pending”时，会发生此活动。此状态变更明确表示流程正在等待外部人员回复。
重要性此活动反映了流程对客户的依赖，并会暂停内部SLA计时。同一工单频繁或反复出现此情况，可能说明初次收集信息不完整，导致解决时间延长。获取途径这是Zendesk Ticket Audits API中明确记录的状态变更事件。“status”字段变为“pending”时，系统会记录此事件。采集在Ticket Audits中查找工单状态被设为“pending”的“Change”事件。事件类型 explicit
	服务请求已关闭	这是最后一项活动，标志着服务请求永久关闭。通常，工单标记为“solved”后经过一段设定时间，且客户没有新回复，系统便会自动关闭工单。
重要性作为最终结束事件，它标志着工单生命周期结束。从“solved”到“closed”的时间段是工单可能重新打开的窗口；“closed”事件则确认解决方案已被接受。获取途径当“status”字段被设为“closed”时，Zendesk Ticket Audits API会明确记录此状态变更。采集在Ticket Audits中查找工单状态被设为“closed”的“Change”事件。事件类型 explicit
	服务请求已创建	此活动标志着客户服务流程的开始：客户通过电子邮件、网页表单或聊天等渠道提交请求后，Zendesk会生成新工单。系统会在创建时记录此事件，并附上唯一工单ID和时间戳。
重要性作为主要开始事件，它是计算案例总时长和分析一段时间内请求量的基础。它也为衡量首次响应时间和总解决时间等关键绩效指标提供基准。获取途径这是Zendesk Ticket Audits API中明确记录的事件，对应工单的“Create”事件，并提供初始创建时间戳。采集从Ticket Audits日志中的工单创建事件获取。事件类型 explicit
	服务请求已解决	客服人员为客户提供解决方案后，会将服务请求标记为“solved”。这只是临时状态，工单在永久关闭前仍可能因客户回复而重新打开。
重要性这是衡量解决时间和客服人员效率的主要节点。它表示客服人员认为工作已完成，也为工单重新打开时分析返工情况提供依据。获取途径当“status”字段被设为“solved”时，Zendesk Ticket Audits API会明确记录此状态变更。采集在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
	客户已回复	客户回复工单时会触发此事件，通常发生在工单处于“pending”状态时。Zendesk会自动将工单状态从“pending”改回“open”，表示客服人员可以继续处理。
重要性此活动标志着等待时间结束，并触发流程继续进行。分析客户的回复耗时，有助于了解客服人员的请求是否清晰。获取途径此事件对应最终用户发布的新公开评论，并会在Ticket Audits API中触发从“pending”到“open”的明确状态变更。采集跟踪状态从“pending”变为“open”的“Change”事件，或识别最终用户发布的新公开评论。事件类型 explicit
	客服人员已发送首条公开回复	此活动标志着客服人员首次向客户发送公开评论，与自动确认消息不同。它是衡量客服人员开始处理客户问题的重要指标。
重要性这是衡量“First Reply Time”SLA的关键节点，也是服务响应速度的重要指标。它标志着沟通从自动回复转为由人工主动调查和提供支持。获取途径在Ticket Comments数据流中，找到由人工客服人员（而非自动化系统用户）发布的第一条公开评论即可识别。采集扫描工单评论，筛选客服人员发布的公开评论，并选取时间戳最早的一条。事件类型 inferred
	已发送满意度调查	指自动向客户发送客户满意度（CSAT）调查的时间点。通常在工单标记为已解决后不久发送。
重要性此活动启动客户反馈流程。了解调查何时发送以及是否发送，有助于结合背景解读满意度评分，并衡量反馈计划的成效。获取途径可根据自动化日志或工单新增的特定标签推断。Ticket Audits API中的“satisfaction_rating”部分也会记录调查发送时间。采集查找“csat_sent”等标签，或使用满意度评分调查发出的时间戳。事件类型 inferred
	已发送首次确认	指发送给客户的自动首次回复，用于确认已收到请求。通常由Zendesk触发器在工单创建后立即发送模板邮件通知。
重要性跟踪此活动对于衡量首次响应速度和管理客户预期至关重要。从请求创建到发送确认的时间，是影响客户满意度的一项关键指标。获取途径如果工单的第一条公开评论由自动化用户发布，或在工单创建后几秒内发布，可据此推断此活动。您可以分析Ticket Comments数据流来识别。采集找出工单创建后由自动化或触发器立即发布的第一条公开评论。事件类型 inferred
	已收到满意度评分	客户提交满意度调查回复时会发生此事件，回复中包含“Good”或“Bad”等评分。评分及相关评论会记录在工单中。
重要性客户直接反馈对于衡量服务质量和客户感受非常重要。结合流程分析这些评分，有助于了解特定活动或客服人员与结果之间的关联。获取途径当“satisfaction_rating”字段记录了客户评分和评论时，Ticket Audits API会将其作为“Change”事件记录。采集筛选Ticket Audit日志，查找“satisfaction_rating”字段的变更。事件类型 explicit
	已触发内部升级	指将服务请求转交给其他内部团队或更高级别的支持团队。通常可根据工单的分配组发生变更来推断。
重要性跟踪升级有助于发现流程薄弱环节、一线支持的知识缺口以及复杂的请求类型。升级率偏高可能意味着需要加强培训或完善流程文档。获取途径可根据Ticket Audits API中的“Change”事件推断：当“group_id”字段发生变更时，即表示团队之间进行了交接。采集监控Ticket Audits日志中“group_id”字段的变更。事件类型 inferred
	请求已分类并确定优先级	客服人员或自动化设置或更新工单字段（如类型、类别和优先级）时，会发生此活动。此步骤会作为变更事件记录在工单历史中。
重要性正确分类并确定优先级，对于高效分派和资源配置至关重要。分析此活动有助于评估初步分流是否准确，以及它对解决时间的影响。获取途径可从Zendesk Ticket Audits API中的“Change”事件获取。查找“priority”“type”或分类相关自定义字段的首次更新即可识别。采集筛选Ticket Audit日志，查找工单创建后关键分类字段的首个“Change”事件。事件类型 explicit

建议填写选填

数据提取指南

如何从Zendesk Support获取数据

步骤

**打开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等工单属性添加筛选条件，将分析范围限定到特定服务或团队。
**计算属性：**此方法主要依靠标准计算属性记录默认属性中没有的事件时间戳，例如状态变更或升级。请确认您有权限创建这些属性。

查询示例 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。进入Admin Center，然后依次打开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_SUBDOMAIN、ZENDESK_EMAIL和API_TOKEN替换为您Zendesk实例的具体信息和凭据。
**设置提取时间范围：**找到脚本中的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令牌。

查询示例 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时间戳，并返回该时间点之后创建或更新的记录。若要完整加载历史数据，请将[根据您的系统配置]设为初始起始时间，并保存返回的结束时间，供后续增量运行使用。日常分析可在运营历史数据充足时采用3至6个月的时间范围。
配置Zendesk子域名、身份验证凭据、起始时间戳和可选的公司专属筛选条件，然后运行Python提取脚本。脚本会请求增量工单数据、处理分页，并在转换前保存原始工单记录及可用的事件详情。
如果增量工单响应未包含足够的字段变更、分配、评论或状态详情，请获取每个工单的审计记录或事件历史。仅使用账户中已启用且获授权的Zendesk API资源。保留返回的源工单ID、审计ID、评论ID、事件时间戳、操作人、字段名称、变更前的值和变更后的值。
将时间戳统一转换为UTC，并将Unix时间戳或ISO 8601值转换为一致的日期时间格式。按ServiceRequest、事件时间戳和源事件标识符对源记录排序。使用可用的最具体稳定源标识符（例如审计ID或评论ID）去重。
为全部14项必需活动明确创建事件日志行。不要依赖ProcessMind根据状态变更或字段值推断活动。根据工单创建记录生成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。
填写必需列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政策和违约时间戳。如果通过计算得出违约状态，请记录所用政策、工作时间、暂停规则和时区。适用工作时间规则时，不要仅根据实际经过时间计算是否违约。
**性能：**逐页处理数据，以流式或批量方式读取记录，缓存工单详情，限制并发数，遵循Zendesk速率限制响应，并在页面成功处理后保存进度。避免反复请求完整历史数据。
**数据保留：**保留原始API响应、提取水位时间、请求元数据和转换日志，以便核对和重新运行。
**前提条件：**需要Zendesk API访问权限、读取相关工单和事件资源的权限、按需启用工单审计和评论；如需CSAT活动，还需访问满意度数据；如需SLA违约活动，还需访问SLA数据。功能是否可用取决于Zendesk方案和账户配置。

查询示例 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，标准化布尔值；源数据未提供属性时保留null。不要为缺失的客服人员、类别、优先级、SLA或渠道数据编造替代值。
**刷新时间戳：**将LastDataUpdate设为提取或转换任务完成的时间戳，并在整个事件日志中使用一致的时区和格式。

查询示例 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 Support运营情况，提升运营效率。

返回改进指南

下载模板

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

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

开始免费试用

无需信用卡

您的客户服务数据模板

您的客户服务数据模板

客户服务属性

客户服务活动

数据提取指南

使用Zendesk Explore界面生成报表并导出数据

步骤

配置

查询示例 config

使用Zendesk Support REST API的Python脚本

步骤

配置

查询示例 python

使用Zendesk Incremental Exports API批量导出数据

步骤

配置

查询示例 python

Zendesk第三方ETL工具连接器

步骤

配置

查询示例 config

准备好开始了吗？

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