如何优化gspread+Pipedrive API的Google工作表批量更新脚本？

核心性能瓶颈分析 #

原脚本耗时主要来自三个核心问题：

• 每个电话号码重复遍历所有Pipedrive Deals页面，产生大量重复API请求
• 逐个单元格更新Google Sheets，单次更新触发一次API调用，累计开销巨大
• 匹配到Deal后单独请求Person信息，额外增加API调用次数

具体优化措施 #

1. 预加载所有Deals并建立本地电话号码映射 #

一次性拉取所有Deals数据（关联Person信息），将格式化后的电话号码与对应业务信息绑定成字典，后续查询直接本地匹配，彻底避免重复请求Pipedrive API。

2. 批量获取Person数据（通过API关联参数） #

利用Pipedrive API的include=persons参数，拉取Deals时直接返回关联的Person数据，无需单独发起Person查询请求。

3. Google Sheets批量更新 #

将所有待更新数据整理成二维数组，通过一次API请求完成全量更新，替代逐个单元格的update_cell调用。

4. 统一电话号码格式化逻辑 #

抽离成独立函数，避免重复代码，确保所有号码格式化规则一致。

5. 细节优化 #

• 替换宽泛的except:为具体异常捕获（如KeyError），避免隐藏潜在问题
• 使用整数运算替代math.ceil计算总页数，减少依赖
• 增加请求异常校验（response.raise_for_status()），及时发现API请求失败情况

优化后的完整代码 #

import httpx
import gspread
from datetime import datetime
from collections import defaultdict

API_TOKEN = "f40227............."
token = {"api_token": API_TOKEN}

stage_statuses = {
    0: "Trying to Contact",
    1: "Took App",
    2: "Rec Docs - Lender Call",
    3: "Financial Sched",
    4: "Compliance Shed",
    5: "Pending Payment",
    6: "PAID",
    7: "Sub'd to Processing",
}

def format_phone_number(phn_no: str) -> str:
    """统一格式化电话号码"""
    if not phn_no:
        return ""
    phn_no = phn_no.strip()
    if "-" in phn_no:
        return phn_no.replace("-", "")
    elif "+1" in phn_no:
        return phn_no.replace("+1", "").replace(" ", "")
    elif "Ext" in phn_no:
        return phn_no.split(" ")[0]
    elif " " in phn_no:
        return phn_no.replace(" ", "")
    return phn_no

def load_all_deals_with_persons() -> dict:
    """一次性加载所有Deals并关联Person数据，建立电话号码到信息的映射"""
    phone_to_info = defaultdict(dict)
    base_url = "https://shc2.pipedrive.com/api/v1/deals?start=0&limit=100&include=persons"
    
    # 初始化请求并获取总页数
    response = httpx.get(base_url, params=token, timeout=30)
    response.raise_for_status()
    data = response.json()
    total_count = data["additional_data"]["summary"]["total_count"]
    total_pages = (total_count + 99) // 100  # 等价于向上取整，避免导入math

    # 处理第一页数据
    process_deal_page(data["data"], phone_to_info)

    # 处理剩余页面
    for page in range(1, total_pages):
        url = f"https://shc2.pipedrive.com/api/v1/deals?start={page*100}&limit=100&include=persons"
        print(f"Loading Deals Page {page+1}/{total_pages}")
        response = httpx.get(url, params=token, timeout=30)
        response.raise_for_status()
        process_deal_page(response.json()["data"], phone_to_info)
    
    return phone_to_info

def process_deal_page(deal_list: list, phone_map: dict):
    """处理单页Deals数据，填充电话号码映射"""
    for deal in deal_list:
        if not deal.get("person_id") or not deal["person_id"].get("data"):
            continue
        person_data = deal["person_id"]["data"]
        
        # 处理该Person的所有电话号码
        for phone_entry in person_data.get("phone", []):
            raw_phone = phone_entry.get("value")
            if not raw_phone:
                continue
            formatted_phone = format_phone_number(raw_phone)
            if not formatted_phone:
                continue
            
            # 仅保留首次匹配的信息（可根据需求调整为保留最新/最高优先级Deal）
            if formatted_phone not in phone_map:
                # 处理阶段状态
                stage_status = stage_statuses.get(deal["stage_order_nr"], "Unknown")
                # 处理won/lost状态
                won_lost = ""
                if deal["status"] == "won":
                    won_lost = "WON"
                elif deal["status"] == "lost":
                    won_lost = "LOST"
                
                # 处理地址
                address = ""
                try:
                    addr_1 = person_data["2a556bd22d2c0374f609f6fafcca7949cf9b2ba2"]
                    addr_2 = person_data["c003c48faccbde63860456ee2f1a5a50f25529a5"]
                    addr_3 = person_data["14d2126d1386f43fdbd18ca803c3faab87315d46"]
                    addr_4 = person_data["2d762978f235765bbd5fc547c55beb173c0a7101"]
                    address = f"{addr_1} {addr_2} {addr_3} {addr_4}".strip()
                except KeyError:
                    address = person_data.get("2a556bd22d2c0374f609f6fafcca7949cf9b2ba2", "")
                
                phone_map[formatted_phone] = {
                    "stage_status": stage_status,
                    "won/lost": won_lost,
                    "benefit_id": person_data["ca8fd59fb797a92665b29c4ee38a45524a6ad51b"],
                    "assigned_to": person_data["owner_id"]["name"],
                    "name": person_data["name"],
                    "address": address
                }

def main():
    # 1. 预加载所有Deals的电话号码映射
    print("Loading all Deals and Person data...")
    phone_info_map = load_all_deals_with_persons()
    print(f"Loaded {len(phone_info_map)} phone number entries")

    # 2. 连接Google Sheets
    sa = gspread.service_account(filename="crm_creds.json")
    sh = sa.open("Form Submissions")
    sheet = sh.worksheet("Mailers")

    # 获取列位置
    phone_col = 2
    last_update_col = sheet.find("Last Updated").col
    benefit_id_col = sheet.find("Benefit ID").col
    stage_status_col = sheet.find("Stage Status").col
    won_lost_col = sheet.find("Won/Lost").col
    assigned_to_col = sheet.find("Assigned To").col
    name_col = sheet.find("Name").col
    address_col = sheet.find("Address").col

    # 获取所有行数据
    all_rows = sheet.get_all_values()
    header_row = all_rows[0]
    data_rows = all_rows[1:]  # 跳过表头
    dt_string = datetime.now().strftime("%d/%m/%Y %H:%M:%S")

    # 准备批量更新数据（按行组织）
    update_data = []
    for row_idx, row in enumerate(data_rows, start=2):
        raw_phone = row[phone_col-1]
        formatted_phone = format_phone_number(raw_phone)
        row_updates = [""] * len(header_row)
        
        # 填充Last Updated
        row_updates[last_update_col-1] = dt_string
        
        if formatted_phone in phone_info_map:
            info = phone_info_map[formatted_phone]
            row_updates[benefit_id_col-1] = str(info["benefit_id"])
            row_updates[stage_status_col-1] = info["stage_status"]
            row_updates[won_lost_col-1] = info["won/lost"]
            row_updates[assigned_to_col-1] = info["assigned_to"]
            row_updates[name_col-1] = info["name"]
            row_updates[address_col-1] = info["address"]
        else:
            # 填充N/A
            row_updates[benefit_id_col-1] = "N/A"
            row_updates[stage_status_col-1] = "N/A"
            row_updates[won_lost_col-1] = "N/A"
            row_updates[assigned_to_col-1] = "N/A"
            row_updates[name_col-1] = "N/A"
            row_updates[address_col-1] = "N/A"
        
        update_data.append(row_updates)
    
    # 执行批量更新
    if update_data:
        range_end = f"{chr(ord('A') + len(header_row)-1)}{len(data_rows)+1}"
        update_range = f"A2:{range_end}"
        print(f"Updating Google Sheets range: {update_range}")
        sheet.update(update_range, update_data, value_input_option="USER_ENTERED")
    
    print("Processing complete!")

if __name__ == "__main__":
    main()

优化效果说明 #

• API请求次数从O(N*P)（N为行数，P为Deals页数）降至O(P)，仅需拉取一次所有Deals数据
• Google Sheets请求次数从O(N*7)降至1次批量更新
• 整体处理时间可从3-4小时缩短至数分钟以内

内容的提问来源于stack exchange，提问作者X-something