基于API响应实现卡车编号转ID查找并批量修改API标签

核心思路 #

先通过API获取**内部卡车编号（name字段）与API识别ID（id字段）**的映射关系，再读取CSV中的卡车编号，批量匹配ID后调用API完成标签添加/移除操作。

分步实现 #

1. 构建编号-ID映射字典 #

把API返回的列表转换成字典（键为卡车编号、值为对应API ID），比列表查找效率更高，同时统一用字符串类型避免数字格式匹配问题：

import requests
import json

def get_truck_id_mapping(api_url, headers):
    response = requests.get(api_url, headers=headers)
    response.raise_for_status()  # 捕获API请求失败的情况
    data = json.loads(response.text)
    
    truck_mapping = {}
    for user in data:
        truck_number = str(user['name'])
        api_id = user['id']
        truck_mapping[truck_number] = api_id
    return truck_mapping

# 调用示例
base_api_url = "https://YYY.YYY.com/comGpsGate/api/v.1/applications/46"
headers = {'Authorization': "YYY"}
truck_mapping = get_truck_id_mapping(f"{base_api_url}/tags/659/users?FromIndex=0&PageSize=1000", headers)

2. 读取CSV中的卡车编号 #

假设CSV仅包含一列内部卡车编号（列名示例为truck_number），用Python内置csv模块读取：

import csv

def read_truck_numbers_from_csv(csv_path):
    truck_numbers = []
    with open(csv_path, 'r', newline='', encoding='utf-8') as csvfile:
        reader = csv.DictReader(csvfile)
        for row in reader:
            # 替换为你CSV实际的列名
            truck_num = str(row['truck_number']).strip()
            if truck_num:  # 跳过空行
                truck_numbers.append(truck_num)
    return truck_numbers

# 调用示例
csv_path = "你的卡车编号列表.csv"
target_trucks = read_truck_numbers_from_csv(csv_path)

3. 批量调用API添加/移除标签 #

封装标签操作函数，支持添加（POST）和移除（需根据API文档调整请求方法与URL）：

def update_truck_tag(api_url, headers, truck_id, action="add"):
    if action == "add":
        # 与你现有代码一致的添加标签API地址
        url = f"{api_url}/tags/2259/users"
        response = requests.post(url, headers=headers, json={'id': truck_id})
    elif action == "remove":
        # 请根据GPS应用API文档调整移除标签的URL和请求方法
        url = f"{api_url}/tags/2259/users/{truck_id}"
        response = requests.delete(url, headers=headers)
    else:
        raise ValueError("action只能是add或remove")
    
    response.raise_for_status()
    return response.json()

# 批量添加标签示例
for truck_num in target_trucks:
    if truck_num in truck_mapping:
        truck_id = truck_mapping[truck_num]
        try:
            result = update_truck_tag(base_api_url, headers, truck_id, action="add")
            print(f"卡车{truck_num}（ID:{truck_id}）标签添加成功")
        except Exception as e:
            print(f"卡车{truck_num}（ID:{truck_id}）标签添加失败：{str(e)}")
    else:
        print(f"未找到卡车编号{truck_num}对应的API ID")

# 每日结束时批量移除标签，只需修改action参数
# for truck_num in target_trucks:
#     if truck_num in truck_mapping:
#         truck_id = truck_mapping[truck_num]
#         try:
#             result = update_truck_tag(base_api_url, headers, truck_id, action="remove")
#             print(f"卡车{truck_num}（ID:{truck_id}）标签移除成功")
#         except Exception as e:
#             print(f"卡车{truck_num}（ID:{truck_id}）标签移除失败：{str(e)}")

4. 完整整合代码 #

将以上模块整合，直接运行即可完成批量操作：

import requests
import json
import csv

def get_truck_id_mapping(api_url, headers):
    response = requests.get(api_url, headers=headers)
    response.raise_for_status()
    data = json.loads(response.text)
    truck_mapping = {}
    for user in data:
        truck_number = str(user['name'])
        api_id = user['id']
        truck_mapping[truck_number] = api_id
    return truck_mapping

def read_truck_numbers_from_csv(csv_path):
    truck_numbers = []
    with open(csv_path, 'r', newline='', encoding='utf-8') as csvfile:
        reader = csv.DictReader(csvfile)
        for row in reader:
            truck_num = str(row['truck_number']).strip()
            if truck_num:
                truck_numbers.append(truck_num)
    return truck_numbers

def update_truck_tag(api_url, headers, truck_id, action="add"):
    if action == "add":
        url = f"{api_url}/tags/2259/users"
        response = requests.post(url, headers=headers, json={'id': truck_id})
    elif action == "remove":
        # 请根据实际API文档调整移除标签的URL和请求方法
        url = f"{api_url}/tags/2259/users/{truck_id}"
        response = requests.delete(url, headers=headers)
    else:
        raise ValueError("action只能是add或remove")
    response.raise_for_status()
    return response.json()

if __name__ == "__main__":
    # 配置参数
    base_api_url = "https://YYY.YYY.com/comGpsGate/api/v.1/applications/46"
    headers = {'Authorization': "YYY"}
    csv_path = "卡车编号列表.csv"
    
    # 获取映射关系
    print("正在获取卡车编号与API ID的映射...")
    truck_mapping = get_truck_id_mapping(f"{base_api_url}/tags/659/users?FromIndex=0&PageSize=1000", headers)
    
    # 读取CSV中的卡车编号
    print("正在读取CSV中的卡车编号...")
    target_trucks = read_truck_numbers_from_csv(csv_path)
    
    # 批量添加标签
    print("开始批量添加标签...")
    for truck_num in target_trucks:
        if truck_num in truck_mapping:
            truck_id = truck_mapping[truck_num]
            try:
                update_truck_tag(base_api_url, headers, truck_id, action="add")
                print(f"✅ 卡车{truck_num}（ID:{truck_id}）标签添加成功")
            except Exception as e:
                print(f"❌ 卡车{truck_num}（ID:{truck_id}）标签添加失败：{str(e)}")
        else:
            print(f"⚠️ 未找到卡车编号{truck_num}对应的API ID")
    
    # 如需批量移除标签，取消下方注释并调整action
    # print("开始批量移除标签...")
    # for truck_num in target_trucks:
    #     if truck_num in truck_mapping:
    #         truck_id = truck_mapping[truck_num]
    #         try:
    #             update_truck_tag(base_api_url, headers, truck_id, action="remove")
    #             print(f"✅ 卡车{truck_num}（ID:{truck_id}）标签移除成功")
    #         except Exception as e:
    #             print(f"❌ 卡车{truck_num}（ID:{truck_id}）标签移除失败：{str(e)}")

注意事项 #

• 确保CSV文件的列名与代码中row['truck_number']一致，若列名不同请修改对应字段
• 移除标签的API地址和请求方法需严格遵循GPS应用官方文档，示例仅为参考
• 若API有请求频率限制，可添加time.sleep(0.5)等延迟避免触发限流
• 可替换print为日志模块（如logging），方便后续排查问题

内容的提问来源于stack exchange，提问作者Yorapios