共計 5226 個字符，預計需要花費 14 分鐘才能閱讀完成。
行業(yè)資訊
數(shù)據(jù)庫
Python 如何實現(xiàn)獲取微信企業(yè)號 access_token 的 Class
丸趣 TV 小編給大家分享一下 Python 如何實現(xiàn)獲取微信企業(yè)號 access_token 的 Class，相信大部分人都還不怎么了解，因此分享這篇文章給大家參考一下，希望大家閱讀完這篇文章后大有收獲，下面讓我們一起去了解一下吧！
微信公眾號共有三種，服務號、訂閱號、企業(yè)號。它們在獲取 AccessToken 上各有不同。其中訂閱號比較坑，它的 AccessToken 是需定時刷新，重復獲取將導致上次獲取的 AccessToken 失效。而企業(yè)號就比較好，AccessToken 有效期同樣為 7200 秒，但有效期內(nèi)重復獲取返回相同結(jié)果。為兼容這兩種方式，因此按照訂閱號的方式處理。
處理辦法與接口文檔中的要求相同：
為了保密 appsecrect，第三方需要一個 access_token 獲取和刷新的中控服務器。而其他業(yè)務邏輯服務器所使用的 access_token 均來自于該中控服務器，不應該各自去刷新，否則會造成 access_token 覆蓋而影響業(yè)務。
下面的代碼以企業(yè)號為例，將 access_token 儲存在 sqlite3 數(shù)據(jù)庫中，相比儲存在文本中, 放在數(shù)據(jù)庫里，可以為后期存放其他數(shù)據(jù)提供向后兼容。如果放在文本中，則不如放在數(shù)據(jù)庫中靈活。
設計思路和使用方法：
自動創(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 中只將真正能被用到的方法和變量設置為 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
if code is None:
print message: %s % msg
else:
print message: %s, code: %s % (msg, code)
# 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])
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
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)
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)
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)
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)
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)
