需求背景
运维平台开发完成后,由运维执行业务变更操作,但是有时候研发那边比较急,而运维也有不在电脑前的时候,这样的话就比较麻烦了,所以想做成钉钉免登陆的方式,当工单任务下发的时候,直接通过手机钉钉登陆到执行页面点击操作,然后再根据项目分执行权限给研发,这样的话就省心多了
前提
1,需要一台外网测试机(阿里云服务器)
2,钉钉上注册一个组织,然后登录钉钉管理后台,创建部门,添加成员,角色信息
浏览器输入https://oa.dingtalk.com/register_new.htm?showmenu=false#/
https://oa.dingtalk.com/contacts.htm#/contacts?deptManage
3,登陆钉钉开发者后台,点击应用开发,选择企业内部应用,然后选择小程序,然后单机目标应用详情页面在基础信息页面可以查看到应用的SuiteKey/SuiteSecret(第三方企业应用)或AppKey/AppSecret(企业内部应用)
open-dev.dingtalk.com/fe/app#/
创建应用后,点击应用进入应用详情页面
然后在权限管理这里开通获取用户个人信息, 查询个人授权记录,通讯录部门信息读权限,成员信息读权限,通讯录部门成员读权限 这些权限开通
3,钉应用上配置登陆与分享回调地址
参考文档
钉钉内免登第三方网站 - 钉钉开放平台 (dingtalk.com)
flask测试钉免密登陆系统项目结构
代码详情
app.py
# -*- coding:UTF-8 -*-
import base64
import hmac
import json
import time
from hashlib import sha256
from urllib.parse import unquote, quote
import click
import requests
from flask import Flask, url_for, render_template, redirect, flash, request, make_response, jsonify
import os
from actlog import HaLog
from flask_login import LoginManager, login_user, login_required, logout_user
from Ding_Dev_Tools.DDtoken import get_token
from Ding_Dev_Tools.DDuser import DingUser
from db import db
from form import LoginForm, RegistrationForm
from models import User
from flask_login import current_user
from Ding_Dev_Tools.DDinfo import *
# 获取当前项目的绝对路径
basedir = os.path.abspath(os.path.dirname(__file__))
# 当前项目服务部署的机器ip server或域名server地址
Ding_Login_Server = "http://xxx.xxx.xxx."
# 初始化logger
logger = HaLog("running.log")
def init_app():
ha_app = Flask(__name__)
# 载入配置项
ha_app.config['DEBUG'] = True
# 尤其在涉及(flask-WTF)登陆页面提交敏感信息时,一定要设置密钥
ha_app.config['SECRET_KEY'] = os.environ.get('SECRET_KEY') or 'westos'
# 数据库引擎配置,用sqlite测试
ha_app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///' + os.path.join(basedir, 'data-dev.sqlite')
# 是否自动提交
ha_app.config['SQLALCHEMY_COMMIT_ON_TEARDOM'] = True
# 是否追踪修改,从Flask-SQLALchemy文档中查看
ha_app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = True
# 初始化绑定DB 与 Flask
db.init_app(ha_app)
return ha_app
app = init_app()
login_manager = LoginManager(app)
# session_protection 属性提供不同的安全等级防止用户会话遭篡改。
login_manager.session_protection = 'basic'
# login_view 属性设置登录页面的端点。
login_manager.login_view = 'login' # required_login 校验不通过,默认跳转
@login_manager.user_loader
def load_user(user_id):
"""
加载用户的回调函数接收以unicode字符串形式表示的用户标识符,如果能找到用户,这个函数必须返回用户对象,否则应该返回None
:param user_id:
:return:
"""
return User.query.get(int(user_id))
def get_login(FULL_URL, form):
"""
:param FULL_URL:
:param form:
:return:
"""
# 如果useragent是钉钉的话直接登录
user_agent = request.headers.get('User-Agent')
if "AliApp(DingTalk" in user_agent:
return dinglogin(FULL_URL)
if current_user.is_authenticated: # 已登录状态
logger.info(f"get_login {current_user.username}用户已登录")
# noinspection PyBroadException
try:
rurl = FULL_URL.split("next=")[1].strip("&") # 截取字符串 获取next 清除结尾的&
except:
rurl = None
if rurl is not None and rurl != '': # redirectUrl 参数存在切不为空
# --start-- 生成response
r = make_response('', 302)
r.headers["Location"] = rurl
# --end-- 生成response
return r
else:
return "欢迎来到主页"
else: # 未登录状态
if FULL_URL is not None and FULL_URL != '': # redirectUrl 参数存在且不为空
return render_template('login.html', form=form, redirectUrl=FULL_URL)
else:
return render_template('login.html', form=form, redirectUrl=None)
def post_login(form):
"""
:param form:
:return:
"""
FULL_URL = unquote(request.url)
logger.info("post_login FULL_URL: {}".format(FULL_URL))
# noinspection PyBroadException
try:
rurl = FULL_URL.split("redirectUrl=")[1].strip("&") # 截取字符串 获取redirectUrl 清除结尾的&
except:
rurl = None
logger.info("post_login rurl: {}".format(rurl))
if form.validate_on_submit():
# 判断用户是否存在且用户密码是否正确
user = User.query.filter_by(email=form.email.data).first()
if user is None:
logger.info("post_login 没有这个用户,需要跳转注册页面")
flash("没有该用户,请先注册用户", category='error')
return redirect(url_for('register'))
if user.verity_password(form.password.data):
login_user(user) # 登录操作
# 利用login_user函数将登录信息存入session中
logger.info(f"post_login {user.username} 通过pc login登录成功")
flash('用户%s登录成功' % user.username, category='success')
if rurl is not None and rurl != '': # redirectUrl 参数存在切不为空
# --start-- 生成response
r = make_response('', 302)
r.headers["Location"] = rurl
# --end-- 生成response
return r
else:
return "欢迎来到主页"
else:
logger.info(f"post_login {user.username} 通过pc login登录失败")
flash('用户%s登录失败' % form.email.data, category='error') # 这里用form.email.data是因为登录信息不正确,所以需要从data中获取
# @login_required 会自动添加一个next parma,如果有这个param 则给redirectUrl加上一个next参数,然后把next挂在上面
if rurl:
return redirect(url_for('login', next=rurl))
else:
return redirect(url_for('login'))
# 验证钉钉免登陆
# 具体流程:
# 1 根据来访请求,判断来源UA是否为钉钉客户端
# 2 如果是,交由dinglogin 函数处理,
#
# 构建跳转链接 形如
# https://oapi.dingtalk.com/connect/oauth2/sns_authorize?
# appid=APPID&response_type=code&scope=snsapi_auth&state=STATE&redirect_uri=REDIRECT_URI
# 请求钉钉 其中rediect_uri 为 Ding_Login_Server
#
# 3 钉钉携带追加临时授权码code及state两个参数 请求redirct_uri 即/dingauth 。 这里需要考虑访问权限相关问题。负载均衡上需要加一定的白名单。
# 4 根据 授权码code 生成 个人免登场景签名
# 5 根据 个人免登场景签名 获取用户信息
# 6 根据unionid获取userid。
# 7 根据userid获取用户详情。
# 验证钉钉免登陆
def dinglogin(rurl):
appId = AppKey
if rurl:
rediect_url_raw = f'{Ding_Login_Server}/dingauth?redirectUrl=' + rurl
else:
rediect_url_raw = f'{Ding_Login_Server}/dingauth'
rediect_uri = quote(rediect_url_raw)
dingjump_uri = 'https://oapi.dingtalk.com/connect/oauth2/sns_authorize?' \
'appid={}&response_type=code&scope=snsapi_auth&redirect_uri={}' \
.format(appId, rediect_uri)
logger.info('检测到钉钉客户端UA dinglogin 跳转 url: {}'.format(dingjump_uri))
return redirect(dingjump_uri)
# 用于接收处理钉钉免登陆回调
@app.route('/dingauth', methods=["GET", "POST"])
def dingauth():
FULL_URL = unquote(request.url)
try:
REDIRECT_URL = FULL_URL.split("redirectUrl=")[1].strip("&") # 截取字符串 获取redirectUrl 清除结尾的&
except:
REDIRECT_URL = None
logger.info('钉钉回调重定向完整url:{}'.format(FULL_URL))
logger.info('用户原始跳转REDIRECT_URL:{}'.format(REDIRECT_URL))
appId = AppKey.encode('utf-8')
appSecret = AppSecret.encode('utf-8')
if request.method == 'GET':
code = request.args.get('code', None)
if not code:
return make_response(jsonify({"success": False, "msg": "缺少合适的参数"}))
# start 通过code 获取用户信息
# 计算签名
timestamp = str(int(time.time() * 1000)).encode('utf-8')
signature = quote(base64.b64encode(hmac.new(appSecret, timestamp, digestmod=sha256).digest()))
logger.info("生成钉钉验证签名:".format(signature))
# 构建数据
data = {'tmp_auth_code': code} # post 数据
json_data = json.dumps(data) # 字典 to json str
# 获取用户unionid信息
unionid_url = 'https://oapi.dingtalk.com/sns/getuserinfo_bycode?' \
+ 'signature=' + signature \
+ '×tamp=' + timestamp.decode('utf-8') \
+ '&accessKey=' + appId.decode('utf-8')
r_unionid = requests.post(unionid_url, data=json_data).json()
logger.info("向钉钉请求用户信息,"+str(r_unionid))
logger.info("通过sns想钉钉请求用户信息返回结果:" + str(r_unionid))
if r_unionid['errcode'] != 0:
logger.error("钉钉数据异常,"+str(unionid_url))
return make_response(jsonify({"success": False, "msg": "钉钉信息获取失败"}))
# 获取钉钉的 unionid
# return r_unionid['user_info']
unionid = r_unionid['user_info']['unionid']
# 通过unionid 获取钉钉id
dingtoken = get_token()
if dingtoken["success"]:
data = {"unionid": unionid}
params = {"access_token": dingtoken['msg']}
r_dingdingID = requests.post('https://oapi.dingtalk.com/topapi/user/getbyunionid',
params=params, data=data).json()
else:
logger.error("获取钉钉token失败")
return make_response(jsonify({"success": False, "msg": "钉钉token失败"}))
if r_dingdingID['errcode'] == 0:
logger.info("获取DDID:" + str(r_dingdingID))
dingding_id = r_dingdingID['result']['userid']
else:
logger.error("获取钉钉id失败")
return make_response(jsonify({"success": False, "msg": "钉钉id失败"}))
# 通过钉钉ID获取详细信息
dinguserinfo = DingUser.get_userinfo(dingtoken['msg'], dingding_id)
logger.info("钉钉请求用户信息:" + str(dinguserinfo))
# end 通过code 获取钉钉信息
if dinguserinfo['success']:
username = dinguserinfo['msg']['name']
# 判断username信息和数据库人员信息进行核对,核对成功才能进行登录操作
user = User.query.filter_by(username=username).first()
if user:
logger.info('获取用户名:{} ding_login继续登录'.format(username))
login_user(user) # 登录操作
# 利用login_user函数将登录信息存入session中
logger.info(f"{user.username} 通过ding_login登录成功")
flash('用户%s登录成功' % user.username, category='success')
if REDIRECT_URL is not None and REDIRECT_URL != '': # redirectUrl 参数存在切不为空
# --start-- 生成response
r = make_response('', 302)
r.headers["Location"] = REDIRECT_URL
# --end-- 生成response
return r
else:
return "欢迎来到主页"
else:
logger.error('{} 钉钉 和本地人员数据无法适配'.format(username))
return make_response(jsonify({"success": False, "msg": '{} 钉钉 和本地人员数据无法适配'.format(username)}))
else:
logger.error('钉钉用户信息获取失败')
return make_response(jsonify({"success": False, "msg": "钉钉用户信息获取失败"}))
@app.route('/login', methods=["GET", "POST"], strict_slashes=False)
def login():
FULL_URL = unquote(request.url)
form = LoginForm()
if request.method == "POST":
return post_login(form)
elif request.method == "GET":
return get_login(FULL_URL, form)
else:
logger.error("不支持的请求方式")
return make_response(jsonify({"success": False, "msg": "不支持的请求方式"}))
@app.route('/register', methods=['GET', 'POST'], strict_slashes=False) # strict_slashes的意思是是否要求百分百符合前面的路由规则
def register():
"""
register:
GET:获取注册页面
POST:获取注册页面提交的数据信息
1)判断是否为POST方法提交数据,并且数据是否通过表单验证
2)如果通过验证,将表单提交的数据存储到数据库中;注册成功,跳转到登陆页面
获取表单提交的数据有两种方式:
i、form.data 是一个字典,通过key值获取
ii、form.email.data form.username.data
"""
form = RegistrationForm()
if form.validate_on_submit():
user = User()
user.username = form.username.data
user.password = form.password.data
user.email = form.email.data
try:
local_object = db.session.merge(user)
db.session.add(local_object)
db.session.commit()
except Exception as e:
db.session.rollback()
click.echo('register user failed: {reason}'.format(reason=str(e)))
flash(u'用户%s注册成功' % user.username, category='success')
return redirect(url_for('login'))
return render_template('register.html', form=form)
@app.route('/logout')
@login_required
# 此装饰器是用来判断用户是否已经登录
# 当一个函数被多个装饰器装饰时,执行的顺序时从上往下执行,被装饰的顺序是从下往上装饰
# 所以这里是先进入logout路由,然后判断是否已经登录,如果已经登录则执行注销
def logout():
logout_user()
return redirect(url_for('login'))
# login_required如果不在登录状态的用户访问这个路由,Flask-Login会拦截请求,把请求发往登录页面(跳转login由login_manager.login_view控制)
@app.route('/wf/demo1/1')
@login_required
def opt_wf():
return {"success": True, "msg": f"当前用户:{current_user.username} 正在操作工单任务"}
if __name__ == "__main__":
app.run(host='0.0.0.0', port=5000, debug=True)
db.py
from flask_sqlalchemy import SQLAlchemy
db = SQLAlchemy()
models.py
import datetime
from werkzeug.security import generate_password_hash, check_password_hash
from db import db
from flask_login import UserMixin
class User(UserMixin, db.Model):
"""
用户信息
"""
__tablename__ = 'users' # 自定义数据表的表名
id = db.Column(db.Integer, primary_key=True, autoincrement=True)
username = db.Column(db.String(100), unique=True, nullable=False)
password_hash = db.Column(db.String(200), nullable=True)
email = db.Column(db.String(50))
create_date = db.Column(db.DATETIME(), default=datetime.datetime.now())
# ........
# 密码不可读
@property
def password(self):
raise AttributeError('password is not a readable attribute')
@password.setter
def password(self, password):
# generate_password_hash(password,method=pbkd2:sha1,salt_length=8):密码加密的散列值,为密码进行哈希加密
self.password_hash = generate_password_hash(password)
def verity_password(self, password):
# check_password_hash(hash,password):密码散列值和用户输入的密码是否一致
return check_password_hash(self.password_hash, password)
def __repr__(self):
return '<User % r>' % self.username
form.py
from flask_wtf import FlaskForm
from wtforms import StringField, PasswordField, SubmitField, ValidationError
from wtforms.validators import DataRequired, Length, Email, Regexp, EqualTo
from models import User
class RegistrationForm(FlaskForm): # 还有一种写法class RegistrationForm (Form):
email = StringField('电子邮箱', validators=[ # email=StringField(
DataRequired(), Length(1, 64), Email()], # label='电子邮箱'
# 给前端的标签添加下面的属性 # validators=[
render_kw={ # validators.data_required(message=u"邮箱不能为空"),
'class': 'layui-input', # validators.length(min=1,max=64),
'placeholder': '电子邮箱' # validators.email],
})
username = StringField('用户名', validators=[
DataRequired(), Length(1, 64), Regexp('^\w*$', message='用户名只能由字母数字或者下划线组成')],
# 给前端的标签添加下面的属性
render_kw={
'class': 'layui-input',
'placeholder': '用户名'
})
password = PasswordField('密码', validators=[
DataRequired(), EqualTo('repassword', message='密码不一致')],
# 给前端的标签添加下面的属性
render_kw={
'class': 'layui-input',
'placeholder': '密码'
})
repassword = PasswordField('确认密码', validators=[
DataRequired()],
# 给前端的标签添加下面的属性
render_kw={
'class': 'layui-input',
'placeholder': '确认密码'
})
submit = SubmitField('注册')
# 两个自动验证的函数,已validate_开头且跟着字段名的方法,这个方法和常规的验证函数一起调用
def validate_email(self, field):
# field是email表单对象,filed.data是email表单里提交的数据信息
if User.query.filter_by(email=field.data).first():
raise ValidationError("邮箱地址%s已经注册" % (field.data))
def validate_username(self, field):
if User.query.filter_by(username=field.data).first():
raise ValueError('用户名%s已经注册' % (field.data))
class LoginForm(FlaskForm):
"""用户登录表单"""
email = StringField('电子邮箱', validators=[
DataRequired(), Length(1, 64), Email()],
# 给前端得标签添加下面的属性
render_kw={
'class': 'layui-input',
'placeholder': '电子邮箱'
})
password = PasswordField('密码', validators=[
DataRequired()],
# 给前端得标签添加下面得属性信息
render_kw={
'class': 'layui-input',
'placeholder': '密码'
})
submit = SubmitField('登录')
actlog.py
import os
import datetime
import sys
basedir = os.path.abspath(os.path.dirname(__file__)) # 获取当前项目的绝对路径
class HaLog:
def __init__(self, file_name, level='info', log_path=os.path.join(basedir, "log")):
self.log_file = os.path.join(log_path, file_name)
self.log_path = log_path
level_dic = ["info", "warning", "error", "cri"]
if level in level_dic:
self.level = level
else:
self.level = "info"
def info(self, msg):
level = 'info'
try:
raise Exception
except:
f = sys.exc_info()[2].tb_frame.f_back
filename = f.f_code.co_filename
fun_co_name = f.f_code.co_name
f_lineno = f.f_lineno
try:
os.mkdir(self.log_path)
except:
pass
now = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')[:-3]
data = "{} - {} {} {} {} : {} \n".format(now, level,filename, fun_co_name, f_lineno, msg)
with open(self.log_file, mode='a', encoding='utf-8') as f:
f.write(data)
def warning(self, msg):
level = 'warning'
try:
raise Exception
except:
f = sys.exc_info()[2].tb_frame.f_back
filename = f.f_code.co_filename
fun_co_name = f.f_code.co_name
f_lineno = f.f_lineno
try:
os.mkdir(self.log_path)
except:
pass
now = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')[:-3]
data = "{} - {} {} {} {} : {} \n".format(now, level, filename, fun_co_name, f_lineno, msg)
with open(self.log_file, mode='a', encoding='utf-8') as f:
f.write(data)
def error(self, msg):
level = 'error'
try:
raise Exception
except:
f = sys.exc_info()[2].tb_frame.f_back
filename = f.f_code.co_filename
fun_co_name = f.f_code.co_name
f_lineno = f.f_lineno
try:
os.mkdir(self.log_path)
except:
pass
now = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')[:-3]
data = "{} - {} {} {} {} : {} \n".format(now, level, filename, fun_co_name, f_lineno, msg)
with open(self.log_file, mode='a', encoding='utf-8') as f:
f.write(data)
def debug(self, msg):
level = 'debug'
try:
raise Exception
except:
f = sys.exc_info()[2].tb_frame.f_back
filename = f.f_code.co_filename
fun_co_name = f.f_code.co_name
f_lineno = f.f_lineno
try:
os.mkdir(self.log_path)
except:
pass
now = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')[:-3]
data = "{} - {} {} {} {} : {} \n".format(now, level, filename, fun_co_name, f_lineno, msg)
with open(self.log_file, mode='a', encoding='utf-8') as f:
f.write(data)
def write(self, msg, level=None):
try:
raise Exception
except:
f = sys.exc_info()[2].tb_frame.f_back
filename = f.f_code.co_filename
fun_co_name = f.f_code.co_name
f_lineno = f.f_lineno
if level:
level = level
else:
level = self.level
try:
os.mkdir(self.log_path)
except:
pass
now = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')[:-3]
data = "{} - {} {} {} {} : {} \n".format(now, level,filename, fun_co_name, f_lineno, msg)
with open(self.log_file, mode='a', encoding='utf-8') as f:
f.write(data)
manager.py
from flask_migrate import Migrate
from app import app
from db import db
# 迁移命令管理与app,建立关系
# sqlite使用此参数render_as_batch=True使用batch操作替换普通操作,因为普通操作不支持表名,列名的改变!
migrate = Migrate(app, db, render_as_batch=True)
DDinfo.py
# 钉钉小程序基础信息
AgentId = 'xxxxxxxxx'
AppKey = 'xxxxxxxxx'
AppSecret = 'xxxxxx'
agentid = AgentId
appkey = AppKey
appsecret = AppSecret
api_prefix = "https://oapi.dingtalk.com"
DDtoken.py
import requests
from Ding_Dev_Tools.DDinfo import *
def get_token():
"""
获取token
:return:
"""
api = "/gettoken"
try:
params = {"appkey": appkey,
"appsecret": appsecret
}
r = requests.get(api_prefix + api, params=params)
result = r.json()
if result["errcode"] == 0 and result["errmsg"] == "ok":
token = result["access_token"]
return {'success': True, 'msg': token}
else:
return {'success': False, 'msg': "Requests Fail"}
except Exception as e:
print(e)
return {'success': False, 'msg': e}
DDuser.py
from Ding_Dev_Tools.DDinfo import *
import requests
from actlog import HaLog
logger = HaLog("running.log")
class DingUser:
@staticmethod
def get_userinfo(dingtoken, userid):
api = '/user/get'
try:
params = {"access_token": dingtoken, "userid": userid}
r = requests.get(api_prefix + api, params=params)
result = r.json()
print(result)
logger.info(str(result))
if result["errcode"] == 0 and result["errmsg"] == "ok":
name = result["name"]
position = result['position']
department = result['department']
userid = result['userid']
# jobnumber = result["jobnumber"]
# hiredDate = result['hiredDate']
return {'success': True, 'msg': {'name': name,
'position': position,
'userid': userid,
'department': department,
}
}
else:
return {'success': False, 'msg': str(result)}
except Exception as e:
print(e)
return {'success': False, 'msg': e}
flash.html
<html>
<head>
<link href="{{ url_for('static',filename='css/bootstrap.min.css') }}" ,rel="stylesheet">
<link rel="stylesheet" href="{{ url_for('static',filename='css/font.css') }}">
<link rel="stylesheet" href="{{ url_for('static',filename='css/xadmin.css') }}">
<link rel="stylesheet" href="{{ url_for('static',filename='css/login.css') }}">
<script type="text/javascript" src="{{ url_for('static',filename="js/jquery.min1.js") }}"></script>
<script src="{{ url_for('static',filename='js/jquery.min2.js') }}"></script>
<script src="{{ url_for('static',filename='js/bootstrap.min1.js') }}"></script>
<script src="{{ url_for('static',filename='lib/layui/layui.js') }}"></script>
<script src="{{ url_for('static',filename='js/html5.min.js') }}"></script>
<script src="{{ url_for('static',filename='js/respind.min.js') }}"></script>
</head>
<body>
{% for message in get_flashed_messages(category_filter=['success']) %}
<div class="alert alert-success alert-dismissible" role="alert">
<button type="button" class="close" data-dismiss="alert" aria-label="Close">
<span aria-hidden="true">×</span>
</button>
<strong>Success! </strong> {{ message }}
</div>
{% endfor %}
{% for message in get_flashed_messages(category_filter=['error']) %}
<div class="alert alert-danger alert-dismissible" role="alert">
<button type="button" class="close" data-dismiss="alert" aria-label="Close">
<span aria-hidden="true">×</span>
</button>
<strong>Warning! </strong> {{ message }}
</div>
{% endfor %}
</body>
</html>
login.html
<html>
<head>
<link href="{{ url_for('static',filename='css/bootstrap.min.css') }}" ,rel="stylesheet">
<link rel="stylesheet" href="{{ url_for('static',filename='css/font.css') }}">
<link rel="stylesheet" href="{{ url_for('static',filename='css/xadmin.css') }}">
<link rel="stylesheet" href="{{ url_for('static',filename='css/login.css') }}">
<script type="text/javascript" src="{{ url_for('static',filename="js/jquery.min1.js") }}"></script>
<script src="{{ url_for('static',filename='js/jquery.min2.js') }}"></script>
<script src="{{ url_for('static',filename='js/bootstrap.min1.js') }}"></script>
<script src="{{ url_for('static',filename='lib/layui/layui.js') }}"></script>
<script src="{{ url_for('static',filename='js/html5.min.js') }}"></script>
<script src="{{ url_for('static',filename='js/respind.min.js') }}"></script>
</head>
<body>
<div class="login layui-anim layui-anim-up">
<div class="title">任务清单系统登录</div>
<div id="darkbannerwrap"></div>
{% include 'flash.html' %}
<form method="post" class="layui-form" action="{{ url_for('login', redirectUrl=redirectUrl) }}">
{{ form.hidden_tag() }}
{{ form.email() }}
{# <input name="username" placeholder="用户名" type="text" lay- verify="required" class="layui-input">#}
{# <p class="error">用户登录失败</p>#}
<p class="error">{{ form.email.errors[0] }}</p>
<hr class="hr15">
{{ form.password() }}
{# <input name="password" lay-verify="required" placeholder="密码" type="password" class="layui-input">#}
{# <p class="error"> 密码失败</p>#}
<p class="error"> {{ form.password.errors[0] }}</p>
<hr class="hr15">
{{ form.submit() }}
{# <input value="登录" style="width:100%;" type="submit">#}
<hr class="hr20">
</form>
</div>
</body>
</html>
register.html
<html>
<head>
<link href="{{ url_for('static',filename='css/bootstrap.min.css') }}" ,rel="stylesheet">
<link rel="stylesheet" href="{{ url_for('static',filename='css/font.css') }}">
<link rel="stylesheet" href="{{ url_for('static',filename='css/xadmin.css') }}">
<link rel="stylesheet" href="{{ url_for('static',filename='css/login.css') }}">
<script type="text/javascript" src="{{ url_for('static',filename="js/jquery.min1.js") }}"></script>
<script src="{{ url_for('static',filename='js/jquery.min2.js') }}"></script>
<script src="{{ url_for('static',filename='js/bootstrap.min1.js') }}"></script>
<script src="{{ url_for('static',filename='lib/layui/layui.js') }}"></script>
<script src="{{ url_for('static',filename='js/html5.min.js') }}"></script>
<script src="{{ url_for('static',filename='js/respind.min.js') }}"></script>
</head>
<body>
<div class="login layui-anim layui-anim-up">
<div class="title">任务清单系统注册</div>
<div id="darkbannerwrap"></div>
{% include 'flash.html' %}
<form method="post" class="layui-form" action="{{ url_for('register') }}">
{{ form.hidden_tag() }}
{{ form.email() }}
<p class="error">{{ form.email.errors[0] }}</p>
<hr class="hr15">
{{ form.username() }}
<p class="error">{{ form.username.errors[0] }}</p>
<hr class="hr15">
{{ form.password() }}
<p class="error"> {{ form.password.errors[0] }}</p>
<hr class="hr15">
{{ form.repassword() }}
<p class="error"> {{ form.repassword.errors[0] }}</p>
<hr class="hr15">
{{ form.submit() }}
<hr class="hr20">
</form>
</div>
</body>
</html>
requirements.txt
alembic==1.7.7
certifi==2023.7.22
charset-normalizer==2.0.12
click==8.0.4
colorama==0.4.5
dataclasses==0.8
dnspython==2.2.1
email-validator==1.3.1
Flask==2.0.3
Flask-Login==0.5.0
Flask-Migrate==3.1.0
Flask-SQLAlchemy==2.5.1
Flask-WTF==1.0.1
greenlet==2.0.2
idna==3.4
importlib-metadata==4.8.3
importlib-resources==5.4.0
itsdangerous==2.0.1
Jinja2==3.0.3
Mako==1.1.6
MarkupSafe==2.0.1
requests==2.27.1
SQLAlchemy==1.4.49
typing_extensions==4.1.1
urllib3==1.26.18
Werkzeug==2.0.3
WTForms==3.0.0
zipp==3.6.0
将代码部署到公网服务器上
这个我们就简单部署下
yum install -y python3,nginx
yum install -y unzip
pip3 install virtualenv -i https://pypi.tuna.tsinghua.edu.cn/simple
unzip 钉钉免登陆demo.zip
cd 钉钉免登陆demo/
virtualenv venv
source venv/bin/activate
pip3 install -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple
export FLASK_APP="manager"
flask db init
flask db migrate
flask db upgrade
nohup python app.py &
nginx配置
server {
listen 80;
server_name 自己的服务器的公网ip;
#charset koi8-r;
#access_log logs/host.access.log main;
#access_log "pipe:rollback logs/host.access_log interval=1d baknum=7 maxsize=2G" main;
location / {
add_header Access-Control-Allow-Origin *;
add_header Access-Control-Allow-Methods 'GET, POST, OPTIONS';
add_header Access-Control-Allow-Headers 'DNT,X-Mx-ReqToken,Keep-Alive,User-Agent,X-Requested-With,If-Modified-Since,Cache-Control,Content-Type,Authorization';
proxy_pass http://127.0.0.1:5000/;
}
}
测试效果
先注册用户
测试pc端访问效果
测试url http://xxxx.xxx.xxx.xx/wf/demo1/1
测试收集钉钉登陆