代码收藏家技术教程 2024-11-17

【python】批量获取企业微信审批列表及详情

最近做的一个企业内部开发企业微信应用需要批量获取审批保存出来，记录一下，本方案用的不是回调方式，只是定时批量。

解决方案的思路：批量获取3天内的审批列表–>获取审批详情(需要根据自己的模板具体处理)–>获取发起人姓名–>获取部门名称–>生成json

企业企微申请模板：

处理结果：

需要使用到的接口：

1、获取access_token

获取access_token – 接口文档 – 企业微信开发者中心

请求方式： GET（HTTPS）
请求地址： https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid=ID&corpsecret=SECRET

提示

此处标注大写的单词 ID 和 SECRET，为需要替换的变量，根据实际获取值更新。其它接口也采用相同的标注，不再说明。

参数说明：

参数	必须	说明
corpid	是	企业ID，获取方式参考：术语说明-corpid
corpsecret	是	应用的凭证密钥，注意应用需要是启用状态，获取方式参考：术语说明-secret

2、批量获取审批单号

批量获取审批单号 – 接口文档 – 企业微信开发者中心

请求方式：POST（HTTPS）
请求地址：https://qyapi.weixin.qq.com/cgi-bin/oa/getapprovalinfo?access_token=ACCESS_TOKEN

请求示例：

{
    "starttime" : "1569546000",
    "endtime" : "1569718800",
    "new_cursor" : "" ,
    "size" : 100 ,
    "filters" : [
        {
            "key": "template_id",
            "value": "ZLqk8pcsAoaXZ1eY56vpAgfX28MPdYU3ayMaSPHaaa"
        },
        {
            "key" : "creator",
            "value" : "WuJunJie"
        },
        {
            "key" : "department",
            "value" : "1"
        },       
        {
            "key" : "sp_status",
            "value" : "1"
        }     
    ]
}

参数说明：

参数	必须	说明
access_token	是	调用接口凭证。必须使用审批应用或企业内自建应用的secret获取，获取方式参考：文档-获取access_token
starttime	是	审批单提交的时间范围，开始时间，UNix时间戳
endtime	是	审批单提交的时间范围，结束时间，Unix时间戳
new_cursor	是	分页查询游标，默认为空串，后续使用返回的new_next_cursor进行分页拉取
size	是	一次请求拉取审批单数量，默认值为100，上限值为100。若accesstoken为自建应用，仅允许获取在应用可见范围内申请人提交的表单，返回的sp_no_list个数可能和size不一致，开发者需用next_cursor判断表单记录是否拉取完
filters	否	筛选条件，可对批量拉取的审批申请设置约束条件，支持设置多个条件
└ key	否	筛选类型，包括： template_id – 模板类型/模板id； creator – 申请人； department – 审批单提单者所在部门； sp_status – 审批状态; record_type – 审批单类型属性，1-请假；2-打卡补卡；3-出差；4-外出；5-加班； 6- 调班；7-会议室预定；8-退款审批；9-红包报销审批注意: 1、仅“部门”支持同时配置多个筛选条件。 2、不同类型的筛选条件之间为“与”的关系，同类型筛选条件之间为“或”的关系 3、record_type筛选类型仅支持2021/05/31以后新提交的审批单，历史单不支持表单类型属性过滤
└ value	否	筛选值，对应为：template_id-模板id；creator-申请人userid ；department-所在部门id；sp_status-审批单状态（1-审批中；2-已通过；3-已驳回；4-已撤销；6-通过后撤销；7-已删除；10-已支付）

3、获取审批申请详情

获取审批申请详情 – 接口文档 – 企业微信开发者中心

请求方式：POST（HTTPS）
请求地址： https://qyapi.weixin.qq.com/cgi-bin/oa/getapprovaldetail?access_token=ACCESS_TOKEN

请求示例：

{
   "sp_no" : "201909270001"
}

参数说明：

参数	必须	说明
access_token	是	调用接口凭证。
sp_no	是	审批单编号。

4、读取成员

读取成员 – 接口文档 – 企业微信开发者中心

请求方式：GET（HTTPS）
请求地址：https://qyapi.weixin.qq.com/cgi-bin/user/get?access_token=ACCESS_TOKEN&userid=USERID

参数说明：

参数	必须	说明
access_token	是	调用接口凭证
userid	是	成员UserID。对应管理端的账号，企业内必须唯一。不区分大小写，长度为1~64个字节

5、获取单个部门详情

获取单个部门详情 – 接口文档 – 企业微信开发者中心

请求方式：GET（HTTPS）
请求地址：https://qyapi.weixin.qq.com/cgi-bin/department/get?access_token=ACCESS_TOKEN&id=ID

参数说明：

参数	必须	说明
access_token	是	调用接口凭证
id	是	部门id。

下面为完整代码：

import http.client
import urllib.parse
from datetime import datetime, timedelta
import time
import json

# 定义函数用于获取微信企业号的 access_token
def get_access_token(corpid, corpsecret):
    """
    此函数用于向微信企业号获取 access_token

    参数:
    corpid (str): 企业号 ID
    corpsecret (str): 企业号密钥

    返回:
    str: 获取到的 access_token，如果获取失败则抛出异常
    """
    conn = http.client.HTTPSConnection("qyapi.weixin.qq.com")
    params = {
        'corpid': corpid,
        'corpsecret': corpsecret
    }
    query_string = urllib.parse.urlencode(params)
    request_path = f"/cgi-bin/gettoken?{query_string}"
    headers = {
        'Accept': "*/*",
        'Accept-Encoding': "gzip, deflate, br",
        'User-Agent': "PostmanRuntime-ApipostRuntime/1.1.0",
        'Connection': "keep-alive"
    }
    conn.request("GET", request_path, headers=headers)
    res = conn.getresponse()
    data = res.read()
    conn.close()
    token_response = json.loads(data.decode("utf-8"))
    if 'access_token' in token_response:
        return token_response['access_token']
    else:
        raise Exception(f"Failed to get access_token: {token_response}")

# 获取当前时间，并计算 3天前的时间
now = datetime.now()
thirty_days_ago = now - timedelta(days=3)

# 将时间转换为 Unix 时间戳（秒）
end_time_timestamp = int(time.mktime(now.timetuple()))
start_time_timestamp = int(time.mktime(thirty_days_ago.timetuple()))

# 构建 payload，用于获取审批信息请求
payload = {
    "starttime": start_time_timestamp,
    "endtime": end_time_timestamp,
    "new_cursor": "",
    "size": 100,
    "filters": [
        {
            "key": "template_id",
            "value": "C4WrCTnZYAzpfHWPAiMfCx1LsddahMpRhANi4Z5vN"
        },
        {
            "key": "sp_status",
            "value": "2"
        }
    ]
}

# 将 payload 转换为 JSON 字符串
payload_json = json.dumps(payload)

# 设置请求头，用于后续请求
headers = {
    'Accept': "*/*",
    'Accept-Encoding': "gzip, deflate, br",
    'User-Agent': "PostmanRuntime-ApipostRuntime/1.1.0",
    'Connection': "keep-alive",
    'Content-Type': "application/json"
}

# 获取企业号 ID 和密钥，并调用函数获取 access_token
corpid = '替换你的企业id'
corpsecret = '替换你的corpsecret'
access_token = get_access_token(corpid, corpsecret)

# 使用 access_token 发送 POST 请求获取审批信息列表
conn = http.client.HTTPSConnection("qyapi.weixin.qq.com")
request_path = f"/cgi-bin/oa/getapprovalinfo?access_token={access_token}"
conn.request("POST", request_path, body=payload_json, headers=headers)
res = conn.getresponse()
data = res.read()
conn.close()

# 解析响应数据
response_data = json.loads(data.decode("utf-8"))

# 检查响应数据中是否有 sp_no_list 且它是一个列表，如果不是则抛出异常
if 'sp_no_list' in response_data and isinstance(response_data['sp_no_list'], list):
    sp_no_list = response_data['sp_no_list']
else:
    raise Exception(f"Failed to get sp_no_list: {response_data}")

# 对每个审批编号进行查询
for sp_no in sp_no_list:
    # 构建用于获取审批详细信息的 payload
    detail_payload = {
        "sp_no": sp_no
    }
    detail_payload_json = json.dumps(detail_payload)

    # 发送 POST 请求获取审批详细信息
    conn = http.client.HTTPSConnection("qyapi.weixin.qq.com")
    detail_request_path = f"/cgi-bin/oa/getapprovaldetail?access_token={access_token}"
    conn.request("POST", detail_request_path, body=detail_payload_json, headers=headers)
    res = conn.getresponse()
    detail_data = res.read()
    conn.close()

    # 解析审批详细信息的 JSON 数据
    data = json.loads(detail_data.decode("utf-8"))

    # 获取审批编号
    approval_number = data['info'].get('sp_no', '无')

    # 将 Unix 时间戳转换为日期时间格式作为提交时间，如果没有提交时间则设为'无'
    apply_time = (datetime.fromtimestamp(data['info'].get('apply_time', 0)).strftime('%Y/%m/%d %H:%M:%S')
                  if data['info'].get('apply_time') else '无')

    # 获取申请人 userid
    applicant_userid = data['info']['applyer'].get('userid', '无')

    # 获取申请人部门 partyid
    applicant_partyid = data['info']['applyer'].get('partyid', '无')



    # 获取差旅类型，如果数据结构不符合预期则设为'无'
    apply_data_contents = data['info']['apply_data']['contents']
    if len(apply_data_contents) > 0:
        value_selector_options = apply_data_contents[0]['value']['selector']['options']
        if value_selector_options and len(value_selector_options) > 0 and value_selector_options[0]['value'] and len(value_selector_options[0]['value']) > 0:
            travel_type = value_selector_options[0]['value'][0].get('text', '无')
        else:
            travel_type = '无'
    else:
        travel_type = '无'

    # 获取出差事由说明，如果数据结构不符合预期则设为'无'
    if len(apply_data_contents) > 1:
        reason_for_travel = apply_data_contents[1]['value'].get('text', '无')
    else:
        reason_for_travel = '无'

    # 处理行程信息
    trips = []
    if len(apply_data_contents) > 2 and 'children' in apply_data_contents[2]['value']:
        for trip in apply_data_contents[2]['value']['children']:
            # 提取行程中的各个字段，如果数据结构不符合预期则设为'无'
            trip_info = {
                '出发地chufadi': trip['list'][0]['value'].get('text', '无'),
                '目的地mudedi': trip['list'][1]['value'].get('text', '无'),
                '出发日期chufariqi': (datetime.fromtimestamp(int(trip['list'][2]['value']['date'].get('s_timestamp', 0))).strftime('%Y/%m/%d')
                           if 'date' in trip['list'][2]['value'] and 's_timestamp' in trip['list'][2]['value']['date'] else '无'),
                '结束日期jieshuriqi': (datetime.fromtimestamp(int(trip['list'][3]['value']['date'].get('s_timestamp', 0))).strftime('%Y/%m/%d')
                           if 'date' in trip['list'][3]['value'] and 's_timestamp' in trip['list'][3]['value']['date'] else '无'),
                '拟拜访客户单位nibaifangkehudanwei': trip['list'][4]['value'].get('text', '无'),
                '客户联系人kehulianxiren': trip['list'][5]['value'].get('text', '无'),
                '交通工具jiaotonggongju': (trip['list'][6]['value'].get('selector', {}).get('options', [{}])[0].get('value', [{}])[0].get('text', '无')
                            if 'list' in trip and len(trip['list']) > 6 and 'value' in trip['list'][6] and 'selector' in trip['list'][6]['value'] and 'options' in trip['list'][6]['value']['selector'] and len(trip['list'][6]['value']['selector']['options']) > 0 and len(trip['list'][6]['value']['selector']['options'][0]['value']) > 0 else '无')
            }
            trips.append(trip_info)
    else:
        trips = []

    # 通过申请人 userid 获取申请人姓名
    user_name = '无'
    if applicant_userid!= '无':
        conn = http.client.HTTPSConnection("qyapi.weixin.qq.com")
        request_path = f"/cgi-bin/user/get?access_token={access_token}&userid={applicant_userid}"
        conn.request("POST", request_path, "", headers)
        res = conn.getresponse()
        data = res.read()
        conn.close()
        user_info = json.loads(data.decode("utf-8"))
        user_name = user_info.get('name', '无')

    # 通过申请人部门 partyid 获取申请人部门名称
    department_name = '无'
    if applicant_partyid!= '无':
        conn = http.client.HTTPSConnection("qyapi.weixin.qq.com")
        request_path = f"/cgi-bin/department/get?access_token={access_token}&id={applicant_partyid}"
        conn.request("POST", request_path, "", headers)
        res = conn.getresponse()
        data = res.read()
        conn.close()
        department_info = json.loads(data.decode("utf-8"))
        if 'department' in department_info and 'name' in department_info['department']:
            department_name = department_info['department']['name']

    # 构建最终输出的字典，包含审批编号、提交时间、申请人部门、差旅类型、出差事由说明、行程信息、申请人 userid、申请人姓名、申请人部门 partyid 和申请人部门名称
    output = {
        "审批编号sp_no": approval_number,
        "提交时间apply_time": apply_time,
         "申请人 userid": applicant_userid,
        "申请人姓名user_name": user_name,
        "申请人部门 partyid": applicant_partyid,
        "申请人部门department_name": department_name,
        "差旅类型travel_type": travel_type,
        "出差事由说明reason_for_travel": reason_for_travel,
        "行程trips": trips
       
    }

    # 将输出内容转换为 JSON 格式，并确保输出的中文字符不乱码，同时设置缩进为 4 个空格
    output_json = json.dumps(output, ensure_ascii=False, indent=4)

    # 打印 JSON 格式的输出
    print(output_json)

作者：志伟呀