小编给大家分享一下Python如何实现获取微信企业号access_token的Class,相信大部分人都还不怎么了解,因此分享这篇文章给大家参考一下,希望大家阅读完这篇文章后大有收获,下面让我们一起去了解一下吧!

微信公众号共有三种,服务号、订阅号、企业号。它们在获取AccessToken上各有不同。其中订阅号比较坑,它的AccessToken是需定时刷新,重复获取将导致上次获取的AccessToken失效。而企业号就比较好,AccessToken有效期同样为7200秒,但有效期内重复获取返回相同结果。为兼容这两种方式,因此按照订阅号的方式处理。

处理办法与接口文档中的要求相同:

为了保密appsecrect,第三方需要一个access_token获取和刷新的中控服务器。而其他业务逻辑服务器所使用的access_token均来自于该中控服务器,不应该各自去刷新,否则会造成access_token覆盖而影响业务。

下面的代码以企业号为例,将access_token储存在sqlite3数据库中,相比储存在文本中,放在数据库里,可以为后期存放其他数据提供向后兼容。如果放在文本中,则不如放在数据库中灵活。

设计思路和使用方法:

自动创建sqlite3数据库,包括表结构和数据,并能在数据库表结构不存在或者数据不存在或遭删除的情况下,创建新的可用的数据

尽可能的保证Class中每一个可执行的函数单独调用都能成功。

Class中只将真正能被用到的方法和变量设置为public的。

使用时只需要修改此文件中的weixin_qy_CorpID和weixin_qy_Secret改成自己的,并import此文件,使用WeiXinTokenClass().get()方法即可得到access_token。

脚本内容可以从github上获取,地址:https://github.com/DingGuodong/LinuxBashShellScriptForOps/blob/master/projects/WeChatOps/OpsDevBestPractice/odbp_getToken.py

脚本内容如下:

#!/usr/bin/python#encoding:utf-8#-*-coding:utf8-*-"""CreatedbyPyCharm.File:LinuxBashShellScriptForOps:odbp_getToken.pyUser:GuodongCreateDate:2016/8/10CreateTime:17:04"""importosimportsqlite3importsysimporturllibimporturllib2importjsonimportdatetime#importtimeenable_debug=Truedefdebug(msg,code=None):ifenable_debug:ifcodeisNone:print"message:%s"%msgelse:print"message:%s,code:%s"%(msg,code)AUTHOR_MAIL="uberurey_ups@163.com"weixin_qy_CorpID="your_corpid"weixin_qy_Secret="your_secret"#Buildpathsinsidetheprojectlikethis:os.path.join(BASE_DIR,...)BASE_DIR=os.path.dirname(os.path.abspath(__file__))#Database#https://docs.djangoproject.com/en/1.9/ref/settings/#databasesDATABASES={'default':{'ENGINE':'db.backends.sqlite3','NAME':os.path.join(BASE_DIR,'.odbp_db.sqlite3'),}}sqlite3_db_file=str(DATABASES['default']['NAME'])defsqlite3_conn(database):try:conn=sqlite3.connect(database)exceptsqlite3.Error:print>>sys.stderr,"""\TherewasaproblemconnectingtoDatabase:%sTheerrorleadingtothisproblemwas:%sIt'spossiblethatthisdatabaseisbrokenorpermissiondenied.Ifyoucannotsolvethisproblemyourself,pleasemailto:%s"""%(database,sys.exc_value,AUTHOR_MAIL)sys.exit(1)else:returnconndefsqlite3_commit(conn):returnconn.commit()defsqlite3_close(conn):returnconn.close()defsqlite3_execute(database,sql):try:sql_conn=sqlite3_conn(database)sql_cursor=sql_conn.cursor()sql_cursor.execute(sql)sql_conn.commit()sql_conn.close()exceptsqlite3.Errorase:printesys.exit(1)defsqlite3_create_table_token():sql_conn=sqlite3_conn(sqlite3_db_file)sql_cursor=sql_conn.cursor()sql_cursor.execute('''CREATETABLE"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)defsqlite3_create_table_account():sql_conn=sqlite3_conn(sqlite3_db_file)sql_cursor=sql_conn.cursor()sql_cursor.execute('''CREATETABLE"main"."weixin_account"("id"INTEGER,"name"TEXT,"corpid"TEXT,"secret"TEXT,"current"INTEGER);''')sqlite3_commit(sql_conn)sqlite3_close(sql_conn)defsqlite3_create_tables():print"sqlite3_create_tables"sql_conn=sqlite3_conn(sqlite3_db_file)sql_cursor=sql_conn.cursor()sql_cursor.execute('''CREATETABLE"main"."weixin_token"("id"INTEGER,"access_token"TEXT,"expires_in"TEXT,"expires_on"TEXT,"is_expired"INTEGER);''')sql_cursor.execute('''CREATETABLE"main"."weixin_account"("id"INTEGER,"name"TEXT,"corpid"TEXT,"secret"TEXT,"current"INTEGER);''')sqlite3_commit(sql_conn)sqlite3_close(sql_conn)defsqlite3_set_credential(corpid,secret):try:sql_conn=sqlite3_conn(sqlite3_db_file)sql_cursor=sql_conn.cursor()sql_cursor.execute('''INSERTINTO"weixin_account"("id","name","corpid","secret","current")VALUES(1,'odbp',?,?,1)''',(corpid,secret))sqlite3_commit(sql_conn)sqlite3_close(sql_conn)exceptsqlite3.Error:sqlite3_create_table_account()sqlite3_set_credential(corpid,secret)defsqlite3_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('''INSERTINTO"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)exceptsqlite3.Error:sqlite3_create_table_token()sqlite3_set_token(access_token,expires_in,expires_on,is_expired)defsqlite3_get_credential():try:sql_conn=sqlite3_conn(sqlite3_db_file)sql_cursor=sql_conn.cursor()credential=sql_cursor.execute('''SELECT"corpid","secret"FROMweixin_accountWHEREcurrent==1;''')result=credential.fetchall()sqlite3_close(sql_conn)exceptsqlite3.Error:sqlite3_set_credential(weixin_qy_CorpID,weixin_qy_Secret)returnsqlite3_get_credential()else:ifresultisnotNoneandlen(result)!=0:returnresultelse:print"unrecoverableproblem,pleasealterto%s"%AUTHOR_MAILsys.exit(1)defsqlite3_get_token():try:sql_conn=sqlite3_conn(sqlite3_db_file)sql_cursor=sql_conn.cursor()credential=sql_cursor.execute('''SELECT"access_token","expires_on"FROMweixin_tokenWHERE"is_expired"==1;''')result=credential.fetchall()sqlite3_close(sql_conn)exceptsqlite3.Error:info=sys.exc_info()printinfo[0],":",info[1]else:ifresultisnotNoneandlen(result)!=0:returnresultelse:#print"unrecoverableproblem,pleasealterto%s"%AUTHOR_MAIL#sys.exit(1)returnNonedefsqlite3_update_token(access_token,expires_on):sql_conn=sqlite3_conn(sqlite3_db_file)sql_cursor=sql_conn.cursor()sql_cursor.execute('''UPDATE"weixin_token"SETaccess_token=?,expires_on=?WHERE_ROWID_=1;''',(access_token,expires_on))sqlite3_commit(sql_conn)sqlite3_close(sql_conn)classWeiXinTokenClass(object):def__init__(self):self.__corpid=Noneself.__corpsecret=Noneself.__use_persistence=Trueself.__access_token=Noneself.__expires_in=Noneself.__expires_on=Noneself.__is_expired=Noneifself.__use_persistence:self.__corpid=sqlite3_get_credential()[0][0]self.__corpsecret=sqlite3_get_credential()[0][1]else:self.__corpid=weixin_qy_CorpIDself.__corpsecret=weixin_qy_Secretdef__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_parametersresponse=urllib2.urlopen(url)result=response.read()token_json=json.loads(result)iftoken_json['access_token']isnotNone:get_time_now=datetime.datetime.now()#TODO(GuodongDing)tokenwillexpiredaheadoftimeornotexpiredafterthetimeexpire_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=1try:token_result_set=sqlite3_get_token()exceptsqlite3.Error:token_result_set=Noneiftoken_result_setisNoneandlen(token_result_set)==0:sqlite3_set_token(self.__access_token,self.__expires_in,self.__expires_on,self.__is_expired)else:ifself.__is_token_expired()isTrue:sqlite3_update_token(self.__access_token,self.__expires_on)else:debug("pass")returnelse:iftoken_json['errcode']isnotNone:print"errcodeis:%s"%token_json['errcode']print"errmsgis:%s"%token_json['errmsg']else:printresultdef__get_token_from_persistence_storage(self):try:token_result_set=sqlite3_get_token()exceptsqlite3.Error:self.__get_token_from_weixin_qy_api()finally:iftoken_result_setisNone: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()ifnow_time<expire_time:#print"Thetokenis%s"%access_token#print"Thetokenwillexpireon%s"%expire_timereturnaccess_tokenelse:self.__get_token_from_weixin_qy_api()returnself.__get_token_from_persistence_storage()@staticmethoddef__is_token_expired():try:token_result_set=sqlite3_get_token()exceptsqlite3.Errorase:printesys.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()ifnow_time<expire_time:returnFalseelse:returnTruedefget(self):returnself.__get_token_from_persistence_storage()

以上是“Python如何实现获取微信企业号access_token的Class”这篇文章的所有内容,感谢各位的阅读!相信大家都有了一定的了解,希望分享的内容对大家有所帮助,如果还想学习更多知识,欢迎关注亿速云行业资讯频道!