Update

.env.example

@@ -0,0 +1,35 @@
+# 应用配置
+APP_ENV=development
+APP_NAME=Smart Question Bank
+APP_VERSION=1.0.0
+APP_DEBUG=true
+APP_SECRET_KEY=change-me-in-production
+APP_TIMEZONE=Asia/Shanghai
+APP_LOCALE=zh_CN
+API_PREFIX=/api/v1
+
+# 数据库配置
+DB_HOST=localhost
+DB_PORT=3306
+DB_USER=root
+DB_PASSWORD=
+DB_NAME=question_bank
+DB_POOL_SIZE=10
+DB_MAX_OVERFLOW=20
+DB_ECHO=false
+DB_POOL_RECYCLE=3600
+
+# Redis配置
+REDIS_HOST=localhost
+REDIS_PORT=6379
+REDIS_PASSWORD=
+REDIS_DB=0
+
+# JWT配置
+JWT_SECRET=change-me-in-production
+JWT_ALGORITHM=HS256
+JWT_ACCESS_TOKEN_EXPIRE_MINUTES=1440
+JWT_REFRESH_TOKEN_EXPIRE_DAYS=30
+
+# CORS配置
+CORS_ORIGINS=http://localhost:3000

.gitignore

@@ -0,0 +1,51 @@
+# IDE
+.vscode/
+.idea/
+
+# Python
+__pycache__/
+*.py[cod]
+*$py.class
+*.so
+.Python
+build/
+develop-eggs/
+dist/
+downloads/
+eggs/
+.eggs/
+lib/
+lib64/
+parts/
+sdist/
+var/
+wheels/
+*.egg-info/
+.installed.cfg
+*.egg
+
+# Logs
+logs/
+*.log
+
+# Environment
+.env
+.venv
+env/
+venv/
+ENV/
+env.bak/
+venv.bak/
+
+# Database
+*.db
+*.sqlite3
+
+# Test
+.coverage
+htmlcov/
+.pytest_cache/
+
+# OS
+.DS_Store
+Thumbs.db

PROJECT_STRUCTURE.md

@@ -0,0 +1,204 @@
+## 项目结构
+
+```
+smart-question-bank/
+├── backend/              # 后端服务
+│   ├── src/              # 源代码目录
+│   │   ├── common/       # 公共模块
+│   │   │   ├── constants.py # 常量定义
+│   │   │   └── exceptions.py # 异常定义
+│   │   ├── exception/    # 异常处理
+│   │   │   ├── api_exception.py # API异常
+│   │   │   ├── business_exception.py # 业务异常
+│   │   │   └── handler.py # 异常处理器
+│   │   ├── models/       # 数据模型
+│   │   │   ├── base_model.py # 基础模型
+│   │   │   ├── user.py   # 用户模型
+│   │   │   ├── question.py # 题目模型
+│   │   │   ├── practice_record.py # 练习记录模型
+│   │   │   ├── knowledge_point.py # 知识点模型
+│   │   │   ├── exam_score.py # 考试成绩模型
+│   │   │   ├── learning_path.py # 学习路径模型
+│   │   │   ├── question_version.py # 题目版本模型
+│   │   │   ├── question_review.py # 题目审核模型
+│   │   │   ├── question_tag.py # 题目标签模型
+│   │   │   ├── wrong_question.py # 错题模型
+│   │   │   ├── wrong_question_reason.py # 错题原因模型
+│   │   │   ├── wrong_question_tag.py # 错题标签模型
+│   │   │   ├── learning_goal.py # 学习目标模型
+│   │   │   ├── learning_style.py # 学习风格模型
+│   │   │   ├── target_score.py # 目标分数模型
+│   │   │   ├── exam.py # 考试模型
+│   │   │   ├── exam_analysis.py # 考试分析模型
+│   │   │   ├── learning_progress.py # 学习进度模型
+│   │   │   ├── knowledge_graph.py # 知识图谱模型
+│   │   │   ├── difficulty_assessment.py # 难度评估模型
+│   │   │   ├── question_recommendation.py # 题目推荐模型
+│   │   │   ├── adaptive_learning.py # 自适应学习模型
+│   │   ├── store/        # 数据持久层
+│   │   │   ├── user_store.py # 用户数据操作
+│   │   │   ├── question_store.py # 题目数据操作
+│   │   │   ├── practice_store.py # 练习数据操作
+│   │   │   ├── knowledge_store.py # 知识点数据操作
+│   │   │   ├── exam_store.py # 考试数据操作
+│   │   │   ├── learning_path_store.py # 学习路径数据操作
+│   │   │   ├── wrong_question_store.py # 错题数据操作
+│   │   │   ├── learning_goal_store.py # 学习目标数据操作
+│   │   │   ├── learning_style_store.py # 学习风格数据操作
+│   │   │   ├── target_score_store.py # 目标分数数据操作
+│   │   │   ├── exam_analysis_store.py # 考试分析数据操作
+│   │   │   ├── learning_progress_store.py # 学习进度数据操作
+│   │   │   ├── knowledge_graph_store.py # 知识图谱数据操作
+│   │   │   ├── difficulty_assessment_store.py # 难度评估数据操作
+│   │   │   ├── question_recommendation_store.py # 题目推荐数据操作
+│   │   │   └── adaptive_learning_store.py # 自适应学习数据操作
+│   │   ├── services/     # 业务逻辑
+│   │   │   ├── auth_service.py # 认证服务
+│   │   │   ├── question_service.py # 题目服务
+│   │   │   ├── practice_service.py # 练习服务
+│   │   │   ├── analysis_service.py # 分析服务
+│   │   │   ├── exam_service.py # 考试服务
+│   │   │   ├── learning_path_service.py # 学习路径服务
+│   │   │   ├── wrong_question_service.py # 错题服务
+│   │   │   ├── learning_goal_service.py # 学习目标服务
+│   │   │   ├── learning_style_service.py # 学习风格服务
+│   │   │   ├── target_score_service.py # 目标分数服务
+│   │   │   ├── exam_analysis_service.py # 考试分析服务
+│   │   │   ├── learning_progress_service.py # 学习进度服务
+│   │   │   ├── knowledge_graph_service.py # 知识图谱服务
+│   │   │   ├── difficulty_assessment_service.py # 难度评估服务
+│   │   │   ├── question_recommendation_service.py # 题目推荐服务
+│   │   │   └── adaptive_learning_service.py # 自适应学习服务
+│   │   ├── controllers/  # API控制器
+│   │   │   ├── auth_controller.py # 认证接口
+│   │   │   ├── question_controller.py # 题目接口
+│   │   │   ├── practice_controller.py # 练习接口
+│   │   │   ├── analysis_controller.py # 分析接口
+│   │   │   ├── exam_controller.py # 考试接口
+│   │   │   ├── learning_path_controller.py # 学习路径接口
+│   │   │   ├── wrong_question_controller.py # 错题接口
+│   │   │   ├── learning_goal_controller.py # 学习目标接口
+│   │   │   ├── learning_style_controller.py # 学习风格接口
+│   │   │   ├── target_score_controller.py # 目标分数接口
+│   │   │   ├── exam_analysis_controller.py # 考试分析接口
+│   │   │   ├── learning_progress_controller.py # 学习进度接口
+│   │   │   ├── knowledge_graph_controller.py # 知识图谱接口
+│   │   │   ├── difficulty_assessment_controller.py # 难度评估接口
+│   │   │   ├── question_recommendation_controller.py # 题目推荐接口
+│   │   │   └── adaptive_learning_controller.py # 自适应学习接口
+│   │   ├── middleware/   # 中间件
+│   │   │   ├── auth_middleware.py # 认证中间件
+│   │   │   ├── logging_middleware.py # 日志中间件
+│   │   │   ├── error_middleware.py # 错误处理中间件
+│   │   │   ├── request_handler.py # 请求处理中间件（合并请求校验、转换、过滤等功能）
+│   │   │   ├── response_handler.py # 响应处理中间件（合并响应格式化、压缩、加密等功能）
+│   │   │   ├── performance_monitor.py # 性能监控中间件（合并性能、指标、响应时间等）
+│   │   │   ├── security_middleware.py # 安全中间件（合并限流、权限、CORS等）
+│   │   │   └── tracing_middleware.py # 追踪中间件（合并请求ID、日志、追踪等）
+│   │   ├── config/       # 配置文件
+│   │   │   ├── config_loader.py # 配置加载器。加载环境变量和配置文件，环境变量优先级高于配置文件
+│   │   │   ├── application.yml # 主配置文件
+│   │   │   ├── application-dev.yml # 开发环境配置
+│   │   │   ├── application-prod.yml # 生产环境配置
+│   │   │   └── application-test.yml # 测试环境配置
+│   │   ├── database/     # 数据库相关
+│   │   │   ├── connection/ # 数据库连接
+│   │   │   │   └── mysql.py # MySQL连接
+│   │   │   ├── migration/ # 数据库迁移
+│   │   │   │   ├── alembic.ini # Alembic配置
+│   │   │   │   └── versions/ # 迁移脚本
+│   │   │   └── utils/    # 数据库工具
+│   │   │       ├── query_builder.py # 查询构建器
+│   │   │       └── pagination.py # 分页工具
+│   │   ├── utils/        # 工具类
+│   │   │   ├── logger.py # 日志配置
+│   │   │   ├── date/     # 日期相关
+│   │   │   │   ├── date_util.py # 日期工具
+│   │   │   │   └── timezone.py # 时区处理
+│   │   │   ├── string/   # 字符串处理
+│   │   │   │   ├── string_util.py # 字符串工具
+│   │   │   │   └── validator.py # 字符串验证
+│   │   │   ├── file/     # 文件处理
+│   │   │   │   ├── file_util.py # 文件工具
+│   │   │   │   └── storage.py # 文件存储
+│   │   │   └── network/  # 网络相关
+│   │   │       ├── request.py # 请求工具
+│   │   │       └── ip_util.py # IP工具
+│   │   ├── tests/        # 测试相关
+│   │   │   ├── unit/     # 单元测试
+│   │   │   │   ├── models/ # 模型测试
+│   │   │   │   ├── services/ # 服务测试
+│   │   │   │   └── utils/ # 工具测试
+│   │   │   ├── integration/ # 集成测试
+│   │   │   │   ├── api/  # API测试
+│   │   │   │   └── database/ # 数据库测试
+│   │   │   ├── fixtures/ # 测试夹具
+│   │   │   │   ├── data/ # 测试数据
+│   │   │   │   └── mocks/ # Mock数据
+│   │   │   └── e2e/      # 端到端测试
+│   │   │       ├── api/  # API端到端测试
+│   │   │       └── web/  # Web端到端测试
+│   ├── requirements.txt  # 依赖包列表
+│   └── Dockerfile        # Docker配置
+├── frontend/             # 前端应用
+│   ├── public/           # 静态资源
+│   │   ├── index.html    # 主页面
+│   │   ├── favicon.ico   # 网站图标
+│   │   └── robots.txt    # 爬虫协议
+│   ├── src/              # 源代码目录
+│   │   ├── assets/       # 静态资源
+│   │   │   ├── images/   # 图片资源
+│   │   │   ├── fonts/    # 字体文件
+│   │   │   └── styles/   # 全局样式
+│   │   ├── components/   # 通用组件
+│   │   │   ├── Header.vue # 头部组件
+│   │   │   ├── Footer.vue # 底部组件
+│   │   │   ├── QuestionCard.vue # 题目卡片
+│   │   │   └── ProgressChart.vue # 进度图表
+│   │   ├── views/        # 页面视图
+│   │   │   ├── Home.vue  # 首页
+│   │   │   ├── Practice.vue # 练习页面
+│   │   │   ├── Analysis.vue # 分析页面
+│   │   │   └── Profile.vue # 个人中心
+│   │   ├── router/       # 路由配置
+│   │   │   └── index.js  # 路由配置
+│   │   ├── store/        # 状态管理
+│   │   │   ├── modules/  # 模块
+│   │   │   │   ├── auth.js # 认证模块
+│   │   │   │   ├── question.js # 题目模块
+│   │   │   │   └── practice.js # 练习模块
+│   │   │   └── index.js  # 状态管理入口
+│   │   ├── api/          # API接口
+│   │   │   ├── auth.js   # 认证接口
+│   │   │   ├── question.js # 题目接口
+│   │   │   └── practice.js # 练习接口
+│   │   ├── utils/        # 工具函数
+│   │   │   ├── request.js # 请求封装
+│   │   │   ├── date.js   # 日期工具
+│   │   │   └── validate.js # 验证工具
+│   │   ├── styles/       # 全局样式
+│   │   │   ├── variables.scss # 样式变量
+│   │   │   ├── mixins.scss # 样式混合
+│   │   │   └── global.scss # 全局样式
+│   │   └── main.js       # 程序入口
+│   ├── tests/            # 测试目录
+│   │   ├── unit/         # 单元测试
+│   │   ├── integration/  # 集成测试
+│   │   └── e2e/          # 端到端测试
+│   ├── package.json      # 项目配置
+│   ├── vue.config.js     # Vue配置
+│   └── Dockerfile        # Docker配置
+├── docs/                 # 文档目录
+│   ├── api/              # API文档
+│   └── design/           # 设计文档
+├── scripts/              # 脚本目录
+│   ├── deploy/           # 部署脚本
+│   ├── migration/        # 数据迁移脚本
+│   │   ├── migrations/   # 迁移文件目录
+│   │   ├── __init__.py   # 初始化文件
+│   │   ├── alembic.ini   # Alembic 配置文件
+│   │   └── env.py        # Alembic 环境配置
+├── docker-compose.yml    # Docker Compose配置
+├── README.md             # 项目说明
+└── .gitignore            # Git忽略配置
+```

README.md

@@ -1,237 +1,70 @@
-# 智能题库学习系统 (Smart Question Bank Learning System)
+# Smart Question Bank 智能题库系统
 
-## 项目简介
+## 配置系统说明
 
-这是一个基于 Python 开发的智能题库学习系统，旨在帮助学生进行高效的题目练习和知识掌握。系统具备题目管理、智能练习、错题本、学习进度追踪等功能，让学习更有针对性和效果。
-
-## 主要功能
-
-### 1. 题库管理
-
-- 支持多学科题目录入（数学、语文、英语等）
-- 题目分类管理（按照难度、知识点、题型等）
-- 支持多种题型（选择题、填空题、简答题等）
-- 每道题目包含详细解析和知识点说明
-- 题目版本控制，记录修改历史
-- 题目审核流程，确保题目质量
-- 题目标签系统，方便分类和搜索
-
-### 2. 智能练习
-
-- 根据学生水平自动推荐适合的题目
-- 支持自选题型和难度进行练习
-- 练习过程中即时查看答案和解析
-- 练习完成后显示得分和错题分析
-- 优化推荐算法，结合用户历史表现和学习目标
-- 个性化推荐，根据用户偏好推荐题目
-- 自适应练习模式，动态调整题目难度
-
-### 3. 错题本功能
-
-- 自动收集做错的题目
-- 错题分类整理
-- 支持错题重复练习
-- 错题数据分析，找出薄弱知识点
-- 错题原因分析，帮助用户理解错误原因
-- 错题标签，方便分类和复习
-- 错题相似题推荐，巩固薄弱知识点
-
-### 4. 学习进度追踪
-
-- 展示每日/每周/每月的练习情况
-- 知识点掌握程度分析
-- 学习时长统计
-- 成绩进步曲线
-- 知识点掌握预测，预估用户掌握程度
-- 学习路径规划，提供个性化学习建议
-- 学习目标设置，帮助用户明确学习方向
-- 手动考试成绩录入，记录各类考试成绩
-- 考试成绩分析，生成成绩报告
-
-## 技术架构
-
-- 后端：Python 3.13
-- 数据库：Mysql
-- 用户界面：Web 界面
-
-## 数据库设计
-
-### 题目表(questions)
-
-- id: 题目 ID
-- subject: 学科
-- type: 题型
-- difficulty: 难度等级
-- content: 题目内容
-- answer: 答案
-- explanation: 解析
-- knowledge_points: 相关知识点
-- version: 题目版本
-- status: 审核状态
-- tags: 题目标签
-- creator_id: 题目创建者
-- reviewer_id: 题目审核者
-
-### 用户表(users)
-
-- id: 用户 ID
-- username: 用户名
-- password: 密码
-- grade: 年级
-- create_time: 创建时间
-- preferences: 学习偏好
-- goals: 学习目标
-- learning_style: 学习风格
-- target_score: 目标分数
-
-### 练习记录表(practice_records)
-
-- id: 记录 ID
-- user_id: 用户 ID
-- question_id: 题目 ID
-- is_correct: 是否正确
-- practice_time: 练习时间
-- time_spent: 用时
-- steps: 答题步骤
-- thinking_time: 思考时间
-- confidence_level: 答题信心
-- feedback: 用户反馈
-
-### 知识点表(knowledge_points)
-
-- id: 知识点 ID
-- name: 知识点名称
-- description: 知识点描述
-- subject: 所属学科
-- difficulty: 难度等级
-- prerequisite: 先修知识点
-
-### 学习路径表(learning_paths)
-
-- id: 路径 ID
-- user_id: 用户 ID
-- target_knowledge_point: 目标知识点
-- path: 学习路径
-- progress: 当前进度
-
-### 考试成绩表(exam_scores)
-
-- id: 成绩 ID
-- user_id: 用户 ID
-- exam_name: 考试名称
-- subject: 考试科目
-- score: 考试成绩
-- total_score: 总分
-- exam_date: 考试日期
-- remark: 备注
-
-## 使用说明
-
-1. 系统启动后，首先需要注册/登录账号
-2. 选择想要练习的学科和题型
-3. 开始答题，系统会记录答题情况
-4. 可以随时查看错题本和学习进度
-5. 在成绩管理页面手动录入考试成绩
-
-## 项目结构
+### 配置文件结构
 
 ```
-smart-question-bank/
-├── backend/              # 后端服务
-│   ├── src/              # 源代码目录
-│   │   ├── common/       # 公共模块
-│   │   ├── exception/    # 异常处理
-│   │   ├── models/       # 数据模型
-│   │   ├── services/     # 业务逻辑
-│   │   ├── controllers/  # API控制器
-│   │   ├── middleware/   # 中间件
-│   │   ├── utils/        # 工具类
-│   │   ├── config/       # 配置文件
-│   │   ├── migrations/   # 数据库迁移
-│   │   └── main.py       # 程序入口
-│   ├── tests/            # 测试目录
-│   ├── requirements.txt  # 依赖包列表
-│   └── Dockerfile        # Docker配置
-├── frontend/             # 前端应用
-│   ├── public/           # 静态资源
-│   ├── src/              # 源代码目录
-│   │   ├── assets/       # 静态资源
-│   │   ├── components/   # 通用组件
-│   │   ├── views/        # 页面视图
-│   │   ├── router/       # 路由配置
-│   │   ├── store/        # 状态管理
-│   │   ├── api/          # API接口
-│   │   ├── utils/        # 工具函数
-│   │   ├── styles/       # 全局样式
-│   │   └── main.js       # 程序入口
-│   ├── tests/            # 测试目录
-│   ├── package.json      # 项目配置
-│   ├── vue.config.js     # Vue配置
-│   └── Dockerfile        # Docker配置
-├── docs/                 # 文档目录
-│   ├── api/              # API文档
-│   └── design/           # 设计文档
-├── scripts/              # 脚本目录
-│   ├── deploy/           # 部署脚本
-│   └── migration/        # 数据迁移脚本
-├── docker-compose.yml    # Docker Compose配置
-├── README.md             # 项目说明
-└── .gitignore            # Git忽略配置
+config/
+├── application.yml          # 基础配置
+├── application-dev.yml      # 开发环境配置
+├── application-prod.yml     # 生产环境配置
+└── config_loader.py         # 配置加载器
 ```
 
-## 安装与运行
+### 环境变量配置
 
-1. 安装依赖
+1. 复制`.env.example`为`.env`
+2. 根据实际环境修改`.env`文件中的配置项
+3. 主要配置项说明：
 
 ```bash
-pip install -r requirements.txt
+# 应用配置
+APP_ENV=development          # 环境类型 (development|production)
+APP_NAME=Smart Question Bank # 应用名称
+APP_VERSION=1.0.0            # 应用版本
+APP_DEBUG=true               # 调试模式
+APP_SECRET_KEY=              # 应用密钥
+APP_TIMEZONE=Asia/Shanghai   # 时区
+APP_LOCALE=zh_CN             # 本地化语言
+API_PREFIX=/api/v1           # API前缀
+
+# 数据库配置
+DB_HOST=localhost            # 数据库主机
+DB_PORT=3306                 # 数据库端口
+DB_USER=root                 # 数据库用户
+DB_PASSWORD=                 # 数据库密码
+DB_NAME=question_bank        # 数据库名称
+DB_POOL_SIZE=10              # 连接池大小
+DB_MAX_OVERFLOW=20           # 最大溢出连接数
+DB_ECHO=false                # SQL日志输出
+DB_POOL_RECYCLE=3600         # 连接回收时间(秒)
+
+# Redis配置
+REDIS_HOST=localhost         # Redis主机
+REDIS_PORT=6379              # Redis端口
+REDIS_PASSWORD=              # Redis密码
+REDIS_DB=0                   # Redis数据库编号
+
+# JWT配置
+JWT_SECRET=                  # JWT密钥
+JWT_ALGORITHM=HS256          # JWT算法
+JWT_ACCESS_TOKEN_EXPIRE_MINUTES=1440 # 访问令牌有效期(分钟)
+JWT_REFRESH_TOKEN_EXPIRE_DAYS=30     # 刷新令牌有效期(天)
+
+# CORS配置
+CORS_ORIGINS=                # 允许跨域请求的源
 ```
 
-2. 配置数据库
-
-- 创建 MySQL 数据库
-- 修改 src/common/config.py 中的数据库连接配置
-
-3. 运行系统
-
-```bash
-python src/main.py
-```
-
-## 已实现功能
-
-### 1. 图形用户界面
-
-- 使用 PyQt5 开发桌面应用界面
-- 提供友好的用户交互体验
-- 支持题目展示、答题、结果查看等功能
-
-### 2. 在线题库更新
-
-- 实现题库云端同步功能
-- 支持自动更新最新题目
-- 提供题目审核机制
-
-### 3. 做题时间限制
-
-- 支持按题型设置时间限制
-- 提供倒计时功能
-- 超时自动提交答案
-
-### 4. 班级管理
-
-- 教师创建和管理班级
-- 学生加入班级
-- 班级练习统计和分析
-
-### 5. 题目难度评估
+### 配置加载顺序
 
-- 基于学生答题数据自动评估题目难度
-- 动态调整题目难度系数
-- 提供难度分布可视化
+1. 加载`application.yml`基础配置
+2. 根据`APP_ENV`加载对应环境配置
+3. 加载`.env`环境变量配置
+4. 环境变量优先级最高，会覆盖配置文件中的相同配置项
 
-### 6. 智能出题算法
+### 最佳实践
 
-- 基于知识图谱的题目推荐
-- 个性化出题策略
-- 自适应难度调整
+- 生产环境务必修改`APP_SECRET_KEY`和`JWT_SECRET`
+- 敏感信息（如数据库密码）应通过环境变量配置
+- 开发环境可使用`.env`文件，生产环境建议直接设置系统环境变量

backend/src/__init__.py

@@ -0,0 +1 @@
+# backend 包初始化文件

backend/src/common/constants.py

@@ -0,0 +1,32 @@
+from fastapi import status
+
+
+class ErrorCode:
+    """错误码常量定义"""
+
+    # 通用错误
+    UNKNOWN_ERROR = 10000
+    VALIDATION_ERROR = 10001
+    DATABASE_ERROR = 10002
+    NETWORK_ERROR = 10003
+    PERMISSION_DENIED = 10004
+    RESOURCE_NOT_FOUND = 10005
+    INVALID_OPERATION = 10006
+
+    # 认证相关
+    AUTH_REQUIRED = 20001
+    INVALID_TOKEN = 20002
+    EXPIRED_TOKEN = 20003
+    INVALID_CREDENTIALS = 20004
+
+    # 业务相关
+    QUESTION_NOT_FOUND = 30001
+    EXAM_NOT_FOUND = 30002
+    USER_NOT_FOUND = 30003
+    INVALID_QUESTION_TYPE = 30004
+    INVALID_EXAM_STATUS = 30005
+
+    # 系统相关
+    CONFIG_ERROR = 40001
+    SERVICE_UNAVAILABLE = 40002
+    RATE_LIMIT_EXCEEDED = 40003

backend/src/common/exceptions.py

@@ -0,0 +1,26 @@
+from typing import Any, Optional
+from fastapi import status
+
+
+class BaseException(Exception):
+    """自定义异常基类"""
+
+    def __init__(self,
+                 code: int = status.HTTP_500_INTERNAL_SERVER_ERROR,
+                 message: str = "Internal Server Error",
+                 detail: Optional[Any] = None):
+        self.code = code
+        self.message = message
+        self.detail = detail
+        super().__init__(message)
+
+    def __str__(self):
+        return f"{self.__class__.__name__}(code={self.code}, message={self.message}, detail={self.detail})"
+
+    def to_dict(self):
+        """将异常转换为字典格式"""
+        return {
+            "code": self.code,
+            "message": self.message,
+            "detail": self.detail
+        }

backend/src/config/__init__.py

@@ -0,0 +1 @@
+# config 包初始化文件

backend/src/config/application-dev.yml

@@ -0,0 +1,25 @@
+# 开发环境配置
+app:
+  debug: true
+  secret_key: 'development-secret-key'
+  cors:
+    origins: 'http://localhost:3000'
+
+database:
+  host: 'localhost'
+  port: 3306
+  user: 'dev_user'
+  password: 'dev_password'
+  name: 'question_bank_dev'
+  echo: true
+
+logging:
+  level: 'DEBUG'
+  file:
+    enabled: false
+
+redis:
+  enabled: true
+
+cache:
+  backend: 'memory'

backend/src/config/application-prod.yml

@@ -0,0 +1,41 @@
+# 生产环境配置
+app:
+  debug: false
+  secret_key: ${APP_SECRET_KEY} # 从环境变量读取
+  cors:
+    origins: ${CORS_ORIGINS} # 从环境变量读取
+
+database:
+  host: ${DB_HOST}
+  port: ${DB_PORT}
+  user: ${DB_USER}
+  password: ${DB_PASSWORD}
+  name: ${DB_NAME}
+  pool_size: 20
+  max_overflow: 50
+  echo: false
+
+logging:
+  level: 'WARNING'
+  file:
+    enabled: true
+    path: '/var/log/question-bank/app.log'
+    max_size: 104857600 # 100MB
+    backup_count: 10
+
+redis:
+  enabled: true
+  host: ${REDIS_HOST}
+  port: ${REDIS_PORT}
+  password: ${REDIS_PASSWORD}
+  db: 0
+
+cache:
+  backend: 'redis'
+  default_ttl: 86400 # 24小时
+
+security:
+  jwt:
+    secret: ${JWT_SECRET}
+    access_token_expire_minutes: 60 # 1小时
+    refresh_token_expire_days: 7

backend/src/config/application.yml

@@ -0,0 +1,43 @@
+app:
+  name: 'smart-question-bank'
+  version: '1.0.0'
+  secret_key: 'your-secret-key-here'
+  debug: false
+  timezone: 'Asia/Shanghai'
+  locale: 'zh_CN'
+  api_prefix: '/api/v1'
+  environment: 'development'
+  log_level: 'INFO'
+  cors_origins: '*'
+  rate_limit: 1000
+
+database:
+  host: 'localhost'
+  port: 3306
+  user: 'root'
+  password: 'password'
+  name: 'question_bank'
+  pool_size: 10
+  max_overflow: 20
+  echo: false
+  pool_recycle: 3600
+
+redis:
+  host: 'localhost'
+  port: 6379
+  password: null
+  db: 0
+  cache_ttl: 3600
+
+auth:
+  token_expire_minutes: 1440
+  refresh_token_expire_days: 7
+  algorithm: 'HS256'
+  issuer: 'smart-question-bank'
+
+logging:
+  level: INFO
+  format: '%(asctime)s %(levelname)s %(name)s:%(lineno)d - %(message)s'
+  datefmt: '%Y-%m-%d %H:%M:%S'
+  console: true
+  file: 'logs/app.log'

backend/src/config/config_loader.py

@@ -0,0 +1,127 @@
+import os
+from typing import Dict, Any, Optional
+import yaml
+from pydantic_settings import BaseSettings
+from pydantic import Field, validator
+from pathlib import Path
+from enum import Enum
+
+
+class Environment(str, Enum):
+    DEVELOPMENT = "development"
+    TESTING = "testing"
+    PRODUCTION = "production"
+
+
+class AppConfig(BaseSettings):
+    name: str = Field(..., env="APP_NAME")
+    version: str = Field(..., env="APP_VERSION")
+    debug: bool = Field(False, env="APP_DEBUG")
+    secret_key: str = Field(..., env="APP_SECRET_KEY")
+    timezone: str = Field("Asia/Shanghai", env="APP_TIMEZONE")
+    locale: str = Field("zh_CN", env="APP_LOCALE")
+    api_prefix: str = Field("/api/v1", env="API_PREFIX")
+    environment: Environment = Field(Environment.DEVELOPMENT, env="APP_ENV")
+    log_level: str = Field("INFO", env="LOG_LEVEL")
+    cors_origins: str = Field("*", env="CORS_ORIGINS")
+    rate_limit: int = Field(1000, env="RATE_LIMIT")
+
+    @validator("environment", pre=True)
+    def validate_environment(cls, v):
+        return Environment(v.lower())
+
+    class Config:
+        env_file = ".env"
+        env_file_encoding = "utf-8"
+
+
+class DatabaseConfig(BaseSettings):
+    host: str = Field(..., env="DB_HOST")
+    port: int = Field(3306, env="DB_PORT")
+    user: str = Field(..., env="DB_USER")
+    password: str = Field(..., env="DB_PASSWORD")
+    name: str = Field(..., env="DB_NAME")
+    pool_size: int = Field(10, env="DB_POOL_SIZE")
+    max_overflow: int = Field(20, env="DB_MAX_OVERFLOW")
+    echo: bool = Field(False, env="DB_ECHO")
+    pool_recycle: int = Field(3600, env="DB_POOL_RECYCLE")
+    ssl_mode: Optional[str] = Field(None, env="DB_SSL_MODE")
+
+
+class RedisConfig(BaseSettings):
+    host: str = Field("localhost", env="REDIS_HOST")
+    port: int = Field(6379, env="REDIS_PORT")
+    password: Optional[str] = Field(None, env="REDIS_PASSWORD")
+    db: int = Field(0, env="REDIS_DB")
+    cache_ttl: int = Field(3600, env="CACHE_TTL")
+
+
+class AuthConfig(BaseSettings):
+    token_expire_minutes: int = Field(1440, env="TOKEN_EXPIRE_MINUTES")
+    refresh_token_expire_days: int = Field(7, env="REFRESH_TOKEN_EXPIRE_DAYS")
+    algorithm: str = Field("HS256", env="TOKEN_ALGORITHM")
+    issuer: str = Field("smart-question-bank", env="TOKEN_ISSUER")
+
+
+class Config(BaseSettings):
+    app: AppConfig
+    database: DatabaseConfig
+    redis: RedisConfig
+    auth: AuthConfig
+
+    class Config:
+        env_nested_delimiter = "__"
+
+
+class ConfigLoader:
+
+    def __init__(self):
+        self._config_data = {}
+        self.config = None
+        self._load_config()
+        self.config = Config(
+            app=AppConfig(**self._get_config_section('app')),
+            database=DatabaseConfig(**self._get_config_section('database')),
+            redis=RedisConfig(**self._get_config_section('redis')),
+            auth=AuthConfig(**self._get_config_section('auth')))
+
+    def _get_config_section(self, section: str) -> dict:
+        """获取配置文件中指定section的内容"""
+        return self._config_data.get(section, {})
+
+    def _load_config(self):
+        # 确定当前环境
+        env = os.getenv("APP_ENV", "development")
+
+        # 按顺序加载配置
+        self._load_config_file("application.yml")  # 基础配置
+        self._load_config_file(f"application-{env}.yml")  # 环境特定配置
+
+    def _load_config_file(self, filename: str):
+        """加载指定配置文件"""
+        config_path = Path(__file__).parent / filename
+        if config_path.exists():
+            with open(config_path, "r", encoding="utf-8") as f:
+                config = yaml.safe_load(f)
+                if config:
+                    self._merge_configs(self._config_data, config)
+
+    def _merge_configs(self, base: Dict, update: Dict):
+        """递归合并配置"""
+        for key, value in update.items():
+            if key in base and isinstance(base[key], dict) and isinstance(
+                    value, dict):
+                self._merge_configs(base[key], value)
+            else:
+                base[key] = value
+
+    def _get_config_section(self, section: str) -> dict:
+        """获取配置文件中指定section的内容"""
+        return self._config_data.get(section, {})
+
+    def get_config(self) -> Config:
+        return self.config
+
+
+config_loader = ConfigLoader()
+config = config_loader.get_config()

backend/src/controllers/__init__.py

@@ -0,0 +1,9 @@
+from fastapi import APIRouter
+from .question_controller import router as question_router
+
+router = APIRouter()
+
+# 注册所有控制器路由
+router.include_router(question_router)
+
+__all__ = ["router"]

backend/src/controllers/auth_controller.py

@@ -0,0 +1,39 @@
+from fastapi import APIRouter, Depends, HTTPException
+from fastapi.security import OAuth2PasswordRequestForm
+from sqlalchemy.orm import Session
+from ..services.auth_service import AuthService
+from ..models.auth_models import Token
+from ..middleware.auth_middleware import get_current_user
+from ..database import get_db
+
+router = APIRouter(prefix="/auth", tags=["Authentication"])
+
+
+@router.post("/login", response_model=Token)
+async def login(form_data: OAuth2PasswordRequestForm = Depends(),
+                db: Session = Depends(get_db)):
+    token = await AuthService.login(db, form_data.username, form_data.password)
+    if not token:
+        raise HTTPException(status_code=401,
+                            detail="Incorrect username or password")
+    return token
+
+
+@router.post("/refresh", response_model=Token)
+async def refresh_token(refresh_token: str, db: Session = Depends(get_db)):
+    try:
+        token = await AuthService.refresh_token(db, refresh_token)
+        if not token:
+            raise HTTPException(status_code=401,
+                                detail="Invalid refresh token")
+        return token
+    except ValueError as e:
+        raise HTTPException(status_code=500, detail=str(e))
+    except Exception as e:
+        raise HTTPException(status_code=500, detail="Internal server error")
+
+
+@router.get("/me")
+async def get_current_user_info(
+        current_user: dict = Depends(get_current_user)):
+    return current_user

backend/src/controllers/question_controller.py

@@ -0,0 +1,173 @@
+from fastapi import APIRouter, Depends, HTTPException
+from ..services.question_service import QuestionService
+from ..models.question import Question
+from ..models.question_bank_permission import QuestionBankPermission
+from typing import List
+
+router = APIRouter(prefix="/api/questions", tags=["questions"])
+
+
+@router.post("/", response_model=Question)
+async def create_question(question_data: dict):
+    """创建题目"""
+    try:
+        return QuestionService().create_question(question_data)
+    except Exception as e:
+        raise HTTPException(status_code=400, detail=str(e))
+
+
+@router.get("/{question_id}", response_model=Question)
+async def get_question(question_id: int):
+    """获取题目详情"""
+    try:
+        return QuestionService().get_question(question_id)
+    except Exception as e:
+        raise HTTPException(status_code=404, detail=str(e))
+
+
+@router.put("/{question_id}", response_model=Question)
+async def update_question(question_id: int, update_data: dict):
+    """更新题目"""
+    try:
+        return QuestionService().update_question(question_id, update_data)
+    except Exception as e:
+        raise HTTPException(status_code=400, detail=str(e))
+
+
+@router.delete("/{question_id}")
+async def delete_question(question_id: int):
+    """删除题目"""
+    try:
+        QuestionService().delete_question(question_id)
+        return {"message": "Question deleted successfully"}
+    except Exception as e:
+        raise HTTPException(status_code=400, detail=str(e))
+
+
+@router.get("/search/", response_model=List[Question])
+async def search_questions(q: str, limit: int = 100):
+    """搜索题目"""
+    try:
+        return QuestionService().search_questions(q, limit)
+    except Exception as e:
+        raise HTTPException(status_code=400, detail=str(e))
+
+
+@router.get("/bank/{bank_id}", response_model=List[Question])
+async def get_questions_by_bank(bank_id: int, limit: int = 100):
+    """获取题库下的题目列表"""
+    try:
+        return QuestionService().get_questions_by_bank(bank_id, limit)
+    except Exception as e:
+        raise HTTPException(status_code=400, detail=str(e))
+
+
+@router.post("/move-to-bank/")
+async def move_questions_to_bank(question_ids: List[int], target_bank_id: int):
+    """将题目移动到指定题库"""
+    try:
+        QuestionService().move_questions_to_bank(question_ids, target_bank_id)
+        return {"message": "Questions moved successfully"}
+    except Exception as e:
+        raise HTTPException(status_code=400, detail=str(e))
+
+
+@router.post("/bank/{bank_id}/permissions",
+             response_model=QuestionBankPermission)
+async def add_bank_permission(bank_id: int, user_id: int, permission_type: str,
+                              granted_by: int):
+    """添加题库权限"""
+    try:
+        return QuestionService().add_bank_permission(bank_id, user_id,
+                                                     permission_type,
+                                                     granted_by)
+    except Exception as e:
+        raise HTTPException(status_code=400, detail=str(e))
+
+
+@router.delete("/bank/permissions/{permission_id}")
+async def remove_bank_permission(permission_id: int):
+    """移除题库权限"""
+    try:
+        QuestionService().remove_bank_permission(permission_id)
+        return {"message": "Permission removed successfully"}
+    except Exception as e:
+        raise HTTPException(status_code=400, detail=str(e))
+
+
+@router.get("/bank/{bank_id}/permissions",
+            response_model=List[QuestionBankPermission])
+async def get_bank_permissions(bank_id: int):
+    """获取题库的所有权限"""
+    try:
+        return QuestionService().get_bank_permissions(bank_id)
+    except Exception as e:
+        raise HTTPException(status_code=400, detail=str(e))
+
+
+@router.get("/bank/{bank_id}/permissions/{user_id}",
+            response_model=List[QuestionBankPermission])
+async def get_user_bank_permissions(bank_id: int, user_id: int):
+    """获取用户对指定题库的权限"""
+    try:
+        return QuestionService().get_user_bank_permissions(bank_id, user_id)
+    except Exception as e:
+        raise HTTPException(status_code=400, detail=str(e))
+
+
+@router.get("/bank/{bank_id}/statistics")
+async def get_bank_statistics(bank_id: int):
+    """获取题库统计信息"""
+    try:
+        return QuestionService().get_bank_statistics(bank_id)
+    except Exception as e:
+        raise HTTPException(status_code=400, detail=str(e))
+
+
+@router.post("/bank/{bank_id}/validate-access")
+async def validate_bank_access(bank_id: int, user_id: int):
+    """验证用户是否有访问题库的权限"""
+    try:
+        QuestionService().validate_bank_access(bank_id, user_id)
+        return {"message": "Access granted"}
+    except Exception as e:
+        raise HTTPException(status_code=403, detail=str(e))
+
+
+@router.post("/bank/{bank_id}/update-statistics")
+async def update_bank_statistics(bank_id: int):
+    """更新题库统计信息"""
+    try:
+        return QuestionService().update_bank_statistics(bank_id)
+    except Exception as e:
+        raise HTTPException(status_code=400, detail=str(e))
+
+
+@router.post("/questions/{question_id}/knowledge_points")
+async def add_knowledge_point(question_id: int, knowledge_point_id: int):
+    """为题目添加知识点"""
+    try:
+        return QuestionService().add_knowledge_point(question_id,
+                                                     knowledge_point_id)
+    except Exception as e:
+        raise HTTPException(status_code=400, detail=str(e))
+
+
+@router.delete(
+    "/questions/{question_id}/knowledge_points/{knowledge_point_id}")
+async def remove_knowledge_point(question_id: int, knowledge_point_id: int):
+    """移除题目的知识点"""
+    try:
+        return QuestionService().remove_knowledge_point(
+            question_id, knowledge_point_id)
+    except Exception as e:
+        raise HTTPException(status_code=400, detail=str(e))
+
+
+@router.get("/questions/{question_id}/knowledge_points")
+async def get_question_knowledge_points(question_id: int):
+    """获取题目的知识点列表"""
+    try:
+        return QuestionService().get_question_knowledge_points(question_id)
+    except Exception as e:
+        raise HTTPException(status_code=400, detail=str(e))

backend/src/controllers/user_controller.py

@@ -0,0 +1,55 @@
+from fastapi import APIRouter, Depends, HTTPException, status
+from typing import List
+from ..services.user_service import UserService
+from ..models.user import UserCreate, UserUpdate, UserResponse
+from ..middleware.auth_middleware import get_current_user, require_admin
+
+router = APIRouter(prefix="/users", tags=["Users"])
+
+
+@router.post("/",
+             response_model=UserResponse,
+             status_code=status.HTTP_201_CREATED)
+async def create_user(user: UserCreate, _=Depends(require_admin)):
+    return await UserService.create_user(user)
+
+
+@router.get("/", response_model=List[UserResponse])
+async def get_users(skip: int = 0, limit: int = 100, _=Depends(require_admin)):
+    return await UserService.get_users(skip, limit)
+
+
+@router.get("/{user_id}", response_model=UserResponse)
+async def get_user(user_id: int,
+                   current_user: dict = Depends(get_current_user)):
+    if current_user["id"] != user_id and not current_user["is_admin"]:
+        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN,
+                            detail="Not authorized to access this user")
+    user = await UserService.get_user(user_id)
+    if not user:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND,
+                            detail="User not found")
+    return user
+
+
+@router.put("/{user_id}", response_model=UserResponse)
+async def update_user(user_id: int,
+                      user: UserUpdate,
+                      current_user: dict = Depends(get_current_user)):
+    if current_user["id"] != user_id and not current_user["is_admin"]:
+        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN,
+                            detail="Not authorized to update this user")
+    updated_user = await UserService.update_user(user_id, user)
+    if not updated_user:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND,
+                            detail="User not found")
+    return updated_user
+
+
+@router.delete("/{user_id}")
+async def delete_user(user_id: int, _=Depends(require_admin)):
+    success = await UserService.delete_user(user_id)
+    if not success:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND,
+                            detail="User not found")
+    return {"message": "User deleted successfully"}

backend/src/data_seeder/__init__.py

@@ -0,0 +1,3 @@
+from .run_seeders import run_all_seeders
+
+__all__ = ['run_all_seeders']

backend/src/data_seeder/base_seeder.py

@@ -0,0 +1,30 @@
+from abc import ABC, abstractmethod
+from typing import Optional
+from ..database import db
+
+
+class BaseSeeder(ABC):
+    """数据种子基类"""
+
+    @abstractmethod
+    def run(self):
+        """执行种子数据初始化"""
+        pass
+
+    def commit(self):
+        """提交数据库事务"""
+        db.session.commit()
+
+    def rollback(self):
+        """回滚数据库事务"""
+        db.session.rollback()
+
+    def execute(self):
+        """执行种子数据初始化，包含事务处理"""
+        try:
+            self.run()
+            self.commit()
+            return True
+        except Exception as e:
+            self.rollback()
+            raise e

backend/src/data_seeder/dict_seeder.py

@@ -0,0 +1,100 @@
+from .base_seeder import BaseSeeder
+from ..models.dict_type_model import DictTypeModel
+from ..models.dict_data_model import DictDataModel
+
+
+class DictSeeder(BaseSeeder):
+    """字典数据种子"""
+
+    def run(self):
+        # 初始化题目类型
+        question_type = DictTypeModel(type_name='题目类型',
+                                      type_code='question_type',
+                                      description='题目类型字典',
+                                      is_system=True)
+
+        # 初始化题目难度
+        question_difficulty = DictTypeModel(type_name='题目难度',
+                                            type_code='question_difficulty',
+                                            description='题目难度字典',
+                                            is_system=True)
+
+        # 初始化题目分类
+        question_category = DictTypeModel(type_name='题目分类',
+                                          type_code='question_category',
+                                          description='题目分类字典',
+                                          is_system=True)
+
+        # 添加字典类型
+        self._add_dict_types(
+            [question_type, question_difficulty, question_category])
+
+        # 初始化默认数据
+        self._init_default_data(question_type, question_difficulty,
+                                question_category)
+
+    def _add_dict_types(self, dict_types):
+        """添加字典类型"""
+        for dict_type in dict_types:
+            if not DictTypeModel.query.filter_by(
+                    type_code=dict_type.type_code).first():
+                self.session.add(dict_type)
+
+    def _init_default_data(self, question_type, question_difficulty,
+                           question_category):
+        """初始化默认字典数据"""
+        # 题目类型默认数据
+        type_data = [
+            {
+                'label': '单选题',
+                'value': 'single_choice'
+            },
+            {
+                'label': '多选题',
+                'value': 'multi_choice'
+            },
+            {
+                'label': '判断题',
+                'value': 'true_false'
+            },
+            {
+                'label': '填空题',
+                'value': 'fill_blank'
+            },
+            {
+                'label': '简答题',
+                'value': 'short_answer'
+            },
+        ]
+
+        # 题目难度默认数据
+        difficulty_data = [
+            {
+                'label': '简单',
+                'value': 'easy'
+            },
+            {
+                'label': '中等',
+                'value': 'medium'
+            },
+            {
+                'label': '困难',
+                'value': 'hard'
+            },
+        ]
+
+        # 添加题目类型数据
+        self._add_dict_data(question_type, type_data)
+
+        # 添加题目难度数据
+        self._add_dict_data(question_difficulty, difficulty_data)
+
+    def _add_dict_data(self, dict_type, data_list):
+        """添加字典数据"""
+        for data in data_list:
+            if not DictDataModel.query.filter_by(dict_type_id=dict_type.id,
+                                                 value=data['value']).first():
+                self.session.add(
+                    DictDataModel(dict_type_id=dict_type.id,
+                                  label=data['label'],
+                                  value=data['value']))

backend/src/data_seeder/run_seeders.py

@@ -0,0 +1,20 @@
+from typing import List
+from .base_seeder import BaseSeeder
+from .dict_seeder import DictSeeder
+from ..database import db
+
+
+def run_all_seeders():
+    """运行所有种子数据初始化"""
+    seeders: List[BaseSeeder] = [DictSeeder()]
+
+    with db.session() as session:
+        for seeder in seeders:
+            try:
+                seeder.execute()
+                print(
+                    f"Seeder {seeder.__class__.__name__} executed successfully"
+                )
+            except Exception as e:
+                print(f"Seeder {seeder.__class__.__name__} failed: {str(e)}")
+                raise e

backend/src/database/__init__.py

@@ -0,0 +1 @@
+# database 包初始化文件

backend/src/exception/__init__.py

@@ -0,0 +1 @@
+# exception 包初始化文件

backend/src/exception/api_exception.py

@@ -0,0 +1,26 @@
+from typing import Any
+from fastapi import status
+from ..common.exceptions import BaseException
+from ..common.constants import ErrorCode
+
+
+class APIException(BaseException):
+    """API异常基类"""
+
+    def __init__(self,
+                 code: int = ErrorCode.UNKNOWN_ERROR,
+                 message: str = "API Error",
+                 detail: Any = None,
+                 http_status: int = status.HTTP_400_BAD_REQUEST):
+        super().__init__(code=code, message=message, detail=detail)
+        self.http_status = http_status
+
+    def to_dict(self):
+        """将异常转换为API响应格式"""
+        return {
+            "error": {
+                "code": self.code,
+                "message": self.message,
+                "detail": self.detail
+            }
+        }

backend/src/exception/business_exception.py

@@ -0,0 +1,18 @@
+from typing import Any
+from fastapi import status
+from .api_exception import APIException
+from ..common.constants import ErrorCode
+
+
+class BusinessException(APIException):
+    """业务异常类"""
+
+    def __init__(self,
+                 code: int = ErrorCode.INVALID_OPERATION,
+                 message: str = "Business Error",
+                 detail: Any = None,
+                 http_status: int = status.HTTP_422_UNPROCESSABLE_ENTITY):
+        super().__init__(code=code,
+                         message=message,
+                         detail=detail,
+                         http_status=http_status)

backend/src/logger/__init__.py

@@ -0,0 +1,14 @@
+# logger 包初始化文件
+from .logger import setup_logging, info, warning, debug, exception, error
+
+# 初始化日志配置
+setup_logging()
+
+__all__ = [
+    "setup_logging",
+    "info",
+    "warning",
+    "debug",
+    "exception",
+    "error",
+]

backend/src/logger/logger.py

@@ -0,0 +1,93 @@
+import logging, os
+import logging.handlers
+from typing import Optional
+from ..config.config_loader import config
+
+
+# 初始化日志系统
+def setup_logging():
+    """初始化日志配置"""
+    logging_config = {
+        "level": config.app.log_level,
+        "console": True,
+        "format":
+        "%(asctime)s %(levelname)s %(name)s:%(lineno)d - %(message)s",
+        "datefmt": "%Y-%m-%d %H:%M:%S",
+        "file":
+        config.logging.get("file") if hasattr(config, "logging") else None
+    }
+
+    logger = logging.getLogger("app")
+    _configure_logger(logger, logging_config)
+    return logger
+
+
+def _configure_logger(logger: logging.Logger, config: dict) -> None:
+    """根据配置初始化日志记录器"""
+    level = config.get("level", "INFO")
+    format = config.get(
+        "format",
+        "%(asctime)s %(levelname)s %(name)s:%(lineno)d - %(message)s")
+    datefmt = config.get("datefmt", "%Y-%m-%d %H:%M:%S")
+
+    logger.setLevel(level)
+    formatter = logging.Formatter(format, datefmt)
+
+    # 控制台输出
+    if config.get("console", True):
+        console_handler = logging.StreamHandler()
+        console_handler.setFormatter(formatter)
+        logger.addHandler(console_handler)
+
+    # 文件输出
+    if config.get("file"):
+        # 将相对路径转换为绝对路径
+        log_file = os.path.join(
+            os.path.dirname(
+                os.path.dirname(os.path.dirname(os.path.dirname(__file__)))),
+            config["file"])
+
+        # 确保日志目录存在
+        log_dir = os.path.dirname(log_file)
+        if not os.path.exists(log_dir):
+            os.makedirs(log_dir, exist_ok=True)
+
+        file_handler = logging.handlers.TimedRotatingFileHandler(
+            log_file,
+            when="midnight",
+            interval=1,
+            backupCount=7,
+            encoding="utf-8")
+        file_handler.setFormatter(formatter)
+        logger.addHandler(file_handler)
+
+
+# 导出常用日志函数
+def debug(msg: str, *args, **kwargs) -> None:
+    """记录调试信息"""
+    logging.getLogger("app").debug(msg, *args, **kwargs)
+
+
+def info(msg: str, *args, **kwargs) -> None:
+    """记录一般信息"""
+    logging.getLogger("app").info(msg, *args, **kwargs)
+
+
+def warning(msg: str, *args, **kwargs) -> None:
+    """记录警告信息"""
+    logging.getLogger("app").warning(msg, *args, **kwargs)
+
+
+def error(msg: str, *args, **kwargs) -> None:
+    """记录错误信息"""
+    logging.getLogger("app").error(msg, *args, **kwargs)
+
+
+def exception(msg: str, *args, **kwargs) -> None:
+    """记录异常信息"""
+    logging.getLogger("app").exception(msg, *args, **kwargs)
+
+
+def get_logger() -> logging.Logger:
+    """获取日志记录器"""
+    return logging.getLogger("app")

backend/src/main.py

@@ -0,0 +1,49 @@
+import logger, data_seeder
+from fastapi import FastAPI
+from fastapi.middleware.cors import CORSMiddleware
+from .middleware.auth_middleware import AuthMiddleware
+from .middleware.exception_handler_middleware import ExceptionHandlerMiddleware
+from .router import register_routers
+
+
+def create_app() -> FastAPI:
+
+    logger.info("欢迎使用智能题库系统后端API文档")
+
+    # 初始化种子数据
+    data_seeder.run_all_seeders()
+
+    app = FastAPI(title="智能题库系统API",
+                  description="智能题库系统后端API文档",
+                  version="1.0.0")
+
+    # 注册中间件
+    def register_middlewares(app: FastAPI):
+        # 异常处理中间件（最先注册）
+        app.add_middleware(ExceptionHandlerMiddleware)
+
+        # CORS中间件
+        app.add_middleware(
+            CORSMiddleware,
+            allow_origins=["*"],
+            allow_credentials=True,
+            allow_methods=["*"],
+            allow_headers=["*"],
+        )
+
+        # HTTP中间件（在路由之前注册）
+        app.middleware("http")(AuthMiddleware.auth_required)
+
+    # 注册路由
+    app.include_router(register_routers())
+
+    # 注册所有中间件
+    register_middlewares(app)
+
+    return app
+
+
+if __name__ == '__main__':
+    import uvicorn
+    app = create_app()
+    uvicorn.run(app, host="0.0.0.0", port=5000, log_level="info")

backend/src/middleware/__init__.py

@@ -0,0 +1,3 @@
+from .exception_handler_middleware import ExceptionHandlerMiddleware
+
+__all__ = ["ExceptionHandlerMiddleware"]

backend/src/middleware/auth_middleware.py

@@ -0,0 +1,90 @@
+from flask import request, g
+from ..services.auth_service import AuthService
+from ..exception.api_exception import ApiException
+from ..common.constants import ErrorCode
+
+
+class AuthMiddleware:
+    """认证中间件"""
+
+    @staticmethod
+    def auth_required(f):
+        """认证装饰器"""
+
+        def wrapper(*args, **kwargs):
+            token = request.headers.get('Authorization')
+            if not token:
+                raise ApiException(ErrorCode.UNAUTHORIZED)
+
+            # 验证token并获取当前用户
+            db = g.db
+            user = AuthService.get_current_user(db, token)
+            if not user:
+                raise ApiException(ErrorCode.UNAUTHORIZED)
+
+            # 将当前用户存入g对象
+            g.current_user = user
+            return f(*args, **kwargs)
+
+        return wrapper
+
+    @staticmethod
+    def permission_required(permission_code: str):
+        """权限装饰器"""
+
+        def decorator(f):
+
+            def wrapper(*args, **kwargs):
+                # 确保用户已认证
+                if not hasattr(g, 'current_user'):
+                    raise ApiException(ErrorCode.UNAUTHORIZED)
+
+                # 获取用户所有权限
+                user_permissions = set()
+                for role in g.current_user.roles:
+                    for permission in role.permissions:
+                        user_permissions.add(permission.code)
+
+                # 检查权限
+                if permission_code not in user_permissions:
+                    raise ApiException(ErrorCode.FORBIDDEN)
+
+                return f(*args, **kwargs)
+
+            return wrapper
+
+        return decorator
+
+    @staticmethod
+    def role_or_permission_required(role_name: str = None,
+                                    permission_code: str = None):
+        """角色或权限装饰器"""
+
+        def decorator(f):
+
+            def wrapper(*args, **kwargs):
+                # 确保用户已认证
+                if not hasattr(g, 'current_user'):
+                    raise ApiException(ErrorCode.UNAUTHORIZED)
+
+                # 检查角色
+                if role_name:
+                    user_roles = [role.name for role in g.current_user.roles]
+                    if role_name in user_roles:
+                        return f(*args, **kwargs)
+
+                # 检查权限
+                if permission_code:
+                    user_permissions = set()
+                    for role in g.current_user.roles:
+                        for permission in role.permissions:
+                            user_permissions.add(permission.code)
+
+                    if permission_code in user_permissions:
+                        return f(*args, **kwargs)
+
+                raise ApiException(ErrorCode.FORBIDDEN)
+
+            return wrapper
+
+        return decorator

backend/src/middleware/exception_handler_middleware.py

@@ -0,0 +1,49 @@
+from fastapi import Request, Response
+from fastapi.responses import JSONResponse
+from starlette.middleware.base import BaseHTTPMiddleware
+from typing import Optional, Dict, Any
+import logging
+
+from ..exception.api_exception import APIException
+from ..exception.business_exception import BusinessException
+from ..common.constants import ErrorCode
+
+logger = logging.getLogger(__name__)
+
+
+class ExceptionHandlerMiddleware(BaseHTTPMiddleware):
+
+    async def dispatch(self, request: Request, call_next):
+        try:
+            response = await call_next(request)
+            return response
+        except BusinessException as e:
+            logger.warning(f"Business Exception: {str(e)}")
+            return JSONResponse(status_code=e.http_status,
+                                content={
+                                    "error": {
+                                        "code": e.code,
+                                        "message": e.message,
+                                        "detail": e.detail
+                                    }
+                                })
+        except APIException as e:
+            logger.error(f"API Exception: {str(e)}")
+            return JSONResponse(status_code=e.http_status,
+                                content={
+                                    "error": {
+                                        "code": e.code,
+                                        "message": e.message,
+                                        "detail": e.detail
+                                    }
+                                })
+        except Exception as e:
+            logger.exception(f"Unexpected Exception: {str(e)}")
+            return JSONResponse(status_code=500,
+                                content={
+                                    "error": {
+                                        "code": ErrorCode.UNKNOWN_ERROR,
+                                        "message": "Internal Server Error",
+                                        "detail": str(e)
+                                    }
+                                })

backend/src/models/__init__.py

@@ -0,0 +1,45 @@
+# models 包初始化文件
+from .base_model import BaseModel, CreateModel
+from .user import User
+from .question import Question
+from .question_bank import QuestionBank
+from .question_bank_permission import QuestionBankPermission
+from .question_difficulty import QuestionDifficulty
+from .question_type import QuestionType
+from .question_tag import QuestionTag
+from .question_version import QuestionVersion
+from .question_relation import QuestionRelation
+from .question_template import QuestionTemplate
+from .question_category import QuestionCategory
+from .question_statistics import QuestionStatistics
+from .user_statistics import UserStatistics
+from .learning_progress import LearningProgress
+from .question_comment import QuestionComment
+from .question_attachment import QuestionAttachment
+from .question_history import QuestionHistory
+from .question_option import QuestionOption
+from .question_feedback import QuestionFeedback
+from .user_behavior import UserBehavior
+from .practice_record import PracticeRecord
+from .knowledge_point import KnowledgePoint
+from .learning_path import LearningPath
+from .question_collection import QuestionCollection
+from .wrong_question import WrongQuestion
+from .question_analysis import QuestionAnalysis
+from .exam_record import ExamRecord
+from .exam_paper import ExamPaper
+from .exam_score import ExamScore
+from .exam_analysis import ExamAnalysis
+from .question_knowledge_point import QuestionKnowledgePoint
+
+__all__ = [
+    'BaseModel', 'CreateModel', 'User', 'Question', 'QuestionBank',
+    'QuestionBankPermission', 'QuestionDifficulty', 'QuestionType',
+    'QuestionTag', 'QuestionVersion', 'QuestionRelation', 'QuestionTemplate',
+    'QuestionCategory', 'QuestionStatistics', 'UserStatistics',
+    'LearningProgress', 'QuestionComment', 'QuestionAttachment',
+    'QuestionHistory', 'QuestionOption', 'QuestionFeedback', 'UserBehavior',
+    'PracticeRecord', 'KnowledgePoint', 'LearningPath', 'QuestionCollection',
+    'WrongQuestion', 'QuestionAnalysis', 'ExamRecord', 'ExamPaper',
+    'ExamScore', 'ExamAnalysis', 'QuestionKnowledgePoint'
+]

backend/src/models/auth_models.py

@@ -0,0 +1,33 @@
+from datetime import datetime
+from sqlalchemy import Column, String, DateTime, ForeignKey
+from sqlalchemy.orm import relationship
+from .base_model import BaseModel
+
+
+class Token(BaseModel):
+    """认证令牌模型"""
+    __tablename__ = "token"
+
+    access_token = Column(String(512), nullable=False, comment="访问令牌")
+    refresh_token = Column(String(512), nullable=False, comment="刷新令牌")
+    token_type = Column(String(50), default="bearer", comment="令牌类型")
+    expires_at = Column(DateTime, nullable=False, comment="过期时间")
+    user_id = Column(Integer,
+                     ForeignKey("user.id"),
+                     nullable=False,
+                     comment="用户ID")
+
+    user = relationship("User", back_populates="tokens")
+
+
+class TokenData(BaseModel):
+    """令牌数据模型"""
+    __tablename__ = "token_data"
+
+    username = Column(String(255), nullable=True, comment="用户名")
+    token_id = Column(Integer,
+                      ForeignKey("token.id"),
+                      nullable=False,
+                      comment="关联令牌ID")
+
+    token = relationship("Token", back_populates="token_data")

backend/src/models/base_model.py

@@ -0,0 +1,87 @@
+from datetime import datetime
+from typing import Any, Dict, Optional
+from sqlalchemy import Column, Integer, DateTime
+from sqlalchemy.ext.declarative import as_declarative, declared_attr
+from sqlalchemy.orm import Session
+from sqlalchemy.sql import func
+
+
+@as_declarative()
+class BaseModel:
+    id = Column(Integer, primary_key=True, index=True, autoincrement=True)
+    created_at = Column(DateTime, server_default=func.now(), comment="创建时间")
+    updated_at = Column(DateTime,
+                        server_default=func.now(),
+                        onupdate=func.now(),
+                        comment="更新时间")
+    deleted_at = Column(DateTime, nullable=True, comment="删除时间")
+
+    @declared_attr
+    def __tablename__(cls) -> str:
+        return cls.__name__.lower()
+
+    def to_dict(self) -> Dict[str, Any]:
+        """将模型实例转换为字典"""
+        return {c.name: getattr(self, c.name) for c in self.__table__.columns}
+
+    @classmethod
+    def create(cls, db: Session, **kwargs) -> "BaseModel":
+        """创建新记录"""
+        instance = cls(**kwargs)
+        db.add(instance)
+        db.commit()
+        db.refresh(instance)
+        return instance
+
+    @classmethod
+    def get(cls, db: Session, id: int) -> Optional["BaseModel"]:
+        """根据ID获取记录"""
+        return db.query(cls).filter(cls.id == id).first()
+
+    @classmethod
+    def update(cls, db: Session, id: int, **kwargs) -> Optional["BaseModel"]:
+        """更新记录"""
+        instance = cls.get(db, id)
+        if instance:
+            for key, value in kwargs.items():
+                setattr(instance, key, value)
+            db.commit()
+            db.refresh(instance)
+        return instance
+
+    @classmethod
+    def delete(cls, db: Session, id: int) -> bool:
+        """软删除记录"""
+        instance = cls.get(db, id)
+        if instance:
+            instance.deleted_at = datetime.now()
+            db.commit()
+            return True
+        return False
+
+    @classmethod
+    def hard_delete(cls, db: Session, id: int) -> bool:
+        """硬删除记录"""
+        instance = cls.get(db, id)
+        if instance:
+            db.delete(instance)
+            db.commit()
+            return True
+        return False
+
+    @classmethod
+    def exists(cls, db: Session, id: int) -> bool:
+        """检查记录是否存在"""
+        return db.query(cls).filter(cls.id == id).first() is not None
+
+    @classmethod
+    def get_all(cls,
+                db: Session,
+                skip: int = 0,
+                limit: int = 100,
+                include_deleted: bool = False) -> list["BaseModel"]:
+        """获取所有记录"""
+        query = db.query(cls)
+        if not include_deleted:
+            query = query.filter(cls.deleted_at.is_(None))
+        return query.offset(skip).limit(limit).all()

backend/src/models/dict_data_model.py

@@ -0,0 +1,21 @@
+from sqlalchemy import Column, String, Boolean, Integer, ForeignKey
+from sqlalchemy.orm import relationship
+from .base_model import BaseModel
+
+
+class DictDataModel(BaseModel):
+    """字典数据模型"""
+    dict_type_id = Column(Integer,
+                          ForeignKey('dicttypemodel.id'),
+                          nullable=False,
+                          comment="字典类型ID")
+    label = Column(String(50), nullable=False, comment="字典标签")
+    value = Column(String(50), nullable=False, comment="字典值")
+    sort = Column(Integer, default=0, nullable=False, comment="排序")
+    status = Column(Boolean, default=True, nullable=False, comment="状态（启用/禁用）")
+
+    # 与 DictTypeModel 的关系
+    dict_type = relationship('DictTypeModel', backref='dict_data')
+
+    def __repr__(self) -> str:
+        return f"<DictDataModel(id={self.id}, label='{self.label}')>"

backend/src/models/dict_type_model.py

@@ -0,0 +1,20 @@
+from sqlalchemy import Column, String, Boolean, Integer
+from .base_model import BaseModel
+
+
+class DictTypeModel(BaseModel):
+    """字典类型模型"""
+    type_name = Column(String(50), nullable=False, comment="字典类型名称")
+    type_code = Column(String(50),
+                       unique=True,
+                       nullable=False,
+                       comment="字典类型代码")
+    description = Column(String(255), nullable=True, comment="描述")
+    is_system = Column(Boolean,
+                       default=False,
+                       nullable=False,
+                       comment="是否系统内置")
+    status = Column(Boolean, default=True, nullable=False, comment="状态（启用/禁用）")
+
+    def __repr__(self) -> str:
+        return f"<DictTypeModel(id={self.id}, type_name='{self.type_name}')>"

backend/src/models/exam_analysis.py

@@ -0,0 +1,25 @@
+from ..models.base_model import BaseModel, CreateModel
+from ..database import db
+
+
+class ExamAnalysis(CreateModel):
+    __tablename__ = 'exam_analyses'
+
+    id = db.Column(db.Integer, primary_key=True)
+    exam_record_id = db.Column(db.Integer,
+                               db.ForeignKey('exam_records.id'),
+                               nullable=False)
+    user_id = db.Column(db.Integer, db.ForeignKey('users.id'), nullable=False)
+    analysis_data = db.Column(db.JSON, nullable=False)  # 分析结果数据
+    create_time = db.Column(db.DateTime, default=db.func.now())
+    update_time = db.Column(db.DateTime, onupdate=db.func.now())
+
+    # 分析类型：weakness/strength/improvement
+    analysis_type = db.Column(db.String(20), nullable=False)
+
+    # 关系定义
+    exam_record = db.relationship('ExamRecord', foreign_keys=[exam_record_id])
+    user = db.relationship('User', foreign_keys=[user_id])
+
+    def __repr__(self):
+        return f'<ExamAnalysis {self.id}: Record {self.exam_record_id} - Type {self.analysis_type}>'

backend/src/models/exam_paper.py

@@ -0,0 +1,27 @@
+from ..models.base_model import BaseModel, CreateModel
+from ..database import db
+from sqlalchemy import JSON
+
+
+class ExamPaper(CreateModel):
+    __tablename__ = 'exam_papers'
+
+    id = db.Column(db.Integer, primary_key=True)
+    name = db.Column(db.String(100), nullable=False)
+    description = db.Column(db.Text)
+    duration = db.Column(db.Integer)  # 考试时长（分钟）
+    total_score = db.Column(db.Float)
+    difficulty = db.Column(db.String(20))
+    status = db.Column(db.String(20), default='draft')
+    tags = db.Column(JSON)
+    creator_id = db.Column(db.Integer, db.ForeignKey('users.id'))
+    create_time = db.Column(db.DateTime, default=db.func.now())
+    update_time = db.Column(db.DateTime, onupdate=db.func.now())
+
+    # 关系定义
+    creator = db.relationship('User', foreign_keys=[creator_id])
+    questions = db.relationship('Question', secondary='exam_paper_questions')
+    exam_records = db.relationship('ExamRecord', backref='exam_paper')
+
+    def __repr__(self):
+        return f'<ExamPaper {self.id}: {self.name}>'

backend/src/models/exam_record.py

@@ -0,0 +1,27 @@
+from ..models.base_model import BaseModel, CreateModel
+from ..database import db
+from sqlalchemy import JSON
+
+
+class ExamRecord(CreateModel):
+    __tablename__ = 'exam_records'
+
+    id = db.Column(db.Integer, primary_key=True)
+    user_id = db.Column(db.Integer, db.ForeignKey('users.id'), nullable=False)
+    exam_paper_id = db.Column(db.Integer,
+                              db.ForeignKey('exam_papers.id'),
+                              nullable=False)
+    start_time = db.Column(db.DateTime, nullable=False)
+    end_time = db.Column(db.DateTime)
+    score = db.Column(db.Float)
+    status = db.Column(db.String(20),
+                       default='in_progress')  # in_progress/completed
+    answers = db.Column(JSON)  # 存储用户答案
+    analysis = db.Column(JSON)  # 考试分析数据
+
+    # 关系定义
+    user = db.relationship('User', foreign_keys=[user_id])
+    exam_paper = db.relationship('ExamPaper', foreign_keys=[exam_paper_id])
+
+    def __repr__(self):
+        return f'<ExamRecord {self.id}: User {self.user_id} - Paper {self.exam_paper_id}>'

backend/src/models/exam_score.py

@@ -0,0 +1,23 @@
+from ..models.base_model import BaseModel, CreateModel
+from ..database import db
+
+
+class ExamScore(CreateModel):
+    __tablename__ = 'exam_scores'
+
+    id = db.Column(db.Integer, primary_key=True)
+    user_id = db.Column(db.Integer, db.ForeignKey('users.id'), nullable=False)
+    exam_record_id = db.Column(db.Integer,
+                               db.ForeignKey('exam_records.id'),
+                               nullable=False)
+    total_score = db.Column(db.Float, nullable=False)
+    score_details = db.Column(db.JSON)  # 各题型得分详情
+    ranking = db.Column(db.Integer)  # 本次考试的排名
+    create_time = db.Column(db.DateTime, default=db.func.now())
+
+    # 关系定义
+    user = db.relationship('User', foreign_keys=[user_id])
+    exam_record = db.relationship('ExamRecord', foreign_keys=[exam_record_id])
+
+    def __repr__(self):
+        return f'<ExamScore {self.id}: User {self.user_id} - Score {self.total_score}>'

backend/src/models/knowledge_point.py

@@ -0,0 +1,32 @@
+from sqlalchemy import Column, Integer, String, Text, ForeignKey
+from sqlalchemy.orm import relationship
+from .base_model import BaseModel
+from .question_knowledge_point import QuestionKnowledgePoint
+
+
+class KnowledgePoint(BaseModel):
+    """知识点模型"""
+    __tablename__ = "knowledge_point"
+
+    name = Column(String(255), nullable=False, comment="知识点名称")
+    description = Column(Text, comment="知识点描述")
+    parent_id = Column(Integer,
+                       ForeignKey("knowledge_point.id"),
+                       comment="父知识点ID")
+    level = Column(Integer, default=1, comment="知识点层级")
+    order = Column(Integer, default=0, comment="排序序号")
+    status = Column(Integer, default=1, comment="状态 1:正常 0:禁用")
+
+    # 关系定义
+    questions = relationship("Question",
+                             secondary="question_knowledge_point",
+                             back_populates="knowledge_points")
+    children = relationship("KnowledgePoint",
+                            back_populates="parent",
+                            remote_side=[parent_id])
+    parent = relationship("KnowledgePoint",
+                          back_populates="children",
+                          remote_side=[parent_id])
+
+    def __repr__(self):
+        return f"<KnowledgePoint(id={self.id}, name={self.name})>"

backend/src/models/learning_path.py

@@ -0,0 +1 @@
+# 学习路径模型

backend/src/models/learning_progress.py

@@ -0,0 +1 @@
+# 学习进度模型

backend/src/models/permission.py

@@ -0,0 +1,26 @@
+from ..models.base_model import BaseModel
+from sqlalchemy import Column, String, Integer
+from sqlalchemy.orm import relationship
+
+
+class Permission(BaseModel):
+    """权限模型"""
+    __tablename__ = 'permissions'
+
+    id = Column(Integer, primary_key=True, autoincrement=True)
+    code = Column(String(50),
+                  unique=True,
+                  nullable=False,
+                  index=True,
+                  comment='权限代码')
+    name = Column(String(50), nullable=False)
+    description = Column(String(255))
+    scope = Column(String(50), nullable=False, default='*', comment='权限作用域')
+
+    # 与角色的关联
+    roles = relationship('Role',
+                         secondary='role_permissions',
+                         back_populates='permissions')
+
+    def __repr__(self):
+        return f"<Permission(id={self.id}, name={self.name})>"

backend/src/models/practice_record.py

@@ -0,0 +1 @@
+# 练习记录模型

backend/src/models/question.py

@@ -0,0 +1,98 @@
+from typing import List, Optional
+from sqlalchemy import Column, String, Text, ForeignKey, JSON
+from sqlalchemy.orm import relationship
+from .base_model import BaseModel
+from .question_type import QuestionType
+from .question_difficulty import QuestionDifficulty
+from .question_category import QuestionCategory
+from .question_tag import QuestionTag
+from .question_option import QuestionOption
+from .question_analysis import QuestionAnalysis
+from .question_feedback import QuestionFeedback
+from .question_comment import QuestionComment
+from .question_attachment import QuestionAttachment
+from .question_version import QuestionVersion
+from .question_relation import QuestionRelation
+
+
+class Question(BaseModel):
+    """题目模型"""
+    __tablename__ = "question"
+
+    title = Column(String(255), nullable=False, comment="题目标题")
+    content = Column(Text, nullable=False, comment="题目内容")
+    type_id = Column(Integer,
+                     ForeignKey("questiontype.id"),
+                     nullable=False,
+                     comment="题目类型ID")
+    difficulty_id = Column(Integer,
+                           ForeignKey("questiondifficulty.id"),
+                           nullable=False,
+                           comment="难度等级ID")
+    category_id = Column(Integer,
+                         ForeignKey("questioncategory.id"),
+                         nullable=True,
+                         comment="分类ID")
+    tags = Column(JSON, nullable=True, comment="标签列表")
+    answer = Column(Text, nullable=True, comment="题目答案")
+    explanation = Column(Text, nullable=True, comment="题目解析")
+    status = Column(Integer, default=1, comment="题目状态 1:正常 0:禁用")
+    is_public = Column(Integer, default=0, comment="是否公开 1:是 0:否")
+    creator_id = Column(Integer, nullable=False, comment="创建者ID")
+    reviewer_id = Column(Integer, nullable=True, comment="审核者ID")
+    review_status = Column(Integer,
+                           default=0,
+                           comment="审核状态 0:待审核 1:审核通过 2:审核不通过")
+    review_comment = Column(Text, nullable=True, comment="审核意见")
+    source = Column(String(255), nullable=True, comment="题目来源")
+    extra_data = Column(JSON, nullable=True, comment="扩展数据")
+
+    # 关系定义
+    type = relationship("QuestionType", back_populates="questions")
+    difficulty = relationship("QuestionDifficulty", back_populates="questions")
+    category = relationship("QuestionCategory", back_populates="questions")
+    options = relationship("QuestionOption", back_populates="question")
+    analyses = relationship("QuestionAnalysis", back_populates="question")
+    feedbacks = relationship("QuestionFeedback", back_populates="question")
+    comments = relationship("QuestionComment", back_populates="question")
+    attachments = relationship("QuestionAttachment", back_populates="question")
+    versions = relationship("QuestionVersion", back_populates="question")
+    relations = relationship("QuestionRelation", back_populates="question")
+    knowledge_points = relationship("KnowledgePoint",
+                                    secondary="question_knowledge_point",
+                                    back_populates="questions")
+
+    def __repr__(self):
+        return f"<Question(id={self.id}, title={self.title})>"
+
+    @classmethod
+    def create_with_options(cls, db: Session, title: str, content: str,
+                            type_id: int, difficulty_id: int,
+                            options: List[Dict[str,
+                                               Any]], **kwargs) -> "Question":
+        """创建题目并添加选项"""
+        question = cls.create(db,
+                              title=title,
+                              content=content,
+                              type_id=type_id,
+                              difficulty_id=difficulty_id,
+                              **kwargs)
+        for option_data in options:
+            QuestionOption.create(db, question_id=question.id, **option_data)
+        return question
+
+    @classmethod
+    def get_with_details(cls, db: Session, id: int) -> Optional["Question"]:
+        """获取题目详情"""
+        return (db.query(cls).options(
+            joinedload(cls.type),
+            joinedload(cls.difficulty),
+            joinedload(cls.category),
+            joinedload(cls.options),
+            joinedload(cls.analyses),
+            joinedload(cls.feedbacks),
+            joinedload(cls.comments),
+            joinedload(cls.attachments),
+            joinedload(cls.versions),
+            joinedload(cls.relations),
+        ).filter(cls.id == id).first())

backend/src/models/question_analysis.py

@@ -0,0 +1 @@
+# 题目分析模型

backend/src/models/question_attachment.py

@@ -0,0 +1,19 @@
+from ..models.base_model import BaseModel
+from ..database import db
+
+
+class QuestionAttachment(BaseModel):
+    __tablename__ = 'question_attachments'
+
+    id = db.Column(db.Integer, primary_key=True)
+    question_id = db.Column(db.Integer,
+                            db.ForeignKey('questions.id'),
+                            nullable=False)
+    file_name = db.Column(db.String(255), nullable=False)
+    file_path = db.Column(db.String(255), nullable=False)
+    file_type = db.Column(db.String(50), nullable=False)
+    file_size = db.Column(db.Integer, nullable=False)
+    upload_time = db.Column(db.DateTime, default=db.func.now())
+
+    def __repr__(self):
+        return f'<QuestionAttachment {self.id}: {self.file_name}>'

backend/src/models/question_bank.py

@@ -0,0 +1,50 @@
+from ..models.base_model import BaseModel, CreateModel
+from ..database import db
+
+
+class QuestionBank(CreateModel):
+    __tablename__ = 'question_banks'
+
+    id = db.Column(db.Integer, primary_key=True)
+    name = db.Column(db.String(100), nullable=False)
+    description = db.Column(db.Text)
+    subject = db.Column(db.String(50), nullable=False)
+    creator_id = db.Column(db.Integer, db.ForeignKey('users.id'))
+    create_time = db.Column(db.DateTime, default=db.func.now())
+    update_time = db.Column(db.DateTime,
+                            default=db.func.now(),
+                            onupdate=db.func.now())
+    is_deleted = db.Column(db.Boolean, default=False)
+    is_public = db.Column(db.Boolean, default=False)  # 题库是否公开
+    question_count = db.Column(db.Integer, default=0)  # 题目数量统计
+    average_difficulty = db.Column(db.Float, default=0.0)  # 平均难度
+    last_used_time = db.Column(db.DateTime)  # 最后使用时间
+
+    # 关系定义
+    creator = db.relationship('User', foreign_keys=[creator_id])
+    questions = db.relationship('Question', backref='question_bank')
+    permissions = db.relationship('QuestionBankPermission',
+                                  backref='question_bank')  # 题库权限
+
+    def __repr__(self):
+        return f'<QuestionBank {self.id}: {self.name}>'
+
+    def update_statistics(self):
+        """更新题库统计信息"""
+        self.question_count = len(self.questions)
+        if self.question_count > 0:
+            total_difficulty = sum(q.difficulty.level for q in self.questions)
+            self.average_difficulty = total_difficulty / self.question_count
+        else:
+            self.average_difficulty = 0.0
+
+    def has_permission(self, user_id: int, permission_type: str) -> bool:
+        """检查用户是否有指定权限"""
+        if self.creator_id == user_id:
+            return True
+
+        permission = next(
+            (p for p in self.permissions
+             if p.user_id == user_id and p.permission_type == permission_type),
+            None)
+        return permission is not None

backend/src/models/question_collection.py

@@ -0,0 +1 @@
+# 题目收藏模型

backend/src/models/question_comment.py

@@ -0,0 +1,22 @@
+from ..models.base_model import BaseModel
+from ..database import db
+
+
+class QuestionComment(BaseModel):
+    __tablename__ = 'question_comments'
+
+    id = db.Column(db.Integer, primary_key=True)
+    question_id = db.Column(db.Integer,
+                            db.ForeignKey('questions.id'),
+                            nullable=False)
+    user_id = db.Column(db.Integer, db.ForeignKey('users.id'), nullable=False)
+    content = db.Column(db.Text, nullable=False)
+    parent_id = db.Column(db.Integer, db.ForeignKey('question_comments.id'))
+    create_time = db.Column(db.DateTime, default=db.func.now())
+
+    # 关系定义
+    replies = db.relationship('QuestionComment',
+                              backref=db.backref('parent', remote_side=[id]))
+
+    def __repr__(self):
+        return f'<QuestionComment {self.id}: {self.content[:50]}...>'

backend/src/models/question_feedback.py

@@ -0,0 +1,18 @@
+from ..models.base_model import BaseModel
+from ..database import db
+
+
+class QuestionFeedback(BaseModel):
+    __tablename__ = 'question_feedbacks'
+
+    id = db.Column(db.Integer, primary_key=True)
+    question_id = db.Column(db.Integer,
+                            db.ForeignKey('questions.id'),
+                            nullable=False)
+    user_id = db.Column(db.Integer, db.ForeignKey('users.id'), nullable=False)
+    content = db.Column(db.Text, nullable=False)
+    rating = db.Column(db.Integer)
+    create_time = db.Column(db.DateTime, default=db.func.now())
+
+    def __repr__(self):
+        return f'<QuestionFeedback {self.id}: {self.content[:50]}...>'

backend/src/models/question_history.py

@@ -0,0 +1,22 @@
+from ..models.base_model import BaseModel
+from ..database import db
+
+
+class QuestionHistory(BaseModel):
+    __tablename__ = 'question_histories'
+
+    id = db.Column(db.Integer, primary_key=True)
+    question_id = db.Column(db.Integer,
+                            db.ForeignKey('questions.id'),
+                            nullable=False)
+    user_id = db.Column(db.Integer, db.ForeignKey('users.id'), nullable=False)
+    action = db.Column(db.String(50),
+                       nullable=False)  # 操作类型：create/update/delete
+    content = db.Column(db.Text)  # 变更内容（JSON格式）
+    operator_id = db.Column(db.Integer,
+                            db.ForeignKey('users.id'),
+                            nullable=False)
+    operate_time = db.Column(db.DateTime, default=db.func.now())
+
+    def __repr__(self):
+        return f'<QuestionHistory {self.id}: {self.action}>'

backend/src/models/question_knowledge_point.py

@@ -0,0 +1,12 @@
+from sqlalchemy import Column, Integer, ForeignKey
+from .base_model import BaseModel
+
+
+class QuestionKnowledgePoint(BaseModel):
+    """题目-知识点关联表"""
+    __tablename__ = "question_knowledge_point"
+
+    question_id = Column(Integer, ForeignKey("question.id"), primary_key=True)
+    knowledge_point_id = Column(Integer,
+                                ForeignKey("knowledge_point.id"),
+                                primary_key=True)

backend/src/models/question_option.py

@@ -0,0 +1,17 @@
+from ..models.base_model import BaseModel
+from ..database import db
+
+
+class QuestionOption(BaseModel):
+    __tablename__ = 'question_options'
+
+    id = db.Column(db.Integer, primary_key=True)
+    question_id = db.Column(db.Integer,
+                            db.ForeignKey('questions.id'),
+                            nullable=False)
+    content = db.Column(db.Text, nullable=False)
+    is_correct = db.Column(db.Boolean, default=False)
+    order = db.Column(db.Integer, default=0)
+
+    def __repr__(self):
+        return f'<QuestionOption {self.id}: {self.content[:50]}...>'

backend/src/models/question_relation.py

@@ -0,0 +1 @@
+# 题目关联模型

backend/src/models/question_statistics.py

@@ -0,0 +1 @@
+# 题目统计模型

backend/src/models/role.py

@@ -0,0 +1,26 @@
+from ..models.base_model import BaseModel
+from sqlalchemy import Column, String, Integer
+from sqlalchemy.orm import relationship
+
+
+class Role(BaseModel):
+    """角色模型"""
+    __tablename__ = 'roles'
+
+    id = Column(Integer, primary_key=True, autoincrement=True)
+    name = Column(String(50), unique=True, nullable=False)  # 角色名称
+    description = Column(String(255))  # 角色描述
+    is_default = Column(Integer, default=0)  # 是否为默认角色
+
+    # 与用户的关联
+    users = relationship('User',
+                         secondary='user_roles',
+                         back_populates='roles')
+
+    # 与权限的关联
+    permissions = relationship('Permission',
+                               secondary='role_permissions',
+                               back_populates='roles')
+
+    def __repr__(self):
+        return f"<Role(id={self.id}, name={self.name})>"

backend/src/models/role_permission.py

@@ -0,0 +1,17 @@
+from ..models.base_model import BaseModel
+from sqlalchemy import Column, Integer, ForeignKey
+
+
+class RolePermission(BaseModel):
+    """角色权限关联模型"""
+    __tablename__ = 'role_permissions'
+
+    id = Column(Integer, primary_key=True, autoincrement=True)
+    role_id = Column(Integer, ForeignKey('roles.id'), nullable=False)
+    permission_id = Column(Integer,
+                           ForeignKey('permissions.id'),
+                           nullable=False)
+    scope = Column(String(50), nullable=False, default='*', comment='权限作用域')
+
+    def __repr__(self):
+        return f"<RolePermission(role_id={self.role_id}, permission_id={self.permission_id})>"

backend/src/models/user.py

@@ -0,0 +1,32 @@
+from ..models.base_model import BaseModel
+from sqlalchemy import Column, String, Integer, DateTime, ForeignKey
+from sqlalchemy.orm import relationship
+from datetime import datetime
+
+
+class User(BaseModel):
+    """用户模型"""
+    __tablename__ = 'users'
+
+    id = Column(Integer, primary_key=True, autoincrement=True)
+    username = Column(String(50), unique=True, nullable=False)
+    password_hash = Column(String(255), nullable=False)  # 密码哈希
+    last_login = Column(DateTime)  # 最后登录时间
+    is_active = Column(Integer, default=1)  # 是否激活
+    created_at = Column(DateTime, default=datetime.now)  # 创建时间
+    reset_token = Column(String(32))  # 密码重置token
+    reset_token_expires = Column(DateTime)  # 重置token过期时间
+
+    # 与角色的关联
+    roles = relationship('Role',
+                         secondary='user_roles',
+                         back_populates='users')
+
+    # 与用户统计信息的关联
+    statistics = relationship('UserStatistics', back_populates='user')
+
+    # 与认证令牌的关联
+    tokens = relationship('Token', back_populates='user')
+
+    def __repr__(self):
+        return f"<User(id={self.id}, username={self.username})>"

backend/src/models/user_behavior.py

@@ -0,0 +1 @@
+# 用户行为模型

backend/src/models/user_role.py

@@ -0,0 +1,14 @@
+from ..models.base_model import BaseModel
+from sqlalchemy import Column, Integer, ForeignKey
+
+
+class UserRole(BaseModel):
+    """用户角色关联模型"""
+    __tablename__ = 'user_roles'
+
+    id = Column(Integer, primary_key=True, autoincrement=True)
+    user_id = Column(Integer, ForeignKey('users.id'), nullable=False)
+    role_id = Column(Integer, ForeignKey('roles.id'), nullable=False)
+
+    def __repr__(self):
+        return f"<UserRole(user_id={self.user_id}, role_id={self.role_id})>"

backend/src/models/user_statistics.py

@@ -0,0 +1 @@
+# 用户统计模型

backend/src/models/wrong_question.py

@@ -0,0 +1 @@
+# 错题模型

backend/src/router/__init__.py

@@ -0,0 +1,3 @@
+from .router import register_routers
+
+__all__ = ["register_routers"]

backend/src/router/router.py

@@ -0,0 +1,22 @@
+from fastapi import APIRouter
+from ..controllers.question_controller import router as question_router
+from ..controllers.user_controller import router as user_router
+from ..controllers.auth_controller import router as auth_router
+
+
+def register_routers() -> APIRouter:
+    """注册所有路由"""
+    router = APIRouter()
+
+    # 注册问题相关路由
+    router.include_router(question_router,
+                          prefix="/api/v1",
+                          tags=["questions"])
+
+    # 注册用户管理路由
+    router.include_router(user_router, prefix="/api/v1/users", tags=["users"])
+
+    # 注册认证路由
+    router.include_router(auth_router, prefix="/api/v1/auth", tags=["auth"])
+
+    return router

backend/src/services/__init__.py

@@ -0,0 +1,10 @@
+from .base_service import BaseService
+from .exam_service import ExamService
+from .learning_service import LearningService
+from .question_service import QuestionService
+from .user_service import UserService
+
+__all__ = [
+    'BaseService', 'ExamService', 'LearningService', 'QuestionService',
+    'UserService'
+]

backend/src/services/auth_service.py

@@ -0,0 +1,72 @@
+import jwt
+from datetime import datetime, timedelta
+from flask import current_app
+from ..models.user import User
+from ..models.auth_models import Token
+from ..services.user_service import UserService
+from ..store.auth_store import AuthStore
+
+
+class AuthService:
+    """认证服务"""
+
+    @staticmethod
+    def generate_token(user_id: int):
+        """生成JWT Token并保存到数据库"""
+        if not current_app.config.get('JWT_EXPIRATION_SECONDS'):
+            raise ValueError('JWT_EXPIRATION_SECONDS config is required')
+        if not current_app.config.get('SECRET_KEY'):
+            raise ValueError('SECRET_KEY config is required')
+
+        payload = {
+            'user_id':
+            user_id,
+            'exp':
+            datetime.utcnow() +
+            timedelta(seconds=current_app.config['JWT_EXPIRATION_SECONDS'])
+        }
+        access_token = jwt.encode(payload,
+                                  current_app.config['SECRET_KEY'],
+                                  algorithm='HS256')
+        refresh_payload = {
+            'user_id': user_id,
+            'exp': datetime.utcnow() + timedelta(days=7)
+        }
+        refresh_token = jwt.encode(refresh_payload,
+                                   current_app.config['SECRET_KEY'],
+                                   algorithm='HS256')
+
+        # 创建并保存Token记录
+        auth_store = AuthStore()
+        token_data = {
+            'access_token': access_token,
+            'refresh_token': refresh_token,
+            'expires_at': payload['exp'],
+            'user_id': user_id
+        }
+        return auth_store.create_token(token_data)
+
+    @staticmethod
+    def login(username: str, password: str):
+        """用户登录"""
+        user = UserService.authenticate(username, password)
+        if user:
+            return AuthService.generate_token(user.id)
+        return None
+
+    @staticmethod
+    def refresh_token(refresh_token: str):
+        """刷新Token"""
+        user_id = AuthService.verify_token(refresh_token)
+        if user_id:
+            return AuthService.generate_token(user_id)
+        return None
+
+    @staticmethod
+    def get_current_user(token: str):
+        """获取当前用户"""
+        user_id = AuthService.verify_token(token)
+        if user_id:
+            auth_store = AuthStore()
+            return auth_store.get_token_by_access_token(token).user
+        return None

backend/src/services/base_service.py

@@ -0,0 +1 @@
+# 基础服务类

backend/src/services/exam_service.py

@@ -0,0 +1,91 @@
+from ..database import db
+from ..models import ExamPaper, ExamRecord, ExamScore, ExamAnalysis
+from ..common.exceptions import BusinessException
+from datetime import datetime
+import json
+
+
+class ExamService:
+
+    def __init__(self):
+        pass
+
+    def create_exam_paper(self, creator_id, paper_data):
+        """创建试卷"""
+        try:
+            paper = ExamPaper(name=paper_data['name'],
+                              description=paper_data.get('description'),
+                              duration=paper_data['duration'],
+                              total_score=paper_data['total_score'],
+                              difficulty=paper_data['difficulty'],
+                              tags=json.dumps(paper_data.get('tags', [])),
+                              creator_id=creator_id)
+            db.session.add(paper)
+            db.session.commit()
+            return paper
+        except Exception as e:
+            db.session.rollback()
+            raise BusinessException(f'创建试卷失败: {str(e)}')
+
+    def start_exam(self, user_id, paper_id):
+        """开始考试"""
+        try:
+            exam_record = ExamRecord(user_id=user_id,
+                                     exam_paper_id=paper_id,
+                                     start_time=datetime.now(),
+                                     status='in_progress')
+            db.session.add(exam_record)
+            db.session.commit()
+            return exam_record
+        except Exception as e:
+            db.session.rollback()
+            raise BusinessException(f'开始考试失败: {str(e)}')
+
+    def submit_exam(self, record_id, answers):
+        """提交考试"""
+        try:
+            record = ExamRecord.query.get(record_id)
+            if not record:
+                raise BusinessException('考试记录不存在')
+
+            record.end_time = datetime.now()
+            record.status = 'completed'
+            record.answers = json.dumps(answers)
+            db.session.commit()
+
+            # 触发成绩计算
+            self.calculate_score(record_id)
+            return record
+        except Exception as e:
+            db.session.rollback()
+            raise BusinessException(f'提交考试失败: {str(e)}')
+
+    def calculate_score(self, record_id):
+        """计算考试成绩"""
+        try:
+            record = ExamRecord.query.get(record_id)
+            if not record:
+                raise BusinessException('考试记录不存在')
+
+            # 这里实现具体的评分逻辑
+            total_score = 100  # 示例分数
+            score = ExamScore(user_id=record.user_id,
+                              exam_record_id=record.id,
+                              total_score=total_score)
+            db.session.add(score)
+            db.session.commit()
+            return score
+        except Exception as e:
+            db.session.rollback()
+            raise BusinessException(f'计算成绩失败: {str(e)}')
+
+    def get_exam_analysis(self, record_id):
+        """获取考试分析"""
+        try:
+            analysis = ExamAnalysis.query.filter_by(
+                exam_record_id=record_id).first()
+            if not analysis:
+                raise BusinessException('考试分析不存在')
+            return analysis
+        except Exception as e:
+            raise BusinessException(f'获取考试分析失败: {str(e)}')

backend/src/services/learning_service.py

@@ -0,0 +1 @@
+# 学习服务类

backend/src/services/question_service.py

@@ -0,0 +1,193 @@
+from typing import List, Optional, Dict, Any
+from datetime import datetime
+from sqlalchemy.orm import Session, joinedload
+from sqlalchemy import or_
+from ..store.question_store import QuestionStore
+
+# 基础模型
+from ..models.question import Question
+from ..models.question_type import QuestionType
+from ..models.question_difficulty import QuestionDifficulty
+from ..models.question_category import QuestionCategory
+from ..models.question_option import QuestionOption
+from ..models.user import User
+from ..models.knowledge_point import KnowledgePoint
+
+# 异常和常量
+from ..common.exceptions import NotFoundException, PermissionDeniedException
+from ..common.constants import QuestionStatus, ReviewStatus
+
+
+class QuestionService:
+
+    def __init__(self, db: Session):
+        self.db = db
+        self.question_store = QuestionStore(db)
+
+    def create_question(self, creator_id: int,
+                        question_data: Dict[str, Any]) -> Question:
+        """创建题目"""
+        question = self.question_store.create_question(
+            creator_id=creator_id, question_data=question_data)
+        return question
+
+    def update_question(self, question_id: int, updater_id: int,
+                        update_data: Dict[str, Any]) -> Question:
+        """更新题目"""
+        question = self.question_store.get_question_by_id(question_id)
+        if not question:
+            raise NotFoundException("题目不存在")
+
+        # 检查更新权限
+        if question.creator_id != updater_id:
+            raise PermissionDeniedException("无权限更新该题目")
+
+        question = self.question_store.update_question(question_id=question_id,
+                                                       updater_id=updater_id,
+                                                       update_data=update_data)
+        return question
+
+    def add_feedback(self, question_id: int, user_id: int, feedback_type: str,
+                     content: str) -> Dict[str, Any]:
+        """添加题目反馈"""
+        result = self.question_store.add_feedback(question_id=question_id,
+                                                  user_id=user_id,
+                                                  feedback_type=feedback_type,
+                                                  content=content)
+        return result
+
+    def get_question_statistics(self, question_id: int) -> Dict[str, Any]:
+        """获取题目统计信息"""
+        return self.question_store.get_question_statistics(question_id)
+
+    def record_question_view(self, question_id: int, viewer_id: int) -> None:
+        """记录题目查看"""
+        self.question_store.record_question_view(question_id, viewer_id)
+
+    def check_question_permission(self, question_id: int, user_id: int,
+                                  required_permission: str) -> bool:
+        """检查用户对题目的权限"""
+        return self.question_store.check_question_permission(
+            question_id, user_id, required_permission)
+
+    def get_question_with_permission(self, question_id: int,
+                                     user_id: int) -> Question:
+        """获取题目详情（带权限检查）"""
+        # 检查查看权限
+        if not self.check_question_permission(question_id, user_id, "view"):
+            raise PermissionDeniedException("无权限查看该题目")
+
+        return self.get_question_detail(question_id)
+
+    def update_question_with_permission(
+            self, question_id: int, updater_id: int,
+            update_data: Dict[str, Any]) -> Question:
+        """更新题目（带权限检查）"""
+        # 检查更新权限
+        if not self.check_question_permission(question_id, updater_id,
+                                              "update"):
+            raise PermissionDeniedException("无权限更新该题目")
+
+        return self.update_question(question_id, updater_id, update_data)
+
+    def delete_question_with_permission(self, question_id: int,
+                                        deleter_id: int) -> None:
+        """删除题目（带权限检查）"""
+        # 检查删除权限
+        if not self.check_question_permission(question_id, deleter_id,
+                                              "delete"):
+            raise PermissionDeniedException("无权限删除该题目")
+
+        self.delete_question(question_id, deleter_id)
+
+    def resolve_feedback(self, feedback_id: int, resolver_id: int,
+                         resolution: str) -> QuestionFeedback:
+        """处理题目反馈"""
+        return self.question_store.resolve_feedback(feedback_id, resolver_id,
+                                                    resolution)
+
+    def add_comment(self, question_id: int, user_id: int,
+                    content: str) -> Dict[str, Any]:
+        """添加题目评论"""
+        return self.question_store.add_comment(question_id=question_id,
+                                               user_id=user_id,
+                                               content=content)
+
+    def add_attachment(self, question_id: int, user_id: int, file_path: str,
+                       file_name: str, file_type: str) -> Dict[str, Any]:
+        """添加题目附件"""
+        return self.question_store.add_attachment(question_id=question_id,
+                                                  user_id=user_id,
+                                                  file_path=file_path,
+                                                  file_name=file_name,
+                                                  file_type=file_type)
+
+    def get_question_versions(self, question_id: int) -> List[QuestionVersion]:
+        """获取题目版本历史"""
+        return self.question_store.get_question_versions(question_id)
+
+    def rollback_question_version(self, version_id: int,
+                                  updater_id: int) -> Question:
+        """回滚题目到指定版本"""
+        return self.question_store.rollback_question_version(
+            version_id, updater_id)
+
+    def delete_attachment(self, attachment_id: int, deleter_id: int) -> None:
+        """删除题目附件"""
+        self.question_store.delete_attachment(attachment_id, deleter_id)
+
+    def delete_comment(self, comment_id: int, deleter_id: int) -> None:
+        """删除题目评论"""
+        self.question_store.delete_comment(comment_id, deleter_id)
+
+    def collect_question(self, question_id: int,
+                         user_id: int) -> Dict[str, Any]:
+        """收藏题目"""
+        return self.question_store.collect_question(question_id, user_id)
+
+    def uncollect_question(self, question_id: int,
+                           user_id: int) -> Dict[str, Any]:
+        """取消收藏题目"""
+        return self.question_store.uncollect_question(question_id, user_id)
+
+    def delete_question(self, question_id: int, deleter_id: int) -> None:
+        """删除题目"""
+        self.question_store.delete_question(question_id, deleter_id)
+
+    def search_questions(self,
+                         search_params: Dict[str, Any],
+                         page: int = 1,
+                         page_size: int = 20) -> Dict[str, Any]:
+        """搜索题目"""
+        return self.question_store.search_questions(search_params, page,
+                                                    page_size)
+
+    def get_question_detail(self, question_id: int) -> Question:
+        """获取题目详情"""
+        return self.question_store.get_question_detail(question_id)
+
+    def add_knowledge_point(self, question_id: int,
+                            knowledge_point_id: int) -> Dict[str, Any]:
+        """为题目添加知识点"""
+        return self.question_store.add_knowledge_point(question_id,
+                                                       knowledge_point_id)
+
+    def remove_knowledge_point(self, question_id: int,
+                               knowledge_point_id: int) -> Dict[str, Any]:
+        """移除题目的知识点"""
+        return self.question_store.remove_knowledge_point(
+            question_id, knowledge_point_id)
+
+    def get_question_knowledge_points(
+            self, question_id: int) -> List[KnowledgePoint]:
+        """获取题目的知识点列表"""
+        return self.question_store.get_question_knowledge_points(question_id)
+
+    def review_question(self,
+                        question_id: int,
+                        reviewer_id: int,
+                        status: str,
+                        comment: Optional[str] = None) -> Question:
+        """审核题目"""
+        return self.question_store.review_question(question_id, reviewer_id,
+                                                   status, comment)

backend/src/services/user_service.py

@@ -0,0 +1,111 @@
+from ..models.user import User
+from ..models.role import Role
+from ..models.user_role import UserRole
+from werkzeug.security import generate_password_hash, check_password_hash
+from datetime import datetime
+import re
+from ..store.user_store import UserStore
+
+
+class UserService:
+    """用户服务"""
+
+    @staticmethod
+    def validate_password_strength(password: str):
+        """验证密码强度"""
+        if len(password) < 8:
+            raise ValueError('密码长度至少8位')
+        if not re.search(r'[A-Z]', password):
+            raise ValueError('密码必须包含大写字母')
+        if not re.search(r'[a-z]', password):
+            raise ValueError('密码必须包含小写字母')
+        if not re.search(r'[0-9]', password):
+            raise ValueError('密码必须包含数字')
+        if not re.search(r'[^A-Za-z0-9]', password):
+            raise ValueError('密码必须包含特殊字符')
+
+    @staticmethod
+    def create_user(username: str, password: str, role_names: list = None):
+        """创建用户"""
+        user_store = UserStore()
+        if user_store.get_user_by_username(username):
+            raise ValueError('用户名已存在')
+
+        UserService.validate_password_strength(password)
+
+        user_data = {
+            'username': username,
+            'password_hash': generate_password_hash(password),
+            'created_at': datetime.now()
+        }
+
+        # 分配默认角色
+        if role_names:
+            user_data['roles'] = role_names
+
+        return user_store.create_user(user_data)
+
+    @staticmethod
+    def authenticate(username: str, password: str):
+        """用户认证"""
+        user_store = UserStore()
+        user = user_store.get_user_by_username(username)
+        if user and check_password_hash(user.password_hash, password):
+            user_store.update_user(user.id, {'last_login': datetime.now()})
+            return user
+        return None
+
+    @staticmethod
+    def get_user_roles(user_id: int):
+        """获取用户角色"""
+        user_store = UserStore()
+        user = user_store.get_user_by_id(user_id)
+        return user.roles if user else []
+
+    @staticmethod
+    def update_user_password(user_id: int, new_password: str):
+        """更新用户密码"""
+        UserService.validate_password_strength(new_password)
+        user_store = UserStore()
+        return user_store.update_user(
+            user_id, {'password_hash': generate_password_hash(new_password)})
+
+    @staticmethod
+    def set_user_status(user_id: int, is_active: bool):
+        """设置用户状态"""
+        user_store = UserStore()
+        return user_store.update_user(user_id, {'is_active': is_active})
+
+    @staticmethod
+    def initiate_password_reset(username: str):
+        """发起密码重置"""
+        user_store = UserStore()
+        user = user_store.get_user_by_username(username)
+        if not user:
+            return None
+
+        # 生成重置token（有效期1小时）
+        reset_token = generate_password_hash(
+            f'{user.id}{datetime.now().timestamp()}')[:32]
+        user_store.update_user(
+            user.id, {
+                'reset_token': reset_token,
+                'reset_token_expires': datetime.now() + timedelta(hours=1)
+            })
+        return reset_token
+
+    @staticmethod
+    def reset_password(reset_token: str, new_password: str):
+        """重置密码"""
+        user_store = UserStore()
+        user = user_store.get_user_by_reset_token(reset_token)
+        if not user or user.reset_token_expires < datetime.now():
+            return False
+
+        UserService.validate_password_strength(new_password)
+        return user_store.update_user(
+            user.id, {
+                'password_hash': generate_password_hash(new_password),
+                'reset_token': None,
+                'reset_token_expires': None
+            })

backend/src/store/__init__.py

@@ -0,0 +1 @@
+# store 包初始化文件

backend/src/store/auth_store.py

@@ -0,0 +1,30 @@
+from ..models.auth_models import Token
+from ..database import get_db
+
+
+class AuthStore:
+
+    def __init__(self):
+        self.db = get_db()
+
+    def create_token(self, token_data: dict) -> Token:
+        """创建新token"""
+        token = Token(**token_data)
+        self.db.add(token)
+        self.db.commit()
+        self.db.refresh(token)
+        return token
+
+    def get_token_by_access_token(self, access_token: str) -> Token:
+        """根据access_token获取token"""
+        return self.db.query(Token).filter(
+            Token.access_token == access_token).first()
+
+    def delete_token(self, token_id: int) -> bool:
+        """删除token"""
+        token = self.db.query(Token).filter(Token.id == token_id).first()
+        if token:
+            self.db.delete(token)
+            self.db.commit()
+            return True
+        return False

backend/src/store/question_store.py

@@ -0,0 +1,450 @@
+from typing import List, Optional, Dict, Any
+from datetime import datetime
+from sqlalchemy.orm import Session, joinedload
+from sqlalchemy import or_
+from ..models.question import Question
+from ..models.question_type import QuestionType
+from ..models.question_difficulty import QuestionDifficulty
+from ..models.question_category import QuestionCategory
+from ..models.question_option import QuestionOption
+from ..models.question_analysis import QuestionAnalysis
+from ..models.question_feedback import QuestionFeedback
+from ..models.question_comment import QuestionComment
+from ..models.question_attachment import QuestionAttachment
+from ..models.question_version import QuestionVersion
+from ..models.question_relation import QuestionRelation
+from ..models.question_statistics import QuestionStatistics
+from ..models.question_bank import QuestionBank
+from ..models.user import User
+from ..models.user_statistics import UserStatistics
+from ..models.learning_progress import LearningProgress
+from ..models.practice_record import PracticeRecord
+from ..models.wrong_question import WrongQuestion
+from ..models.knowledge_point import KnowledgePoint
+from ..models.learning_path import LearningPath
+from ..models.question_collection import QuestionCollection
+from ..models.question_template import QuestionTemplate
+from ..models.question_tag import QuestionTag
+from ..models.question_history import QuestionHistory
+from ..models.question_knowledge_point import QuestionKnowledgePoint
+from ..common.exceptions import NotFoundException, PermissionDeniedException
+from ..common.constants import QuestionStatus, ReviewStatus
+
+
+class QuestionStore:
+
+    def __init__(self, db: Session):
+        self.db = db
+
+    def create_question(self, creator_id: int,
+                        question_data: Dict[str, Any]) -> Question:
+        """创建题目"""
+        question = Question.create_with_options(
+            db=self.db,
+            title=question_data["title"],
+            content=question_data["content"],
+            type_id=question_data["type_id"],
+            difficulty_id=question_data["difficulty_id"],
+            options=question_data.get("options", []),
+            creator_id=creator_id,
+            category_id=question_data.get("category_id"),
+            tags=question_data.get("tags"),
+            answer=question_data.get("answer"),
+            explanation=question_data.get("explanation"),
+            source=question_data.get("source"),
+            extra_data=question_data.get("extra_data"))
+        return question
+
+    def get_question_by_id(self, question_id: int) -> Optional[Question]:
+        """根据ID获取题目"""
+        return self.db.query(Question).get(question_id)
+
+    def update_question(self, question_id: int,
+                        update_data: Dict[str, Any]) -> Question:
+        """更新题目"""
+        question = self.get_question_by_id(question_id)
+        if not question:
+            raise NotFoundException("题目不存在")
+
+        # 更新基础信息
+        if "title" in update_data:
+            question.title = update_data["title"]
+        if "content" in update_data:
+            question.content = update_data["content"]
+        if "type_id" in update_data:
+            question.type_id = update_data["type_id"]
+        if "difficulty_id" in update_data:
+            question.difficulty_id = update_data["difficulty_id"]
+        if "category_id" in update_data:
+            question.category_id = update_data["category_id"]
+        if "answer" in update_data:
+            question.answer = update_data["answer"]
+        if "explanation" in update_data:
+            question.explanation = update_data["explanation"]
+        if "source" in update_data:
+            question.source = update_data["source"]
+        if "extra_data" in update_data:
+            question.extra_data = update_data["extra_data"]
+
+        # 更新选项
+        if "options" in update_data:
+            # 删除旧选项
+            self.db.query(QuestionOption).filter(
+                QuestionOption.question_id == question_id).delete()
+
+            # 添加新选项
+            for option_data in update_data["options"]:
+                QuestionOption.create(db=self.db,
+                                      question_id=question_id,
+                                      content=option_data["content"],
+                                      is_correct=option_data.get(
+                                          "is_correct", False),
+                                      order=option_data.get("order", 0),
+                                      extra_data=option_data.get("extra_data"))
+
+        # 更新标签
+        if "tags" in update_data:
+            # 删除旧标签
+            self.db.query(QuestionTag).filter(
+                QuestionTag.question_id == question_id).delete()
+
+            # 添加新标签
+            for tag_data in update_data["tags"]:
+                QuestionTag.create(db=self.db,
+                                   question_id=question_id,
+                                   name=tag_data["name"],
+                                   description=tag_data.get("description"),
+                                   extra_data=tag_data.get("extra_data"))
+
+        # 创建新版本
+        QuestionVersion.create(db=self.db,
+                               question_id=question_id,
+                               version_data={
+                                   "title":
+                                   question.title,
+                                   "content":
+                                   question.content,
+                                   "type_id":
+                                   question.type_id,
+                                   "difficulty_id":
+                                   question.difficulty_id,
+                                   "category_id":
+                                   question.category_id,
+                                   "answer":
+                                   question.answer,
+                                   "explanation":
+                                   question.explanation,
+                                   "source":
+                                   question.source,
+                                   "extra_data":
+                                   question.extra_data,
+                                   "options": [{
+                                       "content": o.content,
+                                       "is_correct": o.is_correct,
+                                       "order": o.order,
+                                       "extra_data": o.extra_data
+                                   } for o in question.options],
+                                   "tags": [{
+                                       "name": t.name,
+                                       "description": t.description,
+                                       "extra_data": t.extra_data
+                                   } for t in question.tags]
+                               })
+
+        # 更新统计信息
+        stats = self.db.query(QuestionStatistics).filter(
+            QuestionStatistics.question_id == question_id).first()
+        if not stats:
+            stats = QuestionStatistics(question_id=question_id)
+            self.db.add(stats)
+
+        stats.update_count += 1
+        stats.last_update_time = datetime.now()
+
+        # 记录历史
+        QuestionHistory.create(db=self.db,
+                               question_id=question_id,
+                               action="update",
+                               change_data={
+                                   "title": question.title,
+                                   "content": question.content,
+                                   "type_id": question.type_id,
+                                   "difficulty_id": question.difficulty_id,
+                                   "category_id": question.category_id,
+                                   "answer": question.answer,
+                                   "explanation": question.explanation,
+                                   "source": question.source,
+                                   "extra_data": question.extra_data
+                               })
+
+        self.db.commit()
+        self.db.refresh(question)
+        return question
+
+    def delete_question(self, question_id: int) -> None:
+        """删除题目"""
+        question = self.get_question_by_id(question_id)
+        if not question:
+            raise NotFoundException("题目不存在")
+
+        # 记录删除历史
+        QuestionHistory.create(db=self.db,
+                               question_id=question_id,
+                               action="delete",
+                               change_data={
+                                   "title": question.title,
+                                   "content": question.content,
+                                   "type_id": question.type_id,
+                                   "difficulty_id": question.difficulty_id,
+                                   "category_id": question.category_id,
+                                   "answer": question.answer,
+                                   "explanation": question.explanation,
+                                   "source": question.source,
+                                   "extra_data": question.extra_data
+                               })
+
+        # 更新统计信息
+        stats = self.db.query(QuestionStatistics).filter(
+            QuestionStatistics.question_id == question_id).first()
+        if stats:
+            stats.delete_count += 1
+            stats.last_update_time = datetime.now()
+
+        # 删除题目
+        self.db.delete(question)
+        self.db.commit()
+
+    def get_question_detail(self, question_id: int) -> Question:
+        """获取题目详情"""
+        question = self.db.query(Question) \
+                         .options(
+                             joinedload(Question.options),
+                             joinedload(Question.tags),
+                             joinedload(Question.versions),
+                             joinedload(Question.comments),
+                             joinedload(Question.attachments),
+                             joinedload(Question.knowledge_points)
+                         ) \
+                         .get(question_id)
+        if not question:
+            raise NotFoundException("题目不存在")
+        return question
+
+    def search_questions(self,
+                         search_params: Dict[str, Any],
+                         page: int = 1,
+                         page_size: int = 20) -> Dict[str, Any]:
+        """搜索题目"""
+        query = self.db.query(Question)
+
+        # 应用过滤条件
+        if search_params.get("type_id"):
+            query = query.filter(Question.type_id == search_params["type_id"])
+        if search_params.get("difficulty_id"):
+            query = query.filter(
+                Question.difficulty_id == search_params["difficulty_id"])
+        if search_params.get("category_id"):
+            query = query.filter(
+                Question.category_id == search_params["category_id"])
+        if search_params.get("creator_id"):
+            query = query.filter(
+                Question.creator_id == search_params["creator_id"])
+        if search_params.get("bank_id"):
+            query = query.filter(Question.bank_id == search_params["bank_id"])
+        if search_params.get("status"):
+            query = query.filter(Question.status == search_params["status"])
+        if search_params.get("review_status"):
+            query = query.filter(
+                Question.review_status == search_params["review_status"])
+        if search_params.get("keyword"):
+            keyword = search_params["keyword"]
+            query = query.filter((Question.title.contains(keyword))
+                                 | (Question.content.contains(keyword))
+                                 | (Question.answer.contains(keyword))
+                                 | (Question.explanation.contains(keyword)))
+
+        # 时间范围过滤
+        if search_params.get("create_time_start"):
+            query = query.filter(
+                Question.create_time >= search_params["create_time_start"])
+        if search_params.get("create_time_end"):
+            query = query.filter(
+                Question.create_time <= search_params["create_time_end"])
+        if search_params.get("update_time_start"):
+            query = query.filter(
+                Question.update_time >= search_params["update_time_start"])
+        if search_params.get("update_time_end"):
+            query = query.filter(
+                Question.update_time <= search_params["update_time_end"])
+
+        # 排序
+        sort_field = search_params.get("sort_field", "create_time")
+        sort_order = search_params.get("sort_order", "desc")
+        if sort_order == "asc":
+            query = query.order_by(getattr(Question, sort_field).asc())
+        else:
+            query = query.order_by(getattr(Question, sort_field).desc())
+
+        # 计算总数
+        total = query.count()
+
+        # 分页查询
+        questions = query.offset((page - 1) * page_size) \
+                        .limit(page_size) \
+                        .all()
+
+        return {
+            "total": total,
+            "page": page,
+            "page_size": page_size,
+            "questions": questions
+        }
+
+    def delete_attachment(self, attachment_id: int, deleter_id: int) -> None:
+        """删除题目附件"""
+        attachment = self.db.query(QuestionAttachment).get(attachment_id)
+        if not attachment:
+            raise NotFoundException("附件不存在")
+
+        # 检查删除权限
+        if attachment.user_id != deleter_id:
+            raise PermissionDeniedException("无权限删除该附件")
+
+        # 更新统计信息
+        stats = self.db.query(QuestionStatistics).filter(
+            QuestionStatistics.question_id == attachment.question_id).first()
+        if stats:
+            stats.attachment_count -= 1
+
+        user_stats = self.db.query(UserStatistics).filter(
+            UserStatistics.user_id == attachment.user_id).first()
+        if user_stats:
+            user_stats.attachment_count -= 1
+
+        # 删除附件记录
+        self.db.delete(attachment)
+        self.db.commit()
+
+    def delete_comment(self, comment_id: int, deleter_id: int) -> None:
+        """删除题目评论"""
+        comment = self.db.query(QuestionComment).get(comment_id)
+        if not comment:
+            raise NotFoundException("评论不存在")
+
+        # 检查删除权限
+        if comment.user_id != deleter_id:
+            raise PermissionDeniedException("无权限删除该评论")
+
+        # 更新统计信息
+        stats = self.db.query(QuestionStatistics).filter(
+            QuestionStatistics.question_id == comment.question_id).first()
+        if stats:
+            stats.comment_count -= 1
+
+        user_stats = self.db.query(UserStatistics).filter(
+            UserStatistics.user_id == comment.user_id).first()
+        if user_stats:
+            user_stats.comment_count -= 1
+
+        # 删除评论
+        self.db.delete(comment)
+        self.db.commit()
+
+    def collect_question(self, question_id: int,
+                         user_id: int) -> Dict[str, Any]:
+        """收藏题目"""
+        # 检查题目是否存在
+        question = self.db.query(Question).get(question_id)
+        if not question:
+            raise NotFoundException("题目不存在")
+
+        # 检查是否已经收藏
+        existing = self.db.query(QuestionCollection).filter(
+            QuestionCollection.question_id == question_id,
+            QuestionCollection.user_id == user_id).first()
+        if existing:
+            return {
+                "status": "already_collected",
+                "message": "题目已收藏",
+                "collect_count": existing.question.statistics.collect_count,
+                "hot_score": question.hot_score
+            }
+
+        # 创建收藏记录
+        QuestionCollection.create(db=self.db,
+                                  question_id=question_id,
+                                  user_id=user_id,
+                                  collect_time=datetime.now())
+
+        # 更新题目统计信息
+        stats = self.db.query(QuestionStatistics).filter(
+            QuestionStatistics.question_id == question_id).first()
+        if not stats:
+            stats = QuestionStatistics(question_id=question_id)
+            self.db.add(stats)
+
+        stats.collect_count += 1
+        stats.last_collect_time = datetime.now()
+        stats.last_collector_id = user_id
+
+        # 更新用户统计信息
+        user_stats = self.db.query(UserStatistics).filter(
+            UserStatistics.user_id == user_id).first()
+        if not user_stats:
+            user_stats = UserStatistics(user_id=user_id)
+            self.db.add(user_stats)
+
+        user_stats.collect_count += 1
+        user_stats.last_collect_time = datetime.now()
+
+        # 更新题目热度
+        question.hot_score = (question.hot_score or 0) + 1
+
+        self.db.commit()
+
+        return {
+            "status": "success",
+            "message": "收藏成功",
+            "collect_count": stats.collect_count,
+            "last_collect_time": stats.last_collect_time,
+            "hot_score": question.hot_score
+        }
+
+    def uncollect_question(self, question_id: int,
+                           user_id: int) -> Dict[str, Any]:
+        """取消收藏题目"""
+        # 检查题目是否存在
+        question = self.db.query(Question).get(question_id)
+        if not question:
+            raise NotFoundException("题目不存在")
+
+        # 删除收藏记录
+        self.db.query(QuestionCollection).filter(
+            QuestionCollection.question_id == question_id,
+            QuestionCollection.user_id == user_id).delete()
+
+        # 更新题目统计信息
+        stats = self.db.query(QuestionStatistics).filter(
+            QuestionStatistics.question_id == question_id).first()
+        if stats:
+            stats.collect_count -= 1
+            stats.collect_count = max(stats.collect_count, 0)
+
+        # 更新用户统计信息
+        user_stats = self.db.query(UserStatistics).filter(
+            UserStatistics.user_id == user_id).first()
+        if user_stats:
+            user_stats.collect_count -= 1
+            user_stats.collect_count = max(user_stats.collect_count, 0)
+
+        # 更新题目热度
+        question.hot_score = max((question.hot_score or 0) - 1, 0)
+
+        self.db.commit()
+
+        return {
+            "status": "success",
+            "message": "取消收藏成功",
+            "collect_count": stats.collect_count if stats else 0,
+            "last_collect_time": stats.last_collect_time if stats else None
+        }

backend/src/store/user_store.py

@@ -0,0 +1,55 @@
+from ..models.user import User
+from ..models.role import Role
+from ..models.user_role import UserRole
+from ..database import get_db
+
+
+class UserStore:
+
+    def __init__(self):
+        self.db = get_db()
+
+    def get_user_by_username(self, username: str) -> User:
+        """根据用户名获取用户"""
+        return self.db.query(User).filter_by(username=username).first()
+
+    def create_user(self, user_data: dict) -> User:
+        """创建用户"""
+        user = User(username=user_data['username'],
+                    password_hash=user_data['password_hash'],
+                    created_at=user_data['created_at'])
+
+        # 分配角色
+        if 'roles' in user_data:
+            roles = self.db.query(Role).filter(
+                Role.name.in_(user_data['roles'])).all()
+            user.roles.extend(roles)
+        else:
+            default_role = self.db.query(Role).filter_by(is_default=1).first()
+            if default_role:
+                user.roles.append(default_role)
+
+        self.db.add(user)
+        self.db.commit()
+        self.db.refresh(user)
+        return user
+
+    def get_user_by_id(self, user_id: int) -> User:
+        """根据ID获取用户"""
+        return self.db.query(User).get(user_id)
+
+    def update_user(self, user_id: int, update_data: dict) -> bool:
+        """更新用户信息"""
+        user = self.get_user_by_id(user_id)
+        if not user:
+            return False
+
+        for key, value in update_data.items():
+            setattr(user, key, value)
+
+        self.db.commit()
+        return True
+
+    def get_user_by_reset_token(self, reset_token: str) -> User:
+        """根据重置token获取用户"""
+        return self.db.query(User).filter_by(reset_token=reset_token).first()

backend/src/tests/__init__.py

@@ -0,0 +1 @@
+# test 包初始化文件

backend/src/tests/test_logger.py

@@ -0,0 +1,45 @@
+import os
+from ..logger.logger import Logger
+
+
+def test_logger():
+    # 测试日志文件路径
+    test_log_path = "logs/test.log"
+
+    # 如果存在先删除旧日志文件
+    if os.path.exists(test_log_path):
+        os.remove(test_log_path)
+
+    # 配置logger
+    Logger.get_logger().handlers.clear()  # 清除已有handler
+    Logger._instance = None  # 重置单例
+    Logger()._configure_logger({
+        "level": "DEBUG",
+        "console": False,
+        "file": test_log_path
+    })
+
+    # 写入测试日志
+    Logger.debug("This is a debug message")
+    Logger.info("This is an info message")
+    Logger.warning("This is a warning message")
+    Logger.error("This is an error message")
+
+    # 验证日志文件是否创建
+    if not os.path.exists(test_log_path):
+        print("Test failed: Log file not created")
+        return False
+
+    # 验证日志内容
+    with open(test_log_path, "r") as f:
+        content = f.read()
+        if "debug message" not in content or "error message" not in content:
+            print("Test failed: Log content incorrect")
+            return False
+
+    print("Test passed: Logger working correctly")
+    return True
+
+
+if __name__ == "__main__":
+    test_logger()

backend/src/utils/__init__.py

@@ -0,0 +1,7 @@
+# utils 包初始化文件
+from .file.file_util import FileUtil
+from .network.network_util import NetworkUtil
+from .crypto.crypto_util import CryptoUtil
+from .validation.validation_util import ValidationUtil
+
+__all__ = ["FileUtil", "NetworkUtil", "CryptoUtil", "ValidationUtil"]

backend/src/utils/crypto/crypto_util.py

@@ -0,0 +1,107 @@
+import hashlib
+import os
+import base64
+from typing import Optional
+from cryptography.fernet import Fernet
+from cryptography.hazmat.primitives import hashes
+from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
+
+
+class CryptoUtil:
+    """加密工具类"""
+
+    @staticmethod
+    def generate_salt(length: int = 16) -> bytes:
+        """生成随机盐值
+        
+        Args:
+            length: 盐值长度
+            
+        Returns:
+            随机生成的盐值
+        """
+        return os.urandom(length)
+
+    @staticmethod
+    def hash_password(password: str, salt: Optional[bytes] = None) -> str:
+        """哈希密码
+        
+        Args:
+            password: 原始密码
+            salt: 盐值
+            
+        Returns:
+            哈希后的密码字符串
+        """
+        if salt is None:
+            salt = CryptoUtil.generate_salt()
+
+        kdf = PBKDF2HMAC(
+            algorithm=hashes.SHA256(),
+            length=32,
+            salt=salt,
+            iterations=100000,
+        )
+        key = base64.urlsafe_b64encode(kdf.derive(password.encode()))
+        return key.decode()
+
+    @staticmethod
+    def md5(data: str) -> str:
+        """计算MD5哈希值
+        
+        Args:
+            data: 输入数据
+            
+        Returns:
+            MD5哈希值
+        """
+        return hashlib.md5(data.encode()).hexdigest()
+
+    @staticmethod
+    def sha256(data: str) -> str:
+        """计算SHA256哈希值
+        
+        Args:
+            data: 输入数据
+            
+        Returns:
+            SHA256哈希值
+        """
+        return hashlib.sha256(data.encode()).hexdigest()
+
+    @staticmethod
+    def encrypt(data: str, key: str) -> str:
+        """对称加密
+        
+        Args:
+            data: 要加密的数据
+            key: 加密密钥
+            
+        Returns:
+            加密后的字符串
+        """
+        f = Fernet(key.encode())
+        return f.encrypt(data.encode()).decode()
+
+    @staticmethod
+    def decrypt(encrypted_data: str, key: str) -> str:
+        """对称解密
+        
+        Args:
+            encrypted_data: 加密后的数据
+            key: 解密密钥
+            
+        Returns:
+            解密后的字符串
+        """
+        f = Fernet(key.encode())
+        return f.decrypt(encrypted_data.encode()).decode()
+
+    @staticmethod
+    def generate_key() -> str:
+        """生成加密密钥
+        
+        Returns:
+            生成的密钥字符串
+        """
+        return Fernet.generate_key().decode()

backend/src/utils/date/date_util.py

@@ -0,0 +1,93 @@
+from datetime import datetime, timedelta
+from typing import Union, Optional
+import pytz
+
+
+class DateUtil:
+    """日期时间工具类"""
+
+    @staticmethod
+    def format_date(date: Union[datetime, str],
+                    fmt: str = '%Y-%m-%d %H:%M:%S',
+                    timezone: Optional[str] = None) -> str:
+        """格式化日期时间
+        
+        Args:
+            date: 日期时间对象或字符串
+            fmt: 格式化字符串
+            timezone: 时区名称
+            
+        Returns:
+            格式化后的日期时间字符串
+        """
+        if isinstance(date, str):
+            date = datetime.strptime(date, fmt)
+
+        if timezone:
+            tz = pytz.timezone(timezone)
+            date = date.astimezone(tz)
+
+        return date.strftime(fmt)
+
+    @staticmethod
+    def add_days(date: Union[datetime, str],
+                 days: int,
+                 fmt: str = '%Y-%m-%d %H:%M:%S') -> str:
+        """日期加减天数
+        
+        Args:
+            date: 日期时间对象或字符串
+            days: 要加减的天数
+            fmt: 格式化字符串
+            
+        Returns:
+            计算后的日期时间字符串
+        """
+        if isinstance(date, str):
+            date = datetime.strptime(date, fmt)
+
+        new_date = date + timedelta(days=days)
+        return new_date.strftime(fmt)
+
+    @staticmethod
+    def compare_dates(date1: Union[datetime, str],
+                      date2: Union[datetime, str],
+                      fmt: str = '%Y-%m-%d %H:%M:%S') -> int:
+        """比较两个日期
+        
+        Args:
+            date1: 第一个日期
+            date2: 第二个日期
+            fmt: 格式化字符串
+            
+        Returns:
+            -1: date1 < date2
+            0: date1 == date2
+            1: date1 > date2
+        """
+        if isinstance(date1, str):
+            date1 = datetime.strptime(date1, fmt)
+        if isinstance(date2, str):
+            date2 = datetime.strptime(date2, fmt)
+
+        if date1 < date2:
+            return -1
+        elif date1 > date2:
+            return 1
+        return 0
+
+    @staticmethod
+    def get_current_time(timezone: Optional[str] = None) -> str:
+        """获取当前时间
+        
+        Args:
+            timezone: 时区名称
+            
+        Returns:
+            当前时间字符串
+        """
+        now = datetime.now()
+        if timezone:
+            tz = pytz.timezone(timezone)
+            now = now.astimezone(tz)
+        return now.strftime('%Y-%m-%d %H:%M:%S')

backend/src/utils/file/file_util.py

@@ -0,0 +1,142 @@
+import os
+import shutil
+from typing import Optional, Union
+from pathlib import Path
+
+
+class FileUtil:
+    """文件操作工具类"""
+
+    @staticmethod
+    def read_file(path: Union[str, Path]) -> Optional[str]:
+        """读取文件内容
+        
+        Args:
+            path: 文件路径
+            
+        Returns:
+            文件内容字符串，如果文件不存在返回None
+        """
+        if not os.path.exists(path):
+            return None
+        with open(path, 'r', encoding='utf-8') as f:
+            return f.read()
+
+    @staticmethod
+    def write_file(path: Union[str, Path], content: str) -> bool:
+        """写入文件内容
+        
+        Args:
+            path: 文件路径
+            content: 要写入的内容
+            
+        Returns:
+            bool: 是否写入成功
+        """
+        try:
+            with open(path, 'w', encoding='utf-8') as f:
+                f.write(content)
+            return True
+        except Exception:
+            return False
+
+    @staticmethod
+    def ensure_dir_exists(path: Union[str, Path]) -> bool:
+        """确保目录存在
+        
+        Args:
+            path: 目录路径
+            
+        Returns:
+            bool: 是否成功创建目录
+        """
+        try:
+            os.makedirs(path, exist_ok=True)
+            return True
+        except Exception:
+            return False
+
+    @staticmethod
+    def delete_file(path: Union[str, Path]) -> bool:
+        """删除文件
+        
+        Args:
+            path: 文件路径
+            
+        Returns:
+            bool: 是否删除成功
+        """
+        try:
+            if os.path.exists(path):
+                os.remove(path)
+                return True
+            return False
+        except Exception:
+            return False
+
+    @staticmethod
+    def copy_file(src: Union[str, Path], dst: Union[str, Path]) -> bool:
+        """复制文件
+        
+        Args:
+            src: 源文件路径
+            dst: 目标文件路径
+            
+        Returns:
+            bool: 是否复制成功
+        """
+        try:
+            shutil.copy(src, dst)
+            return True
+        except Exception:
+            return False
+
+    @staticmethod
+    def get_file_size(path: Union[str, Path]) -> Optional[int]:
+        """获取文件大小
+        
+        Args:
+            path: 文件路径
+            
+        Returns:
+            文件大小（字节），如果文件不存在返回None
+        """
+        if not os.path.exists(path):
+            return None
+        return os.path.getsize(path)
+
+    @staticmethod
+    def get_file_extension(path: Union[str, Path]) -> str:
+        """获取文件扩展名
+        
+        Args:
+            path: 文件路径
+            
+        Returns:
+            文件扩展名（小写）
+        """
+        return Path(path).suffix.lower()
+
+    @staticmethod
+    def is_file(path: Union[str, Path]) -> bool:
+        """判断路径是否为文件
+        
+        Args:
+            path: 路径
+            
+        Returns:
+            bool: 是否为文件
+        """
+        return os.path.isfile(path)
+
+    @staticmethod
+    def is_dir(path: Union[str, Path]) -> bool:
+        """判断路径是否为目录
+        
+        Args:
+            path: 路径
+            
+        Returns:
+            bool: 是否为目录
+        """
+        return os.path.isdir(path)

backend/src/utils/network/network_util.py

@@ -0,0 +1,109 @@
+import requests
+import socket
+import urllib.parse
+from typing import Optional, Dict, Any
+from urllib.parse import urlparse
+
+
+class NetworkUtil:
+    """网络工具类"""
+
+    @staticmethod
+    def get(url: str,
+            params: Optional[Dict] = None,
+            headers: Optional[Dict] = None) -> Optional[Dict]:
+        """发送GET请求
+        
+        Args:
+            url: 请求URL
+            params: 请求参数
+            headers: 请求头
+            
+        Returns:
+            响应数据字典，如果请求失败返回None
+        """
+        try:
+            response = requests.get(url, params=params, headers=headers)
+            response.raise_for_status()
+            return response.json()
+        except Exception:
+            return None
+
+    @staticmethod
+    def post(url: str,
+             data: Optional[Dict] = None,
+             headers: Optional[Dict] = None) -> Optional[Dict]:
+        """发送POST请求
+        
+        Args:
+            url: 请求URL
+            data: 请求体数据
+            headers: 请求头
+            
+        Returns:
+            响应数据字典，如果请求失败返回None
+        """
+        try:
+            response = requests.post(url, json=data, headers=headers)
+            response.raise_for_status()
+            return response.json()
+        except Exception:
+            return None
+
+    @staticmethod
+    def is_valid_ip(ip: str) -> bool:
+        """验证IP地址是否有效
+        
+        Args:
+            ip: IP地址字符串
+            
+        Returns:
+            bool: 是否为有效IP地址
+        """
+        try:
+            socket.inet_aton(ip)
+            return True
+        except socket.error:
+            return False
+
+    @staticmethod
+    def get_domain(url: str) -> Optional[str]:
+        """从URL中提取域名
+        
+        Args:
+            url: URL字符串
+            
+        Returns:
+            域名，如果URL无效返回None
+        """
+        try:
+            parsed = urlparse(url)
+            return parsed.netloc
+        except Exception:
+            return None
+
+    @staticmethod
+    def is_online() -> bool:
+        """检查网络是否在线
+        
+        Returns:
+            bool: 是否在线
+        """
+        try:
+            socket.create_connection(("www.baidu.com", 80))
+            return True
+        except OSError:
+            return False
+
+    @staticmethod
+    def build_url(base: str, params: Dict[str, Any]) -> str:
+        """构建带参数的URL
+        
+        Args:
+            base: 基础URL
+            params: 参数字典
+            
+        Returns:
+            带参数的完整URL
+        """
+        return f"{base}?{urllib.parse.urlencode(params)}"

backend/src/utils/string/string_util.py

@@ -0,0 +1,98 @@
+import re
+from typing import Optional, Union
+
+
+class StringUtil:
+    """字符串处理工具类"""
+
+    @staticmethod
+    def is_empty(s: Optional[str]) -> bool:
+        """判断字符串是否为空
+        
+        Args:
+            s: 输入字符串
+            
+        Returns:
+            bool: 是否为空
+        """
+        return s is None or len(s.strip()) == 0
+
+    @staticmethod
+    def to_camel_case(s: str) -> str:
+        """将下划线命名转换为驼峰命名
+        
+        Args:
+            s: 输入字符串
+            
+        Returns:
+            驼峰命名字符串
+        """
+        parts = s.split('_')
+        return parts[0] + ''.join(x.title() for x in parts[1:])
+
+    @staticmethod
+    def to_snake_case(s: str) -> str:
+        """将驼峰命名转换为下划线命名
+        
+        Args:
+            s: 输入字符串
+            
+        Returns:
+            下划线命名字符串
+        """
+        s = re.sub('(.)([A-Z][a-z]+)', r'\1_\2', s)
+        return re.sub('([a-z0-9])([A-Z])', r'\1_\2', s).lower()
+
+    @staticmethod
+    def truncate(s: str, length: int, suffix: str = '...') -> str:
+        """截断字符串
+        
+        Args:
+            s: 输入字符串
+            length: 最大长度
+            suffix: 后缀
+            
+        Returns:
+            截断后的字符串
+        """
+        if len(s) <= length:
+            return s
+        return s[:length - len(suffix)] + suffix
+
+    @staticmethod
+    def is_email(s: str) -> bool:
+        """验证是否为有效的邮箱地址
+        
+        Args:
+            s: 输入字符串
+            
+        Returns:
+            bool: 是否为有效邮箱
+        """
+        pattern = r'^[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+$'
+        return bool(re.match(pattern, s))
+
+    @staticmethod
+    def is_phone(s: str) -> bool:
+        """验证是否为有效的手机号码
+        
+        Args:
+            s: 输入字符串
+            
+        Returns:
+            bool: 是否为有效手机号
+        """
+        pattern = r'^1[3-9]\d{9}$'
+        return bool(re.match(pattern, s))
+
+    @staticmethod
+    def join_with_comma(items: list) -> str:
+        """将列表元素用逗号连接
+        
+        Args:
+            items: 输入列表
+            
+        Returns:
+            连接后的字符串
+        """
+        return ', '.join(str(item) for item in items)

backend/src/utils/validation/validation_util.py

@@ -0,0 +1,119 @@
+import re
+from typing import Optional, Union
+from datetime import datetime
+
+
+class ValidationUtil:
+    """验证工具类"""
+
+    @staticmethod
+    def is_email(email: str) -> bool:
+        """验证邮箱格式
+        
+        Args:
+            email: 邮箱地址
+            
+        Returns:
+            bool: 是否为有效邮箱
+        """
+        pattern = r'^[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+$'
+        return bool(re.match(pattern, email))
+
+    @staticmethod
+    def is_phone(phone: str) -> bool:
+        """验证手机号格式
+        
+        Args:
+            phone: 手机号码
+            
+        Returns:
+            bool: 是否为有效手机号
+        """
+        pattern = r'^1[3-9]\d{9}$'
+        return bool(re.match(pattern, phone))
+
+    @staticmethod
+    def is_id_card(id_card: str) -> bool:
+        """验证身份证号格式
+        
+        Args:
+            id_card: 身份证号码
+            
+        Returns:
+            bool: 是否为有效身份证号
+        """
+        pattern = r'^\d{17}[\dXx]$'
+        return bool(re.match(pattern, id_card))
+
+    @staticmethod
+    def is_date(date_str: str, fmt: str = '%Y-%m-%d') -> bool:
+        """验证日期格式
+        
+        Args:
+            date_str: 日期字符串
+            fmt: 日期格式
+            
+        Returns:
+            bool: 是否为有效日期
+        """
+        try:
+            datetime.strptime(date_str, fmt)
+            return True
+        except ValueError:
+            return False
+
+    @staticmethod
+    def is_number(value: Union[str, int, float]) -> bool:
+        """验证是否为数字
+        
+        Args:
+            value: 输入值
+            
+        Returns:
+            bool: 是否为数字
+        """
+        if isinstance(value, (int, float)):
+            return True
+        try:
+            float(value)
+            return True
+        except ValueError:
+            return False
+
+    @staticmethod
+    def is_in_range(value: Union[int, float],
+                    min_val: Optional[Union[int, float]] = None,
+                    max_val: Optional[Union[int, float]] = None) -> bool:
+        """验证数值是否在指定范围内
+        
+        Args:
+            value: 要验证的值
+            min_val: 最小值
+            max_val: 最大值
+            
+        Returns:
+            bool: 是否在范围内
+        """
+        if min_val is not None and value < min_val:
+            return False
+        if max_val is not None and value > max_val:
+            return False
+        return True
+
+    @staticmethod
+    def is_empty(value: Optional[Union[str, list, dict]]) -> bool:
+        """验证值是否为空
+        
+        Args:
+            value: 输入值
+            
+        Returns:
+            bool: 是否为空
+        """
+        if value is None:
+            return True
+        if isinstance(value, str) and not value.strip():
+            return True
+        if isinstance(value, (list, dict)) and not value:
+            return True
+        return False