bwadmin/pear-admin-flask
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

完善开发文档与更新函数注释

Browse Source
This commit is contained in:
wojiaoyishang
2025-01-27 11:16:26 +08:00
parent de4f2c6d52
commit 0db36a6218
17 changed files with 932 additions and 254 deletions
Download Patch File Download Diff File Expand all files Collapse all files
applications/common/admin_log.py
+23 -32
View File
@@ -5,16 +5,15 @@ from applications.extensions import db
from applications.models import AdminLog
def login_log(request, uid, is_access):
def normal_log(method, url, ip, user_agent, desc, uid, is_access):
info = {
'method': request.method,
'url': request.path,
'ip': request.remote_addr,
'user_agent': str_escape(request.headers.get('User-Agent')),
'desc': str_escape(request.form.get('username')),
'method': method,
'url': url,
'ip': ip,
'user_agent': user_agent,
'desc': desc,
'uid': uid,
'success': int(is_access)
}
log = AdminLog(
url=info.get('url'),
@@ -26,33 +25,25 @@ def login_log(request, uid, is_access):
success=info.get('success')
)
db.session.add(log)
db.session.flush()
db.session.commit()
return log.id
def admin_log(request, is_access):
def login_log(request, uid, is_access):
method = request.method
url = request.path
ip = request.remote_addr
user_agent = str_escape(request.headers.get('User-Agent'))
desc = str_escape(request.form.get('username'))
return normal_log(method, url, ip, user_agent, desc, uid, is_access)
def admin_log(request, is_access, desc=None):
method = request.method
url = request.path
ip = request.remote_addr
user_agent = str_escape(request.headers.get('User-Agent'))
request_data = request.json if request.headers.get('Content-Type') == 'application/json' else request.values
info = {
'method': request.method,
'url': request.path,
'ip': request.remote_addr,
'user_agent': str_escape(request.headers.get('User-Agent')),
'desc': str_escape(str(dict(request_data))),
'uid': current_user.id,
'success': int(is_access)
}
log = AdminLog(
url=info.get('url'),
ip=info.get('ip'),
user_agent=info.get('user_agent'),
desc=info.get('desc'),
uid=info.get('uid'),
method=info.get('method'),
success=info.get('success')
)
db.session.add(log)
db.session.commit()
return log.id
if desc is None:
desc = str_escape(str(dict(request_data)))
return normal_log(method, url, ip, user_agent, desc, current_user.id, is_access)
applications/common/utils/cache.py
-1
View File
@@ -53,4 +53,3 @@ def cache_auto_internal(key, call, expired=5):
cache_set_internal(key, data, expired)
return data
applications/common/utils/captcha.py
+39 -25
View File
@@ -7,20 +7,23 @@ class vieCode:
__width = 120 # 画布宽度
__heigth = 45 # 画布高度
__length = 4 # 验证码长度
__draw = None # 画布
__img = None # 图片资源
__draw = None # 画布对象
__img = None # 图片对象
__code = None # 验证码字符
__str = None # 自定义验证码字符集
__inCurve = True # 是否画干扰线
__inNoise = True # 是否画干扰点
__type = 2 # 验证码类型 1、纯字母 2、数字字母混合
__fontPatn = 'applications/common/utils/fonts/captcha.ttf' # 字体
__inCurve = True # 是否绘制干扰曲线
__inNoise = True # 是否绘制干扰点
__type = 2 # 验证码类型：1-纯字母，2-数字字母混合
__fontPatn = 'applications/common/utils/fonts/captcha.ttf' # 字体路径
def GetCodeImage(self, size=80, length=4):
'''获取验证码图片
@param int size 验证码大小
@param int length 验证码长度
'''
"""
生成验证码图片及其对应的验证码字符。
:param size: 验证码字体大小，默认为 80。
:param length: 验证码字符长度，默认为 4。
:return: 返回验证码图片对象和验证码字符。
"""
# 准备基础数据
self.__length = length
self.__fontSize = size
@@ -37,14 +40,18 @@ class vieCode:
return self.__img, self.__code
def __cerateFilter(self):
'''模糊处理'''
"""
对验证码图片进行模糊处理，增加识别难度。
"""
self.__img = self.__img.filter(ImageFilter.BLUR)
filter = ImageFilter.ModeFilter(8)
self.__img = self.__img.filter(filter)
def __createCode(self):
'''创建验证码字符'''
# 是否自定义字符集合
"""
生成验证码字符。
"""
# 是否使用自定义字符集
if not self.__str:
# 源文本
number = "3456789"
@@ -55,32 +62,37 @@ class vieCode:
else:
self.__str = srcLetter + srcUpper + number
# 构造验证码
# 随机生成验证码字符
self.__code = random.sample(self.__str, self.__length)
def __createImage(self):
'''创建画布'''
"""
创建画布并设置背景颜色。
"""
bgColor = (random.randint(200, 255), random.randint(200, 255), random.randint(200, 255))
self.__img = Image.new('RGB', (self.__width, self.__heigth), bgColor)
self.__draw = ImageDraw.Draw(self.__img)
def __createNoise(self):
'''画干扰点'''
"""
在验证码图片上绘制干扰点。
"""
if not self.__inNoise:
return
font = ImageFont.truetype(self.__fontPatn, int(self.__fontSize / 1.5))
for i in range(5):
# 杂点颜色
# 干扰点颜色
noiseColor = (random.randint(150, 200), random.randint(150, 200), random.randint(150, 200))
putStr = random.sample(self.__str, 2)
for j in range(2):
# 绘杂点
# 绘制干扰点
size = (random.randint(-10, self.__width), random.randint(-10, self.__heigth))
self.__draw.text(size, putStr[j], font=font, fill=noiseColor)
pass
def __createCurve(self):
'''画干扰线'''
"""
在验证码图片上绘制干扰曲线。
"""
if not self.__inCurve:
return
x = y = 0
@@ -93,7 +105,7 @@ class vieCode:
xend = random.randint(self.__width / 2, self.__width * 2)
w = (2 * math.pi) / t
# 画曲线
# 绘制曲线
color = (random.randint(30, 150), random.randint(30, 150), random.randint(30, 150))
for x in range(xend):
if w != 0:
@@ -107,15 +119,17 @@ class vieCode:
i -= i
def __printString(self):
'''打印验证码字符串'''
"""
在画布上打印验证码字符。
"""
font = ImageFont.truetype(self.__fontPatn, self.__fontSize)
x = 0
# 打印字符到画板
# 打印字符到画布
for i in range(self.__length):
# 设置字体随机颜色
color = (random.randint(30, 150), random.randint(30, 150), random.randint(30, 150))
# 计算座标
# 计算坐标
x = random.uniform(self.__fontSize * i * 0.95, self.__fontSize * i * 1.1)
y = self.__fontSize * random.uniform(0.3, 0.5)
# 打印字符
self.__draw.text((x, y), self.__code[i], font=font, fill=color)
self.__draw.text((x, y), self.__code[i], font=font, fill=color)
applications/common/utils/http.py
+22 -5
View File
@@ -2,23 +2,40 @@ from flask import jsonify
def success_api(msg: str = "成功"):
""" 成功响应 默认值“成功” """
"""
返回成功的 API 响应。
:param msg: 成功消息内容，默认为 "成功"。
:return: 返回 JSON 格式的响应，包含 `success` 和 `msg` 字段。
"""
return jsonify(success=True, msg=msg)
def fail_api(msg: str = "失败"):
""" 失败响应 默认值“失败” """
"""
返回失败的 API 响应。
:param msg: 失败消息内容，默认为 "失败"。
:return: 返回 JSON 格式的响应，包含 `success` 和 `msg` 字段。
"""
return jsonify(success=False, msg=msg)
def table_api(msg: str = "", count=0, data=None, limit=10):
""" 动态表格渲染响应 """
"""
返回动态表格渲染所需的 API 响应。
:param msg: 响应消息内容，默认为空字符串。
:param count: 数据总数，默认为 0。
:param data: 表格数据，默认为 None。
:param limit: 每页数据条数，默认为 10。
:return: 返回 JSON 格式的响应，包含 `msg`、`code`、`data`、`count` 和 `limit` 字段。
"""
res = {
'msg': msg,
'code': 0,
'data': data,
'count': count,
'limit': limit
}
return jsonify(res)
return jsonify(res)
applications/common/utils/mail.py
+34 -37
View File
@@ -1,5 +1,5 @@
"""
集成了对 Pear Admin Flask 二次开发的的邮件操作，并给了相对应的示例。
集成了对 Pear Admin Flask 二次开发的邮件操作模块，并提供了相应的示例。
"""
from flask import current_app
from flask_mail import Message
@@ -7,32 +7,31 @@ from flask_mail import Message
from applications.common.curd import model_to_dicts
from applications.common.helper import ModelFilter
from applications.extensions import db, flask_mail
from applications.extensions.init_mail import mail
from applications.models import Mail
from applications.schemas import MailOutSchema
def get_all(receiver=None, subject=None, content=None):
"""
获取邮件
获取邮件列表，支持根据接收者、主题和内容进行筛选。
返回的列表中的字典构造如下::
返回的列表中的字典结构如下::
{
"content": "", # html内容
"create_at": "2022-12-25T10:51:17", # 时间
"content": "", # HTML 内容
"create_at": "2022-12-25T10:51:17", # 创建时间
"id": 17, # 邮件ID
"realname": "超级管理", # 创建者
"realname": "超级管理", # 创建者姓名
"receiver": "", # 接收者
"subject": "" # 主题
"subject": "" # 邮件主题
}
:param receiver: 发送者
:param subject: 邮件标题
:param content: 邮件内容
:return: 列表
:param receiver: 接收者邮箱地址，支持模糊查询。
:param subject: 邮件主题，支持模糊查询。
:param content: 邮件内容，支持模糊查询。
:return: 返回符合条件的邮件列表。
"""
# 查询参数构造
# 构造查询条件
mf = ModelFilter()
if receiver:
mf.contains(field_name="receiver", value=receiver)
@@ -40,31 +39,27 @@ def get_all(receiver=None, subject=None, content=None):
mf.contains(field_name="subject", value=subject)
if content:
mf.exact(field_name="content", value=content)
# orm查询
# 使用分页获取data需要.items
# 查询邮件数据并分页
mail = Mail.query.filter(mf.get_filter(Mail)).layui_paginate()
return model_to_dicts(schema=MailOutSchema, data=mail.items)
def add(receiver, subject, content, user_id):
"""
发送一封邮件，若发送成功立刻提交数据库。
发送一封邮件，并将发送记录保存到数据库。 **该方法被邮件发送的视图函数调用。**
:param receiver: 接收者 多个用英文分号隔开
:param subject: 邮件主题
:param content: 邮件 html
:param user_id: 发送用户ID（谁发送的？） 可以用 from flask_login import current_user ; current_user.id 来表示当前登录用户
:return: 成功与否
:param receiver: 接收者邮箱地址，多个邮箱用英文分号隔开。
:param subject: 邮件主题。
:param content: 邮件内容（HTML 格式）。
:param user_id: 发送者用户ID，表示谁发送了这封邮件。
可以使用 `from flask_login import current_user; current_user.id` 获取当前登录用户的ID。
:return: 发送成功返回 True，失败报错。
"""
try:
msg = Message(subject=subject, recipients=receiver.split(";"), html=content)
flask_mail.send(msg)
except BaseException as e:
current_app.log_exception(e)
return False
send_mail(subject=subject, recipients=receiver.split(";"), content=content)
# 保存邮件记录到数据库
mail = Mail(receiver=receiver, subject=subject, content=content, user_id=user_id)
db.session.add(mail)
db.session.commit()
return True
@@ -72,10 +67,10 @@ def add(receiver, subject, content, user_id):
def delete(id):
"""
删除邮件记录，立刻写入数据库。
删除指定的邮件记录。
:param id: 邮件ID
:return: 成功与否
:param id: 邮件ID。
:return: 删除成功返回 True，失败返回 False。
"""
res = Mail.query.filter_by(id=id).delete()
if not res:
@@ -83,14 +78,16 @@ def delete(id):
db.session.commit()
return True
def send_mail(subject, recipients, content):
"""原发送邮件函数，不会记录邮件发送记录
"""
发送邮件（不记录发送日志）。
失败报错，请注意使用 try 拦截。
注意：如果发送失败会抛出异常，请使用 try-except 进行捕获。
:param subject: 主题
:param recipients: 接收者 多个用英文分号隔开
:param content: 邮件 html
:param subject: 邮件主题。
:param recipients: 接收者邮箱地址，多个邮箱用英文分号隔开。
:param content: 邮件内容（HTML 格式）。
"""
message = Message(subject=subject, recipients=recipients, html=content)
mail.send(message)
flask_mail.send(message)
applications/common/utils/rights.py
+3 -1
View File
@@ -5,7 +5,9 @@ from applications.common.admin_log import admin_log
def authorize(power: str, log: bool = False):
"""用户权限判断，用于判断目前会话用户是否拥有访问权限
"""
用户权限判断，用于判断目前会话用户是否拥有访问权限。
在模板中有与之对应的全局非修饰函数 authorize ，此函数定义位于 `applications/extensions/init_template_directives.py` 。
:param power: 权限标识
:type power: str
applications/common/utils/validate.py
+194 -141
View File
@@ -1,187 +1,240 @@
# xss过滤
# XSS 过滤
import validators
from markupsafe import escape
from validators import validator
def str_escape(s):
"""
对字符串进行 XSS 过滤，返回转义后的安全字符串。
:param s: 需要转义的字符串。
:return: 返回转义后的字符串，如果输入为空则返回 None。
"""
if not s:
return None
return str(escape(s))
between = validators.between
"""
验证数字是否介于最小值和/或最大值之间。
这将适用于任何类似的类型，如浮点数、小数和日期，而不仅仅是整数。
between(value, min=None, max=None)
def between(*args, **kwargs):
"""
验证数字是否介于最小值和最大值之间。
适用于整数、浮点数、小数和日期等类型。
min-数字的最小必需值。如果未提供，则不会检查最小值。
max-数字的最大值。如果未提供，将不检查最大值。
:param value: 需要验证的数字。
:param min: 数字的最小值（可选）。
:param max: 数字的最大值（可选）。
:return: 如果验证成功返回 True，否则返回 ValidationFailure。
>>> from datetime import datetime
示例：
>>> between(5, min=2)
True
>>> between(13.2, min=13, max=14)
True
>>> between(500, max=400)
ValidationFailure(func=between, args=...)
>>> between(
... datetime(2000, 11, 11),
... min=datetime(1999, 11, 11)
... )
True
"""
"""
return validators.between(*args, **kwargs)
domain = validators.domain
"""
返回给定值是否为有效域
如果值是有效域名，则此函数返回 True ，否则返回 ValidationFailure
domain(value)
value-要验证的属性域字符串
def domain(*args, **kwargs):
"""
验证给定值是否为有效的域名。
:param value: 需要验证的域名字符串。
:return: 如果验证成功返回 True，否则返回 ValidationFailure。
示例：
>>> domain('example.com')
True
>>> domain('example.com/')
ValidationFailure(func=domain, ...)
"""
"""
return validators.domain(*args, **kwargs)
email = validators.email
"""
验证电子邮件地址。验证成功时返回 True ，验证失败时返回
>>> email('someone@example.com')
True
>>> email('bogus@@')
ValidationFailure(func=email, ...)
"""
iban = validators.iban
"""
返回给定值是否为有效的IBAN代码。
如果值是有效的IBAN，则此函数返回 True ，否则返回 ValidationFailure 。
def email(*args, **kwargs):
"""
验证给定值是否为有效的电子邮件地址。
>>> iban('DE29100500001061045672')
True
>>> iban('123456')
ValidationFailure(func=iban, ...)
"""
:param value: 需要验证的电子邮件地址。
:return: 如果验证成功返回 True，否则返回 ValidationFailure。
ipv4 = validators.ipv4
"""
返回给定值是否为有效的IPv4地址。
示例：
>>> email('someone@example.com')
True
>>> ipv4('123.0.0.7')
True
>>> ipv4('900.80.70.11')
ValidationFailure(func=ipv4, args={'value': '900.80.70.11'})
"""
>>> email('bogus@@')
ValidationFailure(func=email, ...)
"""
return validators.email(*args, **kwargs)
ipv6 = validators.ipv6
"""
返回给定值是否为有效的IP版本6地址。
>>> ipv6('abcd:ef::42:1')
True
>>> ipv6('abc.0.0.1')
ValidationFailure(func=ipv6, args={'value': 'abc.0.0.1'})
"""
length = validators.length
"""
返回给定字符串的长度是否在指定范围内。
>>> length('something', min=2)
True
>>> length('something', min=9, max=9)
True
>>> length('something', max=5)
ValidationFailure(func=length, ...)
"""
def iban(*args, **kwargs):
"""
验证给定值是否为有效的 IBAN 代码。
mac_address = validators.mac_address
"""
返回给定值是否为有效MAC地址。
如果该值是有效的MAC地址，则此函数返回 True ，否则返回 ValidationFailure 。
:param value: 需要验证的 IBAN 代码。
:return: 如果验证成功返回 True，否则返回 ValidationFailure。
>>> mac_address('01:23:45:67:ab:CD')
True
>>> mac_address('00:00:00:00:00')
ValidationFailure(func=mac_address, args={'value': '00:00:00:00:00'})
"""
示例：
>>> iban('DE29100500001061045672')
True
slug = validators.slug
"""
验证给定值是否为有效的块。
有效的短信息只能包含字母数字字符、连字符和下划线。
>>> slug('my.slug')
ValidationFailure(func=slug, args={'value': 'my.slug'})
>>> slug('my-slug-2134')
True
"""
>>> iban('123456')
ValidationFailure(func=iban, ...)
"""
return validators.iban(*args, **kwargs)
#truthy = validators.truthy
"""
验证给定值不是错误值。
"""
url = validators.url
"""
返回给定值是否为有效URL。
如果值是有效URL，则此函数返回 True ，否则返回 ValidationFailure 。
def ipv4(*args, **kwargs):
"""
验证给定值是否为有效的 IPv4 地址。
>>> url('http://foobar.dk')
True
>>> url('http://10.0.0.1')
True
>>> url('http://foobar.d')
ValidationFailure(func=url, ...)
>>> url('http://10.0.0.1', public=True)
ValidationFailure(func=url, ...)
"""
:param value: 需要验证的 IPv4 地址。
:return: 如果验证成功返回 True，否则返回 ValidationFailure。
uuid = validators.uuid
"""
返回给定值是否为有效UUID。
如果值是有效的UUID，则此函数返回 True ，否则返回 ValidationFailure 。
示例：
>>> ipv4('123.0.0.7')
True
>>> uuid('2bc1c94f-0deb-43e9-92a1-4775189ec9f8')
True
>>> uuid('2bc1c94f 0deb-43e9-92a1-4775189ec9f8')
ValidationFailure(func=uuid, ...)
"""
>>> ipv4('900.80.70.11')
ValidationFailure(func=ipv4, args={'value': '900.80.70.11'})
"""
return validators.ipv4(*args, **kwargs)
def ipv6(*args, **kwargs):
"""
验证给定值是否为有效的 IPv6 地址。
:param value: 需要验证的 IPv6 地址。
:return: 如果验证成功返回 True，否则返回 ValidationFailure。
示例：
>>> ipv6('abcd:ef::42:1')
True
>>> ipv6('abc.0.0.1')
ValidationFailure(func=ipv6, args={'value': 'abc.0.0.1'})
"""
return validators.ipv6(*args, **kwargs)
def length(*args, **kwargs):
"""
验证给定字符串的长度是否在指定范围内。
:param value: 需要验证的字符串。
:param min: 字符串的最小长度（可选）。
:param max: 字符串的最大长度（可选）。
:return: 如果验证成功返回 True，否则返回 ValidationFailure。
示例：
>>> length('something', min=2)
True
>>> length('something', min=9, max=9)
True
>>> length('something', max=5)
ValidationFailure(func=length, ...)
"""
return validators.length(*args, **kwargs)
def mac_address(*args, **kwargs):
"""
验证给定值是否为有效的 MAC 地址。
:param value: 需要验证的 MAC 地址。
:return: 如果验证成功返回 True，否则返回 ValidationFailure。
示例：
>>> mac_address('01:23:45:67:ab:CD')
True
>>> mac_address('00:00:00:00:00')
ValidationFailure(func=mac_address, args={'value': '00:00:00:00:00'})
"""
return validators.mac_address(*args, **kwargs)
def slug(*args, **kwargs):
"""
验证给定值是否为有效的 Slug 格式。
有效的 Slug 只能包含字母数字字符、连字符和下划线。
:param value: 需要验证的字符串。
:return: 如果验证成功返回 True，否则返回 ValidationFailure。
示例：
>>> slug('my.slug')
ValidationFailure(func=slug, args={'value': 'my.slug'})
>>> slug('my-slug-2134')
True
"""
return validators.slug(*args, **kwargs)
def url(*args, **kwargs):
"""
验证给定值是否为有效的 URL。
:param value: 需要验证的 URL。
:param public: 是否仅允许公共 URL（可选）。
:return: 如果验证成功返回 True，否则返回 ValidationFailure。
示例：
>>> url('http://foobar.dk')
True
>>> url('http://10.0.0.1')
True
>>> url('http://foobar.d')
ValidationFailure(func=url, ...)
>>> url('http://10.0.0.1', public=True)
ValidationFailure(func=url, ...)
"""
return validators.url(*args, **kwargs)
def uuid(*args, **kwargs):
"""
验证给定值是否为有效的 UUID。
:param value: 需要验证的 UUID。
:return: 如果验证成功返回 True，否则返回 ValidationFailure。
示例：
>>> uuid('2bc1c94f-0deb-43e9-92a1-4775189ec9f8')
True
>>> uuid('2bc1c94f 0deb-43e9-92a1-4775189ec9f8')
ValidationFailure(func=uuid, ...)
"""
return validators.uuid(*args, **kwargs)
@validator
def even(value):
return not (value % 2)
"""
验证给定值是否为偶数。
:param value: 需要验证的数字。
:return: 如果是偶数返回 True，否则返回 ValidationFailure。
"""
一个装饰器，它使给定的函数验证器
每当给定函数被调用并返回 False 值时，这个装饰器返回 ValidationFailure 对象。
>>> @validator
... def even(value):
... return not (value % 2)
示例：
>>> even(4)
True
>>> even(4)
True
>>> even(5)
ValidationFailure(func=even, args={'value': 5})
"""
>>> even(5)
ValidationFailure(func=even, args={'value': 5})
"""
return not (value % 2)
applications/view/system/mail.py
+8 -10
View File
@@ -1,14 +1,16 @@
from flask import Blueprint, render_template, request, current_app
from flask_login import current_user
from flask_mail import Message
from applications.common.curd import model_to_dicts
from applications.common.helper import ModelFilter
from applications.common.utils.http import table_api, fail_api, success_api
from applications.common.utils.rights import authorize
from applications.common.utils.validate import str_escape
from applications.extensions import db, flask_mail
from applications.extensions import db
from applications.models import Mail
from applications.schemas import MailOutSchema
from applications.common.utils import mail
from applications.common.admin_log import admin_log
bp = Blueprint('adminMail', __name__, url_prefix='/mail')
@@ -61,17 +63,13 @@ def save():
user_id = current_user.id
try:
msg = Message(subject=subject, recipients=receiver.split(";"), body=content)
flask_mail.send(msg)
if mail.add(receiver=receiver, subject=subject, content=content, user_id=user_id):
return success_api(msg="增加成功")
except Exception as e:
current_app.log_exception(e)
return fail_api(msg="发送失败，请检查邮件配置或发送人邮箱是否写错")
admin_log(request, False, desc="发送日志失败：" + str(e))
mail = Mail(receiver=receiver, subject=subject, content=content, user_id=user_id)
db.session.add(mail)
db.session.commit()
return success_api(msg="增加成功")
return success_api(msg="发送失败，请检查日志。")
# 删除用户
applications/view/system/passport.py
+1
View File
@@ -75,6 +75,7 @@ def login_post():
# session['role'] = [roles]
return success_api(msg="登录成功")
login_log(request, uid=user.id, is_access=False)
return fail_api(msg="用户名或密码错误")