快捷導(dǎo)航

python獲取微信企業(yè)號(hào)打卡數(shù)據(jù)并生成windows計(jì)劃任務(wù)

更新時(shí)間：2019年04月30日 09:32:39 作者：半杯清茶半杯酒

由于公司的系統(tǒng)用的是Java版本，開(kāi)通了企業(yè)號(hào)打卡之后又沒(méi)有預(yù)算讓供應(yīng)商做數(shù)據(jù)對(duì)接，所以只能自己搗鼓這個(gè)，以下是個(gè)人設(shè)置的一些內(nèi)容，僅供大家參考

安裝python

python的安裝，這里就不詳細(xì)寫(xiě)了，大家可自行度娘或google。

安裝第三方庫(kù)

python安裝好之后別忘記配置環(huán)境變量！另外，所以的內(nèi)容都是安裝在服務(wù)器上的，且服務(wù)器需要能夠上外網(wǎng)，否則，只能配置在本地，因?yàn)樾枰饩W(wǎng)連接微信企業(yè)號(hào)的接口。這里需要用到幾個(gè)第三方庫(kù)：

python的pip命令，一般python安裝好之后都會(huì)默認(rèn)有，如果不確定，可輸入命令查詢，通過(guò)cmd進(jìn)入命令提示符，輸入

pip list

如果提示你需要更新，你可以更新，也可以不更新，更新命令其實(shí)給到你了python -m pip install --upgrade pip

安裝所需要的庫(kù)

Step.1

pip install pymssql

如果安裝pymssql出錯(cuò)，提示什么visual C++ 14，則先安裝wheel，如不報(bào)錯(cuò)則忽略step2、step3

Step.2

pip install wheel

Step.3

下載pymssql-2.1.4.dev5-cp37-cp37m-win_amd64.whl

可去這里下載最新版本的。pymssql下載

下載好之后，進(jìn)入該文件所在的目錄，通過(guò)pip install安裝即可cd D:\

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

step.4

pip install requests

至此，所有第三方庫(kù)都配置好了。

寫(xiě)主程序

# !/usr/bin/python
# -*- coding:utf-8 -*-
# @Time: 2018/7/26 16:05
# @Author: hychen.cc
import json # 因微信企業(yè)號(hào)返回的格式為json，所以引入json
import requests
import pymssql
import math # 引入數(shù)學(xué)方法
import time
import datetime
server = 'XX.XX.XX.XX' # 數(shù)據(jù)庫(kù)服務(wù)器地址
user = 'sa' # 數(shù)據(jù)庫(kù)登錄名，可以用sa
password = '******' # 數(shù)據(jù)庫(kù)用戶對(duì)應(yīng)的密碼
dbName = 'DBNAME' # 數(shù)據(jù)庫(kù)名稱(chēng)
CORP_ID = 'XXXXXX' # 微信企業(yè)號(hào)提供的CORP_ID
CORP_SECRET = 'XXXXX' # 微信企業(yè)號(hào)提供的CORP_SECRET
"""

因微信接口所需要unix時(shí)間戳，所以需要把時(shí)間轉(zhuǎn)為為Unix時(shí)間戳格式

定義時(shí)間轉(zhuǎn)換為Unix時(shí)間方法

"""def datetime_timestamp(dt):
 # dt為字符串
 # 中間過(guò)程，一般都需要將字符串轉(zhuǎn)化為時(shí)間數(shù)組
 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"轉(zhuǎn)化為時(shí)間戳
 s = time.mktime(time.strptime(dt, '%Y-%m-%d %H:%M:%S'))
 return int(s)
# 定義連接數(shù)據(jù)庫(kù)方法
def get_link_server():
 connection = pymssql.connect(server, user, password, database=dbName)
 if connection:
  return connection
 else:
  raise ValueError('Connect DBServer failed.')
"""

定義獲取用戶列表，因?yàn)槲⑿牌髽I(yè)號(hào)一次最大只能獲取100個(gè)，所以需要轉(zhuǎn)換為列表格式，分批次獲取

我這里設(shè)置是從DB中獲取有權(quán)限微信打卡的人員(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，因?yàn)門(mén)oken有時(shí)效（2小時(shí)），所以需要存在本地，這樣不需要頻繁調(diào)用，所以我定義了存儲(chǔ)過(guò)程（sP_GetWX_access_token）來(lái)判斷之前存儲(chǔ)的token是否有效，有效的話就不需要重復(fù)獲取了

"""
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.')
# 調(diào)用接口，獲得數(shù)據(jù)
if __name__ == '__main__':
 today = datetime.date.today()
 oneday = datetime.timedelta(days=3) # days，即獲取幾天內(nèi)的
 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把獲取到的數(shù)據(jù)解析后存入數(shù)據(jù)庫(kù)中
      conn.commit()
     conn.close()
     print ('Get punch card successed.')
    else:
     raise ValueError('No userlist exists')

設(shè)置Windows計(jì)劃任務(wù)

通過(guò)控制面板-管理工具-任務(wù)計(jì)劃程序，右擊選擇創(chuàng)建基本任務(wù)，這里注意的是路徑和程序。

在這里插入圖片描述

程序或腳本：python.exe

添加參數(shù)（可選）（A）：你的py文件目錄

起始于：python目錄，如果不知道python安裝到哪去了，按照下列cmd命令，輸入python后進(jìn)入python命令查詢

import sys
sys.prefix,回車(chē)

到此，配置完成，可自行右擊任務(wù)-執(zhí)行查詢效果，或者通過(guò)python命令執(zhí)行py文件

進(jìn)入到py文件目錄

python xxx.py

總結(jié)

以上所述是小編給大家介紹的python獲取微信企業(yè)號(hào)打卡數(shù)據(jù)并生成windows計(jì)劃任務(wù),希望對(duì)大家有所幫助，如果大家有任何疑問(wèn)請(qǐng)給我留言，小編會(huì)及時(shí)回復(fù)大家的。在此也非常感謝大家對(duì)腳本之家網(wǎng)站的支持！
如果你覺(jué)得本文對(duì)你有幫助，歡迎轉(zhuǎn)載，煩請(qǐng)注明出處，謝謝！

您可能感興趣的文章: