小編給大家分享一下Python如何實現(xiàn)獲取微信企業(yè)號access_token的Class，相信大部分人都還不怎么了解，因此分享這篇文章給大家參考一下，希望大家閱讀完這篇文章后大有收獲，下面讓我們一起去了解一下吧！
微信公眾號共有三種，服務(wù)號、訂閱號、企業(yè)號。它們在獲取AccessToken上各有不同。其中訂閱號比較坑，它的AccessToken是需定時刷新，重復(fù)獲取將導(dǎo)致上次獲取的AccessToken失效。而企業(yè)號就比較好，AccessToken有效期同樣為7200秒，但有效期內(nèi)重復(fù)獲取返回相同結(jié)果。為兼容這兩種方式，因此按照訂閱號的方式處理。
處理辦法與接口文檔中的要求相同：
為了保密appsecrect，第三方需要一個access_token獲取和刷新的中控服務(wù)器。而其他業(yè)務(wù)邏輯服務(wù)器所使用的access_token均來自于該中控服務(wù)器，不應(yīng)該各自去刷新，否則會造成access_token覆蓋而影響業(yè)務(wù)。
下面的代碼以企業(yè)號為例，將access_token儲存在sqlite3數(shù)據(jù)庫中，相比儲存在文本中,放在數(shù)據(jù)庫里，可以為后期存放其他數(shù)據(jù)提供向后兼容。如果放在文本中，則不如放在數(shù)據(jù)庫中靈活。
設(shè)計思路和使用方法：
自動創(chuàng)建sqlite3數(shù)據(jù)庫，包括表結(jié)構(gòu)和數(shù)據(jù)，并能在數(shù)據(jù)庫表結(jié)構(gòu)不存在或者數(shù)據(jù)不存在或遭刪除的情況下，創(chuàng)建新的可用的數(shù)據(jù)
盡可能的保證Class中每一個可執(zhí)行的函數(shù)單獨調(diào)用都能成功。
Class中只將真正能被用到的方法和變量設(shè)置為public的。
使用時只需要修改此文件中的weixin_qy_CorpID和weixin_qy_Secret改成自己的，并import此文件，使用WeiXinTokenClass().get()方法即可得到access_token。
腳本內(nèi)容可以從github上獲取，地址：https://github.com/DingGuodong/LinuxBashShellScriptForOps/blob/master/projects/WeChatOps/OpsDevBestPractice/odbp_getToken.py
腳本內(nèi)容如下：
#!/usr/bin/python # encoding: utf-8 # -*- coding: utf8 -*- """ Created by PyCharm. File: LinuxBashShellScriptForOps:odbp_getToken.py User: Guodong Create Date: 2016/8/10 Create Time: 17:04 """ import os import sqlite3 import sys import urllib import urllib2 import json import datetime # import time enable_debug = True def debug(msg, code=None): if enable_debug: if code is None: print "message: %s" % msg else: print "message: %s, code: %s " % (msg, code) AUTHOR_MAIL = "uberurey_ups@163.com" weixin_qy_CorpID = "your_corpid" weixin_qy_Secret = "your_secret" # Build paths inside the project like this: os.path.join(BASE_DIR, ...) BASE_DIR = os.path.dirname(os.path.abspath(__file__)) # Database # https://docs.djangoproject.com/en/1.9/ref/settings/#databases DATABASES = { 'default': { 'ENGINE': 'db.backends.sqlite3', 'NAME': os.path.join(BASE_DIR, '.odbp_db.sqlite3'), } } sqlite3_db_file = str(DATABASES['default']['NAME']) def sqlite3_conn(database): try: conn = sqlite3.connect(database) except sqlite3.Error: print >> sys.stderr, """\ There was a problem connecting to Database: %s The error leading to this problem was: %s It's possible that this database is broken or permission denied. If you cannot solve this problem yourself, please mail to: %s """ % (database, sys.exc_value, AUTHOR_MAIL) sys.exit(1) else: return conn def sqlite3_commit(conn): return conn.commit() def sqlite3_close(conn): return conn.close() def sqlite3_execute(database, sql): try: sql_conn = sqlite3_conn(database) sql_cursor = sql_conn.cursor() sql_cursor.execute(sql) sql_conn.commit() sql_conn.close() except sqlite3.Error as e: print e sys.exit(1) def sqlite3_create_table_token(): sql_conn = sqlite3_conn(sqlite3_db_file) sql_cursor = sql_conn.cursor() sql_cursor.execute('''CREATE TABLE "main"."weixin_token" ( "id" INTEGER , "access_token" TEXT, "expires_in" TEXT, "expires_on" TEXT, "is_expired" INTEGER ) ; ''') sqlite3_commit(sql_conn) sqlite3_close(sql_conn) def sqlite3_create_table_account(): sql_conn = sqlite3_conn(sqlite3_db_file) sql_cursor = sql_conn.cursor() sql_cursor.execute('''CREATE TABLE "main"."weixin_account" ( "id" INTEGER, "name" TEXT, "corpid" TEXT, "secret" TEXT, "current" INTEGER ) ; ''') sqlite3_commit(sql_conn) sqlite3_close(sql_conn) def sqlite3_create_tables(): print "sqlite3_create_tables" sql_conn = sqlite3_conn(sqlite3_db_file) sql_cursor = sql_conn.cursor() sql_cursor.execute('''CREATE TABLE "main"."weixin_token" ( "id" INTEGER , "access_token" TEXT, "expires_in" TEXT, "expires_on" TEXT, "is_expired" INTEGER ) ; ''') sql_cursor.execute('''CREATE TABLE "main"."weixin_account" ( "id" INTEGER, "name" TEXT, "corpid" TEXT, "secret" TEXT, "current" INTEGER ) ; ''') sqlite3_commit(sql_conn) sqlite3_close(sql_conn) def sqlite3_set_credential(corpid, secret): try: sql_conn = sqlite3_conn(sqlite3_db_file) sql_cursor = sql_conn.cursor() sql_cursor.execute('''INSERT INTO "weixin_account" ("id", "name", "corpid", "secret", "current") VALUES (1, 'odbp', ?, ?, 1) ''', (corpid, secret)) sqlite3_commit(sql_conn) sqlite3_close(sql_conn) except sqlite3.Error: sqlite3_create_table_account() sqlite3_set_credential(corpid, secret) def sqlite3_set_token(access_token, expires_in, expires_on, is_expired): try: sql_conn = sqlite3_conn(sqlite3_db_file) sql_cursor = sql_conn.cursor() sql_cursor.execute('''INSERT INTO "weixin_token" ("id", "access_token", "expires_in", "expires_on", "is_expired") VALUES ( 1, ?, ?, ?, ? ) ''', (access_token, expires_in, expires_on, is_expired)) sqlite3_commit(sql_conn) sqlite3_close(sql_conn) except sqlite3.Error: sqlite3_create_table_token() sqlite3_set_token(access_token, expires_in, expires_on, is_expired) def sqlite3_get_credential(): try: sql_conn = sqlite3_conn(sqlite3_db_file) sql_cursor = sql_conn.cursor() credential = sql_cursor.execute('''SELECT "corpid", "secret" FROM weixin_account WHERE current == 1;''') result = credential.fetchall() sqlite3_close(sql_conn) except sqlite3.Error: sqlite3_set_credential(weixin_qy_CorpID, weixin_qy_Secret) return sqlite3_get_credential() else: if result is not None and len(result) != 0: return result else: print "unrecoverable problem, please alter to %s" % AUTHOR_MAIL sys.exit(1) def sqlite3_get_token(): try: sql_conn = sqlite3_conn(sqlite3_db_file) sql_cursor = sql_conn.cursor() credential = sql_cursor.execute( '''SELECT "access_token", "expires_on" FROM weixin_token WHERE "is_expired" == 1 ;''') result = credential.fetchall() sqlite3_close(sql_conn) except sqlite3.Error: info = sys.exc_info() print info[0], ":", info[1] else: if result is not None and len(result) != 0: return result else: # print "unrecoverable problem, please alter to %s" % AUTHOR_MAIL # sys.exit(1) return None def sqlite3_update_token(access_token, expires_on): sql_conn = sqlite3_conn(sqlite3_db_file) sql_cursor = sql_conn.cursor() sql_cursor.execute('''UPDATE "weixin_token" SET access_token=?, expires_on=? WHERE _ROWID_ = 1;''', (access_token, expires_on) ) sqlite3_commit(sql_conn) sqlite3_close(sql_conn) class WeiXinTokenClass(object): def __init__(self): self.__corpid = None self.__corpsecret = None self.__use_persistence = True self.__access_token = None self.__expires_in = None self.__expires_on = None self.__is_expired = None if self.__use_persistence: self.__corpid = sqlite3_get_credential()[0][0] self.__corpsecret = sqlite3_get_credential()[0][1] else: self.__corpid = weixin_qy_CorpID self.__corpsecret = weixin_qy_Secret def __get_token_from_weixin_qy_api(self): parameters = { "corpid": self.__corpid, "corpsecret": self.__corpsecret } url_parameters = urllib.urlencode(parameters) token_url = "https://qyapi.weixin.qq.com/cgi-bin/gettoken?" url = token_url + url_parameters response = urllib2.urlopen(url) result = response.read() token_json = json.loads(result) if token_json['access_token'] is not None: get_time_now = datetime.datetime.now() # TODO(Guodong Ding) token will expired ahead of time or not expired after the time expire_time = get_time_now + datetime.timedelta(seconds=token_json['expires_in']) token_json['expires_on'] = str(expire_time) self.__access_token = token_json['access_token'] self.__expires_in = token_json['expires_in'] self.__expires_on = token_json['expires_on'] self.__is_expired = 1 try: token_result_set = sqlite3_get_token() except sqlite3.Error: token_result_set = None if token_result_set is None and len(token_result_set) == 0: sqlite3_set_token(self.__access_token, self.__expires_in, self.__expires_on, self.__is_expired) else: if self.__is_token_expired() is True: sqlite3_update_token(self.__access_token, self.__expires_on) else: debug("pass") return else: if token_json['errcode'] is not None: print "errcode is: %s" % token_json['errcode'] print "errmsg is: %s" % token_json['errmsg'] else: print result def __get_token_from_persistence_storage(self): try: token_result_set = sqlite3_get_token() except sqlite3.Error: self.__get_token_from_weixin_qy_api() finally: if token_result_set is None: self.__get_token_from_weixin_qy_api() token_result_set = sqlite3_get_token() access_token = token_result_set[0][0] expire_time = token_result_set[0][1] else: access_token = token_result_set[0][0] expire_time = token_result_set[0][1] expire_time = datetime.datetime.strptime(expire_time, '%Y-%m-%d %H:%M:%S.%f') now_time = datetime.datetime.now() if now_time < expire_time: # print "The token is %s" % access_token # print "The token will expire on %s" % expire_time return access_token else: self.__get_token_from_weixin_qy_api() return self.__get_token_from_persistence_storage() @staticmethod def __is_token_expired(): try: token_result_set = sqlite3_get_token() except sqlite3.Error as e: print e sys.exit(1) expire_time = token_result_set[0][1] expire_time = datetime.datetime.strptime(expire_time, '%Y-%m-%d %H:%M:%S.%f') now_time = datetime.datetime.now() if now_time < expire_time: return False else: return True def get(self): return self.__get_token_from_persistence_storage()以上是“Python如何實現(xiàn)獲取微信企業(yè)號access_token的Class”這篇文章的所有內(nèi)容，感謝各位的閱讀！相信大家都有了一定的了解，希望分享的內(nèi)容對大家有所幫助，如果還想學(xué)習(xí)更多知識，歡迎關(guān)注創(chuàng)新互聯(lián)行業(yè)資訊頻道！
另外有需要云服務(wù)器可以了解下創(chuàng)新互聯(lián)cdcxhl.cn，海內(nèi)外云服務(wù)器15元起步，三天無理由+7*72小時售后在線，公司持有idc許可證，提供“云服務(wù)器、裸金屬服務(wù)器、高防服務(wù)器、香港服務(wù)器、美國服務(wù)器、虛擬主機、免備案服務(wù)器”等云主機租用服務(wù)以及企業(yè)上云的綜合解決方案，具有“安全穩(wěn)定、簡單易用、服務(wù)可用性高、性價比高”等特點與優(yōu)勢，專為企業(yè)上云打造定制，能夠滿足用戶豐富、多元化的應(yīng)用場景需求。
重慶分公司
重慶分公司
