refactor: P4c 批次2 服务层改造（article/category/file_template）

- 3 模块按新范式改造（get_x_page(params)、add/update(data)、delete(ids)等）
- category 为 tree 型：list/data 函数改为 params 化返回 list（手动处理）
- get_info 全部委托 Repository（新增 3 个 repository.py）
- upload 记为例外：文件上传依赖 request.files，不适合数据入参范式
  （其辅助函数已是纯函数可测）

【工具】批次1 codemod 复用 + ast 级未用导入清理（避免 prefix 误删）
【验收】flake8 零告警；应用启动；3 服务运行时正常；12 passed

Co-Authored-By: Claude <noreply@anthropic.com>

src/modules/cms/article/models.py

@@ -9,7 +9,6 @@ from src.common.base_db import BaseDB
 from src.common.base_model import BaseModel
 from config import DB_PREFIX
 from src.core.database import db
-from src.common.utils.case_converter import convert_camel_to_snake
 
 
 # ============================================================

src/modules/cms/article/repository.py

@@ -0,0 +1,15 @@
+"""
+Article 仓储模块
+=============================================================
+
+数据访问层，继承 BaseRepository 获得通用 CRUD 能力。
+"""
+
+from src.common.base_repository import BaseRepository
+from src.modules.cms.article.models import Article
+
+
+class ArticleRepository(BaseRepository):
+    """Article 仓储"""
+
+    model = Article

src/modules/cms/article/router.py

@@ -16,22 +16,24 @@
          application.middleware.check_permission - 权限校验中间件
 """
 
-from flask import Blueprint
+from flask import Blueprint, request
 
 from application.middleware.check_demo import check_demo
 from application.middleware.check_permission import check_permission
 from src.modules.cms.article import service as article
-from src.common.utils.request import parse_batch_ids
+from src.common.utils.request import parse_batch_ids, parse_request_body
-article_blue = Blueprint('article', __name__, url_prefix='/article')
 
 from src.core import response as R
 
+article_blue = Blueprint("article", __name__, url_prefix="/article")
+
 
 # ============================================================
 # 文章分页查询
 # ============================================================
 
-@article_blue.route('/page', methods=["GET"])
+
+@article_blue.route("/page", methods=["GET"])
 @check_permission("sys:article:page")
 def page():
     """
@@ -52,15 +54,15 @@ def page():
     Returns:
         JSON: 文章分页数据，包含 records（数据列表）、total（总记录数）等
     """
-    result = article.get_article_page()
+    return R.ok(data=article.get_article_page(request.args))
-    return result
 
 
 # ============================================================
 # 文章详情查询
 # ============================================================
 
-@article_blue.route('/detail/<int:id>', methods=["GET"])
+
+@article_blue.route("/detail/<int:id>", methods=["GET"])
 def detail(id):
     """
     查询文章详情
@@ -84,7 +86,8 @@ def detail(id):
 # 文章添加
 # ============================================================
 
-@article_blue.route('/add', methods=["POST"])
+
+@article_blue.route("/add", methods=["POST"])
 @check_permission("sys:article:add")
 @check_demo
 def add():
@@ -110,15 +113,19 @@ def add():
     Returns:
         JSON: 操作结果
     """
-    result = article.add_article()
+    json_data, err = parse_request_body()
-    return result
+    if err:
+        return R.failed(msg=err)
+    article.add_article(json_data)
+    return R.ok(msg="添加成功")
 
 
 # ============================================================
 # 文章更新
 # ============================================================
 
-@article_blue.route('/update', methods=["PUT"])
+
+@article_blue.route("/update", methods=["PUT"])
 @check_permission("sys:article:update")
 @check_demo
 def update():
@@ -131,15 +138,19 @@ def update():
     Returns:
         JSON: 操作结果
     """
-    result = article.update_article()
+    json_data, err = parse_request_body()
-    return result
+    if err:
+        return R.failed(msg=err)
+    article.update_article(json_data)
+    return R.ok(msg="更新成功")
 
 
 # ============================================================
 # 文章删除
 # ============================================================
 
-@article_blue.route('/delete/<string:id>', methods=["DELETE"])
+
+@article_blue.route("/delete/<string:id>", methods=["DELETE"])
 @check_permission("sys:article:delete")
 @check_demo
 def delete(id):
@@ -152,15 +163,16 @@ def delete(id):
     Returns:
         JSON: 操作结果
     """
-    result = article.delete_articles(id)
+    count = article.delete_articles(id)
-    return result
+    return R.ok(msg=f"本次共删除{count}条数据")
 
 
 # ============================================================
 # 文章批量删除
 # ============================================================
 
-@article_blue.route('/batchDelete', methods=["DELETE"])
+
+@article_blue.route("/batchDelete", methods=["DELETE"])
 @check_permission("sys:article:batch_delete")
 @check_demo
 def batch_delete():
@@ -186,7 +198,8 @@ def batch_delete():
 # 文章状态设置
 # ============================================================
 
-@article_blue.route('/status', methods=["PUT"])
+
+@article_blue.route("/status", methods=["PUT"])
 @check_permission("sys:article:status")
 @check_demo
 def status():
@@ -199,5 +212,8 @@ def status():
     Returns:
         JSON: 操作结果
     """
-    result = article.update_article_status()
+    json_data, err = parse_request_body()
-    return result
+    if err:
+        return R.failed(msg=err)
+    article.update_article_status(json_data)
+    return R.ok(msg="更新成功")

src/modules/cms/article/service.py

@@ -23,20 +23,18 @@
 import logging
 from datetime import datetime, timezone
 
-from flask import request
-from sqlalchemy import and_
 
 from src.common.constants.common import PAGE_SIZE
 from src.common.constants.dictionary_constants import ARTICLE_STATUS_LIST
 from src.modules.cms.article.schema import ArticleForm, ArticleStatusForm
 from src.modules.cms.article.models import Article
+from src.modules.cms.article.repository import ArticleRepository
+from src.core.exceptions import BizError
 from config import FLASK_FILE_URL
-from src.core.database import db
-from src.core import response as R
 from src.common.utils import regular
 from src.common.utils.case_converter import convert_camel_to_snake
 from src.common.utils.model_utils import models_to_list, model_to_dict
-from src.common.utils.request import parse_request_body, parse_id_list, like_contains
+from src.common.utils.request import parse_id_list, like_contains
 from src.common.utils.rich_text import get_file_url, save_file, save_content
 from src.core.security import get_username
 
@@ -48,7 +46,8 @@ logger = logging.getLogger(__name__)
 # 文章分页查询
 # ============================================================
 
-def get_article_page():
+
+def get_article_page(params: dict) -> dict:
     """
     查询文章分页数据
 
@@ -72,13 +71,12 @@ def get_article_page():
                 "pages": 总页数
             }
     """
-    try:
     # 获取分页参数
-        page_no = int(request.args.get('pageNo', 1))
+    page_no = int(params.get("pageNo", 1))
-        page_size = int(request.args.get('pageSize', PAGE_SIZE))
+    page_size = int(params.get("pageSize", PAGE_SIZE))
 
     # 应用筛选条件
-        query = _apply_filters()
+    query = _apply_filters(params)
 
     # 记录总数
     total = query.count()
@@ -86,21 +84,19 @@ def get_article_page():
     pages = (total + page_size - 1) // page_size if total > 0 else 0
 
     # 分页查询
-        article_list = query.order_by(Article.sort.asc(), Article.create_time.desc()) \
+    article_list = (
-            .limit(page_size) \
+        query.order_by(Article.sort.asc(), Article.create_time.desc())
-            .offset((page_no - 1) * page_size) \
+        .limit(page_size)
+        .offset((page_no - 1) * page_size)
         .all()
+    )
 
     # 构建返回数据
     page_data = _build_page_data(article_list, total, page_size, page_no, pages)
-        return R.ok(data=page_data)
+    return page_data
-
-    except Exception as e:
-        logger.error(f"查询文章分页数据异常: {str(e)}")
-        return R.failed(msg="查询失败，请稍后重试")
 
 
-def _apply_filters():
+def _apply_filters(params: dict):
     """
     应用查询筛选条件
 
@@ -118,34 +114,34 @@ def _apply_filters():
     query = Article.query.filter(Article.is_delete == 0)
 
     # 文章标题模糊筛选
-    title = request.args.get('title')
+    title = params.get("title")
     if title:
-        query = query.filter(Article.title.like(like_contains(title), escape='\\'))
+        query = query.filter(Article.title.like(like_contains(title), escape="\\"))
 
     # 分类ID精确筛选
-    category_id = request.args.get('categoryId')
+    category_id = params.get("categoryId")
     if category_id:
         query = query.filter(Article.category_id == category_id)
 
     # 文章状态精确筛选
-    status = request.args.get('status')
+    status = params.get("status")
     if status:
         query = query.filter(Article.status == status)
 
     # 时间范围筛选
-    start_time = request.args.get('startTime')
+    start_time = params.get("startTime")
-    end_time = request.args.get('endTime')
+    end_time = params.get("endTime")
 
     if start_time:
         try:
-            start_date = datetime.strptime(start_time, '%Y-%m-%d %H:%M:%S')
+            start_date = datetime.strptime(start_time, "%Y-%m-%d %H:%M:%S")
             query = query.filter(Article.create_time >= start_date)
         except ValueError:
             pass
 
     if end_time:
         try:
-            end_date = datetime.strptime(end_time, '%Y-%m-%d %H:%M:%S')
+            end_date = datetime.strptime(end_time, "%Y-%m-%d %H:%M:%S")
             query = query.filter(Article.create_time <= end_date)
         except ValueError:
             pass
@@ -173,23 +169,18 @@ def _build_page_data(article_list, total, page_size, page_no, pages):
     # 为每条记录添加状态文本映射
     for record in records:
         # 文章封面
-        record['cover'] = get_file_url(record.get('cover'))
+        record["cover"] = get_file_url(record.get("cover"))
         # 文章状态
-        record['statusText'] = ARTICLE_STATUS_LIST.get(record.get('status'))
+        record["statusText"] = ARTICLE_STATUS_LIST.get(record.get("status"))
 
-    return {
+    return {"records": records, "total": total, "size": page_size, "current": page_no, "pages": pages}
-        'records': records,
-        'total': total,
-        'size': page_size,
-        'current': page_no,
-        'pages': pages
-    }
 
 
 # ============================================================
 # 文章详情查询
 # ============================================================
 
+
 def get_article_detail(article_id):
     """
     根据ID查询文章详情（返回字典格式）
@@ -212,24 +203,24 @@ def get_article_detail(article_id):
         record = model_to_dict(article)
 
         # 图片封面
-        record['cover'] = get_file_url(record['cover'])
+        record["cover"] = get_file_url(record["cover"])
 
         # 处理图集（逗号分隔转URL列表）
-        images = record.get('images')
+        images = record.get("images")
         if images:
             # 将逗号分隔的路径转换为URL列表
-            image_paths = [path.strip() for path in images.split(',') if path.strip()]
+            image_paths = [path.strip() for path in images.split(",") if path.strip()]
-            record['imagesList'] = [get_file_url(path) for path in image_paths]
+            record["imagesList"] = [get_file_url(path) for path in image_paths]
         else:
-            record['imagesList'] = []
+            record["imagesList"] = []
 
         # 富文本处理
-        content = record.get('content')
+        content = record.get("content")
         if content:
-            record['content'] = content.replace("[IMG_URL]", FLASK_FILE_URL)
+            record["content"] = content.replace("[IMG_URL]", FLASK_FILE_URL)
 
         # 文章状态
-        record['statusText'] = ARTICLE_STATUS_LIST.get(record.get('status'))
+        record["statusText"] = ARTICLE_STATUS_LIST.get(record.get("status"))
 
         return record
 
@@ -242,6 +233,7 @@ def get_article_detail(article_id):
 # 文章信息查询（根据ID）- 核心查询方法
 # ============================================================
 
+
 def get_info(article_id: int):
     """
     根据ID查询文章信息（返回模型对象）
@@ -268,26 +260,10 @@ def get_info(article_id: int):
         if article:
             print(article.title)
     """
-    # 参数有效性验证
+    return ArticleRepository.get_by_id(article_id)
-    if not article_id or article_id <= 0:
-        return None
-
-    try:
-        # 查询未删除的文章
-        article = Article.query.filter(
-            and_(Article.id == article_id, Article.is_delete == 0)
-        ).first()
-        return article
-    except Exception as e:
-        logger.error(f"查询文章信息异常，ID: {article_id}, 错误: {str(e)}")
-        return None
 
 
-# ============================================================
+def add_article(data: dict) -> None:
-# 文章添加
-# ============================================================
-
-def add_article():
     """
     添加文章
 
@@ -300,38 +276,32 @@ def add_article():
     Returns:
         R: 操作结果
     """
-    try:
-        # 获取参数
-        json_data, error = parse_request_body()
-        if error:
-            return R.failed(msg=error)
 
     # 表单验证
-        form = ArticleForm(data=json_data)
+    form = ArticleForm(data=data)
     if not form.validate():
-            err_msg = regular.get_err(form)
+        raise BizError(regular.get_err(form))
-            return R.failed(msg=err_msg)
 
     # 文章封面
-        cover = form.data.get('cover')
+    cover = form.data.get("cover")
     if cover:
         cover = save_file(cover, "article")
         form.cover.data = cover
 
     # 文章图集（逗号分隔的图片路径）
-        images = form.data.get('images')
+    images = form.data.get("images")
     if images:
-            image_paths = [img.strip() for img in images.split(',') if img.strip()]
+        image_paths = [img.strip() for img in images.split(",") if img.strip()]
         processed_paths = []
         for img in image_paths:
             image = save_file(img, "article")
             processed_paths.append(image)
         # 处理后的地址逗号分隔
-            form.images.data = ','.join(processed_paths)
+        form.images.data = ",".join(processed_paths)
 
     # 文章内容
-        title = form.data.get('title', '')
+    title = form.data.get("title", "")
-        content = form.data.get('content', '')
+    content = form.data.get("content", "")
     if content:
         form.content.data = save_content(content, title, "article")
 
@@ -340,18 +310,13 @@ def add_article():
     article.create_user = get_username()
     article.save()
 
-        return R.ok(msg="添加成功")
-
-    except Exception as e:
-        logger.error(f"添加文章异常: {str(e)}")
-        return R.failed(msg="添加文章失败，请稍后重试")
-
 
 # ============================================================
 # 文章更新
 # ============================================================
 
-def update_article():
+
+def update_article(data: dict) -> None:
     """
     更新文章
 
@@ -365,75 +330,65 @@ def update_article():
     Returns:
         R: 操作结果
     """
-    try:
-        # 获取参数
-        json_data, error = parse_request_body()
-        if error:
-            return R.failed(msg=error)
 
     # 表单验证
-        form = ArticleForm(data=json_data)
+    form = ArticleForm(data=data)
     if not form.validate():
-            return R.failed(msg=regular.get_err(form))
+        raise BizError(regular.get_err(form))
 
     # 获取文章ID
-        article_id = form.data.get('id')
+    article_id = form.data.get("id")
     if not article_id or int(article_id) <= 0:
-            return R.failed("文章ID无效")
+        raise BizError("文章ID无效")
     article_id = int(article_id)
 
     # 复用 get_info 方法验证文章是否存在
     article = get_info(article_id)
     if not article:
-            return R.failed("文章不存在")
+        raise BizError("文章不存在")
 
     # 文章封面
-        cover = form.data.get('cover')
+    cover = form.data.get("cover")
     if cover:
         cover = save_file(cover, "article")
         form.cover.data = cover
 
     # 文章图集（逗号分隔的图片路径）
-        images = form.data.get('images')
+    images = form.data.get("images")
     if images:
-            image_paths = [img.strip() for img in images.split(',') if img.strip()]
+        image_paths = [img.strip() for img in images.split(",") if img.strip()]
         processed_paths = []
         for img in image_paths:
             image = save_file(img, "article")
             processed_paths.append(image)
         # 处理后的地址逗号分隔
-            form.images.data = ','.join(processed_paths)
+        form.images.data = ",".join(processed_paths)
 
     # 文章内容
-        title = form.data.get('title', '')
+    title = form.data.get("title", "")
-        content = form.data.get('content', '')
+    content = form.data.get("content", "")
     if content:
         form.content.data = save_content(content, title, "article")
 
     # 更新文章信息（移除ID，避免更新主键）
     form_data = convert_camel_to_snake(form.data.copy())
-        form_data.pop('id', None)
+    form_data.pop("id", None)
 
-        form_data['update_time'] = datetime.now(tz=timezone.utc)
+    form_data["update_time"] = datetime.now(tz=timezone.utc)
-        form_data['update_user'] = get_username()
+    form_data["update_user"] = get_username()
 
     result = Article.query.filter_by(id=article_id).update(form_data)
 
     if not result:
-            return R.failed("更新失败")
+        raise BizError("更新失败")
-
-        return R.ok(msg="更新成功")
-
-    except Exception as e:
-        logger.error(f"更新文章异常: {str(e)}")
-        return R.failed(msg="更新文章失败，请稍后重试")
 
 
 # ============================================================
 # 文章删除
 # ============================================================
 
-def delete_articles(article_ids):
+
+def delete_articles(article_ids: str) -> int:
     """
     删除文章（批量删除）
 
@@ -446,14 +401,18 @@ def delete_articles(article_ids):
     Returns:
         R: 操作结果
     """
-    return Article.batch_delete_r(article_ids)
+    article_ids_list = parse_id_list(article_ids)
+    if not article_ids_list:
+        raise BizError("无效的记录ID")
+    return Article.batch_delete(article_ids_list)
 
 
 # ============================================================
 # 文章状态设置
 # ============================================================
 
-def update_article_status():
+
+def update_article_status(data: dict) -> None:
     """
     设置文章状态
 
@@ -462,35 +421,23 @@ def update_article_status():
     Returns:
         R: 操作结果
     """
-    try:
-        # 获取参数
-        json_data, error = parse_request_body()
-        if error:
-            return R.failed(msg=error)
 
     # 表单验证
-        form = ArticleStatusForm(data=json_data)
+    form = ArticleStatusForm(data=data)
     if not form.validate():
-            err_msg = regular.get_err(form)
+        raise BizError(regular.get_err(form))
-            return R.failed(msg=err_msg)
 
-        article_id = int(form.data['id'])
+    article_id = int(form.data["id"])
-        status = int(form.data['status'])
+    status = int(form.data["status"])
 
     # 复用 get_info 方法验证文章是否存在
     article = get_info(article_id)
     if not article:
-            return R.failed("文章不存在")
+        raise BizError("文章不存在")
 
     # 更新状态
     update_data = {"status": status, "update_time": datetime.now(tz=timezone.utc), "update_user": get_username()}
     result = Article.query.filter_by(id=article_id).update(update_data)
 
     if not result:
-            return R.failed("设置失败")
+        raise BizError("设置失败")
-
-        return R.ok(msg="设置成功")
-
-    except Exception as e:
-        logger.error(f"设置文章状态异常: {str(e)}")
-        return R.failed(msg="设置文章状态失败，请稍后重试")

src/modules/cms/category/models.py

@@ -9,7 +9,6 @@ from src.common.base_db import BaseDB
 from src.common.base_model import BaseModel
 from config import DB_PREFIX
 from src.core.database import db
-from src.common.utils.case_converter import convert_camel_to_snake
 
 
 # ============================================================

src/modules/cms/category/repository.py

@@ -0,0 +1,15 @@
+"""
+Category 仓储模块
+=============================================================
+
+数据访问层，继承 BaseRepository 获得通用 CRUD 能力。
+"""
+
+from src.common.base_repository import BaseRepository
+from src.modules.cms.category.models import Category
+
+
+class CategoryRepository(BaseRepository):
+    """Category 仓储"""
+
+    model = Category

src/modules/cms/category/router.py

@@ -15,21 +15,24 @@
          application.middleware.check_permission - 权限校验中间件
 """
 
-from flask import Blueprint
+from flask import Blueprint, request
 
 from application.middleware.check_demo import check_demo
 from application.middleware.check_permission import check_permission
 from src.modules.cms.category import service as category
-category_blue = Blueprint('category', __name__, url_prefix='/category')
+from src.common.utils.request import parse_request_body
 
 from src.core import response as R
 
+category_blue = Blueprint("category", __name__, url_prefix="/category")
+
 
 # ============================================================
 # 分类列表查询
 # ============================================================
 
-@category_blue.route('/list', methods=["GET"])
+
+@category_blue.route("/list", methods=["GET"])
 @check_permission("sys:category:list")
 def list():
     """
@@ -45,15 +48,15 @@ def list():
     Returns:
         JSON: 分类列表数据
     """
-    result = category.get_category_list()
+    return R.ok(data=category.get_category_list(request.args))
-    return result
 
 
 # ============================================================
 # 分类详情查询
 # ============================================================
 
-@category_blue.route('/detail/<int:id>', methods=["GET"])
+
+@category_blue.route("/detail/<int:id>", methods=["GET"])
 @check_permission("sys:category:detail")
 def detail(id):
     """
@@ -75,7 +78,8 @@ def detail(id):
 # 分类添加
 # ============================================================
 
-@category_blue.route('/add', methods=["POST"])
+
+@category_blue.route("/add", methods=["POST"])
 @check_permission("sys:category:add")
 @check_demo
 def add():
@@ -98,15 +102,19 @@ def add():
     Returns:
         JSON: 操作结果
     """
-    result = category.add_category()
+    json_data, err = parse_request_body()
-    return result
+    if err:
+        return R.failed(msg=err)
+    category.add_category(json_data)
+    return R.ok(msg="添加成功")
 
 
 # ============================================================
 # 分类更新
 # ============================================================
 
-@category_blue.route('/update', methods=["PUT"])
+
+@category_blue.route("/update", methods=["PUT"])
 @check_permission("sys:category:update")
 @check_demo
 def update():
@@ -130,15 +138,19 @@ def update():
     Returns:
         JSON: 操作结果
     """
-    result = category.update_category()
+    json_data, err = parse_request_body()
-    return result
+    if err:
+        return R.failed(msg=err)
+    category.update_category(json_data)
+    return R.ok(msg="更新成功")
 
 
 # ============================================================
 # 分类删除
 # ============================================================
 
-@category_blue.route('/delete/<string:id>', methods=["DELETE"])
+
+@category_blue.route("/delete/<string:id>", methods=["DELETE"])
 @check_permission("sys:category:delete")
 @check_demo
 def delete(id):
@@ -151,15 +163,16 @@ def delete(id):
     Returns:
         JSON: 操作结果
     """
-    result = category.delete_categories(id)
+    count = category.delete_categories(id)
-    return result
+    return R.ok(msg=f"本次共删除{count}条数据")
 
 
 # ============================================================
 # 分类数据列表获取
 # ============================================================
 
-@category_blue.route('/getCategoryList', methods=["GET"])
+
+@category_blue.route("/getCategoryList", methods=["GET"])
 def get_category_list():
     """
     获取分类数据列表

src/modules/cms/category/service.py

@@ -23,16 +23,15 @@
 import logging
 
 from datetime import datetime, timezone
-from flask import request
 from sqlalchemy import and_
 
 from src.modules.cms.category.schema import CategoryForm
 from src.modules.cms.category.models import Category
-from src.core.database import db
+from src.modules.cms.category.repository import CategoryRepository
-from src.core import response as R
+from src.core.exceptions import BizError
 from src.common.utils import regular
 from src.common.utils.case_converter import convert_camel_to_snake
-from src.common.utils.request import parse_request_body, parse_id_list, like_contains
+from src.common.utils.request import parse_id_list, like_contains
 from src.common.utils.model_utils import models_to_list, model_to_dict
 from src.core.security import get_username
 
@@ -44,38 +43,32 @@ logger = logging.getLogger(__name__)
 # 分类列表查询
 # ============================================================
 
-def get_category_list():
+
+def get_category_list(params: dict) -> list:
     """
     查询分类列表（无分页）
 
     用于分类管理页面的列表展示，支持按名称模糊搜索。
 
     Args:
-        request: Flask request对象，包含查询参数
+        params: 查询参数
             - category_name: 分类名称（可选，模糊查询）
             - type: 分类类型（可选，精确查询）
 
     Returns:
-        R: 包含分类列表的JSON响应
+        list: 分类字典列表
     """
-    try:
     # 应用筛选条件
-        query = _apply_filters()
+    query = _apply_filters(params)
 
     # 查询数据
     category_list = query.all()
 
     # 转换为字典列表
-        records = models_to_list(category_list)
+    return models_to_list(category_list)
-
-        return R.ok(data=records)
-
-    except Exception as e:
-        logger.error(f"查询分类列表异常: {str(e)}")
-        return R.failed(msg="查询失败，请稍后重试")
 
 
-def _apply_filters():
+def _apply_filters(params: dict):
     """
     应用查询筛选条件
 
@@ -91,12 +84,12 @@ def _apply_filters():
     query = Category.query.filter(Category.is_delete == 0)
 
     # 分类名称模糊筛选
-    category_name = request.args.get('category_name')
+    category_name = params.get("category_name")
     if category_name:
-        query = query.filter(Category.category_name.like(like_contains(category_name), escape='\\'))
+        query = query.filter(Category.category_name.like(like_contains(category_name), escape="\\"))
 
     # 分类类型精确筛选
-    type = request.args.get('type')
+    type = params.get("type")
     if type:
         query = query.filter(Category.type == type)
 
@@ -108,6 +101,7 @@ def _apply_filters():
 # 分类详情查询
 # ============================================================
 
+
 def get_category_detail(category_id):
     """
     根据ID查询分类详情（返回字典格式）
@@ -138,6 +132,7 @@ def get_category_detail(category_id):
 # 分类信息查询（根据ID）- 核心查询方法
 # ============================================================
 
+
 def get_info(category_id: int):
     """
     根据ID查询分类信息（返回模型对象）
@@ -163,26 +158,10 @@ def get_info(category_id: int):
         if category:
             print(category.category_name)
     """
-    # 参数有效性验证
+    return CategoryRepository.get_by_id(category_id)
-    if not category_id or category_id <= 0:
-        return None
-
-    try:
-        # 查询未删除的分类
-        category = Category.query.filter(
-            and_(Category.id == category_id, Category.is_delete == 0)
-        ).first()
-        return category
-    except Exception as e:
-        logger.error(f"查询分类信息异常，ID: {category_id}, 错误: {str(e)}")
-        return None
 
 
-# ============================================================
+def add_category(data: dict) -> None:
-# 分类添加
-# ============================================================
-
-def add_category():
     """
     添加分类
 
@@ -194,35 +173,24 @@ def add_category():
     Returns:
         R: 操作结果
     """
-    try:
-        # 获取参数
-        json_data, error = parse_request_body()
-        if error:
-            return R.failed(msg=error)
 
     # 表单验证
-        form = CategoryForm(data=json_data)
+    form = CategoryForm(data=data)
     if not form.validate():
-            err_msg = regular.get_err(form)
+        raise BizError(regular.get_err(form))
-            return R.failed(msg=err_msg)
 
     # 创建分类
     category = Category(**form.data)
     category.create_user = get_username()
     category.save()
 
-        return R.ok(msg="添加成功")
-
-    except Exception as e:
-        logger.error(f"添加分类异常: {str(e)}")
-        return R.failed(msg="添加分类失败，请稍后重试")
-
 
 # ============================================================
 # 分类更新
 # ============================================================
 
-def update_category():
+
+def update_category(data: dict) -> None:
     """
     更新分类
 
@@ -235,52 +203,42 @@ def update_category():
     Returns:
         R: 操作结果
     """
-    try:
-        # 获取参数
-        json_data, error = parse_request_body()
-        if error:
-            return R.failed(msg=error)
 
     # 表单验证
-        form = CategoryForm(data=json_data)
+    form = CategoryForm(data=data)
     if not form.validate():
-            return R.failed(msg=regular.get_err(form))
+        raise BizError(regular.get_err(form))
 
     # 获取分类ID
-        category_id = form.data.get('id')
+    category_id = form.data.get("id")
     if not category_id or int(category_id) <= 0:
-            return R.failed("分类ID无效")
+        raise BizError("分类ID无效")
     category_id = int(category_id)
 
     # 复用 get_info 方法验证分类是否存在
     category = get_info(category_id)
     if not category:
-            return R.failed("分类不存在")
+        raise BizError("分类不存在")
 
     # 更新分类信息（移除ID，避免更新主键）
     form_data = convert_camel_to_snake(form.data.copy())
-        form_data.pop('id', None)
+    form_data.pop("id", None)
 
-        form_data['update_time'] = datetime.now(tz=timezone.utc)
+    form_data["update_time"] = datetime.now(tz=timezone.utc)
-        form_data['update_user'] = get_username()
+    form_data["update_user"] = get_username()
 
     result = Category.query.filter_by(id=category_id).update(form_data)
 
     if not result:
-            return R.failed("更新失败")
+        raise BizError("更新失败")
-
-        return R.ok(msg="更新成功")
-
-    except Exception as e:
-        logger.error(f"更新分类异常: {str(e)}")
-        return R.failed(msg="更新分类失败，请稍后重试")
 
 
 # ============================================================
 # 分类删除
 # ============================================================
 
-def delete_categories(category_ids):
+
+def delete_categories(category_ids: str) -> int:
     """
     删除分类（批量删除）
 
@@ -294,28 +252,30 @@ def delete_categories(category_ids):
         R: 操作结果
     """
     if not category_ids:
-        return R.failed("记录ID不存在")
+        raise BizError("记录ID不存在")
 
     # 解析ID列表
     id_list = parse_id_list(category_ids)
     if not id_list:
-        return R.failed("无效的记录ID")
+        raise BizError("无效的记录ID")
 
     # 检查是否有子分类
     for category_id in id_list:
-        child_count = Category.query.filter(
+        child_count = Category.query.filter(and_(Category.parent_id == category_id, Category.is_delete == 0)).count()
-            and_(Category.parent_id == category_id, Category.is_delete == 0)
-        ).count()
         if child_count > 0:
-            return R.failed("存在子分类，请先删除子分类")
+            raise BizError("存在子分类，请先删除子分类")
 
-    return Category.batch_delete_r(category_ids)
+    category_ids_list = parse_id_list(category_ids)
+    if not category_ids_list:
+        raise BizError("无效的记录ID")
+    return Category.batch_delete(category_ids_list)
 
 
 # ============================================================
 # 分类数据获取（用于下拉选择、树形展示）
 # ============================================================
 
+
 def get_category_data():
     """
     获取分类数据列表（无分页）
@@ -328,9 +288,7 @@ def get_category_data():
     """
     try:
         # 查询所有未删除的分类
-        category_list = Category.query.filter(
+        category_list = Category.query.filter(Category.is_delete == 0).order_by(Category.sort.asc()).all()
-            Category.is_delete == 0
-        ).order_by(Category.sort.asc()).all()
 
         # 转换为字典列表
         result = models_to_list(category_list)

src/modules/file_template/models.py

@@ -9,7 +9,6 @@ from src.common.base_db import BaseDB
 from src.common.base_model import BaseModel
 from config import DB_PREFIX
 from src.core.database import db
-from src.common.utils.case_converter import convert_camel_to_snake
 
 
 # ============================================================

src/modules/file_template/repository.py

@@ -0,0 +1,15 @@
+"""
+FileTemplate 仓储模块
+=============================================================
+
+数据访问层，继承 BaseRepository 获得通用 CRUD 能力。
+"""
+
+from src.common.base_repository import BaseRepository
+from src.modules.file_template.models import FileTemplate
+
+
+class FileTemplateRepository(BaseRepository):
+    """FileTemplate 仓储"""
+
+    model = FileTemplate

src/modules/file_template/router.py

@@ -14,22 +14,25 @@
 依赖模块： application.services.file_template - 文件模板业务逻辑层
          application.middleware.check_permission - 权限校验中间件
 """
-from flask import Blueprint
+
+from flask import Blueprint, request
 
 from application.middleware.check_demo import check_demo
 from application.middleware.check_permission import check_permission
 from src.modules.file_template import service as file_template
-from src.common.utils.request import parse_batch_ids
+from src.common.utils.request import parse_batch_ids, parse_request_body
-file_template_blue = Blueprint('file_template', __name__, url_prefix='/file/template')
 
 from src.core import response as R
 
+file_template_blue = Blueprint("file_template", __name__, url_prefix="/file/template")
+
 
 # ============================================================
 # 文件模板分页查询
 # ============================================================
 
-@file_template_blue.route('/page', methods=["GET"])
+
+@file_template_blue.route("/page", methods=["GET"])
 @check_permission("sys:filetemplate:page")
 def page():
     """
@@ -41,15 +44,15 @@ def page():
     Returns:
         JSON: 文件模板分页数据，包含 records（数据列表）、total（总记录数）等
     """
-    result = file_template.get_file_template_page()
+    return R.ok(data=file_template.get_file_template_page(request.args))
-    return result
 
 
 # ============================================================
 # 文件模板详情查询
 # ============================================================
 
-@file_template_blue.route('/detail/<int:id>', methods=["GET"])
+
+@file_template_blue.route("/detail/<int:id>", methods=["GET"])
 @check_permission("sys:filetemplate:detail")
 def detail(id):
     """
@@ -71,7 +74,8 @@ def detail(id):
 # 文件模板添加
 # ============================================================
 
-@file_template_blue.route('/add', methods=["POST"])
+
+@file_template_blue.route("/add", methods=["POST"])
 @check_permission("sys:filetemplate:add")
 @check_demo
 def add():
@@ -84,15 +88,19 @@ def add():
     Returns:
         JSON: 操作结果
     """
-    result = file_template.add_file_template()
+    json_data, err = parse_request_body()
-    return result
+    if err:
+        return R.failed(msg=err)
+    file_template.add_file_template(json_data)
+    return R.ok(msg="添加成功")
 
 
 # ============================================================
 # 文件模板更新
 # ============================================================
 
-@file_template_blue.route('/update', methods=["PUT"])
+
+@file_template_blue.route("/update", methods=["PUT"])
 @check_permission("sys:filetemplate:update")
 @check_demo
 def update():
@@ -105,15 +113,19 @@ def update():
     Returns:
         JSON: 操作结果
     """
-    result = file_template.update_file_template()
+    json_data, err = parse_request_body()
-    return result
+    if err:
+        return R.failed(msg=err)
+    file_template.update_file_template(json_data)
+    return R.ok(msg="更新成功")
 
 
 # ============================================================
 # 文件模板删除
 # ============================================================
 
-@file_template_blue.route('/delete/<string:id>', methods=["DELETE"])
+
+@file_template_blue.route("/delete/<string:id>", methods=["DELETE"])
 @check_permission("sys:filetemplate:delete")
 @check_demo
 def delete(id):
@@ -126,15 +138,16 @@ def delete(id):
     Returns:
         JSON: 操作结果
     """
-    result = file_template.delete_file_templates(id)
+    count = file_template.delete_file_templates(id)
-    return result
+    return R.ok(msg=f"本次共删除{count}条数据")
 
 
 # ============================================================
 # 文件模板批量删除
 # ============================================================
 
-@file_template_blue.route('/batchDelete', methods=["DELETE"])
+
+@file_template_blue.route("/batchDelete", methods=["DELETE"])
 @check_permission("sys:filetemplate:batch_delete")
 @check_demo
 def batch_delete():

src/modules/file_template/service.py

@@ -23,18 +23,16 @@
 import logging
 
 from datetime import datetime, timezone
-from flask import request
-from sqlalchemy import and_
 
 from src.common.constants.common import PAGE_SIZE
 from src.modules.file_template.schema import FileTemplateForm
 from src.modules.file_template.models import FileTemplate
-from src.core.database import db
+from src.modules.file_template.repository import FileTemplateRepository
-from src.core import response as R
+from src.core.exceptions import BizError
 from src.common.utils import regular
 from src.common.utils.case_converter import convert_camel_to_snake
 from src.common.utils.model_utils import models_to_list, model_to_dict
-from src.common.utils.request import parse_request_body, parse_id_list, like_contains
+from src.common.utils.request import parse_id_list, like_contains
 from src.common.utils.rich_text import save_file, get_file_url
 from src.core.security import get_username
 
@@ -46,7 +44,8 @@ logger = logging.getLogger(__name__)
 # 文件模板分页查询
 # ============================================================
 
-def get_file_template_page():
+
+def get_file_template_page(params: dict) -> dict:
     """
     查询文件模板分页数据
 
@@ -67,13 +66,12 @@ def get_file_template_page():
                 "pages": 总页数
             }
     """
-    try:
     # 获取分页参数
-        page_no = int(request.args.get('pageNo', 1))
+    page_no = int(params.get("pageNo", 1))
-        page_size = int(request.args.get('pageSize', PAGE_SIZE))
+    page_size = int(params.get("pageSize", PAGE_SIZE))
 
     # 应用筛选条件
-        query = _apply_filters()
+    query = _apply_filters(params)
 
     # 记录总数
     total = query.count()
@@ -84,14 +82,10 @@ def get_file_template_page():
     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)
+    return page_data
-
-    except Exception as e:
-        logger.error(f"查询文件模板分页数据异常: {str(e)}")
-        return R.failed(msg="查询失败，请稍后重试")
 
 
-def _apply_filters():
+def _apply_filters(params: dict):
     """
     应用查询筛选条件
 
@@ -108,14 +102,14 @@ def _apply_filters():
     query = FileTemplate.query.filter(FileTemplate.is_delete == 0)
 
     # 模板名称模糊筛选
-    name = request.args.get('name')
+    name = params.get("name")
     if name:
-        query = query.filter(FileTemplate.name.like(like_contains(name), escape='\\'))
+        query = query.filter(FileTemplate.name.like(like_contains(name), escape="\\"))
 
     # 模板编码模糊筛选
-    code = request.args.get('code')
+    code = params.get("code")
     if code:
-        query = query.filter(FileTemplate.code.like(like_contains(code), escape='\\'))
+        query = query.filter(FileTemplate.code.like(like_contains(code), escape="\\"))
 
     # 按排序值升序排列
     return query.order_by(FileTemplate.sort.asc())
@@ -138,19 +132,14 @@ def _build_page_data(template_list, total, page_size, page_no, pages):
     # 将模型对象列表转换为字典列表
     records = models_to_list(template_list)
 
-    return {
+    return {"records": records, "total": total, "size": page_size, "current": page_no, "pages": pages}
-        'records': records,
-        'total': total,
-        'size': page_size,
-        'current': page_no,
-        'pages': pages
-    }
 
 
 # ============================================================
 # 文件模板详情查询
 # ============================================================
 
+
 def get_file_template_detail(template_id):
     """
     根据ID查询文件模板详情（返回字典格式）
@@ -172,7 +161,7 @@ def get_file_template_detail(template_id):
         # 转换为字典格式
         data = model_to_dict(template)
         # 头像URL
-        data['filePath'] = get_file_url(data.get('filePath'))
+        data["filePath"] = get_file_url(data.get("filePath"))
         # 返回结果
         return data
 
@@ -185,6 +174,7 @@ def get_file_template_detail(template_id):
 # 文件模板信息查询（根据ID）- 核心查询方法
 # ============================================================
 
+
 def get_info(template_id: int):
     """
     根据ID查询文件模板信息（返回模型对象）
@@ -197,26 +187,10 @@ def get_info(template_id: int):
     Returns:
         FileTemplate/None: 文件模板模型对象，不存在或已删除时返回 None
     """
-    # 参数有效性验证
+    return FileTemplateRepository.get_by_id(template_id)
-    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(data: dict) -> None:
-# 文件模板添加
-# ============================================================
-
-def add_file_template():
     """
     添加文件模板
 
@@ -228,20 +202,14 @@ def add_file_template():
     Returns:
         R: 操作结果
     """
-    try:
-        # 获取参数
-        json_data, error = parse_request_body()
-        if error:
-            return R.failed(msg=error)
 
     # 表单验证
-        form = FileTemplateForm(data=json_data)
+    form = FileTemplateForm(data=data)
     if not form.validate():
-            err_msg = regular.get_err(form)
+        raise BizError(regular.get_err(form))
-            return R.failed(msg=err_msg)
 
     # 文件处理
-        file_path = form.data.get('filePath')
+    file_path = form.data.get("filePath")
     if file_path:
         form.filePath.data = save_file(file_path, "file_template")
 
@@ -250,18 +218,13 @@ def add_file_template():
     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():
+
+def update_file_template(data: dict) -> None:
     """
     更新文件模板
 
@@ -274,57 +237,47 @@ def update_file_template():
     Returns:
         R: 操作结果
     """
-    try:
-        # 获取参数
-        json_data, error = parse_request_body()
-        if error:
-            return R.failed(msg=error)
 
     # 表单验证
-        form = FileTemplateForm(data=json_data)
+    form = FileTemplateForm(data=data)
     if not form.validate():
-            return R.failed(msg=regular.get_err(form))
+        raise BizError(regular.get_err(form))
 
     # 获取模板ID
-        template_id = form.data.get('id')
+    template_id = form.data.get("id")
     if not template_id or int(template_id) <= 0:
-            return R.failed("模板ID无效")
+        raise BizError("模板ID无效")
     template_id = int(template_id)
 
     # 复用 get_info 方法验证模板是否存在
     template = get_info(template_id)
     if not template:
-            return R.failed("模板不存在")
+        raise BizError("模板不存在")
 
     # 文件处理
-        file_path = form.data.get('filePath')
+    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.pop("id", None)
 
-        form_data['update_time'] = datetime.now(tz=timezone.utc)
+    form_data["update_time"] = datetime.now(tz=timezone.utc)
-        form_data['update_user'] = get_username()
+    form_data["update_user"] = get_username()
 
     result = FileTemplate.query.filter_by(id=template_id).update(form_data)
 
     if not result:
-            return R.failed("更新失败")
+        raise BizError("更新失败")
-
-        return R.ok(msg="更新成功")
-
-    except Exception as e:
-        logger.error(f"更新文件模板异常: {str(e)}")
-        return R.failed(msg="更新文件模板失败，请稍后重试")
 
 
 # ============================================================
 # 文件模板删除
 # ============================================================
 
-def delete_file_templates(template_ids):
+
+def delete_file_templates(template_ids: str) -> int:
     """
     删除文件模板（批量删除）
 
@@ -337,4 +290,7 @@ def delete_file_templates(template_ids):
     Returns:
         R: 操作结果
     """
-    return FileTemplate.batch_delete_r(template_ids)
+    template_ids_list = parse_id_list(template_ids)
+    if not template_ids_list:
+        raise BizError("无效的记录ID")
+    return FileTemplate.batch_delete(template_ids_list)