铁科院功能模块

app/api/function_list.py

@@ -0,0 +1,100 @@
+from fastapi import APIRouter, Depends, status
+from sqlalchemy.orm import Session
+from ..core.database import get_db
+from ..core.response_code import ResponseCode
+from ..schemas.function_list import (
+    FunctionListCreate, FunctionListRequest, FunctionGetRequest,
+    FunctionUpdateRequest, FunctionDeleteRequest,
+    FunctionApiResponse, FunctionListApiResponse
+)
+from ..services.function_list import FunctionListService
+
+router = APIRouter(prefix="/functions", tags=["铁科平台功能管理"])
+
+
+@router.post("/create", response_model=FunctionApiResponse, status_code=status.HTTP_201_CREATED)
+def create_function(func_data: FunctionListCreate, db: Session = Depends(get_db)):
+    """创建功能"""
+    # 检查功能名称是否已存在
+    existing = FunctionListService.get_function_by_name(db, func_data.function_name)
+    if existing:
+        return FunctionApiResponse(
+            code=ResponseCode.DATA_EXISTS,
+            message="功能名称已存在",
+            data=None
+        )
+
+    func_response = FunctionListService.create_function(db, func_data)
+    return FunctionApiResponse(
+        code=ResponseCode.SUCCESS,
+        message="功能创建成功",
+        data=func_response.model_dump()
+    )
+
+
+@router.post("/list", response_model=FunctionListApiResponse)
+def get_functions(request: FunctionListRequest, db: Session = Depends(get_db)):
+    """获取功能列表"""
+    functions = FunctionListService.get_functions(db, skip=request.skip, limit=request.limit)
+    return FunctionListApiResponse(
+        code=ResponseCode.SUCCESS,
+        message="查询成功",
+        total=len(functions),
+        data=functions
+    )
+
+
+@router.post("/get", response_model=FunctionListApiResponse)
+def get_function(request: FunctionGetRequest, db: Session = Depends(get_db)):
+    """根据条件查询功能"""
+    functions = FunctionListService.search_functions(
+        db,
+        func_id=request.id,
+        function_name=request.function_name
+    )
+    if not functions:
+        return FunctionListApiResponse(
+            code=ResponseCode.DATA_NOT_FOUND,
+            message="功能不存在",
+            total=0,
+            data=[]
+        )
+    return FunctionListApiResponse(
+        code=ResponseCode.SUCCESS,
+        message="查询成功",
+        total=len(functions),
+        data=functions
+    )
+
+
+@router.post("/update", response_model=FunctionApiResponse)
+def update_function(request: FunctionUpdateRequest, db: Session = Depends(get_db)):
+    """更新功能"""
+    func = FunctionListService.update_function(db, request.id, request.function_data)
+    if not func:
+        return FunctionApiResponse(
+            code=ResponseCode.DATA_NOT_FOUND,
+            message="功能不存在",
+            data=None
+        )
+    return FunctionApiResponse(
+        code=ResponseCode.SUCCESS,
+        message="功能更新成功",
+        data=func.model_dump()
+    )
+
+
+@router.post("/delete", response_model=FunctionApiResponse)
+def delete_function(request: FunctionDeleteRequest, db: Session = Depends(get_db)):
+    """删除功能"""
+    if not FunctionListService.delete_function(db, request.id):
+        return FunctionApiResponse(
+            code=ResponseCode.DATA_NOT_FOUND,
+            message="功能不存在",
+            data=None
+        )
+    return FunctionApiResponse(
+        code=ResponseCode.SUCCESS,
+        message="功能删除成功",
+        data=None
+    )

app/main.py

@@ -16,6 +16,7 @@ from .api.daily import router as daily_router
 from .api.section_data import router as section_data_router
 from .api.level_data import router as level_data_router
 from .api.checkpoint import router as checkpoint_router
+from .api.function_list import router as function_list_router
 from .utils.scheduler import task_scheduler
 
 # 初始化日志系统
@@ -98,6 +99,7 @@ app.include_router(daily_router, prefix="/api")
 app.include_router(section_data_router, prefix="/api")
 app.include_router(level_data_router, prefix="/api")
 app.include_router(checkpoint_router, prefix="/api")
+app.include_router(function_list_router, prefix="/api")
 # app.include_router(test_router, prefix="/api")
 
 # 根路径

app/models/function_list.py

@@ -0,0 +1,18 @@
+from sqlalchemy import Column, BigInteger, String
+from ..core.database import Base
+
+
+class FunctionList(Base):
+    """铁科平台功能列表"""
+    __tablename__ = "function_list"
+
+    id = Column(BigInteger, primary_key=True, index=True, autoincrement=True)
+    function_name = Column(String(255), nullable=False, comment="功能名称")
+    description = Column(String(1000), nullable=True, comment="描述")
+
+    def to_dict(self):
+        """将模型实例转换为字典"""
+        return {
+            column.name: getattr(self, column.name)
+            for column in self.__table__.columns
+        }

app/schemas/function_list.py

@@ -0,0 +1,74 @@
+from pydantic import BaseModel, ConfigDict
+from typing import Optional, List, Any
+
+
+class FunctionListBase(BaseModel):
+    """功能基础模型"""
+    function_name: str
+    description: Optional[str] = None
+
+
+class FunctionListCreate(FunctionListBase):
+    """创建功能请求"""
+    pass
+
+
+class FunctionListUpdate(BaseModel):
+    """更新功能请求"""
+    function_name: Optional[str] = None
+    description: Optional[str] = None
+
+
+class FunctionListResponse(FunctionListBase):
+    """功能响应模型"""
+    id: int
+
+    model_config = ConfigDict(from_attributes=True)
+
+    @classmethod
+    def from_orm_function(cls, func):
+        """从ORM对象创建响应对象"""
+        return cls(
+            id=func.id,
+            function_name=func.function_name,
+            description=func.description
+        )
+
+
+class FunctionListRequest(BaseModel):
+    """列表查询请求"""
+    skip: Optional[int] = 0
+    limit: Optional[int] = 100
+
+
+class FunctionGetRequest(BaseModel):
+    """条件查询请求"""
+    id: Optional[int] = None
+    function_name: Optional[str] = None
+
+
+class FunctionUpdateRequest(BaseModel):
+    """更新请求"""
+    id: int
+    function_data: FunctionListUpdate
+
+
+class FunctionDeleteRequest(BaseModel):
+    """删除请求"""
+    id: int
+
+
+# 统一响应格式
+class FunctionApiResponse(BaseModel):
+    """功能API统一响应格式"""
+    code: int = 0
+    message: str
+    data: Optional[Any] = None
+
+
+class FunctionListApiResponse(BaseModel):
+    """功能列表响应格式"""
+    code: int = 0
+    message: str
+    total: int
+    data: List[FunctionListResponse] = []

app/services/function_list.py

@@ -0,0 +1,80 @@
+from sqlalchemy.orm import Session
+from ..models.function_list import FunctionList
+from ..schemas.function_list import (
+    FunctionListCreate, FunctionListUpdate, FunctionListResponse
+)
+from typing import List, Optional
+import logging
+
+logger = logging.getLogger(__name__)
+
+
+class FunctionListService:
+    """铁科平台功能服务"""
+
+    @staticmethod
+    def create_function(db: Session, func_data: FunctionListCreate) -> FunctionListResponse:
+        """创建功能"""
+        db_func = FunctionList(**func_data.model_dump())
+        db.add(db_func)
+        db.commit()
+        db.refresh(db_func)
+        return FunctionListResponse.from_orm_function(db_func)
+
+    @staticmethod
+    def get_function_by_id(db: Session, func_id: int) -> Optional[FunctionList]:
+        """根据ID获取功能"""
+        return db.query(FunctionList).filter(FunctionList.id == func_id).first()
+
+    @staticmethod
+    def get_function_by_name(db: Session, name: str) -> Optional[FunctionList]:
+        """根据名称获取功能"""
+        return db.query(FunctionList).filter(FunctionList.function_name == name).first()
+
+    @staticmethod
+    def get_functions(db: Session, skip: int = 0, limit: int = 100) -> List[FunctionListResponse]:
+        """获取功能列表"""
+        functions = db.query(FunctionList).offset(skip).limit(limit).all()
+        return [FunctionListResponse.from_orm_function(f) for f in functions]
+
+    @staticmethod
+    def search_functions(
+        db: Session,
+        func_id: Optional[int] = None,
+        function_name: Optional[str] = None
+    ) -> List[FunctionListResponse]:
+        """根据条件搜索功能"""
+        query = db.query(FunctionList)
+
+        if func_id is not None:
+            query = query.filter(FunctionList.id == func_id)
+        if function_name is not None:
+            query = query.filter(FunctionList.function_name.like(f"%{function_name}%"))
+
+        functions = query.all()
+        return [FunctionListResponse.from_orm_function(f) for f in functions]
+
+    @staticmethod
+    def update_function(
+        db: Session, func_id: int, func_data: FunctionListUpdate
+    ) -> Optional[FunctionListResponse]:
+        """更新功能"""
+        db_func = db.query(FunctionList).filter(FunctionList.id == func_id).first()
+        if db_func:
+            update_data = func_data.model_dump(exclude_unset=True)
+            for field, value in update_data.items():
+                setattr(db_func, field, value)
+            db.commit()
+            db.refresh(db_func)
+            return FunctionListResponse.from_orm_function(db_func)
+        return None
+
+    @staticmethod
+    def delete_function(db: Session, func_id: int) -> bool:
+        """删除功能"""
+        db_func = db.query(FunctionList).filter(FunctionList.id == func_id).first()
+        if db_func:
+            db.delete(db_func)
+            db.commit()
+            return True
+        return False