基于flask、flask-apscheduler定时框架建立的定时web定时运行py脚本实例

需求

• 需要定时运行相应py脚本文件，及web上展示详细情况
• 及时返回报错信息，及输出日志记录
• 修改运行中的定时任务信息（名称、时间、及定时时间）
• 动态导入脚本文件
• 定时任务的开始、暂停、结束控制

思路

使用到的p库：flask、flask-scheduler、sqlite3、sqlalchemy、waitress
系统：Windows
py版本：python 3.7
由于运行在Windows上，所以使用了waitress 作为flask 应用的容器实现多线程、多进程。（由于gevent 是单线程的一开始尝试使用，发现怎么都无法进行定时任务。）
前端使用到ajax、boostrap、jquery…
下面介绍的button 都是基本使用ajax去相应事件。

开始动手了

首先使用（HTML、JAVASCRIPT、CSS）先建立好自己的登录模块，登录窗口

建立好相应的登录界面，做好权限控制以及登录检验。
使用flask 检验登录以及sqlalchemy的ORM控制，

有模板的可以套一套。
配置文件

# config.py
app = Flask(__name__, template_folder=r'.\moban3365\moban3365', static_folder=r'.\moban3365\moban3365\assets')
# scheduler = BackgroundScheduler(app)

class  Config(object):
    # JOBS = [
    #     {
    #         'id': 'job1',
    #         'func': 'workdaysub:job1',
    #         'args': (),
    #         'trigger': 'interval',
    #         'seconds': 10
    #     }
    # ]
    SCHEDULER_JOBSTORES = {
        'default': SQLAlchemyJobStore(url='sqlite:///.\yoursqlie.db')
    }
    SCHEDULER_EXECUTORS = {
        'default': {'type': 'threadpool', 'max_workers': 20}
    }
    SCHEDULER_JOB_DEFAULTS = {
        'coalesce': False,
        'max_instances': 3
    }
    # SCHEDULER_API_ENABLED = True
    SCHEDULER_API_ENABLED = False
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = True
# app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///.\workday.db'
app.config['SQLALCHEMY_DATABASE_URI'] = 'mssql+pymssql://yourservice/PYR?charset=utf8'
# app.config['SQLALCHEMY_DATABASE_URI'] = 'mssql+pyodbc://yourservice/PYR?driver=SQL+Server?charset=unicode'

app.config['SECRET_KEY'] = 'alkjfdlkajfdlkjaslkdfjlkj'

app.config['DEBUG'] = True
# app.config['SCHEDULEER_API_ENABLED'] = True
app.permanent_session_lifetime = timedelta(minutes=60)  # 设置session到期时间
# app.config["SQLALCHEMY_NATIVE_UNICODE"] = True
app.config["THREADED"] = True
# app.config["SCHEDULER_JOBSTORES"]  = {'default':SQLAlchemyJobStore(url='sqlite:///.\yoursqlite.db')}
# 设置时区，时区不一致会导致定时任务的时间错误
# SCHEDULER_TIMEZONE = 'Asia/Shanghai'
# app.config["SCHEDULER_EXECUTORS"] = {'default': {'type': 'threadpool', 'max_workers': 10}}
# app.config.from_object(config),use_native_unicode="utf8"

db = SQLAlchemy(app)
db.init_app(app)
scheduler = APScheduler()
app.config.from_object(Config())
scheduler.init_app(app)

# @scheduler.task('interval', id='write_file', seconds=10)

# scheduler.add_job(id="write_xlsx" ,func=job1, trigger='interval', seconds=10)

scheduler.start()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55

使用BluePrint进行视图管理

from app.config import app
from app.Model import db, Script_Status
from User.view import ur
from Contract.view import ct
from Report.veiw import rt
from Scheduler_Job.view import scheduler_job


app.register_blueprint(ct)
app.register_blueprint(ur)
app.register_blueprint(rt)
app.register_blueprint(scheduler_job)
1
2
3
4
5
6
7
8
9
10
11
12

建立shceduler view视图
填充前端代码

设置了刷新功能、添加执行计划功能、脚本刷新功能

功能：

刷新功能：

刷新查询还在计划中的scheduler
刷先显示如下，添加了暂停以及删除功能，还有暂停后重新启动的功能

添加执行计划功能：

可以根据选用的脚本名称，命名脚本以及执行周期选择
执行周期有（每天、每周、一次性任务）三种，（暂时）
以及选择开始执行时间到结束的时间段

当时选择到Other_Script的时候，可以选择脚本路径，和模块名字。（感觉这里应该使用上传脚本文件的方式实现。）但没有关系，只是在自己及公共盘上运行。
模块名称是路径脚本中的方法名字

上面的是基于线程的定时计划。

还有多进程的定时计划

脚本刷新

主要是用于重新导入py脚本文件。方便脚本选择的维护。

錯誤日志生成

目前針對id（時間戳）對日志進行日志輸出


針對輸出正常log的後續針對出現是否有錯誤log進行特定標識。

後續定時任務狀態的統計，因爲需要將定時任務狀態記錄以及log文件；
文件log的形式

後續

需要對每一個定時任務進行修改
周期：1.每天 2.每周 3.一次性任務

修改在定时job 还有触发器上。

权限验证（2021-08-27）

某些访问页面以及按钮方法需要身份验证才能使用，所以需要设置相应的权限群组。
由于使用到ajax使得返回的数据形式为json格式

对应的方式需要不同的验证返回结果

#return jsonfy
def permission_required(permission):

    def decorator(f):
        @wraps(f)
        def decorator_function(*args, **kwargs):
            user = User.query.filter_by(username=session.get("uname")).first()
            # print("test permission")
            if not user.can(permission):
                print("test permission")
                flash('您当前的用户组权限不够', 'warning')
                return redirect('/login')
            return f(*args, **kwargs)
        return decorator_function
    return decorator
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

返回 路径地址或网页template

#return redirect
def permission_required_json(permission):

    def decorator(f):
        @wraps(f)
        def decorator_function(*args, **kwargs):
            user = User.query.filter_by(username=session.get("uname")).first()
            # print("test permission")
            if not user.can(permission):
                # print("test permission")
                flash('您当前的用户组权限不够', 'warning')
                return jsonify({"code": "502", "Msg": "您当前的用户组权限不够!!"})
            return f(*args, **kwargs)
        return decorator_function
    return decorator
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

而上面使用了flash的方法，需要同样的在html上增加上

{% for msg in  get_flashed_messages(category_filter=["warning"]) %}
                    toast_fun('权限警告','{{ msg }}');
                        {% endfor %}
1
2
3

通过装饰器的方法进行权限验证

# @permission_required('监察者')
@permission_required('用户')
def scheduler_page():
	.......
1
2
3
4