Python调用企业微信接口获取打卡数据并配置Windows计划任务

阿澈

发布时间：2026-10-12| 3人浏览过

原创

本文介绍如何利用Python调用企业微信API获取打卡数据，并通过存储Access_Token优化接口调用。同时详细讲解如何在Windows系统中配置计划任务，实现Python脚本的自动化定时执行，适用于无预算对接系统的企业场景。

由于公司现有系统基于Java开发，开通企业微信打卡后缺乏预算让供应商进行数据对接，因此只能自行开发解决方案。以下记录的是个人配置过程中的关键步骤，仅供参考。

安装Python环境

Python的具体安装过程此处不再赘述，建议通过搜索引擎查询相关教程或官方文档进行安装。

安装第三方依赖库

Python安装完成后，务必配置好环境变量。此外，所有操作均在服务器上执行，且服务器必须能够访问外网，否则只能配置在本地环境，因为需要连接微信企业号的接口。此过程需要安装以下第三方库：

Python安装后通常默认包含pip命令。如果不确定是否已安装，可通过CMD进入命令提示符，输入进行查询。

pip list

如果系统提示需要更新，可以选择更新或忽略。更新命令已在上一步中给出：python -m pip install --upgrade pip。

安装所需的具体库文件

步骤1

pip install pymssql

如果在安装pymssql时出现关于Visual C++ 14的错误提示，请先安装wheel库；若未报错，则忽略后续步骤2和步骤3。

步骤2

pip install wheel

步骤3

下载pymssql-2.1.4.dev5-cp37-cp37m-win_amd64.whl文件。

可前往相关资源页面下载最新版本的pymssql库。

下载完成后，进入该文件所在的目录，通过pip install命令进行安装，例如：cd D:\。

pip install pymssql-2.1.4.dev5-cp37-cp37m-win_amd64.whl

步骤4

pip install requests

至此，所有第三方依赖库均已配置完成。

编写主程序逻辑

# !/usr/bin/python
# -*- coding:utf-8 -*-
# @Time: 2018/7/26 16:05
# @Author: hychen.cc
import json # 因微信企业号返回的格式为json，所以引入json
import requests
import pymssql
import math # 引入数学方法
import time
import datetime
server = 'XX.XX.XX.XX' # 数据库服务器地址
user = 'sa' # 数据库登录名，可以用sa
password = '******' # 数据库用户对应的密码
dbName = 'DBNAME' # 数据库名称
CORP_ID = 'XXXXXX' # 微信企业号提供的CORP_ID
CORP_SECRET = 'XXXXX' # 微信企业号提供的CORP_SECRET
"""

由于微信接口要求使用Unix时间戳，因此需要将常规时间格式转换为Unix时间戳格式。

定义将时间转换为Unix时间戳的方法。

"""def datetime_timestamp(dt):
 # dt为字符串
 # 中间过程，一般都需要将字符串转化为时间数组
 time.strptime(dt, '%Y-%m-%d %H:%M:%S')
 ## time.struct_time(tm_year=2018, tm_mon=10, tm_mday=25, tm_hour=10, tm_min=0, tm_sec=0, tm_wday=0, tm_yday=88, tm_isdst=-1)
 # 将"2018-10-25 10:00:00"转化为时间戳
 s = time.mktime(time.strptime(dt, '%Y-%m-%d %H:%M:%S'))
 return int(s)
# 定义连接数据库方法
def get_link_server():
 connection = pymssql.connect(server, user, password, database=dbName)
 if connection:
  return connection
 else:
  raise ValueError('Connect DBServer failed.')
"""

定义获取用户列表的功能。由于微信企业号单次接口调用最大限制为100个用户，因此需要将结果转换为列表格式，并分批次获取数据。

此处示例从数据库中查询有权限进行微信打卡的人员（Select * From Table），实际使用时请替换为获取用户列表的具体方式。

"""
def get_userid_list():
 """
 获取用户列表
 :return:
 """
 conn = get_link_server()
 cursor = conn.cursor()
 sql = "Select * From Table"
 cursor.execute(sql)
 row = cursor.fetchone()
 userlist = []
 while row:
  userlist.append(row[0])
  row = cursor.fetchone()
 if userlist:
  return userlist
 else:
  raise ValueError('Get Userlist failed.')
 conn.close()
"""

获取Access_Token。由于Token具有时效性（通常为2小时），为避免频繁调用接口，将其存储在本地数据库中。此处定义了存储过程（sP_GetWX_access_token）来判断之前存储的Token是否有效，若有效则直接复用，无需重新获取。

"""
def get_access_token(refresh=False):
 """
 获取Access Token
 :return:
 """
 if not refresh:
  API_ACCESS_TOKEN_URL = "https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid=%s&corpsecret=%s" % (
   CORP_ID, CORP_SECRET)
  response = requests.get(API_ACCESS_TOKEN_URL, verify=False)
  if response.status_code == 200:
   rep_dict = json.loads(response.text)
   errcode = rep_dict.get('errcode')
   if errcode:
    raise ValueError('Get wechat Access Token failed, errcode=%s.' % errcode)
   else:
    access_token = rep_dict.get('access_token')
    if access_token:
     conn = get_link_server()
     cursor = conn.cursor()
     cursor.execute('exec sP_GetWX_access_token @Access_Token=%s', access_token)
     conn.commit()
     conn.close()
     return access_token
    else:
     raise ValueError('Get wechat Access Token failed.')
  else:
   raise ValueError('Get wechat Access Token failed.')
 else:
  conn = get_link_server()
  cursor = conn.cursor()
  cursor.execute("Select Access_Token From wx_AccessToken Where ID=1")
  access_token = cursor.fetchone()
  if access_token:
   return access_token[0]
   conn.close()
  else:
   API_ACCESS_TOKEN_URL = "https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid=%s&corpsecret=%s" % (
    CORP_ID, CORP_SECRET)
   response = requests.get(API_ACCESS_TOKEN_URL, verify=False)
   if response.status_code == 200:
    rep_dict = json.loads(response.text)
    errcode = rep_dict.get('errcode')
    if errcode:
     raise ValueError('Get wechat Access Token failed, errcode=%s.' % errcode)
    else:
     access_token = rep_dict.get('access_token')
     if access_token:
      conn = get_link_server()
      cursor = conn.cursor()
      cursor.execute('exec sP_GetWX_access_token @Access_Token=%s', access_token)
      conn.commit()
      conn.close()
      return access_token
     else:
      raise ValueError('Get wechat Access Token failed.')
   else:
    raise ValueError('Get wechat Access Token failed.')
# 获取微信打卡的json格式
def get_punchcard_info(access_token, opencheckindatatype, starttime, endtime, useridlist):
 API_PUNCH_CARD_URL = 'https://qyapi.weixin.qq.com/cgi-bin/checkin/getcheckindata?access_token=' + access_token
 json_str = json.dumps(
  {'opencheckindatatype': opencheckindatatype, 'starttime': starttime, 'endtime': endtime, 'useridlist': useridlist})
 response = requests.post(API_PUNCH_CARD_URL, data=json_str, verify=False)
 if response.status_code == 200:
  rep_dic = json.loads(response.text)
  errcode = rep_dic.get('errcode')
  if errcode == 42001:
   access_token = get_access_token(True)
   API_PUNCH_CARD_URL = 'https://qyapi.weixin.qq.com/cgi-bin/checkin/getcheckindata?access_token=' + access_token
   json_str = json.dumps(
    {'opencheckindatatype': opencheckindatatype, 'starttime': starttime, 'endtime': endtime,
    'useridlist': useridlist})
   response = requests.post(API_PUNCH_CARD_URL, data=json_str, verify=False)
   rep_dic = json.loads(response.text)
   errcode = rep_dic.get('errcode')
   if errcode:
    raise ValueError('Get punch data failed1, errcode=%s' % errcode)
   else:
    value_str = rep_dic.get('checkindata')
    if value_str:
     return value_str
    else:
     raise ValueError('Get punch data failed2.')
  elif errcode:
   raise ValueError ('Get punch data failed3, errcode=%s' % errcode)
  else:
   value_str = rep_dic.get('checkindata')
   if value_str:
    return value_str
   else:
    raise ValueError('I do not find employee punch data.')
 else:
  raise ValueError ('Get punch data failed5.')
# 调用接口，获得数据
if __name__ == '__main__':
 today = datetime.date.today()
 oneday = datetime.timedelta(days=3) # days，即获取几天内的
 yesterday = today - oneday
 starttime = datetime_timestamp(yesterday.strftime('%Y-%m-%d') + ' 00:00:00')
 endtime = datetime_timestamp(today.strftime('%Y-%m-%d') + ' 23:59:59')
 opencheckindatatype = 3
 access_token = get_access_token()
 if access_token:
  useridlist = get_userid_list()
  if useridlist:
   step = 100
   total = len(useridlist)
   n = math.ceil(total/step)
   for i in range(n):
    # print (useridlist[i*step:(i+1)*step])
    punch_card = get_punchcard_info(access_token, opencheckindatatype, starttime, endtime,useridlist[i*step:(i+1)*step])
    # print (punch_card)
    if punch_card:
     conn = get_link_server()
     cursor = conn.cursor()
     for dic_obj in punch_card:
      cursor.execute('exec sp_AnalysisPunchCard @Json=%s',
          (json.dumps(dic_obj, ensure_ascii=False)))
      # print((json.dumps(dic_obj, ensure_ascii=False)))，sp_AnalysisPunchCard把获取到的数据解析后存入数据库中
      conn.commit()
     conn.close()
     print ('Get punch card successed.')
    else:
     raise ValueError('No userlist exists')