共计 5226 个字符，预计需要花费 14 分钟才能阅读完成。
行业资讯
数据库
Python 如何实现获取微信企业号 access_token 的 Class
丸趣 TV 小编给大家分享一下 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 -*-
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)
