您的线索到回款数据模板

Pipedrive

← 返回改进指南

线索到收款属性 (20) 线索到收款活动 (13) 提取指南 (3)

您的线索到回款数据模板

该模板为收集分析Pipedrive线索到回款流程所需的关键数据提供清晰指引，列出需要跟踪的核心属性和关键活动，并说明如何直接从Pipedrive系统中提取这些信息。利用这一资源准备数据，开展全面的流程挖掘和优化。

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

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

下载模板

线索到收款属性

这些建议的数据字段对于构建可靠的事件日志至关重要，可支持对您的线索到收款流程进行全面分析。

3 必需 4 建议 13 可选

	名称	说明
	事件时间 EventTime	表示Activity发生时间的时间戳。
说明 Event Time或Start Time记录特定Activity被记录的准确日期和时间。这些时间数据对于按时间顺序排列事件，以及计算不同流程步骤之间的时长至关重要。在分析中，该时间戳用于计算周期时间、处理时间和等待时间等关键绩效指标。它还支持创建基于时间的仪表板，用于监控绩效并识别销售流程中的延迟。为什么重要该时间戳对于排列事件顺序、计算流程时长，以及根据基于时间的目标分析绩效至关重要。获取位置通过Pipedrive API从Deal、Activity或相关对象的add_time或update_time等字段获取。示例 2023-04-15T10:00:00Z2023-04-16T14:30:00Z2023-05-01T09:05:00Z
	活动名称 ActivityName	销售流程中发生的具体业务事件或任务的名称。
说明该属性描述线索到收款流程中的单个步骤，例如“交易已创建”“方案已发送”或“已收到付款”。活动是用于呈现顺序流的基础元素。分析活动的顺序和频率，有助于识别最常见的流程路径、工作积压形成的瓶颈，以及偏离标准销售流程的情况。这对于了解流程的实际执行方式至关重要。为什么重要它定义流程中的步骤，从而支持呈现和分析顺序流、识别瓶颈以及执行一致性检查。获取位置通常通过将Pipedrive事件映射到预先定义的Activity列表来生成，例如交易阶段变更，或发票等相关对象的创建。示例创建交易发送方案交易赢单发送发票收到付款
	销售机会 SalesOpportunityId	销售商机的唯一标识符，作为跟踪整个Lead to Cash流程的Case ID。
说明 Sales Opportunity ID是连接所有相关活动的主键，从最初创建线索到最终收款均通过它关联。它代表Pipedrive中正在跟踪的一笔潜在销售。在流程分析中，该标识符对于重建每笔交易的端到端旅程至关重要。它可以将Deal Created、Proposal Sent和Payment Received等所有事件归入一个完整Case，从而分析流程变体、周期时间和转化率。为什么重要该标识符是流程视图的基础，将所有相关事件连接起来，为每笔销售交易形成完整的Case历史记录。获取位置通过Pipedrive API从Deal对象的id字段获取。示例 12543125441254512546
	交易状态 DealStatus	交易当前的结果或状态，例如won、lost或open。
说明 Deal Status表示销售商机的最终结果或当前总体状态，是分析销售成功率和流程效率的主要维度。该属性是计算赢单率和丢单率的基础，也用于筛选流程图，对比赢单交易与丢单交易的旅程。这种对比通常能揭示导致成功或失败的流程执行差异，为流程改进提供有价值的洞察。为什么重要它对于计算赢单率和输单率，以及比较成功交易与未成功交易的顺序流至关重要。获取位置通过Pipedrive API从Deal对象的status字段获取。示例赢单丢单开放
	交易负责人 DealOwner	负责管理销售商机的销售代表。
说明 Deal Owner是分配给特定交易的用户。该属性是了解销售团队资源分配和绩效的关键。该属性支持个人和团队层面的绩效分析。仪表板可以按Deal Owner筛选，对比销售周期时间、转化率和平均交易金额等指标，帮助识别高绩效人员和需要辅导的领域。为什么重要支持按销售代表分析绩效，帮助识别高绩效人员和需要改进的领域。获取位置通过Pipedrive API从Deal对象的user_id字段获取，该字段关联User对象。示例 John SmithJane DoePeter Jones
	交易金额 DealValue	销售商机的总金额。
说明 Deal Value表示与销售商机相关的潜在或实际收入，是确定交易优先级和评估销售管道财务影响的基础指标。在流程分析中，该属性用于按金额规模划分Case，以比较高价值交易和低价值交易的流程执行情况。它也是分析销售绩效、折扣效果和整体收入的仪表板中的核心指标。为什么重要支持对销售流程进行财务分析，帮助您优先处理高价值交易并了解其收入影响。获取位置通过Pipedrive API从Deal对象的value字段获取。示例 5000.0025000.501500.75
	折扣百分比 DiscountPercentage	交易所 적용用的折扣比例。
说明该属性表示销售交易中提供的折扣比例（如有），是了解定价策略及其对盈利能力影响的关键因素。分析Discount Percentage有助于评估折扣策略的有效性。您可以将其与Deal Value、Sales Representative或Product等其他属性关联，了解折扣最常出现的场景，以及折扣是否带来更高赢单率，还是仅仅侵蚀利润率。为什么重要这对于分析定价策略、销售利润率和交易整体盈利能力至关重要。获取位置这通常是Pipedrive中Deal对象的自定义字段。有关自定义字段设置，请参阅Pipedrive文档。示例 0.050.100.150.0
	事件结束时间 EventEndTime	表示一项有持续时间的活动完成时间的时间戳。
说明 Event End Time标记具有可测量持续时间的活动完成时间，例如会议或任务。它与Start Time结合使用，可精确计算具体活动的耗时。该属性对于计算单项活动的处理时间至关重要。分析持续时间有助于识别最耗时的任务，并发现优化、自动化或重新分配资源的机会。为什么重要支持精确计算单项活动的持续时间，这是识别耗时任务和瓶颈的关键。获取位置从Pipedrive API中Activity对象的“due_date”等字段或完成时间戳获取。示例 2023-04-15T11:00:00Z2023-04-16T15:00:00Z2023-05-01T09:35:00Z
	交易币种 DealCurrency	与交易金额关联的货币。
说明该属性指定Deal Value所使用的货币，例如USD、EUR或GBP，并为所有财务指标提供必要背景。分析来自多个地区或国家的交易时，该属性对于准确进行财务报告和汇总至关重要。在汇总财务数据前，您可以据此正确应用货币换算，避免得出不准确的结论。为什么重要为Deal Value提供必要背景，确保跨地区开展准确的财务分析和报告。获取位置通过Pipedrive API从Deal对象的currency字段获取。示例 USDEURGBP
	交易阶段 DealStage	交易当前所处的销售管道具体阶段。
说明 Deal Stage表示企业定义的销售管道中的具体步骤，例如Qualified、Proposal Made或Negotiation。与整体Deal Status相比，它能更细致地反映交易进展。分析交易阶段对于了解销售漏斗和识别交易容易停滞的位置至关重要。通过跟踪每个阶段的停留时间，团队可以定位瓶颈，并将改进工作集中在销售周期中问题最突出的环节。为什么重要更细致地呈现交易在销售漏斗中的进展，帮助识别瓶颈和各阶段之间的转化率。获取位置通过Pipedrive API从Deal对象的stage_id字段获取，该字段关联Stage对象。示例线索进入完成联系安排演示Proposal Made开始谈判
	产品 ProductAttached	与销售商机关联的产品或服务。
说明该属性指定交易中销售的产品或服务，支持针对具体产品或服务开展更细致的销售活动分析。纳入Product后，可以按产品线划分分析结果，比较不同产品线的销售流程。这有助于发现某些产品是否具有更长的销售周期、不同的转化率或不同的流程步骤，从而制定更有针对性的销售策略。为什么重要支持按产品线分析销售流程，帮助识别不同产品或服务在销售周期和成功率方面的差异。获取位置通过Pipedrive API从与Deal关联的Product对象获取。示例标准SaaS订阅企业平台许可证专业服务套餐
	付款条款 PaymentTerms	约定的付款条件，例如“Net 30”或“Net 60”。
说明 Payment Terms定义客户在发票发送后应在多长时间内付款。该属性对于管理应收账款和现金流至关重要。在流程挖掘中，该属性是评估付款表现的基准。通过将实际付款日期与根据发票日期和这些付款条件计算出的到期日进行比较，可用于计算按时付款率等KPI。为什么重要为衡量按时付款表现和分析现金回收流程中的延迟提供基准。获取位置通常是Pipedrive中Deal或Organization对象的自定义字段，也可能来自集成的会计系统。示例 30天账期60天账期收到即付
	最近数据更新时间 LastDataUpdate	源系统最近一次刷新数据的时间戳。
说明该属性表示数据最近一次提取并更新到流程挖掘数据集的时间，为所分析数据的新鲜度提供明确参考。对于任何分析，了解数据的时效性都至关重要。该属性有助于您判断当前查看的洞察是否反映流程的最新状态，这对运营监控和决策十分重要。为什么重要反映数据的新鲜度，确保用户了解流程分析所依据的数据有多新。获取位置该值在数据摄取或ETL过程中生成并记录。示例 2023-10-27T02:00:00Z2023-10-28T02:00:00Z
	区域 Region	与客户或销售机会相关的地理区域。
说明该属性按地理区域对交易进行分类，例如北美、EMEA或APAC，从而支持基于位置的销售表现和流程分析。按区域细分流程后，企业可以发现不同区域在销售周期、赢单率和流程变体方面的差异。这些洞察有助于制定区域销售策略、分配资源并推进流程标准化。为什么重要支持对销售流程进行地理分析，突出不同区域在表现和客户行为方面的差异。获取位置通常是Deal或Organization对象的自定义字段，也可以根据Organization的地址字段推导得出。示例北美EMEAAPACLATAM
	发票付款时间 InvoicePaymentTime	从发票发送到收到付款之间的持续时间。
说明 Invoice Payment Time用于衡量应收账款流程的效率，具体跟踪从“Invoice Sent”活动到“Payment Received”活动所经过的时间。该指标对于监控现金流和识别收款延迟至关重要。分析这一持续时间有助于企业了解客户付款行为以及催收和收款策略的有效性，并直接影响从报价到付款的周期。为什么重要直接衡量现金回收流程的效率，对于管理现金流至关重要。获取位置对于指定案例，计算“Payment Received”和“Invoice Sent”活动时间戳之差。示例 28天45天10天
	是否按时付款 IsOnTimePayment	表示付款是否在约定付款条件内收到的布尔标记。
说明该属性是一个二元标记（True/False），表示付款是否按时完成。它通过将实际Invoice Payment Time与指定的Payment Terms进行比较得出。该标记简化了付款行为分析。您可以用它快速计算按时付款率KPI，并筛选逾期付款，从而针对特定客户或交易类型发生付款延迟的原因开展根因分析。为什么重要简化付款遵循情况的分析，并用于计算按时付款率KPI。获取位置通过比较“InvoicePaymentTime”和“PaymentTerms”属性计算。例如，如果PaymentTerms为“Net 30”，且InvoicePaymentTime小于或等于30天，则值为True。示例 truefalse
	是否返工 IsRework	用于识别属于返工循环的活动的标记。
说明 Is Rework是一个布尔标记，通常由流程挖掘工具计算，用于突出流程偏离线性路径并重复之前步骤的情况。例如，交易从“Negotiation”退回“Proposal Sent”就属于返工。识别返工对于流程优化至关重要，因为返工意味着效率低下、重复投入，并可能使客户感到困惑。分析标记为返工的活动，有助于定位流程失效环节并实施针对性改进。为什么重要通过识别并标记重复或循环流程路径中的活动，帮助量化流程低效。获取位置由流程挖掘软件分析每个案例中的活动序列后计算得出。示例 truefalse
	源系统 SourceSystem	标识数据提取自哪个系统。
说明该属性指定数据来源，本例中为Pipedrive。在整合多个系统的数据以形成全局流程视图时，这一属性尤其有用。在单系统分析中，它通常是静态值，但能提供重要背景信息，并支持数据治理和可追溯性。它可以确认相关事件和属性均来自Pipedrive实例。为什么重要提供有关数据来源的重要背景信息，确保数据可追溯且含义清晰，尤其适用于整合多个系统数据的场景。获取位置通常在数据转换过程中添加为静态值。示例 Pipedrive
	线索来源 LeadSource	获取线索的来源或渠道。
说明 Lead Source表示首次识别潜在客户的方式，例如Webinar、Cold Call或Partner Referral。这些信息对制定市场营销和销售策略至关重要。分析Lead Source有助于评估不同营销渠道的效果。您可以比较来自不同来源的线索转化率和销售周期时间，从而优化营销投入和销售工作。为什么重要通过分析各来源线索的转化率和流程绩效，帮助衡量不同营销渠道的有效性。获取位置通过Lead对象的source_name字段获取，该字段随后与Deal关联。示例自然搜索付费社交媒体推荐贸易展会
	组织名称 OrganizationName	与交易关联的客户公司或组织名称。
说明该属性标识与销售商机关联的企业客户，提供销售管道中客户公司的背景信息。按Organization Name分析数据，有助于了解客户行为并识别重点客户。您可以据此划分流程分析，比较战略客户与小型客户的流程差异，或跟踪与特定公司相关的完整销售历史。为什么重要标识客户，支持以客户为中心的分析和销售流程分群。获取位置通过Pipedrive API从Deal对象的org_id字段获取，该字段关联Organization对象。示例 ABC CorporationGlobal Tech Inc.Innovate Solutions

必需建议可选

线索到收款活动

应在事件日志中记录以下关键流程步骤和里程碑，以确保准确发现流程并有效优化。

5 建议 8 可选

	活动	说明
	交易丢单	标志着交易未能成功结束。当客户选择竞争对手、不再感兴趣或因其他原因终止销售流程时，就会发生这一事件。
为什么重要这是主要的失败终点。分析交易何时以及为何丢失，可以为改进销售策略、培训和资格审查流程提供关键洞察。获取位置当交易状态变更为lost时，Pipedrive会记录这一显式事件。Deals实体中的lost_time字段记录该时间戳。采集使用Deals实体中的lost_time时间戳。事件类型 explicit
	交易赢单	标志着交易成功结束，表示客户已口头或书面同意。这是Pipedrive中的标准显式状态变更。
为什么重要这是最重要的成功里程碑，对于计算赢单率、成功交易的销售周期长度和整体销售绩效至关重要。获取位置当交易状态变更为won时，Pipedrive会记录这一显式事件。Deals实体中的won_time字段提供精确时间戳。采集使用Deals实体中的won_time时间戳。事件类型 explicit
	创建交易	标志着销售商机正式开始。当新交易由人工创建，或由转化后的线索自动生成时，就会触发此事件。
为什么重要这是销售周期的主要起始活动。分析从此事件到Deal Won或Deal Lost的时间，可以得出整体销售周期时长。获取位置这是Pipedrive Deals实体中记录的显式事件。交易记录的创建时间戳（add_time）即为事件时间。采集使用Deals实体中的add_time字段。事件类型 explicit
	发送方案	表示正式报价或方案交付给客户的时间点。通常可根据交易是否进入Proposal Made或Quote Sent等特定管道阶段来推断。
为什么重要这是销售流程中的关键里程碑。分析从此事件到Deal Won的时间，有助于了解决策和谈判阶段的持续时间。获取位置根据Deal changelog推断：当stage_id变更为指定的方案阶段时，使用该阶段变更的时间戳。采集检测Deal的stage_id是否更新为特定的Proposal阶段。事件类型 inferred
	收到付款	标志着成功收到客户付款，完成该商机的现金回收周期。这是Lead to Cash流程的最后一个事件，来自财务系统。
为什么重要这是最终实现价值的步骤，也是Lead to Cash流程真正的终点，对于衡量端到端周期时间和分析付款延迟至关重要。获取位置从会计或支付处理系统中获取。必须将付款确认关联回Pipedrive交易。采集来自外部财务系统的事件数据，通过唯一标识符与交易关联。事件类型 explicit
	交易通过资格审查	表示交易已满足预设标准，被视为值得继续推进的有效商机。通常可根据交易是否从Lead In等早期管道阶段进入Contact Made或Needs Defined等更高资格阶段来推断。
为什么重要这一里程碑有助于分析线索资格审查流程的有效性，并确定交易在初始未合格阶段停留的时间。获取位置根据Deal changelog或历史记录推断：当stage_id字段变更为表示已完成资格审查的阶段时，即可识别该事件。采集从交易历史中识别阶段是否变更为预先定义的qualified阶段。事件类型 inferred
	创建销售订单	表示交易赢单后，在ERP或会计平台等下游系统中创建正式销售订单。该活动并非Pipedrive原生活动，需要依赖集成系统的数据。
为什么重要该事件标志着工作从销售团队移交给运营或财务团队。分析从Deal Won到该活动的时间，可以发现订单处理中的潜在延迟。获取位置通常从已连接的ERP或财务系统中获取。需要导入数据，并将其与Pipedrive交易ID关联。采集外部系统通过API推送事件数据，通常会更新Pipedrive交易中的自定义字段。事件类型 explicit
	发送发票	表示向客户发送付款发票的时间。订单履约完成后，通常由外部会计或计费系统触发该事件。
为什么重要这标志着付款周期开始。从该事件到Payment Received的时间，对于监控应收账款和现金流至关重要。获取位置来自Xero或QuickBooks等外部会计或ERP系统。需要集成事件数据，并将其与Pipedrive交易关联。采集来自外部计费系统的事件数据，通过API集成与交易关联。事件类型 explicit
	完成首次联系	表示与该交易相关的潜在客户首次进行有意义的互动。通常记录为已完成的活动，例如与交易关联的电话、电子邮件或会议。
为什么重要跟踪此事件有助于衡量首次联系耗时，确保及时跟进线索，并了解销售团队的响应速度。获取位置从Pipedrive Activities实体中获取。通过查找与交易ID关联的最早已完成活动来识别。采集根据done_time字段，查找与Deal关联的第一个已完成Activity。事件类型 explicit
	开始谈判	表示围绕价格、条款或范围开始正式谈判。通常可根据交易在发送方案后进入Negotiation管道阶段来推断。
为什么重要单独分析谈判阶段，有助于确定该阶段的耗时，以及哪些交易涉及较长时间的谈判。获取位置根据Deal changelog推断：当stage_id更新为明确表示谈判的阶段时，即可识别。采集从交易历史中识别阶段是否变更为预先定义的Negotiation阶段。事件类型 inferred
	订单履行完成	表示产品已发货或服务已交付给客户。该事件来自外部履约或运营系统。
为什么重要跟踪履约情况是了解订单到交付周期时间的关键。此处的延迟会直接影响客户满意度和收入确认。获取位置从外部ERP、库存或服务管理系统中获取。必须将数据关联回对应的Pipedrive交易。采集来自外部系统的事件数据，通过交易ID或其他通用标识符关联。事件类型 explicit
	跟进方案	记录向客户发送方案后发生的销售活动，例如电话或电子邮件。该事件表示客户考虑期间仍在积极跟进。
为什么重要衡量跟进操作的及时性和频率，这对保持推进势头并促成成交至关重要。该指标支持Proposal Follow-up Timeliness仪表板。获取位置从Pipedrive Activities实体中获取。对于同一交易，在Proposal Sent事件之后发生的已完成活动即被识别为该事件。采集识别done_time晚于Proposal Sent活动时间戳的已完成Activity。事件类型 explicit
	重新打开交易	表示此前标记为丢失的交易重新激活，并恢复后续销售活动。当交易状态从lost变回open时，即可推断该事件。
为什么重要该活动体现了返工循环和重新激活客户的努力。分析这些情况，可以发现客户行为或销售跟进效果中的模式。获取位置根据Deal changelog推断。当status字段从lost变更为open时，即可检测到该事件。采集从交易历史中识别状态是否从lost变更为open。事件类型 inferred

建议可选

提取指南

如何从Pipedrive获取数据

步骤

登录Pipedrive：使用具有数据导出权限的用户账户访问Pipedrive。通常需要使用Admin用户。
进入数据导出页面：点击右上角的个人头像，然后选择“Settings”。在左侧导航菜单中找到“Tools and apps”部分，点击“Export data”。
导出Deals数据：在“Export data”页面选择“Deals”，选择“CSV”文件格式。建议按特定日期范围筛选，例如最近6个月，以控制文件大小。点击“Export”，等待文件生成并下载。该文件将包含创建时间、赢单时间和丢单时间等关键交易信息。
导出Activities数据：对“Activities”重复上述操作。选择“Activities”，选择“CSV”，应用与Deals相同的日期范围筛选并导出文件。该文件包含与交易关联的电话、电子邮件、会议和自定义活动记录。
收集外部数据：从下游系统获取对应数据。您需要从ERP、会计或履约系统中导出“Sales Order Created”“Order Fulfilled”“Invoice Sent”和“Payment Received”等事件。请确保这些文件包含可映射回Pipedrive Deal ID的通用标识符。
设置数据处理环境：使用能够处理并合并多个CSV文件的工具，例如搭配pandas库的Python、带Power Query的Microsoft Excel或其他数据准备工具。
加载并合并数据：加载Deals.csv和Activities.csv文件，同时加载上一步收集的外部数据文件。
构建事件日志：通过脚本将原始数据转换为所需的活动格式，创建统一事件日志。针对13项必需活动分别创建对应的数据行。例如：
- “Deal Created”事件来自Deals.csv文件的add_time列。
- “Deal Won”事件来自Deals.csv文件的won_time列。
- “Initial Contact Made”是Activities.csv文件中每笔交易最早完成的活动。
- “Proposal Sent”或“Negotiation Started”等其他活动，则根据Activities.csv中的具体活动主题或类型推断。
映射属性：确保每个事件行都填充所有必需列：SalesOpportunityId、ActivityName和EventTime。通过与交易数据连接，添加DealValue、DealOwner和DealStatus等推荐属性。
最终导出：事件日志构建并验证完成后，将最终表格保存为采用UTF-8编码的单个CSV文件。现在即可将该文件上传到ProcessMind进行分析。

配置

导出对象：至少导出“Deals”和“Activities”。如果需要额外的维度数据，也可以导出“Organizations”或“People”。
文件格式：选择CSV，以便与Python或Power Query等数据处理工具实现最大兼容。
日期范围筛选：使用“Filter by”选项选择特定时间段，例如将“Deal add time”设为“last 180 days”内。这对于控制导出性能和文件大小至关重要。
用户权限：您需要使用Admin用户，或拥有明确允许从Pipedrive账户导出数据的权限集。
自定义活动类型：能否成功提取“Deal Qualified”或“Proposal Sent”等流程步骤，很大程度上取决于组织是否在Pipedrive中使用自定义Activity Types。请确保这些类型已配置，并由销售团队统一使用。

a 示例查询 python

import pandas as pd

class=class="hl-string">"hl-comment"># --- 1. Load Data ---
class=class="hl-string">"hl-comment"># Load Pipedrive exports. Replace class="hl-string">'deals.csv' and class="hl-string">'activities.csv' with your actual file paths.
deals_df = pd.read_csv(class="hl-string">'deals.csv')
activities_df = pd.read_csv(class="hl-string">'activities.csv')

class=class="hl-string">"hl-comment"># Placeholder: Load data from external systems. These files should contain a Deal ID column.
class=class="hl-string">"hl-comment"># Replace with actual file loads from your ERP, Accounting, or Fulfillment systems.
sales_orders_df = pd.DataFrame({class="hl-string">'Deal ID': [], class="hl-string">'Order Created Date': []}) class=class="hl-string">"hl-comment"># Example from ERP
fulfillments_df = pd.DataFrame({class="hl-string">'Deal ID': [], class="hl-string">'Fulfillment Date': []})   class=class="hl-string">"hl-comment"># Example from Fulfillment System
invoices_df = pd.DataFrame({class="hl-string">'Deal ID': [], class="hl-string">'Invoice Sent Date': []})         class=class="hl-string">"hl-comment"># Example from Accounting System
payments_df = pd.DataFrame({class="hl-string">'Deal ID': [], class="hl-string">'Payment Received Date': []})     class=class="hl-string">"hl-comment"># Example from Accounting System

class=class="hl-string">"hl-comment"># --- 2. Prepare and Clean Data ---
class=class="hl-string">"hl-comment"># Rename columns for consistency and convert date columns to datetime objects
deals_df.rename(columns={class="hl-string">'ID': class="hl-string">'SalesOpportunityId', class="hl-string">'Value': class="hl-string">'DealValue', class="hl-string">'Owner name': class="hl-string">'DealOwner', class="hl-string">'Status': class="hl-string">'DealStatus', class="hl-string">'Add time': class="hl-string">'EventTime_Created', class="hl-string">'Won time': class="hl-string">'EventTime_Won', class="hl-string">'Lost time': class="hl-string">'EventTime_Lost'}, inplace=True)
activities_df.rename(columns={class="hl-string">'Deal ID': class="hl-string">'SalesOpportunityId', class="hl-string">'Subject': class="hl-string">'ActivityName', class="hl-string">'Done time': class="hl-string">'EventTime'}, inplace=True)

for col in [class="hl-string">'EventTime_Created', class="hl-string">'EventTime_Won', class="hl-string">'EventTime_Lost']: 
    deals_df[col] = pd.to_datetime(deals_df[col], errors=class="hl-string">'coerce')
activities_df[class="hl-string">'EventTime'] = pd.to_datetime(activities_df[class="hl-string">'EventTime'], errors=class="hl-string">'coerce')


class=class="hl-string">"hl-comment"># --- 3. Extract Activities from Pipedrive Data ---

class=class="hl-string">"hl-comment"># Activity: Deal Created
event_created = deals_df[[class="hl-string">'SalesOpportunityId', class="hl-string">'EventTime_Created']].copy()
event_created.rename(columns={class="hl-string">'EventTime_Created': class="hl-string">'EventTime'}, inplace=True)
event_created[class="hl-string">'ActivityName'] = class="hl-string">'Deal Created'

class=class="hl-string">"hl-comment"># Activity: Deal Won
event_won = deals_df[deals_df[class="hl-string">'EventTime_Won'].notna()][[class="hl-string">'SalesOpportunityId', class="hl-string">'EventTime_Won']].copy()
event_won.rename(columns={class="hl-string">'EventTime_Won': class="hl-string">'EventTime'}, inplace=True)
event_won[class="hl-string">'ActivityName'] = class="hl-string">'Deal Won'

class=class="hl-string">"hl-comment"># Activity: Deal Lost
event_lost = deals_df[deals_df[class="hl-string">'EventTime_Lost'].notna()][[class="hl-string">'SalesOpportunityId', class="hl-string">'EventTime_Lost']].copy()
event_lost.rename(columns={class="hl-string">'EventTime_Lost': class="hl-string">'EventTime'}, inplace=True)
event_lost[class="hl-string">'ActivityName'] = class="hl-string">'Deal Lost'

class=class="hl-string">"hl-comment"># Filter for relevant, completed activities linked to a deal
completed_activities = activities_df[activities_df[class="hl-string">'EventTime'].notna() & activities_df[class="hl-string">'SalesOpportunityId'].notna()].copy()
completed_activities[class="hl-string">'SalesOpportunityId'] = completed_activities[class="hl-string">'SalesOpportunityId'].astype(int)

class=class="hl-string">"hl-comment"># Activity: Initial Contact Made (first completed activity for each deal)
event_initial_contact = completed_activities.sort_values(class="hl-string">'EventTime', ascending=True).drop_duplicates(class="hl-string">'SalesOpportunityId')
event_initial_contact = event_initial_contact[[class="hl-string">'SalesOpportunityId', class="hl-string">'EventTime']].copy()
event_initial_contact[class="hl-string">'ActivityName'] = class="hl-string">'Initial Contact Made'

class=class="hl-string">"hl-comment"># Infer other activities based on activity subject keywords. Adjust keywords as needed.
event_qualified = completed_activities[completed_activities[class="hl-string">'ActivityName'].str.contains(class="hl-string">'qualify|qualification', case=False, na=False)].copy()
event_qualified[class="hl-string">'ActivityName'] = class="hl-string">'Deal Qualified'

event_proposal_sent = completed_activities[completed_activities[class="hl-string">'ActivityName'].str.contains(class="hl-string">'proposal|quote', case=False, na=False)].copy()
event_proposal_sent[class="hl-string">'ActivityName'] = class="hl-string">'Proposal Sent'

class=class="hl-string">"hl-comment"># Proposal Follow-Up: Any call or email activity after a proposal was sent
if not event_proposal_sent.empty:
    merged_followup = pd.merge(completed_activities, event_proposal_sent[[class="hl-string">'SalesOpportunityId', class="hl-string">'EventTime']].rename(columns={class="hl-string">'EventTime':class="hl-string">'ProposalTime'}), on=class="hl-string">'SalesOpportunityId')
    follow_ups = merged_followup[merged_followup[class="hl-string">'EventTime'] > merged_followup[class="hl-string">'ProposalTime']]
    event_follow_up = follow_ups[follow_ups[class="hl-string">'ActivityName'].str.contains(class="hl-string">'call|email|follow-up', case=False, na=False)].copy()
    event_follow_up[class="hl-string">'ActivityName'] = class="hl-string">'Proposal Follow-Up'
else:
    event_follow_up = pd.DataFrame(columns=[class="hl-string">'SalesOpportunityId', class="hl-string">'EventTime', class="hl-string">'ActivityName'])

event_negotiation = completed_activities[completed_activities[class="hl-string">'ActivityName'].str.contains(class="hl-string">'negotiation|negotiate', case=False, na=False)].copy()
event_negotiation[class="hl-string">'ActivityName'] = class="hl-string">'Negotiation Started'

class=class="hl-string">"hl-comment"># Placeholder for Deal Reopened: Logic is complex and not reliably available from standard export.
event_reopened = pd.DataFrame(columns=[class="hl-string">'SalesOpportunityId', class="hl-string">'EventTime', class="hl-string">'ActivityName'])

class=class="hl-string">"hl-comment"># --- 4. Extract Activities from External Data (Placeholders) ---

class=class="hl-string">"hl-comment"># Rename columns and convert dates for external data
class=class="hl-string">"hl-comment"># This section needs to be adapted to your specific file structures
event_so_created = sales_orders_df.rename(columns={class="hl-string">'Deal ID': class="hl-string">'SalesOpportunityId', class="hl-string">'Order Created Date': class="hl-string">'EventTime'}).assign(ActivityName=class="hl-string">'Sales Order Created')
event_fulfilled = fulfillments_df.rename(columns={class="hl-string">'Deal ID': class="hl-string">'SalesOpportunityId', class="hl-string">'Fulfillment Date': class="hl-string">'EventTime'}).assign(ActivityName=class="hl-string">'Order Fulfilled')
event_invoice_sent = invoices_df.rename(columns={class="hl-string">'Deal ID': class="hl-string">'SalesOpportunityId', class="hl-string">'Invoice Sent Date': class="hl-string">'EventTime'}).assign(ActivityName=class="hl-string">'Invoice Sent')
event_payment_received = payments_df.rename(columns={class="hl-string">'Deal ID': class="hl-string">'SalesOpportunityId', class="hl-string">'Payment Received Date': class="hl-string">'EventTime'}).assign(ActivityName=class="hl-string">'Payment Received')


class=class="hl-string">"hl-comment"># --- 5. Combine and Finalize Event Log ---

event_log_frames = [
    event_created,
    event_won,
    event_lost,
    event_initial_contact,
    event_qualified[[class="hl-string">'SalesOpportunityId', class="hl-string">'EventTime', class="hl-string">'ActivityName']],
    event_proposal_sent[[class="hl-string">'SalesOpportunityId', class="hl-string">'EventTime', class="hl-string">'ActivityName']],
    event_follow_up[[class="hl-string">'SalesOpportunityId', class="hl-string">'EventTime', class="hl-string">'ActivityName']],
    event_negotiation[[class="hl-string">'SalesOpportunityId', class="hl-string">'EventTime', class="hl-string">'ActivityName']],
    event_reopened,
    event_so_created,
    event_fulfilled,
    event_invoice_sent,
    event_payment_received
]

event_log = pd.concat(event_log_frames)

class=class="hl-string">"hl-comment"># Merge with deal attributes
final_event_log = pd.merge(event_log, deals_df[[class="hl-string">'SalesOpportunityId', class="hl-string">'DealValue', class="hl-string">'DealOwner', class="hl-string">'Status']], on=class="hl-string">'SalesOpportunityId', how=class="hl-string">'left')

class=class="hl-string">"hl-comment"># Clean up and select final columns
final_event_log.rename(columns={class="hl-string">'Status':class="hl-string">'DealStatus'}, inplace=True)
final_event_log.dropna(subset=[class="hl-string">'EventTime', class="hl-string">'SalesOpportunityId'], inplace=True)

class=class="hl-string">"hl-comment"># Placeholder for DiscountPercentage - Assuming it's a custom field in your deals export
if class="hl-string">'Discount percentage' in deals_df.columns:
    final_event_log = pd.merge(final_event_log, deals_df[[class="hl-string">'SalesOpportunityId', class="hl-string">'Discount percentage']], on=class="hl-string">'SalesOpportunityId', how=class="hl-string">'left')
    final_event_log.rename(columns={class="hl-string">'Discount percentage':class="hl-string">'DiscountPercentage'}, inplace=True)
else:
    final_event_log[class="hl-string">'DiscountPercentage'] = None

class=class="hl-string">"hl-comment"># Reorder columns to match requirements
output_columns = [class="hl-string">'SalesOpportunityId', class="hl-string">'ActivityName', class="hl-string">'EventTime', class="hl-string">'DealValue', class="hl-string">'DealOwner', class="hl-string">'DealStatus', class="hl-string">'DiscountPercentage']
final_event_log = final_event_log[output_columns]

class=class="hl-string">"hl-comment"># --- 6. Save Final Event Log ---
final_event_log.to_csv(class="hl-string">'pipedrive_event_log.csv', index=False, date_format=class="hl-string">'%Y-%m-%d %H:%M:%S')

print(class="hl-string">'Event log created successfully.')

步骤

设置Python环境：确保系统已安装Python 3.6或更高版本。打开终端或命令提示符，运行以下命令安装所需Python库：pip install requests pandas。
获取Pipedrive API凭据：登录Pipedrive。点击个人头像进入Settings，然后选择“Personal preferences”，进入“API”选项卡。复制个人API令牌。您还需要公司域名，即Pipedrive URL中的前缀，例如your-company.pipedrive.com中的your-company。
配置提取脚本：在文本编辑器或IDE中打开提供的Python脚本。找到顶部的配置部分，将API_TOKEN和COMPANY_DOMAIN的占位值替换为您的凭据。
定义提取时间范围：在脚本配置部分，将START_DATE变量设置为要纳入分析的交易最早创建日期。格式应为“YYYY-MM-DD”。
映射销售管道和阶段ID：脚本会根据交易在阶段之间的移动推断若干活动。您必须提供销售管道和关键阶段的具体ID。可以在Pipedrive界面中进入相应阶段并查看URL，或通过API获取这些ID。更新脚本中的PIPELINE_ID、QUALIFIED_STAGE_ID、PROPOSAL_STAGE_ID和NEGOTIATION_STAGE_ID变量。
执行脚本：在终端中运行以下命令：python your_script_name.py。脚本将开始调用Pipedrive API，获取交易、活动和变更日志。
监控流程：脚本会在控制台输出进度更新，显示正在获取的数据。具体耗时取决于数据量和API速率限制，可能需要几分钟。
提取外部数据：对于“Sales Order Created”和“Payment Received”等活动，脚本包含占位函数。您必须修改这些函数，连接ERP或会计平台等外部系统并获取相关事件数据，将Pipedrive的SalesOpportunityId映射到其他系统中的记录。
检查输出：脚本完成后，将生成名为pipedrive_event_log.csv的CSV文件，其中包含提取的结构化事件日志。
准备上传：打开CSV文件，确认文件已正确生成。列应包括SalesOpportunityId、ActivityName、EventTime及其他推荐属性。现在即可将该文件上传到ProcessMind进行分析。

配置

API凭据：脚本需要有效的API_TOKEN和用于向Pipedrive API进行身份验证的COMPANY_DOMAIN。请确保与令牌关联的用户拥有读取交易、活动和用户数据的足够权限。
日期范围：START_DATE变量会筛选在指定日期或之后创建的交易。首次分析建议使用3至6个月的数据，以控制数据量。
管道和阶段ID：要准确推断“Deal Qualified”或“Proposal Sent”等流程步骤，必须为主要销售PIPELINE_ID以及各关键里程碑对应的STAGE_ID配置正确的数字ID。ID不匹配会导致活动缺失。
性能和速率限制：脚本会在API调用之间设置短暂延迟，以遵守Pipedrive的速率限制。对于交易量非常大的组织，可能需要调整延迟，或实施更完善的退避策略，以避免API错误。
自定义字段映射：如果使用DiscountPercentage等重要属性的自定义字段，需要找到该字段的API键（通常是一串较长的哈希字符串，而不是显示名称），并在数据转换逻辑中引用。
外部系统集成：对于来自Pipedrive之外的事件，例如开票或履约，脚本使用占位函数处理。您必须自定义这些部分，以连接具体的外部数据源。

a 示例查询 python

import requests
import pandas as pd
import time
import datetime

class=class="hl-string">"hl-comment"># --- CONFIGURATION ---
class=class="hl-string">"hl-comment"># Replace with your Pipedrive API token and company domain
API_TOKEN = class="hl-string">'YOUR_API_TOKEN'
COMPANY_DOMAIN = class="hl-string">'YOUR_COMPANY_DOMAIN' class=class="hl-string">"hl-comment"># e.g., class="hl-string">'your-company-name'

class=class="hl-string">"hl-comment"># Set the start date for deal extraction (YYYY-MM-DD)
START_DATE = class="hl-string">'2023-01-01'

class=class="hl-string">"hl-comment"># Configure your specific Pipeline and Stage IDs
class=class="hl-string">"hl-comment"># Find these in your Pipedrive URL or via the API /stages endpoint
PIPELINE_ID = 1 class=class="hl-string">"hl-comment"># Replace with your main sales pipeline ID
QUALIFIED_STAGE_ID = 2 class=class="hl-string">"hl-comment"># Replace with your class="hl-string">'Qualified' stage ID
PROPOSAL_STAGE_ID = 3 class=class="hl-string">"hl-comment"># Replace with your class="hl-string">'Proposal Sent' stage ID
NEGOTIATION_STAGE_ID = 4 class=class="hl-string">"hl-comment"># Replace with your class="hl-string">'Negotiation' stage ID

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

BASE_URL = fclass="hl-string">'https://{COMPANY_DOMAIN}.pipedrive.com/api/v1'

def get_paginated_data(endpoint, params={}):
    class="hl-string">"""Fetches all pages of data from a Pipedrive API endpoint."""
    all_data = []
    start = 0
    limit = 500
    while True:
        request_params = {
            class="hl-string">'start': start,
            class="hl-string">'limit': limit,
            class="hl-string">'api_token': API_TOKEN
        }
        request_params.update(params)
        
        try:
            response = requests.get(fclass="hl-string">"{BASE_URL}/{endpoint}", params=request_params)
            response.raise_for_status()
            data = response.json()

            if not data.get(class="hl-string">'data'):
                break

            all_data.extend(data[class="hl-string">'data'])

            if not data.get(class="hl-string">'additional_data', {}).get(class="hl-string">'pagination', {}).get(class="hl-string">'more_items_in_collection'):
                break
            
            start += limit
            time.sleep(0.2) class=class="hl-string">"hl-comment"># Respect API rate limits
        except requests.exceptions.RequestException as e:
            print(fclass="hl-string">"Error fetching {endpoint}: {e}")
            break
            
    return all_data

def get_deal_flow(deal_id):
    class="hl-string">"""Fetches the change history (flow) for a single deal."""
    params = {class="hl-string">'api_token': API_TOKEN}
    try:
        response = requests.get(fclass="hl-string">"{BASE_URL}/deals/{deal_id}/flow", params=params)
        response.raise_for_status()
        data = response.json()
        return data.get(class="hl-string">'data', [])
    except requests.exceptions.RequestException as e:
        print(fclass="hl-string">"Error fetching flow for deal {deal_id}: {e}")
        return []

class=class="hl-string">"hl-comment"># Placeholder functions for external data
def get_sales_order_data(deal_ids):
    print(class="hl-string">"Placeholder: Fetching sales order data from external ERP/CRM...")
    class=class="hl-string">"hl-comment"># Example: Connect to your ERP, query for orders linked to deal_ids
    class=class="hl-string">"hl-comment"># Return a list of dicts: [{class="hl-string">'SalesOpportunityId': id, class="hl-string">'ActivityName': class="hl-string">'Sales Order Created', class="hl-string">'EventTime': timestamp}, ...]
    return []

def get_fulfillment_data(deal_ids):
    print(class="hl-string">"Placeholder: Fetching order fulfillment data...")
    return []

def get_invoice_data(deal_ids):
    print(class="hl-string">"Placeholder: Fetching invoice data from accounting system...")
    return []

def get_payment_data(deal_ids):
    print(class="hl-string">"Placeholder: Fetching payment data from financial system...")
    return []


class=class="hl-string">"hl-comment"># --- MAIN EXTRACTION LOGIC ---
if __name__ == class="hl-string">'__main__':
    print(class="hl-string">"Starting Pipedrive data extraction...")
    events = []

    class=class="hl-string">"hl-comment"># 1. Fetch all deals created after the START_DATE and in the specified pipeline
    print(class="hl-string">"Fetching deals...")
    deals = get_paginated_data(class="hl-string">'deals', {class="hl-string">'start_date': START_DATE, class="hl-string">'pipeline_id': PIPELINE_ID})
    if not deals:
        print(class="hl-string">"No deals found for the specified criteria. Exiting.")
        exit()

    deal_ids = [d[class="hl-string">'id'] for d in deals]
    print(fclass="hl-string">"Found {len(deals)} deals.")

    class=class="hl-string">"hl-comment"># 2. Fetch all activities related to the deals
    print(class="hl-string">"Fetching activities...")
    all_activities = get_paginated_data(class="hl-string">'activities', {class="hl-string">'user_id': 0}) class=class="hl-string">"hl-comment"># Get for all users
    deal_activities = [act for act in all_activities if act.get(class="hl-string">'deal_id') in deal_ids and act[class="hl-string">'done']]

    activities_by_deal = {}
    for act in deal_activities:
        if act[class="hl-string">'deal_id'] not in activities_by_deal:
            activities_by_deal[act[class="hl-string">'deal_id']] = []
        activities_by_deal[act[class="hl-string">'deal_id']].append(act)

    class=class="hl-string">"hl-comment"># 3. Process each deal to create events
    print(class="hl-string">"Processing deals and activities to generate event log...")
    for deal in deals:
        deal_id = deal[class="hl-string">'id']
        deal_owner_name = deal.get(class="hl-string">'owner_name', class="hl-string">'')
        base_attributes = {
            class="hl-string">'SalesOpportunityId': deal_id,
            class="hl-string">'DealValue': deal.get(class="hl-string">'value', 0),
            class="hl-string">'DealOwner': deal_owner_name,
            class="hl-string">'DealStatus': deal.get(class="hl-string">'status', class="hl-string">'open'),
            class="hl-string">'DiscountPercentage': deal.get(class="hl-string">'discount_percentage', 0)
        }

        class=class="hl-string">"hl-comment"># Activity: Deal Created
        events.append({**base_attributes, class="hl-string">'ActivityName': class="hl-string">'Deal Created', class="hl-string">'EventTime': deal[class="hl-string">'add_time']})

        class=class="hl-string">"hl-comment"># Activity: Deal Won / Lost
        if deal[class="hl-string">'status'] == class="hl-string">'won' and deal[class="hl-string">'won_time']:
            events.append({**base_attributes, class="hl-string">'ActivityName': class="hl-string">'Deal Won', class="hl-string">'EventTime': deal[class="hl-string">'won_time']})
        elif deal[class="hl-string">'status'] == class="hl-string">'lost' and deal[class="hl-string">'lost_time']:
            events.append({**base_attributes, class="hl-string">'ActivityName': class="hl-string">'Deal Lost', class="hl-string">'EventTime': deal[class="hl-string">'lost_time']})

        class=class="hl-string">"hl-comment"># Activity: Initial Contact Made & Proposal Follow-Up
        proposal_sent_time = None
        if deal_id in activities_by_deal:
            sorted_activities = sorted(activities_by_deal[deal_id], key=lambda x: x[class="hl-string">'marked_as_done_time'])
            if sorted_activities:
                first_activity = sorted_activities[0]
                events.append({**base_attributes, class="hl-string">'ActivityName': class="hl-string">'Initial Contact Made', class="hl-string">'EventTime': first_activity[class="hl-string">'marked_as_done_time']})
            
            class=class="hl-string">"hl-comment"># Find proposal follow-ups (requires proposal time)
            class=class="hl-string">"hl-comment"># This part is revisited after stage changes are processed

        class=class="hl-string">"hl-comment"># Activities from Stage Changes: Qualified, Proposal Sent, Negotiation, Reopened
        time.sleep(0.2) class=class="hl-string">"hl-comment"># Rate limit for flow endpoint
        flow = get_deal_flow(deal_id)
        if not flow:
            continue

        lost_then_reopened = False
        was_lost = False

        for change in sorted(flow, key=lambda x: x[class="hl-string">'add_time']):
            class=class="hl-string">"hl-comment"># Deal Reopened
            if change.get(class="hl-string">'field_key') == class="hl-string">'status' and change.get(class="hl-string">'old_value') == class="hl-string">'lost' and change.get(class="hl-string">'new_value') == class="hl-string">'open':
                events.append({**base_attributes, class="hl-string">'ActivityName': class="hl-string">'Deal Reopened', class="hl-string">'EventTime': change[class="hl-string">'add_time']})
            
            if change.get(class="hl-string">'field_key') != class="hl-string">'stage_id':
                continue

            stage_change_time = change[class="hl-string">'add_time']

            class=class="hl-string">"hl-comment"># Deal Qualified
            if change[class="hl-string">'new_value'] == QUALIFIED_STAGE_ID:
                events.append({**base_attributes, class="hl-string">'ActivityName': class="hl-string">'Deal Qualified', class="hl-string">'EventTime': stage_change_time})

            class=class="hl-string">"hl-comment"># Proposal Sent
            if change[class="hl-string">'new_value'] == PROPOSAL_STAGE_ID:
                events.append({**base_attributes, class="hl-string">'ActivityName': class="hl-string">'Proposal Sent', class="hl-string">'EventTime': stage_change_time})
                proposal_sent_time = stage_change_time

            class=class="hl-string">"hl-comment"># Negotiation Started
            if change[class="hl-string">'new_value'] == NEGOTIATION_STAGE_ID:
                events.append({**base_attributes, class="hl-string">'ActivityName': class="hl-string">'Negotiation Started', class="hl-string">'EventTime': stage_change_time})

        class=class="hl-string">"hl-comment"># Re-check for Proposal Follow-Up now that we have proposal_sent_time
        if proposal_sent_time and deal_id in activities_by_deal:
            for act in activities_by_deal[deal_id]:
                if act[class="hl-string">'marked_as_done_time'] > proposal_sent_time:
                    events.append({**base_attributes, class="hl-string">'ActivityName': class="hl-string">'Proposal Follow-Up', class="hl-string">'EventTime': act[class="hl-string">'marked_as_done_time']})
                    break class=class="hl-string">"hl-comment"># Only capture the first follow-up

    class=class="hl-string">"hl-comment"># 4. Add events from external systems (placeholders)
    print(class="hl-string">"Adding data from external system placeholders...")
    events.extend(get_sales_order_data(deal_ids))
    events.extend(get_fulfillment_data(deal_ids))
    events.extend(get_invoice_data(deal_ids))
    events.extend(get_payment_data(deal_ids))

    class=class="hl-string">"hl-comment"># 5. Create and export the final event log
    if not events:
        print(class="hl-string">"No events were generated. The output file will be empty.")
    else:
        print(class="hl-string">"Creating CSV file...")
        df = pd.DataFrame(events)
        class=class="hl-string">"hl-comment"># Ensure required columns are present
        df = df[[class="hl-string">'SalesOpportunityId', class="hl-string">'ActivityName', class="hl-string">'EventTime', class="hl-string">'DealValue', class="hl-string">'DealOwner', class="hl-string">'DealStatus', class="hl-string">'DiscountPercentage']]
        df.sort_values(by=[class="hl-string">'SalesOpportunityId', class="hl-string">'EventTime'], inplace=True)
        
        output_filename = class="hl-string">'pipedrive_event_log.csv'
        df.to_csv(output_filename, index=False)
        print(fclass="hl-string">"Extraction complete. Event log saved to {output_filename}")

步骤

选择并购买带受支持Pipedrive连接器的商业ETL或自动化服务，例如Fivetran或Make。确认所选产品能够提取交易、活动、交易阶段历史、交易状态变更、人员、组织和自定义字段，并能写入目标数据仓库或数据湖。
使用连接器支持的身份验证方式，创建具有读取权限的Pipedrive连接。根据所选工具配置源账户、工作区或组织。不要假设固定的菜单路径或API端点，因为不同产品和连接器版本可能有所不同。
配置目标连接，为同步的源对象创建落地区域。使用连接器生成的表名或集合名，或将脚本中的名称替换为您环境中创建的名称。保留源标识符、时间戳、状态值、阶段值、所有者标识符、金额值和自定义字段。
为所需日期范围配置增量同步，通常为前3至6个月；连接器支持时，请设置重叠窗口。纳入识别重新打开交易所需的历史记录，并包含阶段或状态历史，而不仅是当前交易快照。
同步Pipedrive对象，以及Sales Order Created、Order Fulfilled、Invoice Sent和Payment Received所需的集成下游对象。这四项活动不是Pipedrive原生事件，需要从ERP、履约、会计、计费或财务系统获取源数据。如果连接器不提供这些对象，请通过其他受支持的连接器将其加载到目标系统。
创建标准化视图或表，将连接器输出映射到提取脚本引用的逻辑源对象。将脚本中的每个占位表名替换为目标系统中对应的表或视图。在指定配置部分设置源系统对应的阶段名称、状态值、活动类型和外部系统关系键。
运行提取脚本。脚本会明确创建Deal Created、Initial Contact Made、Deal Qualified、Proposal Sent、Proposal Follow-Up、Negotiation Started、Deal Won、Sales Order Created、Order Fulfilled、Invoice Sent、Payment Received、Deal Lost和Deal Reopened的数据行。只有当源历史包含所需转换时，才会输出根据阶段或状态历史推断的事件。上传后，ProcessMind不会推断缺失事件。
验证SalesOpportunityId、ActivityName和EventTime均已填充，确保时间戳已统一为一致的时区，并使用源事件标识符和事件时间戳删除或控制重复源记录。检查推荐属性；源系统未提供的值请保留NULL。
将结果导出为扁平CSV或其他ProcessMind支持的事件日志格式。使用SalesOpportunityId作为案例标识符，ActivityName作为活动列，EventTime作为事件时间戳。添加DealValue、DealOwner、DealStatus和DiscountPercentage等属性。将文件上传到ProcessMind，或在ProcessMind中连接目标数据集，然后确认活动名称与所需标签完全一致。

配置

**源连接器：**在所选ETL或自动化服务中配置受支持的Pipedrive连接器。连接器能力、同步模式、生成的表名和身份验证选项会因产品和版本而异。
**必需源数据：**包括交易、活动、交易阶段历史、交易状态历史、所有者或用户，以及用于识别已转化线索的字段。还应包括集成系统中的下游销售订单、履约、发票和付款数据。
**目标对象：**将[Your deals table]、[Your activities table]、[Your deal stage history table]、[Your deal status history table]、[Your sales orders table]、[Your fulfillment table]、[Your invoices table]和[Your payments table]替换为连接器生成的表或整理后的视图。
**日期范围：**先使用3至6个月的数据。当流程周期更长、需要分析重新打开的交易，或需要历史阶段和状态转换时，再扩展日期范围。
**日期参数：**按照目标系统支持的日期格式设置[Start date]和[End date]。增量加载时使用重叠时间段，以捕获延迟更新、修正后的时间戳和延迟写入的下游交易。
**阶段配置：**配置Pipedrive账户中用于资格确认、提案和谈判的准确阶段名称或标准化阶段类别。除非示例阶段名称与源数据值一致，否则不要直接使用。
**活动配置：**根据连接器的标准化值，配置代表初次联系和提案跟进的准确已完成活动类型，例如电话、电子邮件或会议。
**外部系统关联：**配置将下游销售订单关联到Pipedrive交易的关系键。如果下游系统使用不同的键，请在运行提取前创建映射视图。
**筛选器：**仅在确有需要时应用公司、组织、管道、团队、货币或业务单元筛选器。请记录每个筛选器，因为它会影响案例和事件的完整性。
**性能：**对于大型数据集，使用增量同步、源端日期筛选、分区目标表，以及针对案例和时间戳列的索引或聚簇。避免反复扫描完整历史表。
**去重：**在可用时保留稳定的源事件标识符。否则，使用案例标识符、活动名称、事件时间戳和源记录标识符进行去重，同时保留多次跟进等合理的重复活动。
**前提条件：**需要连接器许可证、Pipedrive授权、目标系统写入权限、交易历史和活动访问权限，以及下游ERP、履约、会计、计费或财务数据的访问权限。自定义字段和历史变更数据可能还需要额外的产品权限或连接器功能。

a 示例查询 python

import os
from datetime import datetime, timezone
import pandas as pd
from sqlalchemy import create_engine, text

CONNECTION_STRING = os.getenv(class="hl-string">"DESTINATION_CONNECTION_STRING", class="hl-string">"[Connection string]")
START_DATE = os.getenv(class="hl-string">"START_DATE", class="hl-string">"[Start date]")
END_DATE = os.getenv(class="hl-string">"END_DATE", class="hl-string">"[End date]")

DEALS_TABLE = class="hl-string">"[Your deals table]"
ACTIVITIES_TABLE = class="hl-string">"[Your activities table]"
DEAL_STAGE_HISTORY_TABLE = class="hl-string">"[Your deal stage history table]"
DEAL_STATUS_HISTORY_TABLE = class="hl-string">"[Your deal status history table]"
SALES_ORDERS_TABLE = class="hl-string">"[Your sales orders table]"
FULFILLMENT_TABLE = class="hl-string">"[Your fulfillment table]"
INVOICES_TABLE = class="hl-string">"[Your invoices table]"
PAYMENTS_TABLE = class="hl-string">"[Your payments table]"

engine = create_engine(CONNECTION_STRING)

sql = fclass="hl-string">"""
WITH
base_deals AS (
    SELECT
        CAST(d.id AS VARCHAR(255)) AS SalesOpportunityId,
        d.id AS deal_source_id,
        d.add_time AS deal_created_time,
        d.value AS DealValue,
        d.owner_name AS DealOwner,
        d.status AS DealStatus,
        d.discount_percentage AS DiscountPercentage
    FROM {DEALS_TABLE} d
    WHERE d.add_time >= :start_date
      AND d.add_time < :end_date
),
completed_activities AS (
    SELECT
        CAST(a.deal_id AS VARCHAR(255)) AS SalesOpportunityId,
        a.id AS activity_source_id,
        a.done AS activity_done,
        a.activity_type,
        a.subject,
        a.done_time AS activity_time,
        a.add_time AS activity_created_time
    FROM {ACTIVITIES_TABLE} a
    WHERE a.deal_id IS NOT NULL
      AND COALESCE(a.done_time, a.add_time) >= :start_date
      AND COALESCE(a.done_time, a.add_time) < :end_date
      AND a.done = TRUE
),
stage_history AS (
    SELECT
        CAST(h.deal_id AS VARCHAR(255)) AS SalesOpportunityId,
        h.stage_name,
        h.changed_at,
        LAG(h.stage_name) OVER (PARTITION BY h.deal_id ORDER BY h.changed_at, h.id) AS previous_stage
    FROM {DEAL_STAGE_HISTORY_TABLE} h
    WHERE h.changed_at >= :start_date
      AND h.changed_at < :end_date
),
status_history AS (
    SELECT
        CAST(h.deal_id AS VARCHAR(255)) AS SalesOpportunityId,
        LOWER(h.status) AS status,
        LOWER(LAG(h.status) OVER (PARTITION BY h.deal_id ORDER BY h.changed_at, h.id)) AS previous_status,
        h.changed_at
    FROM {DEAL_STATUS_HISTORY_TABLE} h
    WHERE h.changed_at >= :start_date
      AND h.changed_at < :end_date
),
external_sales_orders AS (
    SELECT
        CAST(o.deal_id AS VARCHAR(255)) AS SalesOpportunityId,
        o.created_at AS event_time,
        o.order_value AS DealValue,
        o.owner_name AS DealOwner,
        o.status AS DealStatus,
        o.discount_percentage AS DiscountPercentage
    FROM {SALES_ORDERS_TABLE} o
    WHERE o.deal_id IS NOT NULL
      AND o.created_at >= :start_date
      AND o.created_at < :end_date
),
external_fulfillment AS (
    SELECT
        CAST(f.deal_id AS VARCHAR(255)) AS SalesOpportunityId,
        f.fulfilled_at AS event_time,
        f.order_value AS DealValue,
        f.owner_name AS DealOwner,
        f.status AS DealStatus,
        f.discount_percentage AS DiscountPercentage
    FROM {FULFILLMENT_TABLE} f
    WHERE f.deal_id IS NOT NULL
      AND f.fulfilled_at >= :start_date
      AND f.fulfilled_at < :end_date
),
external_invoices AS (
    SELECT
        CAST(i.deal_id AS VARCHAR(255)) AS SalesOpportunityId,
        i.sent_at AS event_time,
        i.invoice_value AS DealValue,
        i.owner_name AS DealOwner,
        i.status AS DealStatus,
        i.discount_percentage AS DiscountPercentage
    FROM {INVOICES_TABLE} i
    WHERE i.deal_id IS NOT NULL
      AND i.sent_at >= :start_date
      AND i.sent_at < :end_date
),
external_payments AS (
    SELECT
        CAST(p.deal_id AS VARCHAR(255)) AS SalesOpportunityId,
        p.received_at AS event_time,
        p.payment_value AS DealValue,
        p.owner_name AS DealOwner,
        p.status AS DealStatus,
        p.discount_percentage AS DiscountPercentage
    FROM {PAYMENTS_TABLE} p
    WHERE p.deal_id IS NOT NULL
      AND p.received_at >= :start_date
      AND p.received_at < :end_date
),
events AS (
    SELECT b.SalesOpportunityId, 'Deal Created' AS ActivityName, b.deal_created_time AS EventTime, b.DealValue, b.DealOwner, b.DealStatus, b.DiscountPercentage
    FROM base_deals b
    WHERE b.deal_created_time IS NOT NULL
    UNION ALL
    SELECT a.SalesOpportunityId, 'Initial Contact Made', COALESCE(a.activity_time, a.activity_created_time), b.DealValue, b.DealOwner, b.DealStatus, b.DiscountPercentage
    FROM completed_activities a
    JOIN base_deals b ON b.SalesOpportunityId = a.SalesOpportunityId
    WHERE LOWER(COALESCE(a.activity_type, '')) IN ('call', 'email', 'meeting')
       OR LOWER(COALESCE(a.subject, '')) LIKE '%initial contact%'
    UNION ALL
    SELECT h.SalesOpportunityId, 'Deal Qualified', h.changed_at, b.DealValue, b.DealOwner, b.DealStatus, b.DiscountPercentage
    FROM stage_history h
    JOIN base_deals b ON b.SalesOpportunityId = h.SalesOpportunityId
    WHERE LOWER(COALESCE(h.stage_name, '')) IN ('contact made', 'needs defined', 'qualified')
      AND LOWER(COALESCE(h.previous_stage, '')) IN ('lead in', 'lead', 'new')
    UNION ALL
    SELECT h.SalesOpportunityId, 'Proposal Sent', h.changed_at, b.DealValue, b.DealOwner, b.DealStatus, b.DiscountPercentage
    FROM stage_history h
    JOIN base_deals b ON b.SalesOpportunityId = h.SalesOpportunityId
    WHERE LOWER(COALESCE(h.stage_name, '')) IN ('proposal made', 'quote sent', 'proposal sent')
    UNION ALL
    SELECT a.SalesOpportunityId, 'Proposal Follow-Up', COALESCE(a.activity_time, a.activity_created_time), b.DealValue, b.DealOwner, b.DealStatus, b.DiscountPercentage
    FROM completed_activities a
    JOIN base_deals b ON b.SalesOpportunityId = a.SalesOpportunityId
    WHERE LOWER(COALESCE(a.activity_type, '')) IN ('call', 'email', 'meeting')
      AND COALESCE(a.activity_time, a.activity_created_time) > COALESCE((SELECT MIN(h.changed_at) FROM stage_history h WHERE h.SalesOpportunityId = a.SalesOpportunityId AND LOWER(COALESCE(h.stage_name, '')) IN ('proposal made', 'quote sent', 'proposal sent')), TIMESTAMP '1900-01-01')
    UNION ALL
    SELECT h.SalesOpportunityId, 'Negotiation Started', h.changed_at, b.DealValue, b.DealOwner, b.DealStatus, b.DiscountPercentage
    FROM stage_history h
    JOIN base_deals b ON b.SalesOpportunityId = h.SalesOpportunityId
    WHERE LOWER(COALESCE(h.stage_name, '')) IN ('negotiation', 'negotiating')
      AND EXISTS (SELECT 1 FROM stage_history p WHERE p.SalesOpportunityId = h.SalesOpportunityId AND LOWER(COALESCE(p.stage_name, '')) IN ('proposal made', 'quote sent', 'proposal sent') AND p.changed_at <= h.changed_at)
    UNION ALL
    SELECT h.SalesOpportunityId, 'Deal Won', h.changed_at, b.DealValue, b.DealOwner, b.DealStatus, b.DiscountPercentage
    FROM status_history h
    JOIN base_deals b ON b.SalesOpportunityId = h.SalesOpportunityId
    WHERE h.status = 'won' AND COALESCE(h.previous_status, '') <> 'won'
    UNION ALL
    SELECT o.SalesOpportunityId, 'Sales Order Created', o.event_time, COALESCE(o.DealValue, b.DealValue), COALESCE(o.DealOwner, b.DealOwner), COALESCE(o.DealStatus, b.DealStatus), COALESCE(o.DiscountPercentage, b.DiscountPercentage)
    FROM external_sales_orders o
    LEFT JOIN base_deals b ON b.SalesOpportunityId = o.SalesOpportunityId
    UNION ALL
    SELECT f.SalesOpportunityId, 'Order Fulfilled', f.event_time, COALESCE(f.DealValue, b.DealValue), COALESCE(f.DealOwner, b.DealOwner), COALESCE(f.DealStatus, b.DealStatus), COALESCE(f.DiscountPercentage, b.DiscountPercentage)
    FROM external_fulfillment f
    LEFT JOIN base_deals b ON b.SalesOpportunityId = f.SalesOpportunityId
    UNION ALL
    SELECT i.SalesOpportunityId, 'Invoice Sent', i.event_time, COALESCE(i.DealValue, b.DealValue), COALESCE(i.DealOwner, b.DealOwner), COALESCE(i.DealStatus, b.DealStatus), COALESCE(i.DiscountPercentage, b.DiscountPercentage)
    FROM external_invoices i
    LEFT JOIN base_deals b ON b.SalesOpportunityId = i.SalesOpportunityId
    UNION ALL
    SELECT p.SalesOpportunityId, 'Payment Received', p.event_time, COALESCE(p.DealValue, b.DealValue), COALESCE(p.DealOwner, b.DealOwner), COALESCE(p.DealStatus, b.DealStatus), COALESCE(p.DiscountPercentage, b.DiscountPercentage)
    FROM external_payments p
    LEFT JOIN base_deals b ON b.SalesOpportunityId = p.SalesOpportunityId
    UNION ALL
    SELECT h.SalesOpportunityId, 'Deal Lost', h.changed_at, b.DealValue, b.DealOwner, b.DealStatus, b.DiscountPercentage
    FROM status_history h
    JOIN base_deals b ON b.SalesOpportunityId = h.SalesOpportunityId
    WHERE h.status = 'lost' AND COALESCE(h.previous_status, '') <> 'lost'
    UNION ALL
    SELECT h.SalesOpportunityId, 'Deal Reopened', h.changed_at, b.DealValue, b.DealOwner, b.DealStatus, b.DiscountPercentage
    FROM status_history h
    JOIN base_deals b ON b.SalesOpportunityId = h.SalesOpportunityId
    WHERE h.status = 'open' AND h.previous_status = 'lost'
)
SELECT SalesOpportunityId, ActivityName, EventTime, DealValue, DealOwner, DealStatus, DiscountPercentage
FROM events
WHERE SalesOpportunityId IS NOT NULL
  AND EventTime IS NOT NULL
ORDER BY SalesOpportunityId, EventTime, ActivityName
"""

with engine.connect() as connection:
    event_log = pd.read_sql(text(sql), connection, params={class="hl-string">"start_date": START_DATE, class="hl-string">"end_date": END_DATE})

event_log[class="hl-string">"EventTime"] = pd.to_datetime(event_log[class="hl-string">"EventTime"], utc=True, errors=class="hl-string">"coerce")
event_log = event_log.dropna(subset=[class="hl-string">"SalesOpportunityId", class="hl-string">"ActivityName", class="hl-string">"EventTime"])
event_log = event_log.sort_values([class="hl-string">"SalesOpportunityId", class="hl-string">"EventTime", class="hl-string">"ActivityName"])
event_log.to_csv(class="hl-string">"processmind_lead_to_cash_event_log.csv", index=False)
print(event_log.to_string(index=False))
print(fclass="hl-string">"Exported {len(event_log)} event rows to processmind_lead_to_cash_event_log.csv")

准备好开始了吗？

使用此模板高效准备线索到回款数据，开始发现有价值的流程洞察。立即优化您的Pipedrive运营。

返回改进指南下载模板

提升赢单率：立即优化Pipedrive线索到回款流程

提升20%的赢单率并缩短销售周期。立即开始。

开始免费试用

无需信用卡。免费试用14天。

您的线索到回款数据模板

您的线索到回款数据模板

线索到收款属性

线索到收款活动

提取指南

将Pipedrive原生数据导出为CSV/Excel

步骤

配置

a 示例查询 python

通过Pipedrive REST API使用Python脚本提取数据

步骤

配置

a 示例查询 python

使用带Pipedrive连接器的第三方ETL工具

步骤

配置

a 示例查询 python

准备好开始了吗？

提升赢单率：立即优化Pipedrive线索到回款流程

您的线索到回款数据模板

您的线索到回款数据模板

线索到收款属性

线索到收款活动

提取指南

将Pipedrive原生数据导出为CSV/Excel

步骤

配置

a 示例查询 python

通过Pipedrive REST API使用Python脚本提取数据

步骤

配置

a 示例查询 python

使用带Pipedrive连接器的第三方ETL工具

步骤

配置

a 示例查询 python

准备好开始了吗？

提升赢单率：立即优化Pipedrive线索到回款流程

我们重视您的隐私