tips:
某天，項(xiàng)目經(jīng)理說，項(xiàng)目上除了本身自帶的登陸注冊，也需要第三方的登陸注冊。方便用戶使用我們的產(chǎn)品。于是便開始加上微信登陸注冊的功能。
flask、flask-sqlalchemy、PyMySQLappid和secret，web和app的appid、secret不一樣。user（用戶表，存儲用戶信息）
| 字段 | 類型 | 含義 |
|---|---|---|
| id | int | 用戶主鍵ID |
| name | string | 用戶名稱 |
| age | int | 年齡 |
| … | … | … |
user_login_method（用戶登陸方式表，存儲不同登陸方式）
| 字段 | 類型 | 含義 |
|---|---|---|
| id | int | 用戶登陸方式主鍵ID |
| user_id | int | 用戶主鍵ID |
| login_method | string | 用戶登陸方式，WX微信，P手機(jī) |
| identification | string | 用戶登陸標(biāo)識，微信ID或手機(jī)號 |
| access_code | string | 用戶登陸通行碼，密碼或token |
| … | … | … |
model.pyfrom flask_sqlalchemy import SQLAlchemydb = SQLAlchemy()class User(db.Model): """ 用戶表 """ __tablename__ = 'user' id = db.Column(db.Integer, autoincrement=True, primary_key=True) # 用戶姓名 name = db.Column(db.String(20), nullable=False) # 用戶年齡 age = db.Column(db.Integer, nullable=False) class UserLoginMethod(db.Model): """ 用戶登陸驗(yàn)證表 """ __tablename__ = 'user_login_method' # 用戶登陸方式主鍵ID id = db.Column(db.Integer, autoincrement=True, primary_key=True) # 用戶主鍵ID user_id = db.Column(db.Integer, nullable=False) # 用戶登陸方式，WX微信，P手機(jī) login_method = db.Column(db.String(36), nullable=False) # 用戶登陸標(biāo)識，微信ID或手機(jī)號 identification = db.Column(db.String(36), nullable=False) # 用戶登陸通行碼，密碼或token access_code = db.Column(db.String(36), nullable=True) 前端拉取微信認(rèn)證授權(quán)二維碼，參考網(wǎng)站應(yīng)用微信登錄開發(fā)指南。此處只需要第一步，用戶掃碼確認(rèn)之后獲取到code傳入后端。
獲取授權(quán)憑證wx_login_or_register.get_access_code
import jsonfrom urllib import parse, requestfrom model import UserLoginMethod, User, dbdef get_access_code(code, flag): """ 獲取微信授權(quán)碼 :param code:前端或app拉取的到臨時授權(quán)碼 :param flag:web端或app端 :return:None 或 微信授權(quán)數(shù)據(jù) """ # 判斷是web端登陸還是app端登陸，采用不同的密鑰。 if flag == "web": appid = "web_appid" secret = "web_secret" elif flag == "app": appid = "app_appid" secret = "app_secret" else: return None try: # 把查詢條件轉(zhuǎn)成url中形式 fields = parse.urlencode( {"appid": appid, "secret": secret, "code": code, "grant_type": "authorization_code"} ) # 拼接請求鏈接 url = 'https://api.weixin.qq.com/sns/oauth2/access_token?{}'.format(fields) print(url) req = request.Request(url=url, method="GET") # 請求數(shù)據(jù) res = request.urlopen(req, timeout=10) # 解析數(shù)據(jù) access_data = json.loads(res.read().decode()) print(access_data) except Exception as e: print(e) return None # 拉取微信授權(quán)成功返回 # { # "access_token": "ACCESS_TOKEN", "expires_in": 7200,"refresh_token": "REFRESH_TOKEN", # "openid": "OPENID","scope": "SCOPE" # } if "openid" in access_data: return access_data # 拉取微信授權(quán)失敗 # { # "errcode":40029,"errmsg":"invalid code" # } else: return None獲取微信用戶信息wx_login_or_register.get_wx_user_info
def get_wx_user_info(access_data: dict): """ 獲取微信用戶信息 :return: """ openid = access_data.get("openid") access_token = access_data.get("access_token") try: # 把查詢條件轉(zhuǎn)成url中形式 fields = parse.urlencode({"access_token": access_token, "openid": openid}) # 拼接請求鏈接 url = 'https://api.weixin.qq.com/sns/userinfo?{}'.format(fields) print(url) req = request.Request(url=url, method="GET") # 請求數(shù)據(jù),超時10s res = request.urlopen(req, timeout=10) # 解析數(shù)據(jù) wx_user_info = json.loads(res.read().decode()) print(wx_user_info) except Exception as e: print(e) return None # 獲取成功 # { # "openid":"OPENID", # "nickname":"NICKNAME", # "sex":1, # "province":"PROVINCE", # "city":"CITY", # "country":"COUNTRY", # "headimgurl": "test.png", # "privilege":[ # "PRIVILEGE1", # "PRIVILEGE2" # ], # "unionid": " o6_bmasdasdsad6_2sgVt7hMZOPfL" # # } if "openid" in wx_user_info: return wx_user_info # 獲取失敗 # {"errcode":40003,"errmsg":"invalid openid"} else: return None驗(yàn)證本地用戶信息wx_login_or_register.login_or_register，關(guān)鍵的是比較微信用戶信息中的unionid。
def login_or_register(wx_user_info): """ 驗(yàn)證該用戶是否注冊本平臺，如果未注冊便注冊后登陸，否則直接登陸。 :param wx_user_info:拉取到的微信用戶信息 :return: """ # 微信統(tǒng)一ID unionid = wx_user_info.get("unionid") # 用戶昵稱 nickname = wx_user_info.get("nickname") # 拉取微信用戶信息失敗 if unionid is None: return None # 判斷用戶是否存在與本系統(tǒng) user_login = db.session(UserLoginMethod). filter(UserLoginMethod.login_method == "WX", UserLoginMethod.identification == unionid, ).first() # 存在則直接返回用戶信息 if user_login: user = db.session.query(User.id, User.name). filter(User.id == user_login.user_id).first() data = dict(zip(user.keys(), user)) return data # 不存在則先新建用戶然后返回用戶信息 else: try: # 新建用戶信息 new_user = User(name=nickname, age=20) db.session.add(new_user) db.session.flush() # 新建用戶登陸方式 new_user_login = UserLoginMethod(user_id=new_user.id, login_method="WX", identification=unionid, access_code=None) db.session.add(new_user_login) db.session.flush() # 提交 db.session.commit() except Exception as e: print(e) return None data = dict(id=new_user.id, name=User.name) return dataFlask接口
from flask import Flaskfrom wx_login_or_register import get_access_code, get_wx_user_info, login_or_registerfrom model import dbapp = Flask(__name__)app.config["SQLALCHEMY_DATABASE_URI"]='mysql+pymysql://root:mad123@localhost:3306/test?charset=utf8mb4'# 注冊數(shù)據(jù)庫連接db.app = appdb.init_app(app)@app.route("/testWXLoginOrRegister",methods=["GET"])def test_wx_login_or_register(): """ 測試微信登陸注冊 :return: """ # 前端獲取到的臨時授權(quán)碼 code = request.args.get("code") # 標(biāo)識web端還是app端登陸或注冊 flag = request.args.get("flag") # 參數(shù)錯誤 if code is None or flag is None: return "參數(shù)錯誤" # 獲取微信用戶授權(quán)碼 access_code = get_access_code(code=code, flag=flag) if access_code is None: return "獲取微信授權(quán)失敗" # 獲取微信用戶信息 wx_user_info = get_wx_user_info(access_data=access_code) if wx_user_info is None: return "獲取微信授權(quán)失敗" # 驗(yàn)證微信用戶信息本平臺是否有， data = login_or_register(wx_user_info=wx_user_info) if data is None: return "登陸失敗" return dataif ___name__ =="__main__": app.run()測試訪問http://127.0.0.1:5000/testWXLoginOrRegister即可。
