使用python 调用钉钉宜搭接口 获取表单数据（通用模板）
·
"""
宜搭数据请求读取
"""
import requests
import json
import pandas as pd
import datetime
from dateutil import parser
class yida_request():
"""
宜搭数据请求类
用于从宜搭平台查询指定日期范围内的表单数据
"""
def __init__(self, app_key, app_secret, system_token,
start_date_str, end_date_str, app_id, form_id, tel_list=None, page=1, page_size=50):
"""
初始化宜搭数据请求类
Args:
app_key: 宜搭应用的app_key
app_secret: 宜搭应用的app_secret
system_token: 宜搭应用的system_token
start_date_str: 查询数据的开始日期，格式为'YYYY-MM-DD'
end_date_str: 查询数据的结束日期，格式为'YYYY-MM-DD'
app_id: 宜搭应用的app_id
form_id: 宜搭表单的form_id
tel_list: 手机号列表，用于筛选数据
page: 查询的页码，默认第一页 #已修改为循环读取，此参数无效
page_size: 每页查询数量，默认50条
"""
self.app_key = app_key
self.app_secret = app_secret
self.system_token = system_token
self.start_date_str = start_date_str
self.end_date_str = end_date_str
self.app_id = app_id
self.form_id = form_id
self.tel_list = tel_list or [] # 初始化手机号列表
self.access_token = None # 初始化access_token
self.page = page
self.page_size = page_size
# 1. 获取钉钉开放平台AccessToken
def get_access_token(self):
"""
获取钉钉开放平台的AccessToken
:return: 返回获取到的AccessToken字符串
"""
url = "https://oapi.dingtalk.com/gettoken"
params = {"appkey": self.app_key, "appsecret": self.app_secret}
resp = requests.get(url, params=params)
self.access_token = resp.json().get("access_token")
print('已获取token:', self.access_token)
return self.access_token
# 2. 通过手机号获取钉钉用户ID
def get_user_ids(self):
"""
通过手机号获取钉钉用户ID
:return: 返回对应的用户ID列表
返回数据固定格式formUuid,version,createTimeGMT,modifiedTimeGMT,title,actionExecutor,data,insert_time
"""
if not self.access_token:
self.get_access_token()
if not self.tel_list:
raise ValueError("手机号列表不能为空")
url = f'https://oapi.dingtalk.com/topapi/v2/user/getbymobile?access_token={self.access_token}'
user_ids = []
for tel in self.tel_list:
cont = {"mobile": tel}
resp = requests.post(url, json=cont)
data = json.loads(resp.text)
user_ids.append(data['result']['userid'])
print('已获取USERID:', user_ids)
return user_ids
# 3. 获取宜搭表单实例ID
def get_formInstanceId(self, user_id, start_time=None, end_time=None):
"""
获取表单实例ID
:return:
"""
if not self.access_token:
self.get_access_token()
start_time = start_time or self.start_date_str
end_time = end_time or self.end_date_str
print(f'查询数据时间：{start_time} 至 {end_time}')
# 准备基础参数
headers = {
"x-acs-dingtalk-access-token": self.access_token,
"Content-Type": "application/json"
}
payload = {
"systemToken": self.system_token,
"userId": user_id,
"formUuid": self.form_id,
"appType": self.app_id,
"modifiedFromTimeGMT": start_time,
"modifiedToTimeGMT": end_time,
}
all_data = [] # 存储所有分页数据
current_page = 1 # 当前页码
total_pages = None # 总页数
while total_pages is None or total_pages>=current_page:
# 更新分页参数
params = {
"pageSize": self.page_size,
"pageNumber": current_page
}
# 发送请求
url = "https://api.dingtalk.com/v1.0/yida/processes/instanceIds/"
resp = requests.post(url, headers=headers, params=params, json=payload)
resp.raise_for_status()
# 解析响应
response_data = resp.json()
# 首次请求时获取总记录数并计算总页数
if total_pages is None:
total_count = response_data.get("totalCount", 0)
if total_count == 0:
print('未查询到任何记录')
return [],0,0
total_pages = (total_count + self.page_size - 1) // self.page_size
print(f'共发现 {total_count} 条记录 | 分 {total_pages} 页')
# 获取当前页数据
page_data = response_data.get("data", [])
all_data.extend(page_data)
# 终止条件：已达总页数 或 当前页返回的数据少于每页大小
if current_page >= total_pages or len(page_data) < self.page_size:
break
current_page += 1
return all_data,total_pages,total_count
@staticmethod
# 统一处理可能为JSON的字段
def format_field(data):
if isinstance(data, (dict, list)):
return json.dumps(data, ensure_ascii=False)
return str(data or "")
# 4. 查询宜搭表单数据
def fetch_yida_forms_data(self, user_id):
"""
查询宜搭表单数据
:param user_id: 操作人钉钉用户ID
:return: 返回包含表单数据的元组列表，字段顺序为：
(formInstanceId, version, createTimeGMT, modifiedTimeGMT,
creatorUserId, modifier, modifyUser, originator, title, formData, instanceValue)
"""
formInstanceId,total_pages,total_count = self.get_formInstanceId(user_id) # 表单实例ID列表
if not self.access_token:
self.get_access_token()
start_time = self.start_date_str
end_time = self.end_date_str
all_results = []
current_page = 1 # 从第一页开始
print(f'数据时间：{start_time} 至 {end_time}')
while total_pages>0 and current_page == 1 : ## current_page < total_pages: 这个API 不会分页 多少instances 读取多少数据
headers = {
"x-acs-dingtalk-access-token": self.access_token,
"Content-Type": "application/json"
}
url = f"https://api.dingtalk.com/v1.0/yida/forms/instances/ids/query?pageSize={self.page_size}&pageNumber={current_page}"
conn_data = {
"systemToken": self.system_token,
"formUuid": self.form_id,
"appType": self.app_id,
"formInstanceIdList": formInstanceId,
"userId": user_id,
"modifiedFromTimeGMT": start_time,
"modifiedToTimeGMT": end_time,
}
current_page += 1
try:
resp = requests.post(url, headers=headers, json=conn_data)
resp.raise_for_status()
resp_data = resp.json().get("result", [])
except requests.exceptions.RequestException as e:
print(f"API请求失败: {e}")
raise
except (json.JSONDecodeError, KeyError) as e:
print(f"响应数据解析失败: {e}")
raise
for record in resp_data:
# modifyUser = self.format_field(record.get('modifyUser'))
# instanceValue = self.format_field(record.get('instanceValue', {}))
all_results.append((
str(record.get('formInstanceId', '')),
int(record.get('version', 0)),
parser.parse(record.get('createTimeGMT', '')).strftime("%Y-%m-%d %H:%M:%S"),
parser.parse(record.get('modifiedTimeGMT', '')).strftime("%Y-%m-%d %H:%M:%S"),
str(record.get('creatorUserId', '')),
str(record.get('modifier', '')),
self.format_field(record.get('title')),
self.format_field(record.get('originator')),
self.format_field(record.get('formData', {}))
))
return all_results
# 5. 查询宜搭流程数据
def fetch_yida_data(self, user_id):
"""
查询宜搭表单/流程数据
Args:
user_id: 操作人钉钉用户ID
start_time: 查询开始时间，默认为类初始化时设置的start_date_str
end_time: 查询结束时间，默认为类初始化时设置的end_date_str
Returns:
包含表单数据的元组列表，字段顺序为：
(formUuid, version, createTime, updateTime, creatorId, modifier, title, actionExecutor, formData)
"""
# 获取访问令牌和时间参数
self.access_token = self.access_token or self.get_access_token()
start_time = self.start_date_str
end_time = self.end_date_str
all_results = []
current_page = 1 # 从第一页开始
total_pages = None # 总页数初始未知
print(f'数据时间：{start_time} 至 {end_time}')
while total_pages is None or current_page <= total_pages:
# 准备请求
headers = {
"x-acs-dingtalk-access-token": self.access_token,
"Content-Type": "application/json"
}
params = {
"pageSize": self.page_size,
"pageNumber": current_page
}
payload = {
"appType": self.app_id,
"systemToken": self.system_token,
"formUuid": self.form_id,
"userId": user_id,
"modifiedFromTimeGMT": start_time,
"modifiedToTimeGMT": end_time
}
resp = requests.post(
"https://api.dingtalk.com/v2.0/yida/processes/instances",
headers=headers,
params=params,
json=payload
)
current_page+=1
resp.raise_for_status()
response_data = resp.json()
if total_pages is None:
total_count = response_data.get("totalCount", 0)
total_pages = (total_count + self.page_size - 1) // self.page_size # 向上取整
print(f'共 {total_count} 条记录 | 分 {total_pages} 页')
records = response_data.get("data", [])
for record in records:
# 处理时间字段
create_time = parser.parse(record.get('createTimeGMT', '')).strftime("%Y-%m-%d %H:%M:%S")
update_time = parser.parse(record.get('modifiedTimeGMT', '')).strftime("%Y-%m-%d %H:%M:%S")
all_results.append((
str(record.get('formUuid', '')),
int(record.get('version', 0)),
create_time,
update_time,
str(record.get('creatorUserId', '')),
str(record.get('modifier', '')),
self.format_field(record.get('title')),
self.format_field(record.get('actionExecutor')),
self.format_field(record.get('data', {}))
))
return all_results
# 6. 获取所有用户数据
def get_data(self,type='form'):
"""
获取所有用户的数据
:param type: processes/form 默认form
:return: 返回所有用户的数据列表
"""
if not self.tel_list:
raise ValueError("手机号列表不能为空")
self.get_access_token()
# 1. 获取手机号对应的钉钉用户ID
user_ids = self.get_user_ids()
# 时间戳
current_time = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
all_data = []
for user_id in user_ids:
try:
if type == 'processes':
data = self.fetch_yida_data(user_id) # 流程数据
elif type == 'form':
data = self.fetch_yida_forms_data(user_id) # 表单数据
for record in data:
# 将元组转为列表以便修改
record_list = list(record)
record_list.append(current_time) # 添加时间戳
all_data.append(tuple(record_list))
print("获取到的数据条数:", len(data))
print("数据示例:", data[0] if data else "无数据")
except Exception as e:
print(f"获取用户 {user_id} 数据时出错:", str(e))
return all_data
文件名命名为yida_request.py
调用方法：
import yida_request
if __name__ == "__main__":
# 1. 初始化类实例
yida = yida_request.yida_request(
app_key=, # 替换为你的AppKey
app_secret=, # 替换为你的AppSecret
system_token=, # 替换为你的宜搭系统Token
start_date_str=, # 查询开始时间
end_date_str=, # 查询结束时间
app_id=, # 替换为你的宜搭应用ID
form_id=, # 替换为你的表单ID
tel_list= # 替换为需要查询的手机号列表
# page = # 第几页 ，已修改为循环读取，可以不入参数
# page_size = # 一页？条数据 ，已修改为循环读取，可以不入参数
)
# 2. 调用 get_data() 获取数据
try:
data = yida.get_data() #processes/form;流程数据/表单数据
# 流程数据返回的formInstanceId 是formid，表单返回的是formUuid
# 默认 form
except Exception as e:
print("调用失败:", str(e))
原理:
通过 app_key 和 app_secret 获取 access_token
通过 tel_list 获取 钉钉用户ID
1.如果是 form 需要先获取表单实例ID UUID, 再获取实例ID 的数据
2.如果是 processes 可以直接获取 表格数据 ，但是获取的数据没有主键
get_data() 返回结果 (formuuid, version, createtimegmt, modifiedtimegmt, title,creator,modifier, actionexecutor, data)
更多推荐
4
0- 0
已为社区贡献1条内容
所有评论(0)