安裝python

python的安裝，這裡就不詳細寫了，大家可自行度娘或google。

安裝第三方庫

python安裝好之後別忘記配置環境變數！另外，所以的內容都是安裝在伺服器上的，且伺服器需要能夠上外網，否則，只能配置在本地，因為需要外網連線微信企業號的介面。這裡需要用到幾個第三方庫：

1. python的pip命令，一般python安裝好之後都會預設有，如果不確定，可輸入命令查詢，通過cmd進入命令提示符，輸入

   pip list
   

   如果提示你需要更新，你可以更新，也可以不更新，更新命令其實給到你了

   python -m pip install --upgrade pip
    
   

2. 安裝所需要的庫
   Step.1

   pip install pymssql
   

   如果安裝pymssql出錯，提示什麼visual C++ 14，則先安裝wheel，如不報錯則忽略step2、step3
   Step.2

   pip install wheel
   

   Step.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
   

   step.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個，所以需要轉換為列表格式，分批次獲取
我這裡設定是從DB中獲取有許可權微信打卡的人員(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')

設定Windows計劃任務

通過控制面板-管理工具-任務計劃程式，右擊選擇建立基本任務，這裡注意的是路徑和程式。





程式或指令碼：python.exe
新增引數（可選）（A）：你的py檔案目錄
起始於：python目錄，如果不知道python安裝到哪去了，按照下列cmd命令，輸入python後進入python命令查詢

import sys
sys.prefix,回車

到此，配置完成，可自行右擊任務-執行查詢效果，或者通過python命令執行py檔案

1. 進入到py檔案目錄