flask-app,定时执行、实时监控脚本

# Flask-WTF表单定义模块 ## 📋 模块概述 `forms` 模块包含了应用中所有的Flask-WTF表单定义，负责处理用户输入验证、表单渲染和数据处理。 ## 📁 目录结构 ``` forms/ ├── __init__.py # 模块初始化文件 ├── admin.py # 管理员相关表单 ├── auth.py # 用户认证相关表单 ├── groups.py # 用户组相关表单 ├── scripts.py # 脚本管理相关表单 ├── tasks.py # 任务管理相关表单 └── __pycache__/ # Python缓存目录 ``` ## 📄 文件详细说明 ### `__init__.py` **功能**：模块初始化文件 - 定义模块的公共接口 - 导入常用的表单类 - 设置模块级别的配置 ### `admin.py` **功能**：管理员功能相关表单 **主要表单类**： - **`CreateUserForm`**：创建用户表单 - 用户名、密码、确认密码 - 用户类型选择 - 权限设置（执行、下载、预览、可视化） - 用户组分配 - **`EditUserForm`**：编辑用户表单 - 用户基本信息修改 - 权限状态调整 - 账户状态管理 - **`CreateGroupForm`**：创建用户组表单 - 组名、描述 - 权限设置 - 成员管理 **表单特性**： ```python class CreateUserForm(FlaskForm): username = StringField('用户名', validators=[DataRequired(), Length(min=3, max=64)]) password = PasswordField('密码', validators=[DataRequired(), Length(min=6)]) password2 = PasswordField('确认密码', validators=[DataRequired(), EqualTo('password')]) user_type = SelectField('用户类型', choices=[ ('admin', '管理员'), ('normal', '普通用户'), ('preview_only', '仅预览用户') ]) has_execute_permission = BooleanField('执行权限', default=True) has_download_permission = BooleanField('下载权限', default=False) has_preview_permission = BooleanField('预览权限', default=True) has_visualization_permission = BooleanField('可视化权限', default=False) ``` ### `auth.py` **功能**：用户认证相关表单 **主要表单类**： - **`LoginForm`**：用户登录表单 - 用户名输入 - 密码输入 - 记住我选项 - **`RegistrationForm`**：用户注册表单 - 用户名、密码、确认密码 - 用户名唯一性验证 - 密码强度验证 - **`ChangePasswordForm`**：修改密码表单 - 当前密码验证 - 新密码输入 - 新密码确认 **验证特性**： ```python class ChangePasswordForm(FlaskForm): current_password = PasswordField('当前密码', validators=[DataRequired()]) new_password = PasswordField('新密码', validators=[DataRequired(), Length(min=6)]) new_password2 = PasswordField('确认新密码', validators=[ DataRequired(), EqualTo('new_password', message='两次输入的密码不一致') ]) def __init__(self, user, *args, **kwargs): super(ChangePasswordForm, self).__init__(*args, **kwargs) self.user = user def validate_current_password(self, current_password): if not self.user.check_password(current_password.data): raise ValidationError('当前密码错误') ``` ### `groups.py` **功能**：用户组管理相关表单 **主要表单类**： - **`CreateGroupForm`**：创建用户组表单 - 组名、描述 - 权限设置 - 初始成员 - **`EditGroupForm`**：编辑用户组表单 - 组信息修改 - 成员管理 - 权限调整 - **`JoinGroupForm`**：加入用户组表单 - 组选择 - 申请理由 **表单结构**： ```python class CreateGroupForm(FlaskForm): name = StringField('组名', validators=[DataRequired(), Length(min=2, max=64)]) description = TextAreaField('描述', validators=[Length(max=200)]) is_public = BooleanField('公开组', default=True) can_preview_shared = BooleanField('可预览共享内容', default=True) submit = SubmitField('创建用户组') def validate_name(self, name): group = UserGroup.query.filter_by(name=name.data).first() if group is not None: raise ValidationError('该组名已被使用') ``` ### `scripts.py` **功能**：脚本管理相关表单 **主要表单类**： - **`UploadScriptForm`**：脚本上传表单 - 脚本文件上传 - 脚本名称、描述 - 分类标签 - **`EditScriptForm`**：编辑脚本表单 - 脚本信息修改 - 参数配置 - 权限设置 - **`ScriptExecutionForm`**：脚本执行表单（动态生成） - 根据脚本参数动态生成字段 - 文件上传字段 - 参数验证 **动态表单生成**： ```python def create_execution_form(script): """根据脚本参数动态创建执行表单""" class DynamicExecutionForm(FlaskForm): pass # 解析脚本参数 params = parse_script_parameters(script.content) for param in params: if param['type'] == 'file': field = FileField(param['name'], validators=[DataRequired()]) elif param['type'] == 'int': field = IntegerField(param['name'], default=param.get('default')) elif param['type'] == 'str': field = StringField(param['name'], default=param.get('default')) elif param['type'] == 'bool': field = BooleanField(param['name'], default=param.get('default', False)) setattr(DynamicExecutionForm, param['name'], field) setattr(DynamicExecutionForm, 'submit', SubmitField('执行脚本')) return DynamicExecutionForm ``` ### `tasks.py` **功能**：任务管理相关表单 **主要表单类**： - **`CreateTaskForm`**：创建定时任务表单 - 任务名称、描述 - 脚本选择 - 调度配置（cron表达式） - 参数设置 - **`EditTaskForm`**：编辑定时任务表单 - 任务信息修改 - 调度时间调整 - 启用/禁用状态 - **`CreateFormForm`**：创建表单表单 - 表单名称、描述 - 脚本关联 - 参数配置 **调度配置**： ```python class CreateTaskForm(FlaskForm): name = StringField('任务名称', validators=[DataRequired(), Length(min=2, max=100)]) description = TextAreaField('任务描述', validators=[Length(max=500)]) script_id = SelectField('选择脚本', coerce=int, validators=[DataRequired()]) # 调度配置 schedule_type = SelectField('调度类型', choices=[ ('cron', 'Cron表达式'), ('interval', '间隔执行'), ('date', '指定时间') ]) cron_expression = StringField('Cron表达式', validators=[Optional()]) interval_seconds = IntegerField('间隔秒数', validators=[Optional()]) run_date = DateTimeField('执行时间', validators=[Optional()]) is_active = BooleanField('启用任务', default=True) submit = SubmitField('创建任务') ``` ## 🔧 表单验证 ### 内置验证器 **常用验证器**： - **`DataRequired()`**：必填字段验证 - **`Length(min, max)`**：长度验证 - **`Email()`**：邮箱格式验证 - **`EqualTo(fieldname)`**：字段相等验证 - **`NumberRange(min, max)`**：数值范围验证 - **`Optional()`**：可选字段验证 ### 自定义验证器 **验证方法**： ```python class CustomForm(FlaskForm): username = StringField('用户名', validators=[DataRequired()]) def validate_username(self, username): """自定义用户名验证""" if User.query.filter_by(username=username.data).first(): raise ValidationError('用户名已存在') if not username.data.isalnum(): raise ValidationError('用户名只能包含字母和数字') ``` **跨字段验证**： ```python def validate(self): """表单级别验证""" if not super().validate(): return False if self.start_date.data > self.end_date.data: self.end_date.errors.append('结束时间不能早于开始时间') retu