From 8c5b39d03b3e3ca7fd25fc8476532082f028d9cc Mon Sep 17 00:00:00 2001
From: HuangHuiKang <HuangHuiKang@qq.com>
Date: Tue, 7 Jul 2026 16:07:26 +0800
Subject: [PATCH] =?UTF-8?q?feat:=E6=96=B0=E5=A2=9E=E8=A7=84=E5=88=99?=
 =?UTF-8?q?=E5=88=97=E8=A1=A8=E5=8F=8A=E8=A7=84=E5=88=99=E5=88=86=E7=B1=BB?=
 =?UTF-8?q?=E6=8E=A5=E5=8F=A3?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 .gitignore                                    |  14 ++
 .idea/claudeCodeTabState.xml                  |   2 +-
 app/api/v1/module_rule/__init__.py            |   9 ++
 app/api/v1/module_rule/rule/__init__.py       |   3 +
 app/api/v1/module_rule/rule/controller.py     |  95 ++++++++++++++
 app/api/v1/module_rule/rule/crud.py           |  12 ++
 app/api/v1/module_rule/rule/model.py          |  34 +++++
 app/api/v1/module_rule/rule/schema.py         |  75 +++++++++++
 app/api/v1/module_rule/rule/service.py        |  85 ++++++++++++
 .../v1/module_rule/rule_category/__init__.py  |   3 +
 .../module_rule/rule_category/controller.py   | 115 +++++++++++++++++
 app/api/v1/module_rule/rule_category/crud.py  |  12 ++
 app/api/v1/module_rule/rule_category/model.py |  44 +++++++
 .../v1/module_rule/rule_category/schema.py    |  72 +++++++++++
 .../v1/module_rule/rule_category/service.py   | 122 ++++++++++++++++++
 app/init_app.py                               |   2 +
 rule_module_api.txt                           |  99 ++++++++++++++
 17 files changed, 797 insertions(+), 1 deletion(-)
 create mode 100644 .gitignore
 create mode 100644 app/api/v1/module_rule/__init__.py
 create mode 100644 app/api/v1/module_rule/rule/__init__.py
 create mode 100644 app/api/v1/module_rule/rule/controller.py
 create mode 100644 app/api/v1/module_rule/rule/crud.py
 create mode 100644 app/api/v1/module_rule/rule/model.py
 create mode 100644 app/api/v1/module_rule/rule/schema.py
 create mode 100644 app/api/v1/module_rule/rule/service.py
 create mode 100644 app/api/v1/module_rule/rule_category/__init__.py
 create mode 100644 app/api/v1/module_rule/rule_category/controller.py
 create mode 100644 app/api/v1/module_rule/rule_category/crud.py
 create mode 100644 app/api/v1/module_rule/rule_category/model.py
 create mode 100644 app/api/v1/module_rule/rule_category/schema.py
 create mode 100644 app/api/v1/module_rule/rule_category/service.py
 create mode 100644 rule_module_api.txt

diff --git a/.gitignore b/.gitignore
new file mode 100644
index 0000000..9ea6a47
--- /dev/null
+++ b/.gitignore
@@ -0,0 +1,14 @@
+.venv/
+__pycache__/
+*.py[cod]
+*$py.class
+
+.pytest_cache/
+.ruff_cache/
+.mypy_cache/
+
+logs/
+*.log
+
+env/.env.dev
+env/.env.prod
diff --git a/.idea/claudeCodeTabState.xml b/.idea/claudeCodeTabState.xml
index 32f92d5..0eb1afb 100644
--- a/.idea/claudeCodeTabState.xml
+++ b/.idea/claudeCodeTabState.xml
@@ -10,7 +10,7 @@
               <option name="sessionId" value="019f2248-8e43-7ea3-81b1-1a0a3d35311b" />
               <option name="cwd" value="$PROJECT_DIR$" />
               <option name="model" value="gpt-5.5" />
-              <option name="permissionMode" value="default" />
+              <option name="permissionMode" value="acceptEdits" />
               <option name="reasoningEffort" value="high" />
             </TabSessionState>
           </value>
diff --git a/app/api/v1/module_rule/__init__.py b/app/api/v1/module_rule/__init__.py
new file mode 100644
index 0000000..4056710
--- /dev/null
+++ b/app/api/v1/module_rule/__init__.py
@@ -0,0 +1,9 @@
+from fastapi import APIRouter
+
+from app.api.v1.module_rule.rule.controller import RuleRouter
+from app.api.v1.module_rule.rule_category.controller import RuleCategoryRouter
+
+rule_router = APIRouter(prefix="/rule")
+
+rule_router.include_router(RuleRouter)
+rule_router.include_router(RuleCategoryRouter)
diff --git a/app/api/v1/module_rule/rule/__init__.py b/app/api/v1/module_rule/rule/__init__.py
new file mode 100644
index 0000000..3fcdd42
--- /dev/null
+++ b/app/api/v1/module_rule/rule/__init__.py
@@ -0,0 +1,3 @@
+from .controller import RuleRouter
+
+__all__ = ["RuleRouter"]
diff --git a/app/api/v1/module_rule/rule/controller.py b/app/api/v1/module_rule/rule/controller.py
new file mode 100644
index 0000000..0fb7054
--- /dev/null
+++ b/app/api/v1/module_rule/rule/controller.py
@@ -0,0 +1,95 @@
+from typing import Annotated
+
+from fastapi import APIRouter, Body, Depends, Path
+from fastapi.responses import JSONResponse
+
+from app.common.response import ResponseSchema, SuccessResponse
+from app.core import cache_util
+from app.core.base_params import PaginationQueryParam
+from app.core.base_schema import AuthSchema, PageResultSchema
+from app.core.cache_util import cache
+from app.core.dependencies import AuthPermission
+from app.core.router_class import OperationLogRoute
+
+from .schema import RuleCreateSchema, RuleOutSchema, RuleQueryParam, RuleUpdateSchema
+from .service import RuleService
+
+RuleRouter = APIRouter(route_class=OperationLogRoute, prefix="", tags=["规则列表"])
+
+_RULE_NS = "rule"
+
+
+@RuleRouter.get(
+    "/list",
+    summary="查询规则列表",
+    response_model=ResponseSchema[PageResultSchema[RuleOutSchema]],
+)
+@cache(expire=300, namespace=_RULE_NS)
+async def get_rule_list_controller(
+    page: Annotated[PaginationQueryParam, Depends()],
+    search: Annotated[RuleQueryParam, Depends()],
+    auth: Annotated[AuthSchema, Depends(AuthPermission(["module_rule:rule:query"]))],
+) -> JSONResponse:
+    result = await RuleService(auth).page(
+        page_no=page.page_no,
+        page_size=page.page_size,
+        search=search,
+        order_by=page.order_by or [{"id": "desc"}],
+    )
+    return SuccessResponse(data=result, msg="查询规则列表成功")
+
+
+@RuleRouter.get(
+    "/detail/{id}",
+    summary="查询规则详情",
+    response_model=ResponseSchema[RuleOutSchema],
+)
+async def get_rule_detail_controller(
+    id: Annotated[int, Path(description="规则ID")],
+    auth: Annotated[AuthSchema, Depends(AuthPermission(["module_rule:rule:detail"]))],
+) -> JSONResponse:
+    result = await RuleService(auth).detail(id=id)
+    return SuccessResponse(data=result, msg="查询规则详情成功")
+
+
+@RuleRouter.post(
+    "/create",
+    summary="创建规则",
+    response_model=ResponseSchema[RuleOutSchema],
+)
+async def create_rule_controller(
+    data: RuleCreateSchema,
+    auth: Annotated[AuthSchema, Depends(AuthPermission(["module_rule:rule:create"]))],
+) -> JSONResponse:
+    result = await RuleService(auth).create(data=data)
+    await cache_util.clear(namespace=_RULE_NS)
+    return SuccessResponse(data=result, msg="创建规则成功")
+
+
+@RuleRouter.put(
+    "/update/{id}",
+    summary="修改规则",
+    response_model=ResponseSchema[RuleOutSchema],
+)
+async def update_rule_controller(
+    data: RuleUpdateSchema,
+    id: Annotated[int, Path(description="规则ID")],
+    auth: Annotated[AuthSchema, Depends(AuthPermission(["module_rule:rule:update"]))],
+) -> JSONResponse:
+    result = await RuleService(auth).update(id=id, data=data)
+    await cache_util.clear(namespace=_RULE_NS)
+    return SuccessResponse(data=result, msg="修改规则成功")
+
+
+@RuleRouter.delete(
+    "/delete",
+    summary="删除规则",
+    response_model=ResponseSchema[None],
+)
+async def delete_rule_controller(
+    ids: Annotated[list[int], Body(description="ID列表")],
+    auth: Annotated[AuthSchema, Depends(AuthPermission(["module_rule:rule:delete"]))],
+) -> JSONResponse:
+    await RuleService(auth).delete(ids=ids)
+    await cache_util.clear(namespace=_RULE_NS)
+    return SuccessResponse(msg="删除规则成功")
diff --git a/app/api/v1/module_rule/rule/crud.py b/app/api/v1/module_rule/rule/crud.py
new file mode 100644
index 0000000..7d197aa
--- /dev/null
+++ b/app/api/v1/module_rule/rule/crud.py
@@ -0,0 +1,12 @@
+from app.core.base_crud import CRUDBase
+from app.core.base_schema import AuthSchema
+
+from .model import RuleModel
+from .schema import RuleCreateSchema, RuleUpdateSchema
+
+
+class RuleCRUD(CRUDBase[RuleModel, RuleCreateSchema, RuleUpdateSchema]):
+    """规则列表数据层"""
+
+    def __init__(self, auth: AuthSchema) -> None:
+        super().__init__(model=RuleModel, auth=auth)
diff --git a/app/api/v1/module_rule/rule/model.py b/app/api/v1/module_rule/rule/model.py
new file mode 100644
index 0000000..4c646fc
--- /dev/null
+++ b/app/api/v1/module_rule/rule/model.py
@@ -0,0 +1,34 @@
+from sqlalchemy import ForeignKey, Integer, String, Text, UniqueConstraint
+from sqlalchemy.orm import Mapped, mapped_column, relationship
+
+from app.core.base_model import ModelMixin, TenantMixin, UserMixin
+
+
+class RuleModel(ModelMixin, TenantMixin, UserMixin):
+    """规则模型"""
+
+    __tablename__: str = "rule_list"
+    __table_args__ = (UniqueConstraint("tenant_id", "name"), {"comment": "规则列表表"})
+    __loader_options__: list[str] = [
+        "category",
+        "created_by",
+        "updated_by",
+        "deleted_by",
+        "tenant_by",
+    ]
+
+    name: Mapped[str] = mapped_column(String(128), nullable=False, comment="规则名称", index=True)
+    category_id: Mapped[int | None] = mapped_column(
+        Integer,
+        ForeignKey("rule_category.id", ondelete="SET NULL", onupdate="CASCADE"),
+        default=None,
+        nullable=True,
+        index=True,
+        comment="规则分类ID",
+    )
+    category_name_snapshot: Mapped[str | None] = mapped_column(String(64), default=None, nullable=True, comment="规则分类名称快照")
+    description: Mapped[str | None] = mapped_column(Text, default=None, nullable=True, comment="规则描述")
+    prompt_template: Mapped[str] = mapped_column(Text, nullable=False, comment="提示词模板")
+    status: Mapped[int] = mapped_column(Integer, default=0, nullable=False, comment="状态 0:启用 1:停用", index=True)
+
+    category = relationship("RuleCategoryModel", lazy="selectin", foreign_keys=[category_id], uselist=False)
diff --git a/app/api/v1/module_rule/rule/schema.py b/app/api/v1/module_rule/rule/schema.py
new file mode 100644
index 0000000..056ba2c
--- /dev/null
+++ b/app/api/v1/module_rule/rule/schema.py
@@ -0,0 +1,75 @@
+from dataclasses import dataclass
+
+from fastapi import Query
+from pydantic import BaseModel, ConfigDict, Field, field_validator
+
+from app.common.enums import QueueEnum
+from app.core.base_params import BaseQueryParam, TenantByQueryParam, UserByQueryParam
+from app.core.base_schema import BaseSchema, TenantBySchema, UserBySchema
+
+
+class RuleCreateSchema(BaseModel):
+    """规则创建模型"""
+
+    name: str = Field(..., min_length=1, max_length=128, description="规则名称")
+    category_id: int | None = Field(default=None, ge=0, description="规则分类ID")
+    description: str | None = Field(default=None, max_length=2000, description="规则描述")
+    prompt_template: str = Field(..., min_length=1, description="提示词模板")
+    status: int = Field(default=0, ge=0, le=1, description="状态 0:启用 1:停用")
+
+    @field_validator("name")
+    @classmethod
+    def validate_name(cls, value: str) -> str:
+        value = value.strip()
+        if not value:
+            raise ValueError("规则名称不能为空")
+        return value
+
+    @field_validator("category_id")
+    @classmethod
+    def normalize_category_id(cls, value: int | None) -> int | None:
+        return None if value == 0 else value
+
+    @field_validator("prompt_template")
+    @classmethod
+    def validate_prompt_template(cls, value: str) -> str:
+        value = value.strip()
+        if not value:
+            raise ValueError("提示词模板不能为空")
+        return value
+
+    @field_validator("status")
+    @classmethod
+    def validate_status(cls, value: int) -> int:
+        if value not in {0, 1}:
+            raise ValueError("状态仅支持 0(启用) 或 1(停用)")
+        return value
+
+
+class RuleUpdateSchema(RuleCreateSchema):
+    """规则更新模型"""
+
+
+class RuleOutSchema(RuleCreateSchema, BaseSchema, UserBySchema, TenantBySchema):
+    """规则响应模型"""
+
+    model_config = ConfigDict(from_attributes=True)
+
+    category_name_snapshot: str | None = Field(default=None, max_length=64, description="规则分类名称快照")
+    category_name: str | None = Field(default=None, max_length=64, description="规则分类名称")
+
+
+@dataclass
+class RuleQueryParam(BaseQueryParam, UserByQueryParam, TenantByQueryParam):
+    """规则查询参数"""
+
+    name: str | None = Query(None, description="规则名称")
+    category_id: int | None = Query(None, ge=0, description="规则分类ID")
+    status: int | None = Query(None, ge=0, le=1, description="状态 0:启用 1:停用")
+
+    def __post_init__(self) -> None:
+        self.name = (QueueEnum.like.value, self.name)
+        if isinstance(self.category_id, int):
+            self.category_id = (QueueEnum.eq.value, None if self.category_id == 0 else self.category_id)
+        if isinstance(self.status, int):
+            self.status = (QueueEnum.eq.value, self.status)
diff --git a/app/api/v1/module_rule/rule/service.py b/app/api/v1/module_rule/rule/service.py
new file mode 100644
index 0000000..7060e11
--- /dev/null
+++ b/app/api/v1/module_rule/rule/service.py
@@ -0,0 +1,85 @@
+from app.api.v1.module_rule.rule_category.crud import RuleCategoryCRUD
+from app.core.base_schema import AuthSchema
+from app.core.exceptions import CustomException
+
+from .crud import RuleCRUD
+from .schema import RuleCreateSchema, RuleOutSchema, RuleQueryParam, RuleUpdateSchema
+
+
+class RuleService:
+    """规则列表服务"""
+
+    def __init__(self, auth: AuthSchema) -> None:
+        self.auth = auth
+
+    async def detail(self, id: int) -> RuleOutSchema:
+        rule = await RuleCRUD(self.auth).get_or_404(id=id)
+        return self._to_out_schema(rule)
+
+    async def page(
+        self,
+        page_no: int,
+        page_size: int,
+        search: RuleQueryParam | None = None,
+        order_by: list[dict[str, str]] | None = None,
+    ) -> dict:
+        offset = (page_no - 1) * page_size
+        page_result = await RuleCRUD(self.auth).page(
+            offset=offset,
+            limit=page_size,
+            order_by=order_by or [{"id": "desc"}],
+            search=vars(search) if search else None,
+            out_schema=RuleOutSchema,
+        )
+        for item in page_result.items:
+            item["category_name"] = item.get("category_name_snapshot")
+        return page_result
+
+    async def get_list(
+        self,
+        search: RuleQueryParam | None = None,
+        order_by: list[dict] | None = None,
+    ) -> list[RuleOutSchema]:
+        rule_list = await RuleCRUD(self.auth).get_list(search=vars(search) if search else None, order_by=order_by or [{"id": "desc"}])
+        return [self._to_out_schema(rule) for rule in rule_list]
+
+    async def create(self, data: RuleCreateSchema) -> RuleOutSchema:
+        await self._validate_unique_name(name=data.name)
+        category_name_snapshot = await self._get_category_name_snapshot(category_id=data.category_id)
+        rule = await RuleCRUD(self.auth).create(data={**data.model_dump(), "category_name_snapshot": category_name_snapshot})
+        return self._to_out_schema(rule)
+
+    async def update(self, id: int, data: RuleUpdateSchema) -> RuleOutSchema:
+        await RuleCRUD(self.auth).get_or_404(id=id, msg="更新失败，该规则不存在")
+        await self._validate_unique_name(name=data.name, exclude_id=id)
+        category_name_snapshot = await self._get_category_name_snapshot(category_id=data.category_id)
+        rule = await RuleCRUD(self.auth).update(id=id, data={**data.model_dump(), "category_name_snapshot": category_name_snapshot})
+        return self._to_out_schema(rule)
+
+    async def delete(self, ids: list[int]) -> None:
+        if len(ids) < 1:
+            raise CustomException(msg="删除失败，删除对象不能为空")
+        rules = await RuleCRUD(self.auth).get_list(search={"id": ("in", ids)})
+        rule_map = {rule.id: rule for rule in rules}
+        for rule_id in ids:
+            if rule_id not in rule_map:
+                raise CustomException(msg="删除失败，该规则不存在")
+        await RuleCRUD(self.auth).delete(ids=ids)
+
+    def _to_out_schema(self, rule) -> RuleOutSchema:
+        rule_out = RuleOutSchema.model_validate(rule)
+        rule_out.category_name = rule_out.category_name_snapshot
+        return rule_out
+
+    async def _validate_unique_name(self, name: str, exclude_id: int | None = None) -> None:
+        rule = await RuleCRUD(self.auth).get(name=name)
+        if rule and rule.id != exclude_id:
+            raise CustomException(msg="规则名称已存在")
+
+    async def _get_category_name_snapshot(self, category_id: int | None) -> str | None:
+        if category_id is None:
+            return None
+        category = await RuleCategoryCRUD(self.auth).get(id=category_id)
+        if not category:
+            raise CustomException(msg="规则分类不存在")
+        return category.name
diff --git a/app/api/v1/module_rule/rule_category/__init__.py b/app/api/v1/module_rule/rule_category/__init__.py
new file mode 100644
index 0000000..209d7c8
--- /dev/null
+++ b/app/api/v1/module_rule/rule_category/__init__.py
@@ -0,0 +1,3 @@
+from .controller import RuleCategoryRouter
+
+__all__ = ["RuleCategoryRouter"]
diff --git a/app/api/v1/module_rule/rule_category/controller.py b/app/api/v1/module_rule/rule_category/controller.py
new file mode 100644
index 0000000..dade1a0
--- /dev/null
+++ b/app/api/v1/module_rule/rule_category/controller.py
@@ -0,0 +1,115 @@
+from typing import Annotated
+
+from fastapi import APIRouter, Body, Depends, Path
+from fastapi.responses import JSONResponse
+
+from app.common.response import ResponseSchema, SuccessResponse
+from app.core import cache_util
+from app.core.base_params import PaginationQueryParam
+from app.core.base_schema import AuthSchema, PageResultSchema
+from app.core.cache_util import cache
+from app.core.dependencies import AuthPermission
+from app.core.router_class import OperationLogRoute
+
+from .schema import (
+    RuleCategoryCreateSchema,
+    RuleCategoryOutSchema,
+    RuleCategoryQueryParam,
+    RuleCategoryTreeOutSchema,
+    RuleCategoryUpdateSchema,
+)
+from .service import RuleCategoryService
+
+RuleCategoryRouter = APIRouter(route_class=OperationLogRoute, prefix="/category", tags=["规则分类"])
+
+_RULE_CATEGORY_NS = "rule_category"
+
+
+@RuleCategoryRouter.get(
+    "/list",
+    summary="查询规则分类列表",
+    response_model=ResponseSchema[PageResultSchema[RuleCategoryOutSchema]],
+)
+@cache(expire=300, namespace=_RULE_CATEGORY_NS)
+async def get_rule_category_list_controller(
+    page: Annotated[PaginationQueryParam, Depends()],
+    search: Annotated[RuleCategoryQueryParam, Depends()],
+    auth: Annotated[AuthSchema, Depends(AuthPermission(["module_rule:rule_category:query"]))],
+) -> JSONResponse:
+    result = await RuleCategoryService(auth).page(
+        page_no=page.page_no,
+        page_size=page.page_size,
+        search=search,
+        order_by=page.order_by or [{"order": "asc"}, {"id": "asc"}],
+    )
+    return SuccessResponse(data=result, msg="查询规则分类列表成功")
+
+
+@RuleCategoryRouter.get(
+    "/tree",
+    summary="查询规则分类树",
+    response_model=ResponseSchema[list[RuleCategoryTreeOutSchema]],
+)
+@cache(expire=300, namespace=_RULE_CATEGORY_NS)
+async def get_rule_category_tree_controller(
+    search: Annotated[RuleCategoryQueryParam, Depends()],
+    auth: Annotated[AuthSchema, Depends(AuthPermission(["module_rule:rule_category:query"]))],
+) -> JSONResponse:
+    result = await RuleCategoryService(auth).tree(search=search)
+    return SuccessResponse(data=result, msg="查询规则分类树成功")
+
+
+@RuleCategoryRouter.get(
+    "/detail/{id}",
+    summary="查询规则分类详情",
+    response_model=ResponseSchema[RuleCategoryOutSchema],
+)
+async def get_rule_category_detail_controller(
+    id: Annotated[int, Path(description="规则分类ID")],
+    auth: Annotated[AuthSchema, Depends(AuthPermission(["module_rule:rule_category:detail"]))],
+) -> JSONResponse:
+    result = await RuleCategoryService(auth).detail(id=id)
+    return SuccessResponse(data=result, msg="查询规则分类详情成功")
+
+
+@RuleCategoryRouter.post(
+    "/create",
+    summary="创建规则分类",
+    response_model=ResponseSchema[RuleCategoryOutSchema],
+)
+async def create_rule_category_controller(
+    data: RuleCategoryCreateSchema,
+    auth: Annotated[AuthSchema, Depends(AuthPermission(["module_rule:rule_category:create"]))],
+) -> JSONResponse:
+    result = await RuleCategoryService(auth).create(data=data)
+    await cache_util.clear(namespace=_RULE_CATEGORY_NS)
+    return SuccessResponse(data=result, msg="创建规则分类成功")
+
+
+@RuleCategoryRouter.put(
+    "/update/{id}",
+    summary="修改规则分类",
+    response_model=ResponseSchema[RuleCategoryOutSchema],
+)
+async def update_rule_category_controller(
+    data: RuleCategoryUpdateSchema,
+    id: Annotated[int, Path(description="规则分类ID")],
+    auth: Annotated[AuthSchema, Depends(AuthPermission(["module_rule:rule_category:update"]))],
+) -> JSONResponse:
+    result = await RuleCategoryService(auth).update(id=id, data=data)
+    await cache_util.clear(namespace=_RULE_CATEGORY_NS)
+    return SuccessResponse(data=result, msg="修改规则分类成功")
+
+
+@RuleCategoryRouter.delete(
+    "/delete",
+    summary="删除规则分类",
+    response_model=ResponseSchema[None],
+)
+async def delete_rule_category_controller(
+    ids: Annotated[list[int], Body(description="ID列表")],
+    auth: Annotated[AuthSchema, Depends(AuthPermission(["module_rule:rule_category:delete"]))],
+) -> JSONResponse:
+    await RuleCategoryService(auth).delete(ids=ids)
+    await cache_util.clear(namespace=_RULE_CATEGORY_NS)
+    return SuccessResponse(msg="删除规则分类成功")
diff --git a/app/api/v1/module_rule/rule_category/crud.py b/app/api/v1/module_rule/rule_category/crud.py
new file mode 100644
index 0000000..ce74236
--- /dev/null
+++ b/app/api/v1/module_rule/rule_category/crud.py
@@ -0,0 +1,12 @@
+from app.core.base_crud import CRUDBase
+from app.core.base_schema import AuthSchema
+
+from .model import RuleCategoryModel
+from .schema import RuleCategoryCreateSchema, RuleCategoryUpdateSchema
+
+
+class RuleCategoryCRUD(CRUDBase[RuleCategoryModel, RuleCategoryCreateSchema, RuleCategoryUpdateSchema]):
+    """规则分类数据层"""
+
+    def __init__(self, auth: AuthSchema) -> None:
+        super().__init__(model=RuleCategoryModel, auth=auth)
diff --git a/app/api/v1/module_rule/rule_category/model.py b/app/api/v1/module_rule/rule_category/model.py
new file mode 100644
index 0000000..1fa8a0b
--- /dev/null
+++ b/app/api/v1/module_rule/rule_category/model.py
@@ -0,0 +1,44 @@
+from sqlalchemy import ForeignKey, Integer, String, Text, UniqueConstraint
+from sqlalchemy.orm import Mapped, mapped_column, relationship
+
+from app.core.base_model import ModelMixin, TenantMixin, UserMixin
+
+
+class RuleCategoryModel(ModelMixin, TenantMixin, UserMixin):
+    """规则分类模型"""
+
+    __tablename__: str = "rule_category"
+    __table_args__ = (UniqueConstraint("tenant_id", "parent_id", "name"), {"comment": "规则分类表"})
+    __tree_children_attr__: str = "children"
+    __loader_options__: list[str] = [
+        "children",
+        "created_by",
+        "updated_by",
+        "deleted_by",
+        "tenant_by",
+    ]
+
+    name: Mapped[str] = mapped_column(String(64), nullable=False, comment="分类名称", index=True)
+    parent_id: Mapped[int | None] = mapped_column(
+        Integer,
+        ForeignKey("rule_category.id", ondelete="SET NULL", onupdate="CASCADE"),
+        default=None,
+        nullable=True,
+        index=True,
+        comment="父级分类ID",
+    )
+    order: Mapped[int] = mapped_column(Integer, nullable=False, default=999, comment="显示排序")
+    status: Mapped[int] = mapped_column(Integer, default=0, nullable=False, comment="状态 0:启用 1:停用", index=True)
+    description: Mapped[str | None] = mapped_column(Text, default=None, nullable=True, comment="描述")
+
+    parent: Mapped["RuleCategoryModel | None"] = relationship(
+        back_populates="children",
+        remote_side="RuleCategoryModel.id",
+        foreign_keys=[parent_id],
+        uselist=False,
+    )
+    children: Mapped[list["RuleCategoryModel"]] = relationship(
+        back_populates="parent",
+        foreign_keys=[parent_id],
+        lazy="selectin",
+    )
diff --git a/app/api/v1/module_rule/rule_category/schema.py b/app/api/v1/module_rule/rule_category/schema.py
new file mode 100644
index 0000000..61d5282
--- /dev/null
+++ b/app/api/v1/module_rule/rule_category/schema.py
@@ -0,0 +1,72 @@
+from dataclasses import dataclass
+
+from fastapi import Query
+from pydantic import BaseModel, ConfigDict, Field, field_validator
+
+from app.common.enums import QueueEnum
+from app.core.base_params import BaseQueryParam, TenantByQueryParam, UserByQueryParam
+from app.core.base_schema import BaseSchema, TenantBySchema, UserBySchema
+
+
+class RuleCategoryCreateSchema(BaseModel):
+    """规则分类创建模型"""
+
+    name: str = Field(..., min_length=1, max_length=64, description="分类名称")
+    parent_id: int | None = Field(default=None, ge=0, description="父级分类ID")
+    order: int = Field(default=999, ge=0, description="显示排序")
+    status: int = Field(default=0, ge=0, le=1, description="状态 0:启用 1:停用")
+    description: str | None = Field(default=None, max_length=500, description="描述")
+
+    @field_validator("name")
+    @classmethod
+    def validate_name(cls, value: str) -> str:
+        value = value.strip()
+        if not value:
+            raise ValueError("分类名称不能为空")
+        return value
+
+    @field_validator("parent_id")
+    @classmethod
+    def normalize_parent_id(cls, value: int | None) -> int | None:
+        return None if value == 0 else value
+
+    @field_validator("status")
+    @classmethod
+    def validate_status(cls, value: int) -> int:
+        if value not in {0, 1}:
+            raise ValueError("状态仅支持 0(启用) 或 1(停用)")
+        return value
+
+
+class RuleCategoryUpdateSchema(RuleCategoryCreateSchema):
+    """规则分类更新模型"""
+
+
+class RuleCategoryOutSchema(RuleCategoryCreateSchema, BaseSchema, UserBySchema, TenantBySchema):
+    """规则分类响应模型"""
+
+    model_config = ConfigDict(from_attributes=True)
+
+    parent_name: str | None = Field(default=None, max_length=64, description="父级分类名称")
+
+
+class RuleCategoryTreeOutSchema(RuleCategoryOutSchema):
+    """规则分类树响应模型"""
+
+    children: list["RuleCategoryTreeOutSchema"] | None = Field(default=None, description="子分类列表")
+
+
+@dataclass
+class RuleCategoryQueryParam(BaseQueryParam, UserByQueryParam, TenantByQueryParam):
+    """规则分类查询参数"""
+
+    name: str | None = Query(None, description="分类名称")
+    parent_id: int | None = Query(None, ge=0, description="父级分类ID")
+    status: int | None = Query(None, ge=0, le=1, description="状态 0:启用 1:停用")
+
+    def __post_init__(self) -> None:
+        self.name = (QueueEnum.like.value, self.name)
+        if isinstance(self.parent_id, int):
+            self.parent_id = (QueueEnum.eq.value, None if self.parent_id == 0 else self.parent_id)
+        if isinstance(self.status, int):
+            self.status = (QueueEnum.eq.value, self.status)
diff --git a/app/api/v1/module_rule/rule_category/service.py b/app/api/v1/module_rule/rule_category/service.py
new file mode 100644
index 0000000..641ca23
--- /dev/null
+++ b/app/api/v1/module_rule/rule_category/service.py
@@ -0,0 +1,122 @@
+from app.core.base_schema import AuthSchema
+from app.core.exceptions import CustomException
+from app.utils.common_util import get_child_id_map, get_parent_id_map, get_parent_recursion
+from app.api.v1.module_rule.rule.crud import RuleCRUD
+
+from .crud import RuleCategoryCRUD
+from .schema import (
+    RuleCategoryCreateSchema,
+    RuleCategoryOutSchema,
+    RuleCategoryQueryParam,
+    RuleCategoryTreeOutSchema,
+    RuleCategoryUpdateSchema,
+)
+
+
+class RuleCategoryService:
+    """规则分类服务"""
+
+    def __init__(self, auth: AuthSchema) -> None:
+        self.auth = auth
+
+    async def detail(self, id: int) -> RuleCategoryOutSchema:
+        category = await RuleCategoryCRUD(self.auth).get_or_404(id=id)
+        return await self._to_out_schema(category)
+
+    async def get_list(
+        self,
+        search: RuleCategoryQueryParam | None = None,
+        order_by: list[dict] | None = None,
+    ) -> list[RuleCategoryOutSchema]:
+        category_list = await RuleCategoryCRUD(self.auth).get_list(search=vars(search) if search else None, order_by=order_by or [{"order": "asc"}, {"id": "asc"}])
+        return [await self._to_out_schema(category) for category in category_list]
+
+    async def page(
+        self,
+        page_no: int,
+        page_size: int,
+        search: RuleCategoryQueryParam | None = None,
+        order_by: list[dict[str, str]] | None = None,
+    ) -> dict:
+        offset = (page_no - 1) * page_size
+        return await RuleCategoryCRUD(self.auth).page(
+            offset=offset,
+            limit=page_size,
+            order_by=order_by or [{"order": "asc"}, {"id": "asc"}],
+            search=vars(search) if search else None,
+            out_schema=RuleCategoryOutSchema,
+        )
+
+    async def tree(
+        self,
+        search: RuleCategoryQueryParam | None = None,
+        order_by: list[dict] | None = None,
+    ) -> list[dict]:
+        category_list = await RuleCategoryCRUD(self.auth).tree_list(search=vars(search) if search else None, order_by=order_by or [{"order": "asc"}, {"id": "asc"}])
+        category_dict_list = [RuleCategoryTreeOutSchema.model_validate(category).model_dump() for category in category_list]
+        return [category for category in category_dict_list if category.get("parent_id") is None]
+
+    async def create(self, data: RuleCategoryCreateSchema) -> RuleCategoryOutSchema:
+        await self._validate_parent(parent_id=data.parent_id)
+        await self._validate_unique_name(name=data.name, parent_id=data.parent_id)
+        category = await RuleCategoryCRUD(self.auth).create(data=data)
+        return await self._to_out_schema(category)
+
+    async def update(self, id: int, data: RuleCategoryUpdateSchema) -> RuleCategoryOutSchema:
+        await RuleCategoryCRUD(self.auth).get_or_404(id=id, msg="更新失败，该规则分类不存在")
+        await self._validate_parent(parent_id=data.parent_id, current_id=id)
+        await self._validate_unique_name(name=data.name, parent_id=data.parent_id, exclude_id=id)
+        category = await RuleCategoryCRUD(self.auth).update(id=id, data=data)
+        return await self._to_out_schema(category)
+
+    async def delete(self, ids: list[int]) -> None:
+        if len(ids) < 1:
+            raise CustomException(msg="删除失败，删除对象不能为空")
+
+        categories = await RuleCategoryCRUD(self.auth).get_list(search={"id": ("in", ids)})
+        category_map = {category.id: category for category in categories}
+        for category_id in ids:
+            if category_id not in category_map:
+                raise CustomException(msg="删除失败，该规则分类不存在")
+
+        all_categories = await RuleCategoryCRUD(self.auth).get_list()
+        child_id_map = get_child_id_map(model_list=all_categories)
+        for category_id in ids:
+            if child_id_map.get(category_id):
+                raise CustomException(msg="存在子分类，不允许删除父级分类")
+
+        rules = await RuleCRUD(self.auth).get_list(search={"category_id": ("in", ids)})
+        if rules:
+            raise CustomException(msg="分类下存在规则，不允许删除")
+
+        await RuleCategoryCRUD(self.auth).delete(ids=ids)
+
+    async def _to_out_schema(self, category) -> RuleCategoryOutSchema:
+        category_out = RuleCategoryOutSchema.model_validate(category)
+        if category_out.parent_id:
+            parent = await RuleCategoryCRUD(self.auth).get(id=category_out.parent_id)
+            if parent:
+                category_out.parent_name = parent.name
+        return category_out
+
+    async def _validate_parent(self, parent_id: int | None, current_id: int | None = None) -> None:
+        if parent_id is None:
+            return
+        if current_id is not None and parent_id == current_id:
+            raise CustomException(msg="父级分类不能选择自身")
+
+        parent = await RuleCategoryCRUD(self.auth).get(id=parent_id)
+        if not parent:
+            raise CustomException(msg="父级分类不存在")
+
+        if current_id is not None:
+            all_categories = await RuleCategoryCRUD(self.auth).get_list()
+            parent_id_map = get_parent_id_map(model_list=all_categories)
+            parent_ids = get_parent_recursion(id=parent_id, id_map=parent_id_map)
+            if current_id in parent_ids:
+                raise CustomException(msg="父级分类不能选择当前分类的子级")
+
+    async def _validate_unique_name(self, name: str, parent_id: int | None, exclude_id: int | None = None) -> None:
+        category = await RuleCategoryCRUD(self.auth).get(name=name, parent_id=parent_id)
+        if category and category.id != exclude_id:
+            raise CustomException(msg="同一父级下分类名称已存在")
diff --git a/app/init_app.py b/app/init_app.py
index d05e5ed..78098e4 100644
--- a/app/init_app.py
+++ b/app/init_app.py
@@ -96,11 +96,13 @@ def register_routers(app: FastAPI) -> None:
     from app.api.v1.module_common import common_router
     from app.api.v1.module_monitor import monitor_router
     from app.api.v1.module_platform import platform_router
+    from app.api.v1.module_rule import rule_router
     from app.api.v1.module_system import system_router
 
     app.include_router(common_router, dependencies=[Depends(RateLimiter(times=200, seconds=10))])
     app.include_router(monitor_router, dependencies=[Depends(RateLimiter(times=200, seconds=10))])
     app.include_router(platform_router, dependencies=[Depends(RateLimiter(times=200, seconds=10))])
+    app.include_router(rule_router, dependencies=[Depends(RateLimiter(times=200, seconds=10))])
     app.include_router(system_router, dependencies=[Depends(RateLimiter(times=200, seconds=10))])
 
     from app.plugin.module_ai.chat.ws import WS_AI
diff --git a/rule_module_api.txt b/rule_module_api.txt
new file mode 100644
index 0000000..37da289
--- /dev/null
+++ b/rule_module_api.txt
@@ -0,0 +1,99 @@
+规则管理接口说明
+
+规则管理下目前有两个模块：规则分类和规则列表。若部署时配置了 ROOT_PATH=/api/v1，则实际请求地址需要在以下路径前追加 /api/v1，例如 /api/v1/rule/category/list。
+
+统一返回结构：所有接口返回 JSON 格式 { code, msg, data, status_code, success }。分页接口的 data 结构为 { page_no, page_size, total, has_next, items }。
+
+一、规则分类模块
+
+接口前缀：/rule/category
+
+1. 查询规则分类分页列表
+GET /rule/category/list
+权限标识：module_rule:rule_category:query
+Query 参数：page_no、page_size、order_by、name、parent_id、status、created_time、updated_time、created_id、updated_id、tenant_id。
+说明：name 为分类名称模糊查询；parent_id 为父级分类 ID，传 0 会按顶级分类 null 处理；status 为 0 启用、1 停用。
+返回：data 为分页结构，items 中每项为规则分类对象。
+
+2. 查询规则分类树
+GET /rule/category/tree
+权限标识：module_rule:rule_category:query
+Query 参数：name、parent_id、status、created_time、updated_time、created_id、updated_id、tenant_id。
+说明：按 parent_id 组装树形结构，返回顶级分类及其 children。
+返回：data 为规则分类树数组。
+
+3. 查询规则分类详情
+GET /rule/category/detail/{id}
+权限标识：module_rule:rule_category:detail
+Path 参数：id，规则分类 ID。
+返回：data 为规则分类对象。
+
+4. 创建规则分类
+POST /rule/category/create
+权限标识：module_rule:rule_category:create
+Body 参数：name、parent_id、order、status、description。
+字段说明：name 必填，1-64 字符；parent_id 可为空，0/null 表示顶级分类；order 默认为 999；status 默认为 0，0 启用、1 停用；description 最大 500 字符。
+返回：data 为创建后的规则分类对象。
+示例 Body：{ "name": "合同审核", "parent_id": null, "order": 1, "status": 0, "description": "合同审核相关规则" }
+
+5. 修改规则分类
+PUT /rule/category/update/{id}
+权限标识：module_rule:rule_category:update
+Path 参数：id，规则分类 ID。
+Body 参数：name、parent_id、order、status、description。
+说明：父级分类不能选择自身，也不能选择当前分类的子级；同一父级下分类名称不能重复。
+返回：data 为修改后的规则分类对象。
+
+6. 删除规则分类
+DELETE /rule/category/delete
+权限标识：module_rule:rule_category:delete
+Body 参数：ID 数组，例如 [1, 2, 3]。
+说明：如果分类下存在子分类，或分类下存在规则，不允许删除。
+返回：data 为 null。
+
+规则分类对象字段：id、uuid、name、parent_id、parent_name、order、status、description、tenant_id、tenant_by、created_id、created_by、updated_id、updated_by、deleted_id、deleted_by、created_time、updated_time、deleted_time、is_deleted。
+
+二、规则列表模块
+
+接口前缀：/rule
+
+1. 查询规则分页列表
+GET /rule/list
+权限标识：module_rule:rule:query
+Query 参数：page_no、page_size、order_by、name、category_id、status、created_time、updated_time、created_id、updated_id、tenant_id。
+说明：name 为规则名称模糊查询；category_id 为规则分类 ID，传 0 会按未分类 null 处理；status 为 0 启用、1 停用。
+返回：data 为分页结构，items 中每项为规则对象。
+
+2. 查询规则详情
+GET /rule/detail/{id}
+权限标识：module_rule:rule:detail
+Path 参数：id，规则 ID。
+返回：data 为规则对象。
+
+3. 创建规则
+POST /rule/create
+权限标识：module_rule:rule:create
+Body 参数：name、category_id、description、prompt_template、status。
+字段说明：name 必填，1-128 字符；category_id 可为空，0/null 表示不选择分类；description 最大 2000 字符；prompt_template 必填，用于保存提示词模板；status 默认为 0，0 启用、1 停用。
+说明：创建时会校验规则名称不能重复；如果传 category_id，会校验分类存在，并保存 category_name_snapshot 作为分类名称快照。
+返回：data 为创建后的规则对象。
+示例 Body：{ "name": "合同风险识别", "category_id": 1, "description": "识别合同中的风险条款", "prompt_template": "请根据以下合同内容识别风险点：{{content}}", "status": 0 }
+
+4. 修改规则
+PUT /rule/update/{id}
+权限标识：module_rule:rule:update
+Path 参数：id，规则 ID。
+Body 参数：name、category_id、description、prompt_template、status。
+说明：会校验规则存在、规则名称不能与其他规则重复、分类存在，并重新保存 category_name_snapshot。
+返回：data 为修改后的规则对象。
+
+5. 删除规则
+DELETE /rule/delete
+权限标识：module_rule:rule:delete
+Body 参数：ID 数组，例如 [1, 2, 3]。
+说明：执行软删除。
+返回：data 为 null。
+
+规则对象字段：id、uuid、name、category_id、category_name_snapshot、category_name、description、prompt_template、status、tenant_id、tenant_by、created_id、created_by、updated_id、updated_by、deleted_id、deleted_by、created_time、updated_time、deleted_time、is_deleted。
+
+补充说明：category_id 是规则与规则分类的真实关联字段；category_name_snapshot 是创建或修改规则时保存的分类名称快照；category_name 当前返回值等于 category_name_snapshot。prompt_template 在数据库中使用 Text 类型保存，适合较长提示词模板。