企业微信API实战：构建自动化的客户档案管理系统

一、准备工作：获取API调用凭证与权限配置

在开始编写代码之前，必须完成企业微信管理后台的配置，这是所有接口调用的基础。请严格按照以下步骤操作，否则后续代码将无法运行。

1. 获取企业ID与Secret

登录企业微信管理后台（https://work.weixin.qq.com/），点击“我的企业”页面的“企业信息”栏，复制并保存企业ID。

进入“应用管理” -> “应用” -> “自建”或“应用市场”，找到或创建一个应用（通常使用“联系我”或具有客户联系权限的应用）。点击进入应用详情页，在“开发者接口”信息栏中，复制并保存Secret。

2. 配置可信IP与通讯录权限

在应用详情页的“开发者接口”部分，找到“企业可信IP”配置项。将你部署代码的服务器公网IP地址填入并保存。若不配置此项，调用接口时会报“ip not in whitelist”错误。

前往“管理工具” -> “通讯录同步”，开启“API接口同步”权限。确保该应用拥有“读取客户联系”与“写客户联系”的权限。这通常需要管理员在“权限管理”中为该应用分配“客户联系”->“客户基础信息”的读取权限。

二、数据库设计：构建本地客户档案表

为了将企业微信中的客户档案数据落地存储，我们需要在本地数据库（以MySQL为例）设计一张表。该表将包含客户的基础信息、联系方式以及重要的标签数据。

执行以下SQL语句创建数据表：

```sql CREATE TABLE `wecustomer_archive` ( `id` bigint(20) NOT NULL AUTO_INCREMENT COMMENT '主键ID', `external_userid` varchar(64) NOT NULL COMMENT '外部联系人userid', `userid` varchar(64) DEFAULT NULL COMMENT '添加了此外部联系人的企业成员userid', `name` varchar(100) DEFAULT NULL COMMENT '外部联系人的名称', `avatar` varchar(255) DEFAULT NULL COMMENT '外部联系人头像url', `type` int(2) DEFAULT NULL COMMENT '外部联系人的类型：1-微信用户，2-企业微信用户', `gender` int(2) DEFAULT NULL COMMENT '性别：0-未知，1-男性，2-女性', `unionid` varchar(64) DEFAULT NULL COMMENT '外部联系人在微信开放平台的唯一身份标识', `position` varchar(100) DEFAULT NULL COMMENT '外部联系人的职位', `corp_name` varchar(100) DEFAULT NULL COMMENT '外部联系人所属企业的名称', `corp_full_name` varchar(200) DEFAULT NULL COMMENT '外部联系人所属企业的全称', `external_profile` text COMMENT '外部联系人的自定义展示信息（JSON格式）', `remark` varchar(500) DEFAULT NULL COMMENT '此成员对此外部联系人的备注', `remark_mobiles` varchar(500) DEFAULT NULL COMMENT '此成员对此外部联系人的备注手机号（JSON数组转字符串）', `remark_company_name` varchar(200) DEFAULT NULL COMMENT '此成员对此外部联系人的备注公司名称', `description` varchar(500) DEFAULT NULL COMMENT '此成员对此外部联系人的描述', `add_way` int(2) DEFAULT NULL COMMENT '添加此人的来源方式', `oper_userid` varchar(64) DEFAULT NULL COMMENT '发起添加的userid', `state` varchar(100) DEFAULT NULL COMMENT '自定义state', `tags` text COMMENT '此成员打上的标签信息（JSON格式）', `create_time` datetime DEFAULT CURRENT_TIMESTAMP COMMENT '记录创建时间', `update_time` datetime DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '记录更新时间', PRIMARY KEY (`id`), UNIQUE KEY `uk_external_userid` (`external_userid`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='企业微信客户档案表'; ```

三、核心代码实现：Python脚本全流程开发

本指南使用Python语言，依赖requests库进行HTTP请求，依赖pymysql操作数据库。请提前安装依赖：pip install requests pymysql。

新建文件wecom_sync.py，完整代码如下：

```python import requests import pymysql import json import time ================= 配置区 ================= CORP_ID = 'wwxxxxxxxxxxxxxxxxxx' 替换为你的企业ID SECRET = 'xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx' 替换为你的Secret DB_HOST = '127.0.0.1' DB_USER = 'root' DB_PASS = 'password' DB_NAME = 'wecom_db' =========================================== 数据库连接配置 db_conn = pymysql.connect(host=DB_HOST, user=DB_USER, password=DB_PASS, database=DB_NAME, charset='utf8mb4') cursor = db_conn.cursor() def get_access_token(): """获取企业微信Access Token""" url = f"https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid={CORP_ID}&corpsecret={SECRET}" response = requests.get(url).json() if response['errcode'] == 0: return response['access_token'] else: print(f"获取Token失败: {response}") exit(1) def get_follow_user_list(access_token): """获取配置了客户联系功能的成员列表""" url = f"https://qyapi.weixin.qq.com/cgi-bin/externalcontact/get_follow_user_list?access_token={access_token}" response = requests.get(url).json() if response['errcode'] == 0: return response['user_list'] else: print(f"获取成员列表失败: {response}") return [] def get_external_contact_list(access_token, userid): """获取某个成员的客户列表""" url = f"https://qyapi.weixin.qq.com/cgi-bin/externalcontact/list?access_token={access_token}&userid={userid}" response = requests.get(url).json() if response['errcode'] == 0: return response['external_userid_list'] else: print(f"获取客户列表失败(userid:{userid}): {response}") return [] def get_customer_detail(access_token, external_userid, cursor): """获取客户详情并入库""" url = f"https://qyapi.weixin.qq.com/cgi-bin/externalcontact/get?access_token={access_token}&external_userid={external_userid}" response = requests.get(url).json() if response['errcode'] != 0: print(f"获取客户详情失败(external_userid:{external_userid}): {response}") return 解析基础信息 external_contact = response.get('external_contact', {}) 解析跟进人信息（取第一个跟进人的数据作为示例） follow_user = response.get('follow_user', [{}])[0] data = { 'external_userid': external_userid, 'userid': follow_user.get('userid'), 'name': external_contact.get('name'), 'avatar': external_contact.get('avatar'), 'type': external_contact.get('type'), 'gender': external_contact.get('gender'), 'unionid': external_contact.get('unionid'), 'position': external_contact.get('position'), 'corp_name': external_contact.get('corp_name'), 'corp_full_name': external_contact.get('corp_full_name'), 'external_profile': json.dumps(external_contact.get('external_profile', {}), ensure_ascii=False), 'remark': follow_user.get('remark'), 'remark_mobiles': json.dumps(follow_user.get('remark_mobiles', []), ensure_ascii=False), 'remark_company_name': follow_user.get('remark_company_name'), 'description': follow_user.get('description'), 'add_way': follow_user.get('add_way'), 'oper_userid': follow_user.get('oper_userid'), 'state': follow_user.get('state'), 'tags': json.dumps(follow_user.get('tags', []), ensure_ascii=False) } 插入或更新数据库 (使用 ON DUPLICATE KEY UPDATE) sql = """ INSERT INTO wecustomer_archive ( external_userid, userid, name, avatar, type, gender, unionid, position, corp_name, corp_full_name, external_profile, remark, remark_mobiles, remark_company_name, description, add_way, oper_userid, state, tags ) VALUES ( %(external_userid)s, %(userid)s, %(name)s, %(avatar)s, %(type)s, %(gender)s, %(unionid)s, %(position)s, %(corp_name)s, %(corp_full_name)s, %(external_profile)s, %(remark)s, %(remark_mobiles)s, %(remark_company_name)s, %(description)s, %(add_way)s, %(oper_userid)s, %(state)s, %(tags)s ) ON DUPLICATE KEY UPDATE name = VALUES(name), avatar = VALUES(avatar), gender = VALUES(gender), remark = VALUES(remark), remark_mobiles = VALUES(remark_mobiles), remark_company_name = VALUES(remark_company_name), description = VALUES(description), tags = VALUES(tags), update_time = NOW() """ try: cursor.execute(sql, data) db_conn.commit() print(f"同步客户成功: {data['name']} ({external_userid})") except Exception as e: db_conn.rollback() print(f"数据库操作失败: {e}") def main(): print("开始同步企业微信客户档案...") token = get_access_token() user_list = get_follow_user_list(token) if not user_list: print("未获取到任何配置了客户权限的成员，请检查后台配置。") return for userid in user_list: print(f"正在处理成员: {userid} 的客户...") 防止频率限制，稍微休眠 time.sleep(0.1) external_list = get_external_contact_list(token, userid) for ext_id in external_list: 防止频率限制 time.sleep(0.1) get_customer_detail(token, ext_id, cursor) print("所有客户档案同步完成。") cursor.close() db_conn.close() if __name__ == "__main__": main() ```

四、关键数据清洗与处理逻辑

上述代码实现了基础的数据同步，但在实际业务中，标签和备注手机号是两个最关键的“档案”维度，需要特别注意。

1. 处理客户标签

企业微信返回的tags字段是一个JSON数组，包含tag_name和tag_id。在数据库中我们将其存储为JSON字符串。如果你需要查询“所有被打上‘VIP’标签的客户”，可以使用MySQL的JSON函数：

```sql SELECT FROM wecustomer_archive WHERE JSON_CONTAINS(tags->'$.tag_name', '"VIP"'); ```

或者，在Python代码解析阶段，提取出所有标签名称拼接成一个字符串存入单独的字段tag_names（需修改表结构），以便于常规的LIKE查询。

2. 处理多手机号

企业微信允许成员为同一个客户添加多个备注手机号（remark_mobiles）。返回的数据结构为数组，例如["13800000000", "13900000000"]。在上述代码中，我们将其直接JSON化存储。若你需要进行手机号去重或验证，可以在get_customer_detail函数中加入以下逻辑：

```python 原有代码基础上增加清洗逻辑 raw_mobiles = follow_user.get('remark_mobiles', []) 去重并过滤空值 clean_mobiles = list(set([m for m in raw_mobiles if m])) data['remark_mobiles'] = json.dumps(clean_mobiles, ensure_ascii=False) ```

五、档案回写：如何更新客户备注

构建档案系统的另一个核心需求是能够反向更新企业微信中的数据。例如，CRM系统计算出了客户的等级，需要回写到企业微信的“描述”字段中。

在代码中添加以下函数，实现更新客户档案的功能：

```python def update_customer_remark(access_token, external_userid, userid, new_description): """ 更新客户备注和描述 :param external_userid: 客户external_userid :param userid: 成员userid :param new_description: 新的描述内容 """ url = f"https://qyapi.weixin.qq.com/cgi-bin/externalcontact/remark?access_token={access_token}" payload = { "userid": userid, "external_userid": external_userid, "description": new_description 还可以更新 remark, remark_mobiles, remark_company_name 等字段 } response = requests.post(url, json=payload).json() if response['errcode'] == 0: print(f"更新客户 {external_userid} 描述成功") else: print(f"更新失败: {response}") ```

使用该功能时，请注意企业微信的接口频率限制。建议在批量更新时，每执行一次请求后休眠200毫秒以上，避免触发“API调用频率限制”导致账号接口被封禁。