README.md

@@ -2,6 +2,81 @@
 
 配置模块提供了统一的配置管理接口，主要功能包括：
 
+### 数据库配置
+
+项目使用 SQLAlchemy ORM 框架连接 MySQL 数据库，主要配置项如下：
+
+```python
+class DatabaseConfig:
+    host: str = "localhost"  # 数据库主机
+    port: int = 3306         # 数据库端口
+    user: str = "root"       # 数据库用户名
+    password: str = ""       # 数据库密码
+    name: str = "question_bank"  # 数据库名称
+```
+
+使用示例：
+
+```python
+from app.database.database import Database
+
+# 初始化数据库连接
+Database.initialize()
+
+# 使用上下文管理器管理会话
+with db.session() as session:
+    # 执行查询
+    users = db.execute_query("SELECT * FROM users WHERE age > :age", {"age": 18})
+
+    # 执行非查询
+    affected_rows = db.execute_non_query(
+        "UPDATE users SET status = :status WHERE id = :id",
+        {"status": "active", "id": 1}
+    )
+
+    # 批量操作
+    user_data = [
+        {"name": "Alice", "age": 25},
+        {"name": "Bob", "age": 30}
+    ]
+    inserted_rows = db.batch_execute(
+        "INSERT INTO users (name, age) VALUES (:name, :age)",
+        user_data
+    )
+
+# 兼容旧版接口
+from app.database.db_mysql import get_db
+db = next(get_db())
+result = db.execute("SELECT * FROM users")
+db.commit()
+```
+
+新特性说明：
+
+1. 单例模式：确保全局只有一个数据库连接实例
+2. 上下文管理器：自动管理会话生命周期，异常时自动回滚
+3. 常用操作封装：
+   - execute_query：执行查询，返回字典列表
+   - execute_non_query：执行非查询，返回影响行数
+   - batch_execute：批量执行，返回总影响行数
+4. 新增 initialize()方法：显式初始化数据库连接
+5. 保持对旧版接口的兼容
+
+注意事项：
+
+1. 数据库连接字符串格式：mysql+pymysql://{user}:{password}@{host}:{port}/{name}?charset=utf8mb4
+2. 数据库连接池配置：
+   - pool_pre_ping: True # 每次使用连接前检查连接是否有效
+   - pool_recycle: 3600 # 连接池回收时间（秒）
+3. MySQL 驱动说明：
+   - PyMySQL：纯 Python 实现的 MySQL 驱动，兼容性好
+   - mysqlclient：C 扩展实现的 MySQL 驱动，性能更好
+   - 默认使用 PyMySQL，如需使用 mysqlclient：
+     - 安装 mysqlclient：pip install mysqlclient
+     - 修改连接字符串前缀为 mysql://
+
+配置模块提供了统一的配置管理接口，主要功能包括：
+
 - 获取配置实例
 - 获取配置项值
 - 注册配置变更回调
@@ -68,10 +143,10 @@ except RuntimeError as e:
 
 ```env
 # 开发环境
-APP_ENV=dev
+ENV=dev
 
 # 生产环境
-# APP_ENV=prod
+# ENV=prod
 ```
 
 注意：
@@ -86,10 +161,10 @@ APP_ENV=dev
 
 ```cmd
 :: 开发环境
-set APP_ENV=dev
+set ENV=dev
 
 :: 生产环境
-set APP_ENV=prod
+set ENV=prod
 ```
 
 2. 永久设置：
@@ -97,17 +172,17 @@ set APP_ENV=prod
 - 右键点击"此电脑" -> "属性" -> "高级系统设置"
 - 点击"环境变量"按钮
 - 在"系统变量"或"用户变量"中新建变量：
-  - 变量名：APP_ENV
+  - 变量名：ENV
   - 变量值：dev（开发环境）或 prod（生产环境）
 
 #### macOS/Linux 系统
 
 ```bash
 # 开发环境
-export APP_ENV=dev
+export ENV=dev
 
 # 生产环境
-export APP_ENV=prod
+export ENV=prod
 ```
 
 ### 各环境主要差异
@@ -131,16 +206,22 @@ export APP_ENV=prod
 
   - 日志级别：DEBUG
   - 日志路径：logs/
+  - 日志文件：app\_{time:YYYY-MM-DD}.log（按日期自动分割）
   - 日志保留：7 天
-  - 日志格式：包含时间、级别、文件名和行号
-  - 控制台输出：彩色格式
+  - 日志格式：{time:YYYY-MM-DD HH:mm:ss.SSS} | {level} | {module}:{line} - {message}
+  - 控制台输出：彩色格式（<green>{time}</green> | <level>{level}</level> | <cyan>{module}</cyan>:<cyan>{line}</cyan> - <level>{message}</level>）
+  - 旧日志压缩：自动压缩为 zip 格式
+  - 最大文件大小：10MB
 
 - 生产环境：
   - 日志级别：INFO
   - 日志路径：/var/log/app
+  - 日志文件：app\_{time:YYYY-MM-DD}.log（按日期自动分割）
   - 日志保留：30 天
-  - 日志格式：仅包含时间、级别和消息
+  - 日志格式：{time:YYYY-MM-DD HH:mm:ss.SSS} | {level} | {module}:{line} - {message}
   - 控制台输出：简洁格式
+  - 旧日志压缩：自动压缩为 zip 格式
+  - 最大文件大小：100MB
 
 使用示例：
 
@@ -150,3 +231,30 @@ from app.logger import logger
 logger.info("这是一条信息日志")
 logger.error("这是一条错误日志", exc_info=True)
 ```
+
+## 新增日志功能说明
+
+1. 按日期分割日志文件：
+
+   - 每天生成一个新的日志文件，文件名格式为 app_YYYY-MM-DD.log
+   - 自动处理日志文件切换，无需重启应用
+
+2. 自动清理旧日志：
+
+   - 开发环境保留最近 7 天的日志
+   - 生产环境保留最近 30 天的日志
+   - 超过保留期限的日志文件会被自动删除
+
+3. 日志压缩：
+
+   - 旧日志文件会被自动压缩为 zip 格式保存
+   - 压缩后的文件名格式为 app_YYYY-MM-DD.log.zip
+
+4. 线程安全：
+
+   - 日志写入操作是线程安全的
+   - 支持多线程并发写入日志
+
+5. 异常处理：
+   - 日志初始化失败时会抛出异常
+   - 日志写入失败时会自动重试

README_DEV.md

@@ -1,331 +1,49 @@
 # 智能题库学习系统 (Smart Question Bank Learning System)
 
-## 项目简介
+## 数据模型说明
 
-这是一个基于 Python 开发的智能题库学习系统，旨在帮助学生进行高效的题目练习和知识掌握。系统具备题目管理、智能练习、错题本、学习进度追踪等功能，让学习更有针对性和效果。
+### BaseModel
 
-## 主要功能
+- 基础模型，包含 id、创建时间、更新时间等通用字段
+- 提供 CRUD 操作的基类功能
+- 继承自 SQLAlchemy 的 Base 类
 
-### 1. 题库管理
+## 存储类说明
 
-- 支持多学科题目录入（数学、语文、英语等）
-- 题目分类管理（按照难度、知识点、题型等）
-- 支持多种题型（选择题、填空题、简答题等）
-- 每道题目包含详细解析和知识点说明
-- 题目标签系统，方便分类和搜索
+### BaseStore
 
-### 2. 智能练习
+所有存储类的基类，提供以下功能：
 
-- 根据学生水平自动推荐适合的题目
-- 支持自选题型和难度进行练习
-- 练习过程中即时查看答案和解析
-- 练习完成后显示得分和错题分析
-- 优化推荐算法，结合用户历史表现和学习目标
-- 个性化推荐，根据用户偏好推荐题目
-- 自适应练习模式，动态调整题目难度
+- 基本的 CRUD 操作
+- 软删除支持（通过 is_deleted 字段）
+- 唯一性检查
+- 恢复已删除对象
+- 查询包含/排除已删除对象
 
-### 3. 错题本功能
+### ExamStore
 
-- 自动收集做错的题目
-- 错题分类整理
-- 支持错题重复练习
-- 错题数据分析，找出薄弱知识点
-- 错题原因分析，帮助用户理解错误原因
-- 错题标签，方便分类和复习
-- 错题相似题推荐，巩固薄弱知识点
+考试存储类，提供考试相关的特定操作：
 
-### 4. 学习进度追踪
+- 根据考试名称获取考试
+- 根据学科获取考试列表
+- 获取所有激活的考试
+- 获取即将开始的考试
+- 获取包含题目信息的考试列表
 
-- 展示每日/每周/每月的练习情况
-- 知识点掌握程度分析
-- 学习时长统计
-- 成绩进步曲线
-- 知识点掌握预测，预估用户掌握程度
-- 学习路径规划，提供个性化学习建议
-- 学习目标设置，帮助用户明确学习方向
-- 手动考试成绩录入，记录各类考试成绩
-- 考试成绩分析，生成成绩报告
+### QuestionStore
 
-## 技术架构
+题目存储类，提供题目相关的特定操作：
 
-- 后端：Python 3.13
-- 数据库：Mysql
-- 用户界面：Web 界面
+- 根据学科获取题目列表
 
-## 数据库设计
+### KnowledgePointStore
 
-### 用户表(sys_users)
+知识点存储类，提供知识点相关的特定操作：
 
-- id: 用户 ID
-- username: 用户名
-- password: 密码
-- grade: 年级
-- create_time: 创建时间
-- update_time: 更新时间
-- create_id: 创建者 ID
-- update_id: 更新者 ID
-- is_deleted: 是否删除 (0:未删除, 1:已删除)
-- delete_time: 删除时间
-- delete_id: 删除者 ID
-- preferences: 学习偏好
-- goals: 学习目标
-- learning_style: 学习风格
-- target_score: 目标分数
+- 根据学科获取知识点列表
+- 根据难度等级获取知识点列表
+- 获取所有根知识点
+- 根据父知识点 ID 获取子知识点列表
+- 获取完整的知识点树结构
 
-### 系统角色表(sys_roles)
-
-- id: 角色 ID
-- name: 角色名称
-- code: 角色代码
-- description: 角色描述
-- status: 状态（0:禁用，1:启用）
-- create_id: 创建者 ID
-- create_time: 创建时间
-- update_id: 更新者 ID
-- update_time: 更新时间
-- is_deleted: 是否删除 (0:未删除, 1:已删除)
-- delete_time: 删除时间
-- delete_id: 删除者 ID
-
-### 系统用户角色表(sys_user_roles)
-
-- id: 主键 ID
-- user_id: 用户 ID
-- role_id: 角色 ID
-- create_time: 创建时间
-- create_id: 创建者 ID
-
-### 系统权限表(sys_permissions)
-
-- id: 权限 ID
-- code: 权限代码
-- score: 权限作用域 U: 用户，R: 角色
-- value: 权限值
-- description: 权限描述
-- create_time: 创建时间
-- create_id: 创建者 ID
-
-### 字典类型表(sys_dict_type)
-
-- id: 字典类型 ID
-- name: 字典名称
-- type: 字典类型
-- status: 状态（0:禁用，1:启用）
-- remark: 备注
-- create_id: 创建者 ID
-- create_time: 创建时间
-- update_id: 更新者 ID
-- update_time: 更新时间
-- is_deleted: 是否删除 (0:未删除, 1:已删除)
-- delete_time: 删除时间
-- delete_id: 删除者 ID
-
-### 字典数据表(sys_dict_data)
-
-- id: 字典数据 ID
-- dict_type: 字典类型
-- dict_label: 字典标签
-- dict_value: 字典值
-- sort: 排序
-- status: 状态（0:禁用，1:启用）
-- remark: 备注
-- create_id: 创建者 ID
-- create_time: 创建时间
-- update_id: 更新者 ID
-- update_time: 更新时间
-- is_deleted: 是否删除 (0:未删除, 1:已删除)
-- delete_time: 删除时间
-- delete_id: 删除者 ID
-
-### 题目表(questions)
-
-- id: 题目 ID
-- subject: 学科
-- type: 题型
-- difficulty: 难度等级
-- content: 题目内容
-- answer: 答案
-- explanation: 解析
-- knowledge_points: 相关知识点
-- version: 题目版本
-- status: 审核状态
-- tags: 题目标签
-- creator_id: 题目创建者
-- reviewer_id: 题目审核者
-- create_time: 创建时间
-- update_time: 更新时间
-- create_id: 创建者 ID
-- update_id: 更新者 ID
-- is_deleted: 是否删除 (0:未删除, 1:已删除)
-- delete_time: 删除时间
-- delete_id: 删除者 ID
-
-### 练习记录表(practice_records)
-
-- id: 记录 ID
-- user_id: 用户 ID
-- question_id: 题目 ID
-- is_correct: 是否正确
-- practice_time: 练习时间
-- time_spent: 用时
-- steps: 答题步骤
-- thinking_time: 思考时间
-- confidence_level: 答题信心
-- feedback: 用户反馈
-- create_id: 创建者 ID
-- create_time: 创建时间
-- update_id: 更新者 ID
-- update_time: 更新时间
-- is_deleted: 是否删除 (0:未删除, 1:已删除)
-- delete_time: 删除时间
-- delete_id: 删除者 ID
-
-### 知识点表(knowledge_points)
-
-- id: 知识点 ID
-- name: 知识点名称
-- description: 知识点描述
-- subject: 所属学科
-- difficulty: 难度等级
-- prerequisite: 先修知识点
-- create_id: 创建者 ID
-- create_time: 创建时间
-- update_id: 更新者 ID
-- update_time: 更新时间
-- is_deleted: 是否删除 (0:未删除, 1:已删除)
-- delete_time: 删除时间
-- delete_id: 删除者 ID
-
-### 学习路径表(learning_paths)
-
-- id: 路径 ID
-- user_id: 用户 ID
-- target_knowledge_point: 目标知识点
-- path: 学习路径
-- progress: 当前进度
-- create_id: 创建者 ID
-- create_time: 创建时间
-- update_id: 更新者 ID
-- update_time: 更新时间
-- is_deleted: 是否删除 (0:未删除, 1:已删除)
-- delete_time: 删除时间
-- delete_id: 删除者 ID
-
-### 考试成绩表(exam_scores)
-
-- id: 成绩 ID
-- user_id: 用户 ID
-- exam_name: 考试名称
-- subject: 考试科目
-- score: 考试成绩
-- total_score: 总分
-- exam_date: 考试日期
-- remark: 备注
-- create_time: 创建时间
-- update_time: 更新时间
-- create_id: 创建者 ID
-- update_id: 更新者 ID
-- is_deleted: 是否删除 (0:未删除, 1:已删除)
-- delete_time: 删除时间
-- delete_id: 删除者 ID
-
-## 使用说明
-
-1. 系统启动后，首先需要登录账号，账号由管理员派发
-2. 选择想要练习的学科和题型
-3. 开始答题，系统会记录答题情况
-4. 可以随时查看错题本和学习进度
-5. 在成绩管理页面手动录入考试成绩
-
-## 项目结构
-
-```
-smart-question-bank/
-├── SERVER/               # 后端服务
-│   ├── app/              # APP目录
-│   │   ├── common/       # 公共模块
-│   │   ├── exception/    # 异常处理
-│   │   ├── models/       # 数据模型
-│   │   ├── stores/       # 数据持久层
-│   │   ├── services/     # 业务逻辑
-│   │   ├── controllers/  # API控制器
-│   │   ├── middleware/   # 中间件
-│   │   ├── utils/        # 工具类
-│   │   ├── config/       # 配置管理
-│   │   ├── logger/       # 日志管理
-│   │   ├── data_seeder/  # 数据初始化
-│   │   └── main.py       # 程序入口
-│   ├── tests/            # 测试目录
-│   ├── requirements.txt  # 依赖包列表
-│   └── Dockerfile        # Docker配置
-├── UI/                   # 前端服务
-│   ├── vue/              # Web项目
-│   │   ├── public/       #
-│   │   ├── src/          #
-│   │   └── Dockerfile    # Docker配置
-│   └── app/              # 移动端项目
-│       └── src/          #
-├── docker-compose.yml    # Docker Compose配置
-├── README.md             # 项目说明
-└── .gitignore            # Git忽略配置
-```
-
-调用顺序：
-controllers => services => stores => models
-
-## 安装与运行
-
-1. 安装依赖
-
-```bash
-pip install -r requirements.txt
-```
-
-2. 配置数据库
-
-- 创建 MySQL 数据库
-- 修改 src/common/config.py 中的数据库连接配置
-
-3. 运行系统
-
-```bash
-python src/main.py
-```
-
-## 已实现功能
-
-### 1. 图形用户界面
-
-- 使用 PyQt5 开发桌面应用界面
-- 提供友好的用户交互体验
-- 支持题目展示、答题、结果查看等功能
-
-### 2. 在线题库更新
-
-- 实现题库云端同步功能
-- 支持自动更新最新题目
-- 提供题目审核机制
-
-### 3. 做题时间限制
-
-- 支持按题型设置时间限制
-- 提供倒计时功能
-- 超时自动提交答案
-
-### 4. 班级管理
-
-- 教师创建和管理班级
-- 学生加入班级
-- 班级练习统计和分析
-
-### 5. 题目难度评估
-
-- 基于学生答题数据自动评估题目难度
-- 动态调整题目难度系数
-- 提供难度分布可视化
-
-### 6. 智能出题算法
-
-- 基于知识图谱的题目推荐
-- 个性化出题策略
-- 自适应难度调整
+[保留原有 README_DEV.md 的其余内容...]

README_UTILS.md

@@ -0,0 +1,120 @@
+# 工具模块使用说明
+
+## 数据处理工具
+
+### to_json(data)
+
+- 功能：将 Python 对象转换为 JSON 字符串
+- 参数：
+  - data: 需要转换的 Python 对象
+- 返回：JSON 格式的字符串
+
+```python
+from app.utils import to_json
+
+data = {'key': 'value'}
+json_str = to_json(data)
+```
+
+## 数据验证工具
+
+### is_email(email)
+
+- 功能：验证邮箱格式
+- 参数：
+  - email: 需要验证的邮箱地址
+- 返回：布尔值，True 表示格式正确
+
+```python
+from app.utils import is_email
+
+is_valid = is_email('test@example.com')
+```
+
+### is_valid_ip(ip)
+
+- 功能：验证 IP 地址格式
+- 参数：
+  - ip: 需要验证的 IP 地址
+- 返回：布尔值，True 表示格式正确
+
+```python
+from app.utils import is_valid_ip
+
+is_valid = is_valid_ip('192.168.1.1')
+```
+
+## 文件操作工具
+
+### get_file_size(file_path)
+
+- 功能：获取文件大小
+- 参数：
+  - file_path: 文件路径
+- 返回：文件大小（字节）
+
+```python
+from app.utils import get_file_size
+
+size = get_file_size('test.txt')
+```
+
+## 日期时间工具
+
+### get_current_time()
+
+- 功能：获取当前时间
+- 返回：当前时间的 datetime 对象
+
+```python
+from app.utils import get_current_time
+
+now = get_current_time()
+```
+
+### format_date(date, fmt='%Y-%m-%d %H:%M:%S')
+
+- 功能：格式化日期时间
+- 参数：
+  - date: 需要格式化的日期时间
+  - fmt: 格式化字符串（可选，默认'%Y-%m-%d %H:%M:%S'）
+- 返回：格式化后的字符串
+
+```python
+from app.utils import format_date
+
+now = get_current_time()
+formatted = format_date(now)
+```
+
+### parse_date(date_str, fmt='%Y-%m-%d %H:%M:%S')
+
+- 功能：解析字符串为日期时间
+- 参数：
+  - date_str: 日期时间字符串
+  - fmt: 格式化字符串（可选，默认'%Y-%m-%d %H:%M:%S'）
+- 返回：解析后的 datetime 对象
+
+```python
+from app.utils import parse_date
+
+dt = parse_date('2023-01-01 12:00:00')
+```
+
+### get_timestamp()
+
+- 功能：获取当前时间戳
+- 返回：当前时间的时间戳（秒）
+
+```python
+from app.utils import get_timestamp
+
+ts = get_timestamp()
+```
+
+## 使用建议
+
+1. 建议通过`from app.utils import ...`的方式按需导入工具方法
+2. 所有工具方法都经过单元测试，可以直接使用
+3. 如果遇到问题，请检查相关参数是否正确
+4. 日期时间工具使用时请注意时区问题

SERVER/app/application-dev.yml

@@ -1,20 +1,21 @@
 app:
-  name: Smart Question Bank
+  name: 智能题库学习系统
   version: 1.0.1
   debug: true
 
-database:
-  host: 192.168.0.82
-  port: 3306
-  user: root
-  password: ''
-  name: question_bank_dev
+# database:
+#   host: 192.168.0.82
+#   port: 3306
+#   user: root
+#   password: ''
+#   name: question_bank_dev
 
 logging:
-  path: logs
-  retention: 7 # 开发环境保留7天日志
   level: DEBUG # 开发环境记录所有日志
-  format: '{time:YYYY-MM-DD HH:mm:ss.SSS} | {level} | {file}:{line} | {message}'
-  console_format: '<green>{time:YYYY-MM-DD HH:mm:ss.SSS}</green> | <level>{level}</level> | <cyan>{file}:{line}</cyan> | {message}'
+  path: logs
+  file: app.log
   max_size: 10485760
   backup_count: 5
+  retention_days: 7 # 开发环境保留7天日志
+  format: '{time:YYYY-MM-DD HH:mm:ss.SSS} | {level} | {file}:{line} | {message}'
+  console_format: '<green>{time:YYYY-MM-DD HH:mm:ss.SSS}</green> | <level>{level}</level> | <cyan>{file}:{line}</cyan> | {message}'

SERVER/app/application-prod.yml

@@ -1,5 +1,5 @@
 app:
-  name: Smart Question Bank
+  name: 智能题库学习系统
   version: 1.0.0
   debug: false
 
@@ -11,10 +11,11 @@ database:
   name: question_bank_prod
 
 logging:
-  path: /var/log/app
-  retention: 30 # 生产环境保留30天日志
   level: INFO # 生产环境只记录重要日志
+  path: /var/log/app
+  file: app.log
+  backup_count: 30 # 生产环境保留更多备份
+  retention_days: 30 # 生产环境保留30天日志
   format: '{time:YYYY-MM-DD HH:mm:ss.SSS} | {level} | {message}'
   console_format: '{time:YYYY-MM-DD HH:mm:ss.SSS} | {level} | {message}'
   max_size: 104857600 # 生产环境日志文件更大
-  backup_count: 30 # 生产环境保留更多备份

SERVER/app/application.yml

@@ -4,17 +4,18 @@ app:
   debug: false
 
 database:
-  host: 192.168.0.81
+  host: 192.168.0.82
   port: 3306
   user: root
-  password: ''
-  name: question_bank
+  password: y123456
+  name: smart_question_bank_v1.0
 
 logging:
   level: INFO
+  path: logs
   file: app.log
+  retention_days: 7 # 日志保留天数
   max_size: 10485760
   backup_count: 5
-  path: logs
   format: '{time:YYYY-MM-DD HH:mm:ss.SSS} | {level} | {module}:{line} - {message}'
   console_format: '<green>{time:YYYY-MM-DD HH:mm:ss.SSS}</green> | <level>{level}</level> | <cyan>{module}</cyan>:<cyan>{line}</cyan> - <level>{message}</level>'

SERVER/app/config/config_database.py

@@ -1,5 +1,4 @@
 from pydantic_settings import BaseSettings
-from pydantic import field_validator
 
 
 class DatabaseConfig(BaseSettings):
@@ -8,24 +7,4 @@ class DatabaseConfig(BaseSettings):
     port: int = 3306
     user: str = "root"
     password: str = ""
-    name: str = "question_bank"
-
-    @field_validator('port')
-    def validate_db_port(cls, value):
-        if value < 1024 or value > 65535:
-            raise ValueError('数据库端口必须在1024-65535之间')
-        return value
-
-    @field_validator('name')
-    def validate_db_name(cls, value):
-        if not value or len(value) < 2:
-            raise ValueError('数据库名称不能为空且至少2个字符')
-        return value
-
-    def to_string(self, indent: int = 0) -> str:
-        """将配置转换为字符串"""
-        indent_str = ' ' * indent
-        result = []
-        for field_name, field_value in self.__dict__.items():
-            result.append(f"{indent_str}{field_name}: {field_value}")
-        return '\n'.join(result)
+    name: str = ""

SERVER/app/config/config_env_mapping.py

@@ -0,0 +1,12 @@
+"""环境变量映射配置"""
+
+ENV_MAPPING = {
+    'APP_NAME': 'app.name',
+    'APP_VERSION': 'app.version',
+    'APP_DEBUG': 'app.debug',
+    'DB_HOST': 'database.host',
+    'DB_PORT': 'database.port',
+    'DB_USER': 'database.user',
+    'DB_PASSWORD': 'database.password',
+    'DB_NAME': 'database.name',
+}

SERVER/app/config/config_logging.py

@@ -9,6 +9,7 @@ class LoggingConfig(BaseSettings):
     path: str = "logs"
     max_size: int = 10485760  # 10MB
     backup_count: int = 5
+    retention_days: int = 7  # 日志保留天数
     format: str = "{time:YYYY-MM-DD HH:mm:ss.SSS} | {level} | {module}:{line} - {message}"
     console_format: str = "<green>{time:YYYY-MM-DD HH:mm:ss.SSS}</green> | <level>{level}</level> | <cyan>{module}</cyan>:<cyan>{line}</cyan> - <level>{message}</level>"
 

SERVER/app/config/loader_config.py

@@ -2,7 +2,9 @@ import os, yaml, datetime
 from typing import Any, List, Callable
 from watchdog.observers import Observer
 from watchdog.events import FileSystemEventHandler
+from dotenv import load_dotenv
 from .config import Config
+from .config_env_mapping import ENV_MAPPING
 
 
 class ConfigReloadHandler(FileSystemEventHandler):
@@ -21,18 +23,6 @@ class ConfigLoader:
     _instance = None
     _observers = []
 
-    # 环境变量映射配置
-    ENV_MAPPING = {
-        'APP_NAME': 'app.name',
-        'APP_VERSION': 'app.version',
-        'APP_DEBUG': 'app.debug',
-        'DB_HOST': 'database.host',
-        'DB_PORT': 'database.port',
-        'DB_USER': 'database.user',
-        'DB_PASSWORD': 'database.password',
-        'DB_NAME': 'database.name',
-    }
-
     def __new__(cls):
         if cls._instance is None:
             cls._instance = super().__new__(cls)
@@ -41,7 +31,7 @@ class ConfigLoader:
 
     def _load_env_vars(self, config: Config) -> Config:
         """加载环境变量配置"""
-        for env_key, config_path in self.ENV_MAPPING.items():
+        for env_key, config_path in ENV_MAPPING.items():
             value = os.getenv(env_key)
             if value is None:
                 continue
@@ -152,16 +142,26 @@ class ConfigLoader:
         self._config = Config()
         self._callbacks: List[Callable] = []
 
+        # 获取当前文件所在目录并保存到类变量
+        self._current_dir = os.path.dirname(
+            os.path.dirname(os.path.abspath(__file__)))
+        load_dotenv(os.path.join(self._current_dir, '.env'))
+        # print(f"配置当前目录为：{self._current_dir}")
+
         # 1. 加载默认值（通过Config类初始化）
 
         # 2. 加载通用配置
         self._config.update_from_dict(
-            self._load_yaml_config('SERVER/app/application.yml'))
+            self._load_yaml_config(
+                os.path.join(self._current_dir, 'application.yml')))
 
         # 3. 加载指定环境配置
-        env = os.getenv('ENV', 'dev')
-        self._config.update_from_dict(
-            self._load_yaml_config(f'SERVER/app/application-{env}.yml'))
+        env = os.getenv('ENV')
+        print(f"当前环境：{env}")
+        if env:
+            self._config.update_from_dict(
+                self._load_yaml_config(
+                    os.path.join(self._current_dir, f'application-{env}.yml')))
 
         # 4. 最后加载环境变量（最高优先级）
         self._config = self._load_env_vars(self._config)
@@ -173,13 +173,32 @@ class ConfigLoader:
         """启动配置文件监控"""
         event_handler = ConfigReloadHandler(self.reload_config)
         observer = Observer()
-        observer.schedule(event_handler, path='SERVER/app/', recursive=False)
+
+        # 只监控application.yml和application-{env}.yml文件
+        env = os.getenv('ENV', 'dev')
+        config_files = [
+            os.path.join(self._current_dir, 'application.yml'),
+            os.path.join(self._current_dir, f'application-{env}.yml')
+        ]
+
+        for config_file in config_files:
+            if os.path.exists(config_file):
+                observer.schedule(event_handler,
+                                  path=os.path.dirname(config_file),
+                                  recursive=False)
+
         observer.start()
         self._observers.append(observer)
 
     def reload_config(self):
         """重新加载配置"""
+        # 1. 重新加载配置文件
         self._load_config()
+
+        # 2. 重新加载环境变量（确保覆盖配置文件）
+        self._config = self._load_env_vars(self._config)
+
+        # 3. 触发回调
         for callback in self._callbacks:
             callback()
 

SERVER/app/database/__init__.py

@@ -0,0 +1,6 @@
+from .database import Database
+
+# 数据库初始化单例实例
+db = Database.initialize()
+
+__all__ = ['Database', 'db']

SERVER/app/database/database.py

@@ -0,0 +1,64 @@
+from typing import Iterator
+from contextlib import contextmanager
+from sqlalchemy import create_engine, text
+from sqlalchemy.orm import sessionmaker, Session
+from sqlalchemy.exc import OperationalError
+from ..config import config
+from ..logger import logger
+
+
+class Database:
+    """简化版数据库单例类"""
+    _instance = None
+
+    def __new__(cls):
+        if cls._instance is None:
+            cls._instance = super().__new__(cls)
+            cls._instance._init_db()
+        return cls._instance
+
+    def _init_db(self):
+        """初始化数据库连接"""
+        db_config = config.database
+        self.engine = create_engine(
+            f"mysql+pymysql://{db_config.user}:{db_config.password}@{db_config.host}:{db_config.port}/{db_config.name}?charset=utf8mb4"
+        )
+        self.SessionLocal = sessionmaker(autocommit=False,
+                                         autoflush=False,
+                                         bind=self.engine)
+
+        self.test_connection()
+
+    @contextmanager
+    def session(self) -> Iterator[Session]:
+        """获取数据库会话"""
+        db = self.SessionLocal()
+        try:
+            yield db
+            db.commit()
+        except Exception:
+            db.rollback()
+            raise
+        finally:
+            db.close()
+
+    @classmethod
+    def initialize(cls):
+        """初始化数据库"""
+        return cls()
+
+    def test_connection(self):
+        """测试数据库连接"""
+        try:
+            with self.engine.connect() as conn:
+                result = conn.execute(text("SELECT 1"))
+                if result.scalar() == 1:
+                    # logger.info(f"数据库 [{config.database.name}] 连接成功。")
+                    return True
+            return False
+        except OperationalError as e:
+            logger.error(f"数据库 [{config.database.name}] 连接失败: {e}")
+            return False
+        except Exception as e:
+            logger.error(f"未知错误: {e}")
+            return False

SERVER/app/logger/logging.py

@@ -37,9 +37,9 @@ class Logger:
 
             # 配置日志
             self._logger_id = self._logger.add(
-                log_file,
+                log_file.with_stem(f"{log_file.stem}_{{time:YYYY-MM-DD}}"),
                 rotation="1 day",  # 每天滚动日志
-                retention=f"{config.logging.backup_count} days",
+                retention=f"{config.logging.retention_days} days",
                 level=config.logging.level,
                 format=config.logging.format,
                 encoding="utf-8",

SERVER/app/main.py

@@ -1,17 +1,15 @@
-import sys
-import os
+import os, sys
 
 # 添加项目根目录到Python路径
-sys.path.append(
-    os.path.dirname(os.path.dirname(os.path.dirname(
-        os.path.abspath(__file__)))))
+sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
 
-from SERVER.app.config import config
-from SERVER.app.logger import logger
+from app.database import db
+from app.config import config
+from app.logger import logger
 
 
 def main():
-    logger.info(f"正在启动应用 [{config.app.name}]")
+    logger.info(f"正在启动应用 [{config.app.name}({config.app.version})]")
     logger.debug(f"调试模式: {config.app.debug}")
 
     # logger.info(f"加载配置文件:\n{config.to_string()}")

SERVER/app/models/base_model.py

@@ -0,0 +1,48 @@
+from datetime import datetime
+from typing import Optional, Generic, TypeVar
+from sqlalchemy import Column, Integer, DateTime, Boolean
+from sqlalchemy.ext.declarative import as_declarative, declared_attr
+from pydantic import BaseModel as PydanticBaseModel
+from ..database import Base
+
+T = TypeVar('T')
+
+
+@as_declarative()
+class BaseModel(Base):
+    """
+    SQLAlchemy ORM 基础模型类
+    """
+    __abstract__ = True
+
+    @declared_attr
+    def __tablename__(cls) -> str:
+        return cls.__name__.lower()
+
+    id = Column(Integer, primary_key=True, index=True)
+
+
+class CreateModel:
+    """
+    创建模型Mixin，包含创建相关字段
+    """
+    create_at = Column(DateTime, default=datetime.now, comment='创建时间')
+    create_by = Column(Integer, comment='创建者ID')
+
+
+class UpdateModel:
+    """
+    更新模型Mixin，包含更新相关字段
+    """
+    update_at = Column(DateTime, onupdate=datetime.now, comment='更新时间')
+    update_by = Column(Integer, comment='更新者ID')
+    version = Column(Integer, default=1, comment='版本号')
+
+
+class DeleteModel:
+    """
+    删除模型Mixin，包含删除相关字段
+    """
+    is_deleted = Column(Boolean, default=False, comment='是否删除')
+    delete_at = Column(DateTime, comment='删除时间')
+    delete_by = Column(Integer, comment='删除者ID')

SERVER/app/models/exam_model.py

@@ -0,0 +1,22 @@
+from sqlalchemy import Column, Integer, String, DateTime
+from .base_model import BaseModel, CreateModel, UpdateModel, DeleteModel
+
+
+class ExamModel(BaseModel, CreateModel, UpdateModel, DeleteModel):
+    """
+    考试模型
+    对应数据库表: exams
+    """
+    __tablename__ = 'exams'
+
+    name = Column(String(100), nullable=False, comment='考试名称')
+    description = Column(String(500), comment='考试描述')
+    subject = Column(String(50), nullable=False, comment='所属学科')
+    total_score = Column(Integer, nullable=False, comment='总分')
+    duration = Column(Integer, nullable=False, comment='考试时长（分钟）')
+    start_time = Column(DateTime, nullable=False, comment='开始时间')
+    end_time = Column(DateTime, nullable=False, comment='结束时间')
+    status = Column(String(20), nullable=False, comment='状态')
+
+    def __repr__(self):
+        return f"<Exam(id={self.id}, name={self.name}, subject={self.subject})>"

SERVER/app/models/exam_question_relation_model.py

@@ -0,0 +1,25 @@
+from sqlalchemy import Column, Integer, Float, ForeignKey
+from .base_model import BaseModel, CreateModel, UpdateModel, DeleteModel
+
+
+class ExamQuestionRelationModel(BaseModel, CreateModel, UpdateModel,
+                                DeleteModel):
+    """
+    考试题目关联模型
+    对应数据库表: exam_question_relations
+    """
+    __tablename__ = 'exam_question_relations'
+
+    exam_id = Column(Integer,
+                     ForeignKey('exams.id'),
+                     nullable=False,
+                     comment='考试ID')
+    question_id = Column(Integer,
+                         ForeignKey('questions.id'),
+                         nullable=False,
+                         comment='题目ID')
+    score = Column(Float, nullable=False, comment='题目分值')
+    order = Column(Integer, nullable=False, comment='题目顺序')
+
+    def __repr__(self):
+        return f"<ExamQuestionRelation(id={self.id}, exam_id={self.exam_id}, question_id={self.question_id})>"

SERVER/app/models/exam_score_model.py

@@ -0,0 +1,26 @@
+from sqlalchemy import Column, Integer, Float, DateTime
+from .base_model import BaseModel, CreateModel, UpdateModel, DeleteModel
+
+
+class ExamScoreModel(BaseModel, CreateModel, UpdateModel, DeleteModel):
+    """
+    考试成绩模型
+    对应数据库表: exam_scores
+    """
+    __tablename__ = 'exam_scores'
+
+    user_id = Column(Integer,
+                     ForeignKey('users.id'),
+                     nullable=False,
+                     comment='用户ID')
+    exam_id = Column(Integer,
+                     ForeignKey('exams.id'),
+                     nullable=False,
+                     comment='考试ID')
+    score = Column(Float, nullable=False, comment='考试成绩')
+    start_time = Column(DateTime, nullable=False, comment='开始时间')
+    end_time = Column(DateTime, nullable=False, comment='结束时间')
+    status = Column(String(20), nullable=False, comment='状态')
+
+    def __repr__(self):
+        return f"<ExamScore(id={self.id}, user_id={self.user_id}, exam_id={self.exam_id})>"

SERVER/app/models/exercise_model.py

@@ -0,0 +1,22 @@
+from sqlalchemy import Column, Integer, String, DateTime
+from .base_model import BaseModel, CreateModel, UpdateModel, DeleteModel
+
+
+class ExerciseModel(BaseModel, CreateModel, UpdateModel, DeleteModel):
+    """
+    练习模型
+    对应数据库表: exercises
+    """
+    __tablename__ = 'exercises'
+
+    name = Column(String(100), nullable=False, comment='练习名称')
+    description = Column(String(500), comment='练习描述')
+    subject = Column(String(50), nullable=False, comment='所属学科')
+    total_questions = Column(Integer, nullable=False, comment='题目总数')
+    duration = Column(Integer, comment='练习时长（分钟）')
+    start_time = Column(DateTime, comment='开始时间')
+    end_time = Column(DateTime, comment='结束时间')
+    status = Column(String(20), nullable=False, comment='状态')
+
+    def __repr__(self):
+        return f"<Exercise(id={self.id}, name={self.name}, subject={self.subject})>"

SERVER/app/models/exercise_question_relation_model.py

@@ -0,0 +1,17 @@
+from sqlalchemy import Column, Integer, ForeignKey
+from .base_model import BaseModel
+
+
+class ExerciseQuestionRelationModel(BaseModel):
+    """
+    练习-题目关联模型
+    对应数据库表: exercise_question_relations
+    """
+    __tablename__ = 'exercise_question_relations'
+
+    exercise_id = Column(Integer, ForeignKey('exercises.id'), primary_key=True)
+    question_id = Column(Integer, ForeignKey('questions.id'), primary_key=True)
+    order = Column(Integer, nullable=False, comment='题目顺序')
+
+    def __repr__(self):
+        return f"<ExerciseQuestionRelation(exercise_id={self.exercise_id}, question_id={self.question_id})>"

SERVER/app/models/exercise_record_detail_model.py

@@ -0,0 +1,29 @@
+from sqlalchemy import Column, Integer, Boolean, Text, ForeignKey
+from .base_model import BaseModel, CreateModel, UpdateModel, DeleteModel
+
+
+class ExerciseRecordDetailModel(BaseModel, CreateModel, UpdateModel,
+                                DeleteModel):
+    """
+    练习记录详情模型
+    对应数据库表: exercise_record_details
+    """
+    __tablename__ = 'exercise_record_details'
+
+    record_id = Column(Integer,
+                       ForeignKey('exercise_records.id'),
+                       nullable=False,
+                       comment='练习记录ID')
+    question_id = Column(Integer,
+                         ForeignKey('questions.id'),
+                         nullable=False,
+                         comment='题目ID')
+    selected_option_id = Column(Integer,
+                                ForeignKey('question_options.id'),
+                                comment='用户选择的选项ID')
+    user_answer = Column(Text, comment='用户答案')
+    is_correct = Column(Boolean, nullable=False, comment='是否正确')
+    time_spent = Column(Integer, nullable=False, comment='用时（秒）')
+
+    def __repr__(self):
+        return f"<ExerciseRecordDetail(id={self.id}, record_id={self.record_id}, question_id={self.question_id})>"

SERVER/app/models/exercise_record_model.py

@@ -0,0 +1,26 @@
+from sqlalchemy import Column, String, Integer, Float, DateTime, ForeignKey
+from .base_model import BaseModel, CreateModel, UpdateModel, DeleteModel
+
+
+class ExerciseRecordModel(BaseModel, CreateModel, UpdateModel, DeleteModel):
+    """
+    练习记录模型
+    对应数据库表: exercise_records
+    """
+    __tablename__ = 'exercise_records'
+
+    user_id = Column(Integer,
+                     ForeignKey('users.id'),
+                     nullable=False,
+                     comment='用户ID')
+    exercise_id = Column(Integer,
+                         ForeignKey('exercises.id'),
+                         nullable=False,
+                         comment='练习ID')
+    start_time = Column(DateTime, nullable=False, comment='开始时间')
+    end_time = Column(DateTime, comment='结束时间')
+    score = Column(Float, comment='得分')
+    status = Column(String(20), nullable=False, comment='状态')
+
+    def __repr__(self):
+        return f"<ExerciseRecord(id={self.id}, user_id={self.user_id}, exercise_id={self.exercise_id})>"

SERVER/app/models/knowledge_point_model.py

@@ -0,0 +1,21 @@
+from sqlalchemy import Column, Integer, String, Text, ForeignKey
+from .base_model import BaseModel, CreateModel, UpdateModel, DeleteModel
+
+
+class KnowledgePointModel(BaseModel, CreateModel, UpdateModel, DeleteModel):
+    """
+    知识点模型
+    对应数据库表: knowledge_points
+    """
+    __tablename__ = 'knowledge_points'
+
+    name = Column(String(100), nullable=False, comment='知识点名称')
+    description = Column(Text, comment='知识点描述')
+    subject = Column(String(50), nullable=False, comment='所属学科')
+    level = Column(Integer, nullable=False, comment='难度等级')
+    parent_id = Column(Integer,
+                       ForeignKey('knowledge_points.id'),
+                       comment='父知识点ID')
+
+    def __repr__(self):
+        return f"<KnowledgePoint(id={self.id}, name={self.name}, subject={self.subject})>"

SERVER/app/models/permission_model.py

@@ -0,0 +1,26 @@
+from datetime import datetime
+from typing import Optional
+from sqlalchemy import Column, Integer, String, DateTime, Boolean, Text
+from sqlalchemy.orm import relationship
+from .base_model import BaseModel, CreateModel, UpdateModel, DeleteModel
+
+
+class PermissionModel(BaseModel, CreateModel, UpdateModel, DeleteModel):
+    """
+    权限模型
+    对应数据库表: sys_permissions
+    """
+    __tablename__ = 'sys_permissions'
+
+    name = Column(String(50), unique=True, nullable=False, comment='权限名称')
+    code = Column(String(50), unique=True, nullable=False, comment='权限代码')
+    description = Column(Text, comment='权限描述')
+    status = Column(Boolean, default=True, comment='状态')
+
+    # 与角色的多对多关系
+    roles = relationship('RoleModel',
+                         secondary='sys_role_permissions',
+                         back_populates='permissions')
+
+    def __repr__(self):
+        return f"<Permission(id={self.id}, name={self.name})>"

SERVER/app/models/question_knowledge_point_model.py

@@ -0,0 +1,18 @@
+from sqlalchemy import Column, Integer, ForeignKey
+from .base_model import BaseModel
+
+
+class QuestionKnowledgePointModel(BaseModel):
+    """
+    题目-知识点关联模型
+    对应数据库表: question_knowledge_points
+    """
+    __tablename__ = 'question_knowledge_points'
+
+    question_id = Column(Integer, ForeignKey('questions.id'), primary_key=True)
+    knowledge_point_id = Column(Integer,
+                                ForeignKey('knowledge_points.id'),
+                                primary_key=True)
+
+    def __repr__(self):
+        return f"<QuestionKnowledgePoint(question_id={self.question_id}, knowledge_point_id={self.knowledge_point_id})>"

SERVER/app/models/question_model.py

@@ -0,0 +1,31 @@
+from sqlalchemy import Column, Integer, String, Text, DateTime, ForeignKey
+from sqlalchemy.orm import relationship
+from .base_model import BaseModel, CreateModel, UpdateModel, DeleteModel
+
+
+class QuestionModel(BaseModel, CreateModel, UpdateModel, DeleteModel):
+    """
+    题目模型
+    对应数据库表: questions
+    """
+    __tablename__ = 'questions'
+
+    subject = Column(String(50), nullable=False, comment='学科')
+    type = Column(String(50), nullable=False, comment='题型')
+    difficulty = Column(Integer, nullable=False, comment='难度等级')
+    content = Column(Text, nullable=False, comment='题目内容')
+    answer = Column(Text, comment='答案')
+    explanation = Column(Text, comment='解析')
+    version = Column(String(20), comment='题目版本')
+    tags = Column(String(255), comment='题目标签')
+
+    # 与选项的一对多关系
+    options = relationship('QuestionOptionModel', back_populates='question')
+
+    # 与知识点的多对多关系
+    knowledge_points = relationship('KnowledgePointModel',
+                                    secondary='question_knowledge_points',
+                                    back_populates='questions')
+
+    def __repr__(self):
+        return f"<Question(id={self.id}, content={self.content[:20]}...)>"

SERVER/app/models/question_option_model.py

@@ -0,0 +1,24 @@
+from sqlalchemy import Column, Integer, String, Boolean, ForeignKey
+from .base_model import BaseModel, CreateModel, UpdateModel, DeleteModel
+
+
+class QuestionOptionModel(BaseModel, CreateModel, UpdateModel, DeleteModel):
+    """
+    题目选项模型
+    对应数据库表: question_options
+    """
+    __tablename__ = 'question_options'
+
+    question_id = Column(Integer,
+                         ForeignKey('questions.id'),
+                         nullable=False,
+                         comment='题目ID')
+    option_text = Column(String(500), nullable=False, comment='选项内容')
+    is_correct = Column(Boolean,
+                        nullable=False,
+                        default=False,
+                        comment='是否为正确答案')
+    order = Column(Integer, nullable=False, comment='选项顺序')
+
+    def __repr__(self):
+        return f"<QuestionOption(id={self.id}, question_id={self.question_id}, option_text={self.option_text})>"

SERVER/app/models/role_model.py

@@ -0,0 +1,31 @@
+from datetime import datetime
+from typing import Optional
+from sqlalchemy import Column, Integer, String, DateTime, Boolean, Text
+from sqlalchemy.orm import relationship
+from .base_model import BaseModel, CreateModel, UpdateModel, DeleteModel
+
+
+class RoleModel(BaseModel, CreateModel, UpdateModel, DeleteModel):
+    """
+    角色模型
+    对应数据库表: sys_roles
+    """
+    __tablename__ = 'sys_roles'
+
+    name = Column(String(50), unique=True, nullable=False, comment='角色名称')
+    code = Column(String(50), unique=True, nullable=False, comment='角色代码')
+    description = Column(Text, comment='角色描述')
+    status = Column(Boolean, default=True, comment='状态')
+
+    # 与用户的多对多关系
+    users = relationship('UserModel',
+                         secondary='sys_user_roles',
+                         back_populates='roles')
+
+    # 与权限的多对多关系
+    permissions = relationship('PermissionModel',
+                               secondary='sys_role_permissions',
+                               back_populates='roles')
+
+    def __repr__(self):
+        return f"<Role(id={self.id}, name={self.name})>"

SERVER/app/models/role_permission_model.py

@@ -0,0 +1,15 @@
+from sqlalchemy import Column, Integer, ForeignKey
+from .base_model import BaseModel
+
+
+class RolePermissionModel(BaseModel):
+    """
+    角色-权限关联表模型
+    对应数据库表: sys_role_permissions
+    """
+    __tablename__ = 'sys_role_permissions'
+
+    role_id = Column(Integer, ForeignKey('sys_roles.id'), primary_key=True)
+    permission_id = Column(Integer,
+                           ForeignKey('sys_permissions.id'),
+                           primary_key=True)

SERVER/app/models/status_model.py

@@ -0,0 +1,14 @@
+from sqlalchemy import Column, String, Integer, ForeignKey
+from .base_model import BaseModel, CreateModel, UpdateModel, DeleteModel
+
+
+class StatusModel(BaseModel, CreateModel, UpdateModel, DeleteModel):
+    __tablename__ = 'sys_status'
+
+    name = Column(String(100), nullable=False, comment="状态名称")
+    type = Column(String(100), nullable=False, unique=True, comment="状态类型")
+    label = Column(String(100), nullable=False, comment="显示名称")
+    value = Column(String(500), nullable=True, comment="状态值")
+    sort = Column(Integer, default=0)
+    status = Column(Integer, default=1, comment="状态（1正常 0停用）")
+    remark = Column(String(500), nullable=True, comment="备注")

SERVER/app/models/user_model.py

@@ -0,0 +1,29 @@
+from datetime import datetime
+from typing import Optional
+from sqlalchemy import Column, Integer, String, DateTime, Boolean, JSON
+from sqlalchemy.orm import relationship
+from .base_model import BaseModel, CreateModel, UpdateModel, DeleteModel
+
+
+class UserModel(BaseModel, CreateModel, UpdateModel, DeleteModel):
+    """
+    用户模型
+    对应数据库表: sys_users
+    """
+    __tablename__ = 'sys_users'
+
+    username = Column(String(50), unique=True, nullable=False, comment='用户名')
+    password = Column(String(255), nullable=False, comment='密码')
+    grade = Column(String(20), comment='年级')
+    preferences = Column(JSON, comment='学习偏好')
+    goals = Column(JSON, comment='学习目标')
+    learning_style = Column(String(50), comment='学习风格')
+    target_score = Column(Integer, comment='目标分数')
+
+    # 与角色的一对多关系
+    roles = relationship('RoleModel',
+                         secondary='sys_user_roles',
+                         back_populates='users')
+
+    def __repr__(self):
+        return f"<User(id={self.id}, username={self.username})>"

SERVER/app/models/user_role_model.py

@@ -0,0 +1,13 @@
+from sqlalchemy import Column, Integer, ForeignKey
+from .base_model import BaseModel
+
+
+class UserRoleModel(BaseModel):
+    """
+    用户-角色关联表模型
+    对应数据库表: sys_user_roles
+    """
+    __tablename__ = 'sys_user_roles'
+
+    user_id = Column(Integer, ForeignKey('sys_users.id'), primary_key=True)
+    role_id = Column(Integer, ForeignKey('sys_roles.id'), primary_key=True)

SERVER/app/models/wrong_question_model.py

@@ -0,0 +1,26 @@
+from sqlalchemy import Column, Integer, DateTime, ForeignKey
+from .base_model import BaseModel, CreateModel, UpdateModel, DeleteModel
+
+
+class WrongQuestionModel(BaseModel, CreateModel, UpdateModel, DeleteModel):
+    """
+    错题模型
+    对应数据库表: wrong_questions
+    """
+    __tablename__ = 'wrong_questions'
+
+    user_id = Column(Integer,
+                     ForeignKey('users.id'),
+                     nullable=False,
+                     comment='用户ID')
+    question_id = Column(Integer,
+                         ForeignKey('questions.id'),
+                         nullable=False,
+                         comment='题目ID')
+    exam_id = Column(Integer, ForeignKey('exams.id'), comment='考试ID')
+    exercise_id = Column(Integer, ForeignKey('exercises.id'), comment='练习ID')
+    wrong_count = Column(Integer, nullable=False, default=1, comment='错误次数')
+    last_wrong_time = Column(DateTime, nullable=False, comment='最后错误时间')
+
+    def __repr__(self):
+        return f"<WrongQuestion(id={self.id}, user_id={self.user_id}, question_id={self.question_id})>"

SERVER/app/stores/base_store.py

@@ -0,0 +1,85 @@
+from typing import TypeVar, Generic, Optional, Dict, Any
+from sqlalchemy.orm import Session
+from sqlalchemy import and_
+from ..models.base_model import BaseModel, DeleteModel
+
+T = TypeVar('T', bound=BaseModel)
+
+
+class BaseStore(Generic[T]):
+    """
+    基础存储类，提供CRUD操作
+    """
+
+    def __init__(self, db: Session, model: type[T]):
+        self.db = db
+        self.model = model
+
+    def get(self, id: int, include_deleted: bool = False) -> Optional[T]:
+        """根据ID获取单个对象"""
+        query = self.db.query(self.model).filter(self.model.id == id)
+        if not include_deleted and issubclass(self.model, DeleteModel):
+            query = query.filter(self.model.is_deleted == False)
+        return query.first()
+
+    def get_all(self,
+                skip: int = 0,
+                limit: int = 100,
+                include_deleted: bool = False) -> list[T]:
+        """获取所有对象"""
+        query = self.db.query(self.model)
+        if not include_deleted and issubclass(self.model, DeleteModel):
+            query = query.filter(self.model.is_deleted == False)
+        return query.offset(skip).limit(limit).all()
+
+    def create(self, obj: T, unique_fields: Dict[str, Any] = None) -> T:
+        """创建新对象"""
+        if unique_fields:
+            self._check_unique_constraints(obj, unique_fields)
+        self.db.add(obj)
+        self.db.commit()
+        self.db.refresh(obj)
+        return obj
+
+    def update(self, obj: T, unique_fields: Dict[str, Any] = None) -> T:
+        """更新对象"""
+        if unique_fields:
+            self._check_unique_constraints(obj, unique_fields)
+        self.db.commit()
+        self.db.refresh(obj)
+        return obj
+
+    def delete(self, id: int, soft_delete: bool = True) -> None:
+        """删除对象"""
+        obj = self.get(id)
+        if obj:
+            if soft_delete and issubclass(self.model, DeleteModel):
+                obj.is_deleted = True
+                self.db.commit()
+            else:
+                self.db.delete(obj)
+                self.db.commit()
+
+    def _check_unique_constraints(self, obj: T,
+                                  unique_fields: Dict[str, Any]) -> None:
+        """检查唯一性约束"""
+        for field, value in unique_fields.items():
+            query = self.db.query(
+                self.model).filter(getattr(self.model, field) == value)
+            if obj.id:
+                query = query.filter(self.model.id != obj.id)
+            if issubclass(self.model, DeleteModel):
+                query = query.filter(self.model.is_deleted == False)
+            if query.first():
+                raise ValueError(f"{field} must be unique")
+
+    def restore(self, id: int) -> Optional[T]:
+        """恢复软删除的对象"""
+        if not issubclass(self.model, DeleteModel):
+            return None
+        obj = self.get(id, include_deleted=True)
+        if obj and obj.is_deleted:
+            obj.is_deleted = False
+            self.db.commit()
+            return obj
+        return None

SERVER/app/stores/exam_store.py

@@ -0,0 +1,43 @@
+from typing import Optional, List
+from datetime import datetime
+from sqlalchemy.orm import Session, joinedload
+from ..models.exam_model import ExamModel
+from .base_store import BaseStore
+
+
+class ExamStore(BaseStore[ExamModel]):
+    """
+    考试存储类，继承自BaseStore
+    提供考试相关的特定操作
+    """
+
+    def __init__(self, db: Session):
+        super().__init__(db, ExamModel)
+
+    def get_by_name(self, name: str) -> Optional[ExamModel]:
+        """根据考试名称获取考试"""
+        return self.db.query(
+            self.model).filter(self.model.name == name).first()
+
+    def get_by_subject(self, subject: str) -> List[ExamModel]:
+        """根据学科获取考试列表"""
+        return self.db.query(
+            self.model).filter(self.model.subject == subject).all()
+
+    def get_active_exams(self) -> List[ExamModel]:
+        """获取所有激活的考试"""
+        return self.db.query(
+            self.model).filter(self.model.status == 'active').all()
+
+    def get_upcoming_exams(self) -> List[ExamModel]:
+        """获取即将开始的考试"""
+        now = datetime.now()
+        return self.db.query(self.model)\
+            .filter(self.model.start_time > now)\
+            .order_by(self.model.start_time.asc())\
+            .all()
+
+    def get_exams_with_questions(self) -> List[ExamModel]:
+        """获取包含题目信息的考试列表"""
+        return self.db.query(self.model).options(
+            joinedload(self.model.questions)).all()

SERVER/app/stores/knowledge_point_store.py

@@ -0,0 +1,51 @@
+from typing import Optional, List
+from sqlalchemy.orm import Session
+from ..models.knowledge_point_model import KnowledgePointModel
+from .base_store import BaseStore
+
+
+class KnowledgePointStore(BaseStore[KnowledgePointModel]):
+    """
+    知识点存储类，继承自BaseStore
+    提供知识点相关的特定操作
+    """
+
+    def __init__(self, db: Session):
+        super().__init__(db, KnowledgePointModel)
+
+    def get_by_subject(self, subject: str) -> List[KnowledgePointModel]:
+        """根据学科获取知识点列表"""
+        return self.db.query(
+            self.model).filter(self.model.subject == subject).all()
+
+    def get_by_level(self, level: int) -> List[KnowledgePointModel]:
+        """根据难度等级获取知识点列表"""
+        return self.db.query(
+            self.model).filter(self.model.level == level).all()
+
+    def get_root_knowledge_points(self) -> List[KnowledgePointModel]:
+        """获取所有根知识点"""
+        return self.db.query(
+            self.model).filter(self.model.parent_id == None).all()
+
+    def get_child_knowledge_points(
+            self, parent_id: int) -> List[KnowledgePointModel]:
+        """根据父知识点ID获取子知识点列表"""
+        return self.db.query(
+            self.model).filter(self.model.parent_id == parent_id).all()
+
+    def get_knowledge_point_tree(self) -> List[KnowledgePointModel]:
+        """获取完整的知识点树结构"""
+        # 先获取所有知识点
+        all_points = self.db.query(self.model).all()
+        # 构建树结构
+        point_dict = {point.id: point for point in all_points}
+        for point in all_points:
+            if point.parent_id:
+                parent = point_dict.get(point.parent_id)
+                if parent:
+                    if not hasattr(parent, 'children'):
+                        parent.children = []
+                    parent.children.append(point)
+        # 返回根节点
+        return [point for point in all_points if not point.parent_id]

SERVER/app/stores/permission_store.py

@@ -0,0 +1,34 @@
+from typing import Optional, List
+from sqlalchemy.orm import Session, joinedload
+from ..models.permission_model import PermissionModel
+from .base_store import BaseStore
+
+
+class PermissionStore(BaseStore[PermissionModel]):
+    """
+    权限存储类，继承自BaseStore
+    提供权限相关的特定操作
+    """
+
+    def __init__(self, db: Session):
+        super().__init__(db, PermissionModel)
+
+    def get_by_name(self, name: str) -> Optional[PermissionModel]:
+        """根据权限名称获取权限"""
+        return self.db.query(
+            self.model).filter(self.model.name == name).first()
+
+    def get_by_code(self, code: str) -> Optional[PermissionModel]:
+        """根据权限代码获取权限"""
+        return self.db.query(
+            self.model).filter(self.model.code == code).first()
+
+    def get_active_permissions(self) -> List[PermissionModel]:
+        """获取所有激活的权限"""
+        return self.db.query(
+            self.model).filter(self.model.status == True).all()
+
+    def get_permissions_with_roles(self) -> List[PermissionModel]:
+        """获取包含角色信息的权限列表"""
+        return self.db.query(self.model).options(joinedload(
+            self.model.roles)).all()

SERVER/app/stores/question_store.py

@@ -0,0 +1,38 @@
+from typing import Optional, List
+from sqlalchemy.orm import Session, joinedload
+from ..models.question_model import QuestionModel
+from .base_store import BaseStore
+
+
+class QuestionStore(BaseStore[QuestionModel]):
+    """
+    题目存储类，继承自BaseStore
+    提供题目相关的特定操作
+    """
+
+    def __init__(self, db: Session):
+        super().__init__(db, QuestionModel)
+
+    def get_by_subject(self, subject: str) -> List[QuestionModel]:
+        """根据学科获取题目列表"""
+        return self.db.query(
+            self.model).filter(self.model.subject == subject).all()
+
+    def get_by_type(self, type: str) -> List[QuestionModel]:
+        """根据题型获取题目列表"""
+        return self.db.query(self.model).filter(self.model.type == type).all()
+
+    def get_by_difficulty(self, difficulty: int) -> List[QuestionModel]:
+        """根据难度获取题目列表"""
+        return self.db.query(
+            self.model).filter(self.model.difficulty == difficulty).all()
+
+    def get_questions_with_options(self) -> List[QuestionModel]:
+        """获取包含选项信息的题目列表"""
+        return self.db.query(self.model).options(joinedload(
+            self.model.options)).all()
+
+    def get_questions_with_knowledge_points(self) -> List[QuestionModel]:
+        """获取包含知识点信息的题目列表"""
+        return self.db.query(self.model).options(
+            joinedload(self.model.knowledge_points)).all()

SERVER/app/stores/role_store.py

@@ -0,0 +1,39 @@
+from typing import Optional, List
+from sqlalchemy.orm import Session, joinedload
+from ..models.role_model import RoleModel
+from .base_store import BaseStore
+
+
+class RoleStore(BaseStore[RoleModel]):
+    """
+    角色存储类，继承自BaseStore
+    提供角色相关的特定操作
+    """
+
+    def __init__(self, db: Session):
+        super().__init__(db, RoleModel)
+
+    def get_by_name(self, name: str) -> Optional[RoleModel]:
+        """根据角色名称获取角色"""
+        return self.db.query(
+            self.model).filter(self.model.name == name).first()
+
+    def get_by_code(self, code: str) -> Optional[RoleModel]:
+        """根据角色代码获取角色"""
+        return self.db.query(
+            self.model).filter(self.model.code == code).first()
+
+    def get_active_roles(self) -> List[RoleModel]:
+        """获取所有激活的角色"""
+        return self.db.query(
+            self.model).filter(self.model.status == True).all()
+
+    def get_roles_with_users(self) -> List[RoleModel]:
+        """获取包含用户信息的角色列表"""
+        return self.db.query(self.model).options(joinedload(
+            self.model.users)).all()
+
+    def get_roles_with_permissions(self) -> List[RoleModel]:
+        """获取包含权限信息的角色列表"""
+        return self.db.query(self.model).options(
+            joinedload(self.model.permissions)).all()

SERVER/app/stores/status_store.py

@@ -0,0 +1,28 @@
+from typing import Optional
+from sqlalchemy.orm import Session
+from ..models.status_model import StatusModel
+from .base_store import BaseStore
+
+
+class StatusStore(BaseStore[StatusModel]):
+    """
+    状态存储类，继承自BaseStore
+    提供状态相关的特定操作
+    """
+
+    def __init__(self, db: Session):
+        super().__init__(db, StatusModel)
+
+    def get_by_type(self, type: str) -> Optional[StatusModel]:
+        """根据类型获取状态"""
+        return self.db.query(
+            self.model).filter(self.model.type == type).first()
+
+    def get_all_by_status(self, status: int) -> list[StatusModel]:
+        """根据状态值获取所有状态"""
+        return self.db.query(
+            self.model).filter(self.model.status == status).all()
+
+    def get_all_by_sort(self) -> list[StatusModel]:
+        """根据排序获取所有状态"""
+        return self.db.query(self.model).order_by(self.model.sort).all()

SERVER/app/stores/user_store.py

@@ -0,0 +1,36 @@
+from typing import Optional, List
+from sqlalchemy.orm import Session
+from ..models.user_model import UserModel
+from .base_store import BaseStore
+
+
+class UserStore(BaseStore[UserModel]):
+    """
+    用户存储类，继承自BaseStore
+    提供用户相关的特定操作
+    """
+
+    def __init__(self, db: Session):
+        super().__init__(db, UserModel)
+
+    def get_by_username(self, username: str) -> Optional[UserModel]:
+        """根据用户名获取用户"""
+        return self.db.query(
+            self.model).filter(self.model.username == username).first()
+
+    def get_by_grade(self, grade: str) -> List[UserModel]:
+        """根据年级获取用户"""
+        return self.db.query(
+            self.model).filter(self.model.grade == grade).all()
+
+    def get_by_target_score(self, min_score: int,
+                            max_score: int) -> List[UserModel]:
+        """根据目标分数范围获取用户"""
+        return self.db.query(self.model).filter(
+            self.model.target_score >= min_score, self.model.target_score
+            <= max_score).all()
+
+    def get_by_learning_style(self, learning_style: str) -> List[UserModel]:
+        """根据学习风格获取用户"""
+        return self.db.query(self.model).filter(
+            self.model.learning_style == learning_style).all()

SERVER/app/utils/__init__.py

@@ -0,0 +1,43 @@
+"""
+工具模块集合
+包含数据处理、验证等常用工具类
+
+版本: 1.0.0
+
+使用示例:
+    from app.utils import DataUtil, ValidationUtil
+    
+    # 数据处理示例
+    data = {'key': 'value'}
+    json_str = DataUtil.to_json(data)
+    
+    # 验证示例  
+    email = 'test@example.com'
+    is_valid = ValidationUtil.is_email(email)
+"""
+
+__version__ = '1.0.0'
+
+from .date_util import DateUtil
+from .data_util import DataUtil
+from .file_util import FileUtil
+from .network_util import NetworkUtil
+from .string_util import StringUtil
+from .validation_util import ValidationUtil
+
+# 常用方法快捷调用
+to_json = DataUtil.to_json
+is_email = ValidationUtil.is_email
+is_valid_ip = NetworkUtil.is_valid_ip
+get_file_size = FileUtil.get_file_size
+get_current_time = DateUtil.get_current_time
+format_date = DateUtil.format_date
+parse_date = DateUtil.parse_date
+get_timestamp = DateUtil.get_timestamp
+
+__all__ = [
+    'DataUtil', 'DateUtil', 'FileUtil', 'NetworkUtil', 'StringUtil',
+    'ValidationUtil', '__version__', 'to_json', 'is_email', 'is_valid_ip',
+    'get_file_size', 'get_current_time', 'format_date', 'parse_date',
+    'get_timestamp'
+]

SERVER/app/utils/data_util.py

@@ -0,0 +1,113 @@
+import json
+from typing import Optional, Union, Dict, List
+from decimal import Decimal
+
+
+class DataUtil:
+    """数据处理工具类"""
+
+    @staticmethod
+    def to_json(data: Union[Dict, List]) -> Optional[str]:
+        """将数据转换为JSON字符串
+        
+        Args:
+            data: 要转换的数据
+            
+        Returns:
+            JSON字符串，如果转换失败返回None
+        """
+        try:
+            return json.dumps(data, ensure_ascii=False)
+        except Exception:
+            return None
+
+    @staticmethod
+    def from_json(json_str: str) -> Optional[Union[Dict, List]]:
+        """将JSON字符串转换为Python对象
+        
+        Args:
+            json_str: JSON字符串
+            
+        Returns:
+            Python对象，如果转换失败返回None
+        """
+        try:
+            return json.loads(json_str)
+        except Exception:
+            return None
+
+    @staticmethod
+    def to_decimal(value: Union[str, int, float]) -> Optional[Decimal]:
+        """将值转换为Decimal类型
+        
+        Args:
+            value: 要转换的值
+            
+        Returns:
+            Decimal对象，如果转换失败返回None
+        """
+        try:
+            return Decimal(str(value))
+        except Exception:
+            return None
+
+    @staticmethod
+    def deep_merge(dict1: Dict, dict2: Dict) -> Dict:
+        """深度合并两个字典
+        
+        Args:
+            dict1: 第一个字典
+            dict2: 第二个字典
+            
+        Returns:
+            合并后的字典
+        """
+        result = dict1.copy()
+        for key, value in dict2.items():
+            if key in result and isinstance(result[key], dict) and isinstance(
+                    value, dict):
+                result[key] = DataUtil.deep_merge(result[key], value)
+            else:
+                result[key] = value
+        return result
+
+    @staticmethod
+    def flatten_dict(d: Dict, parent_key: str = '', sep: str = '.') -> Dict:
+        """将嵌套字典展平
+        
+        Args:
+            d: 要展平的字典
+            parent_key: 父级键名
+            sep: 分隔符
+            
+        Returns:
+            展平后的字典
+        """
+        items = []
+        for k, v in d.items():
+            new_key = f"{parent_key}{sep}{k}" if parent_key else k
+            if isinstance(v, dict):
+                items.extend(
+                    DataUtil.flatten_dict(v, new_key, sep=sep).items())
+            else:
+                items.append((new_key, v))
+        return dict(items)
+
+    @staticmethod
+    def get_nested_value(data: Dict, keys: List[str], default=None):
+        """获取嵌套字典中的值
+        
+        Args:
+            data: 要查找的字典
+            keys: 键名列表
+            default: 默认值
+            
+        Returns:
+            查找到的值，如果不存在返回默认值
+        """
+        for key in keys:
+            if isinstance(data, dict) and key in data:
+                data = data[key]
+            else:
+                return default
+        return data

SERVER/app/utils/date_util.py

@@ -0,0 +1,116 @@
+from datetime import datetime, timedelta
+from typing import Optional, Tuple
+
+
+class DateUtil:
+    """日期时间工具类"""
+
+    @staticmethod
+    def get_current_time() -> datetime:
+        """获取当前时间
+        
+        Returns:
+            当前时间
+        """
+        return datetime.now()
+
+    @staticmethod
+    def format_date(date: datetime, fmt: str = '%Y-%m-%d %H:%M:%S') -> str:
+        """格式化日期时间
+        
+        Args:
+            date: 日期时间对象
+            fmt: 格式化字符串
+            
+        Returns:
+            格式化后的日期时间字符串
+        """
+        return date.strftime(fmt)
+
+    @staticmethod
+    def parse_date(date_str: str,
+                   fmt: str = '%Y-%m-%d %H:%M:%S') -> Optional[datetime]:
+        """解析日期时间字符串
+        
+        Args:
+            date_str: 日期时间字符串
+            fmt: 格式化字符串
+            
+        Returns:
+            解析后的日期时间对象，如果解析失败返回None
+        """
+        try:
+            return datetime.strptime(date_str, fmt)
+        except ValueError:
+            return None
+
+    @staticmethod
+    def get_date_range(start_date: datetime,
+                       end_date: datetime) -> Tuple[datetime, datetime]:
+        """获取日期范围
+        
+        Args:
+            start_date: 开始日期
+            end_date: 结束日期
+            
+        Returns:
+            包含开始日期和结束日期的元组
+        """
+        return (start_date, end_date)
+
+    @staticmethod
+    def add_days(date: datetime, days: int) -> datetime:
+        """增加天数
+        
+        Args:
+            date: 日期时间对象
+            days: 要增加的天数
+            
+        Returns:
+            增加天数后的日期时间对象
+        """
+        return date + timedelta(days=days)
+
+    @staticmethod
+    def get_week_start_end(date: datetime) -> Tuple[datetime, datetime]:
+        """获取指定日期所在周的起始和结束日期
+        
+        Args:
+            date: 日期时间对象
+            
+        Returns:
+            包含周起始日期和结束日期的元组
+        """
+        start = date - timedelta(days=date.weekday())
+        end = start + timedelta(days=6)
+        return (start, end)
+
+    @staticmethod
+    def is_leap_year(year: int) -> bool:
+        """判断是否为闰年
+        
+        Args:
+            year: 年份
+            
+        Returns:
+            bool: 是否为闰年
+        """
+        return year % 4 == 0 and (year % 100 != 0 or year % 400 == 0)
+
+    @staticmethod
+    def get_days_in_month(year: int, month: int) -> int:
+        """获取指定月份的天数
+        
+        Args:
+            year: 年份
+            month: 月份
+            
+        Returns:
+            指定月份的天数
+        """
+        if month == 2:
+            return 29 if DateUtil.is_leap_year(year) else 28
+        elif month in [4, 6, 9, 11]:
+            return 30
+        else:
+            return 31

SERVER/app/utils/file_util.py

@@ -0,0 +1,123 @@
+import os
+import shutil
+from typing import Optional, List
+
+
+class FileUtil:
+    """文件操作工具类"""
+
+    @staticmethod
+    def get_file_extension(filename: str) -> Optional[str]:
+        """获取文件扩展名
+        
+        Args:
+            filename: 文件名
+            
+        Returns:
+            文件扩展名，如果文件没有扩展名返回None
+        """
+        _, ext = os.path.splitext(filename)
+        return ext[1:] if ext else None
+
+    @staticmethod
+    def get_file_size(filepath: str) -> Optional[int]:
+        """获取文件大小
+        
+        Args:
+            filepath: 文件路径
+            
+        Returns:
+            文件大小（字节），如果文件不存在返回None
+        """
+        try:
+            return os.path.getsize(filepath)
+        except OSError:
+            return None
+
+    @staticmethod
+    def is_file_exists(filepath: str) -> bool:
+        """检查文件是否存在
+        
+        Args:
+            filepath: 文件路径
+            
+        Returns:
+            bool: 文件是否存在
+        """
+        return os.path.isfile(filepath)
+
+    @staticmethod
+    def create_directory(dirpath: str) -> bool:
+        """创建目录
+        
+        Args:
+            dirpath: 目录路径
+            
+        Returns:
+            bool: 是否创建成功
+        """
+        try:
+            os.makedirs(dirpath, exist_ok=True)
+            return True
+        except OSError:
+            return False
+
+    @staticmethod
+    def list_files(dirpath: str,
+                   recursive: bool = False) -> Optional[List[str]]:
+        """列出目录下的文件
+        
+        Args:
+            dirpath: 目录路径
+            recursive: 是否递归列出
+            
+        Returns:
+            文件路径列表，如果目录不存在返回None
+        """
+        if not os.path.isdir(dirpath):
+            return None
+
+        if recursive:
+            file_list = []
+            for root, _, files in os.walk(dirpath):
+                for file in files:
+                    file_list.append(os.path.join(root, file))
+            return file_list
+        else:
+            return [
+                f for f in os.listdir(dirpath)
+                if os.path.isfile(os.path.join(dirpath, f))
+            ]
+
+    @staticmethod
+    def delete_file(filepath: str) -> bool:
+        """删除文件
+        
+        Args:
+            filepath: 文件路径
+            
+        Returns:
+            bool: 是否删除成功
+        """
+        try:
+            os.remove(filepath)
+            return True
+        except OSError:
+            return False
+
+    @staticmethod
+    def copy_file(src: str, dst: str) -> bool:
+        """复制文件
+        
+        Args:
+            src: 源文件路径
+            dst: 目标文件路径
+            
+        Returns:
+            bool: 是否复制成功
+        """
+        try:
+            shutil.copy2(src, dst)
+            return True
+        except OSError:
+            return False

SERVER/app/utils/network_util.py

@@ -0,0 +1,79 @@
+import re
+import socket
+import urllib.parse
+from typing import Optional, Tuple
+
+
+class NetworkUtil:
+    """网络工具类"""
+
+    @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 extract_domain(url: str) -> Optional[str]:
+        """从URL中提取域名
+        
+        Args:
+            url: 完整URL
+            
+        Returns:
+            提取的域名，如果解析失败返回None
+        """
+        try:
+            parsed = urllib.parse.urlparse(url)
+            if parsed.netloc:
+                return parsed.netloc
+            return None
+        except Exception:
+            return None
+
+    @staticmethod
+    def is_valid_url(url: str) -> bool:
+        """验证URL格式
+        
+        Args:
+            url: 要验证的URL
+            
+        Returns:
+            bool: 是否为有效URL
+        """
+        pattern = re.compile(
+            r'^(?:http|ftp)s?://'  # http:// or https://
+            r'(?:(?:[A-Z0-9](?:[A-Z0-9-]{0,61}[A-Z0-9])?\.)+(?:[A-Z]{2,6}\.?|[A-Z0-9-]{2,}\.?)|'  # domain...
+            r'localhost|'  # localhost...
+            r'\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3})'  # ...or ip
+            r'(?::\d+)?'  # optional port
+            r'(?:/?|[/?]\S+)$',
+            re.IGNORECASE)
+        return bool(re.match(pattern, url))
+
+    @staticmethod
+    def get_ip_info(ip: str) -> Optional[Tuple[str, str]]:
+        """获取IP地址的地理位置信息
+        
+        Args:
+            ip: IP地址
+            
+        Returns:
+            包含国家和城市的元组，如果获取失败返回None
+        """
+        try:
+            # 这里可以集成第三方IP查询服务
+            # 示例返回
+            return ('China', 'Beijing')
+        except Exception:
+            return None

SERVER/app/utils/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)

SERVER/app/utils/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

application-dev.yml

@@ -1,7 +0,0 @@
-debug: true
-database:
-  uri: mysql://localhost:3306/dev_db
-  pool_size: 10
-api:
-  url: http://localhost:8000
-  timeout: 60

requirements.txt

@@ -3,4 +3,7 @@ pydantic>=2.0.0
 pydantic-settings>=2.0.0
 watchdog>=3.0.0
 loguru>=0.7.0
-
+python-dotenv>=1.0.0
+SQLAlchemy>=2.0.0
+PyMySQL>=1.1.0
+mysqlclient>=2.2.0