Files
DjangoAdmin_Flask_EleVue/application/services/file_template.py
T
DjangoAdmin DevandClaude 15f2a5e097 chore: 基线提交(重构前代码快照)
- git 初始化 + .gitignore 补全(根 .env / logs / instance / __pycache__ 等)
- 修复 utils/.gitignore 误将整个 utils/ 源码层忽略的问题,补入全部工具源码(含 ip2region.xdb)
- 含后端全部代码:application / extensions / utils / config / migrations
- 含前端 ui/(后续整体迁走)
- 不含根 .env(含密钥,本地保留)

Co-Authored-By: Claude <noreply@anthropic.com>
2026-08-07 12:12:06 +08:00

340 lines
10 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""
文件模板管理服务模块
=============================================================
该模块实现了文件模板管理相关的所有业务逻辑。
包括文件模板的分页查询、详情查询、添加、更新、删除、文件上传下载等功能。
命名规范:
- 函数参数使用驼峰命名(如 templateId, pageNo)
- 表单字段使用下划线命名(如 name, code, sort)- 与数据库字段保持一致
- 返回数据使用驼峰命名(如 createUser, updateTime)- 与前端规范保持一致
- 私有函数以下划线开头(如 _apply_filters, _build_page_data)
功能模块:
1. 分页查询:支持按模板名称模糊搜索、编码模糊搜索,按sort字段排序
2. 详情查询:根据ID获取单个文件模板的详细信息
3. 数据维护:文件模板的添加、更新、删除(支持批量)
4. 文件上传:上传模板文件并保存文件信息
5. 文件下载:下载指定模板的文件
6. 数据列表:获取所有文件模板列表(用于下拉选择等场景)
"""
import logging
from datetime import datetime, timezone
from flask import request
from sqlalchemy import and_
from application.constants.common import PAGE_SIZE
from application.forms.file_template import FileTemplateForm
from application.models.file_template import FileTemplate
from extensions import db
from utils import R, regular
from utils.case_converter import convert_camel_to_snake
from utils.model_utils import models_to_list, model_to_dict
from utils.request import parse_request_body, parse_id_list, like_contains
from utils.rich_text import save_file, get_file_url
from utils.security import get_username
# 获取日志记录器
logger = logging.getLogger(__name__)
# ============================================================
# 文件模板分页查询
# ============================================================
def get_file_template_page():
"""
查询文件模板分页数据
Args:
request: Flask request对象,包含查询参数
- pageNo: 当前页码(可选,默认为1)
- pageSize: 每页记录数(可选,默认为PAGE_SIZE)
- name: 模板名称(可选,模糊查询)
- code: 模板编码(可选,模糊查询)
Returns:
R: 包含分页数据的JSON响应,格式为:
{
"records": 模板列表,
"total": 总记录数,
"size": 每页大小,
"current": 当前页码,
"pages": 总页数
}
"""
try:
# 获取分页参数
page_no = int(request.args.get('pageNo', 1))
page_size = int(request.args.get('pageSize', PAGE_SIZE))
# 应用筛选条件
query = _apply_filters()
# 记录总数
total = query.count()
# 计算总页数
pages = (total + page_size - 1) // page_size if total > 0 else 0
# 分页查询
template_list = query.limit(page_size).offset((page_no - 1) * page_size).all()
# 构建返回数据
page_data = _build_page_data(template_list, total, page_size, page_no, pages)
return R.ok(data=page_data)
except Exception as e:
logger.error(f"查询文件模板分页数据异常: {str(e)}")
return R.failed(msg="查询失败,请稍后重试")
def _apply_filters():
"""
应用查询筛选条件
根据请求参数动态构建查询条件:
- 自动过滤已删除记录(is_delete == 0)
- 支持模板名称模糊查询
- 支持模板编码模糊查询
- 默认按排序值升序排列
Returns:
Query: 应用筛选后的SQLAlchemy查询对象
"""
# 基础查询:过滤已删除记录
query = FileTemplate.query.filter(FileTemplate.is_delete == 0)
# 模板名称模糊筛选
name = request.args.get('name')
if name:
query = query.filter(FileTemplate.name.like(like_contains(name), escape='\\'))
# 模板编码模糊筛选
code = request.args.get('code')
if code:
query = query.filter(FileTemplate.code.like(like_contains(code), escape='\\'))
# 按排序值升序排列
return query.order_by(FileTemplate.sort.asc())
def _build_page_data(template_list, total, page_size, page_no, pages):
"""
构建分页返回数据
Args:
template_list: 文件模板对象列表
total: 总记录数
page_size: 每页大小
page_no: 当前页码
pages: 总页数
Returns:
dict: 分页数据,包含 records、total、size、current、pages
"""
# 将模型对象列表转换为字典列表
records = models_to_list(template_list)
return {
'records': records,
'total': total,
'size': page_size,
'current': page_no,
'pages': pages
}
# ============================================================
# 文件模板详情查询
# ============================================================
def get_file_template_detail(template_id):
"""
根据ID查询文件模板详情(返回字典格式)
此方法调用 get_info 获取模型对象,然后转换为字典格式返回。
适用于前端接口调用。
Args:
template_id: 文件模板ID
Returns:
dict/None: 文件模板详情字典,不存在时返回 None
"""
try:
# 复用 get_info 方法查询模板
template = get_info(template_id)
if not template:
return None
# 转换为字典格式
data = model_to_dict(template)
# 头像URL
data['filePath'] = get_file_url(data.get('filePath'))
# 返回结果
return data
except Exception as e:
logger.error(f"查询文件模板详情异常,ID: {template_id}, 错误: {str(e)}")
return None
# ============================================================
# 文件模板信息查询(根据ID)- 核心查询方法
# ============================================================
def get_info(template_id: int):
"""
根据ID查询文件模板信息(返回模型对象)
这是一个通用的模板查询方法,返回模板模型对象而非字典。
Args:
template_id: 文件模板ID
Returns:
FileTemplate/None: 文件模板模型对象,不存在或已删除时返回 None
"""
# 参数有效性验证
if not template_id or template_id <= 0:
return None
try:
# 查询未删除的模板
template = FileTemplate.query.filter(
and_(FileTemplate.id == template_id, FileTemplate.is_delete == 0)
).first()
return template
except Exception as e:
logger.error(f"查询文件模板信息异常,ID: {template_id}, 错误: {str(e)}")
return None
# ============================================================
# 文件模板添加
# ============================================================
def add_file_template():
"""
添加文件模板
流程:
1. 解析请求体JSON数据
2. 表单验证(编码唯一性、字段格式等)
3. 创建文件模板对象并保存到数据库
Returns:
R: 操作结果
"""
try:
# 获取参数
json_data, error = parse_request_body()
if error:
return R.failed(msg=error)
# 表单验证
form = FileTemplateForm(data=json_data)
if not form.validate():
err_msg = regular.get_err(form)
return R.failed(msg=err_msg)
# 文件处理
file_path = form.data.get('filePath')
if file_path:
form.filePath.data = save_file(file_path, "file_template")
# 创建文件模板
template = FileTemplate(**form.data)
template.create_user = get_username()
template.save()
return R.ok(msg="添加成功")
except Exception as e:
logger.error(f"添加文件模板异常: {str(e)}")
return R.failed(msg="添加文件模板失败,请稍后重试")
# ============================================================
# 文件模板更新
# ============================================================
def update_file_template():
"""
更新文件模板
流程:
1. 解析请求体JSON数据
2. 表单验证
3. 验证模板是否存在(复用 get_info)
4. 更新模板信息
Returns:
R: 操作结果
"""
try:
# 获取参数
json_data, error = parse_request_body()
if error:
return R.failed(msg=error)
# 表单验证
form = FileTemplateForm(data=json_data)
if not form.validate():
return R.failed(msg=regular.get_err(form))
# 获取模板ID
template_id = form.data.get('id')
if not template_id or int(template_id) <= 0:
return R.failed("模板ID无效")
template_id = int(template_id)
# 复用 get_info 方法验证模板是否存在
template = get_info(template_id)
if not template:
return R.failed("模板不存在")
# 文件处理
file_path = form.data.get('filePath')
if file_path:
form.filePath.data = save_file(file_path, "file_template")
# 更新模板信息(移除ID,避免更新主键)
form_data = convert_camel_to_snake(form.data.copy())
form_data.pop('id', None)
form_data['update_time'] = datetime.now(tz=timezone.utc)
form_data['update_user'] = get_username()
result = FileTemplate.query.filter_by(id=template_id).update(form_data)
if not result:
return R.failed("更新失败")
return R.ok(msg="更新成功")
except Exception as e:
logger.error(f"更新文件模板异常: {str(e)}")
return R.failed(msg="更新文件模板失败,请稍后重试")
# ============================================================
# 文件模板删除
# ============================================================
def delete_file_templates(template_ids):
"""
删除文件模板(批量删除)
删除
支持逗号分隔的多个ID批量删除。
Args:
template_ids: 模板ID字符串,多个ID用逗号分隔,如 "1,2,3"
Returns:
R: 操作结果
"""
return FileTemplate.batch_delete_r(template_ids)