history_manager.py 3.6 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110
  1. from typing import Dict, Any, Optional
  2. from core.db import db
  3. from models.history import History, GameTypeEnum
  4. class HistoryManager:
  5. """历史记录管理器,负责记录和查询历史记录"""
  6. def __init__(self):
  7. """初始化历史记录管理器"""
  8. pass
  9. @staticmethod
  10. def record_question(result: Dict[str, Any]) -> int:
  11. """记录问题生成历史
  12. Args:
  13. result: 生成的问题结果
  14. Returns:
  15. 历史记录ID
  16. """
  17. # 只有生成题目时记录日志
  18. try:
  19. # 确保question_type是有效的题目类型(A-E)
  20. game_type = GameTypeEnum(result.get('type'))
  21. # 创建HistoryDTO对象
  22. history_dto = History(
  23. game_type=game_type,
  24. question=result.get('question', ''),
  25. numbers=','.join(map(str, result.get('numbers', []))),
  26. answers= str(result.get('answers', ''))
  27. )
  28. # 添加历史记录
  29. return db.add_history(history_dto)
  30. except (ValueError, KeyError) as e:
  31. # 记录错误但不中断流程
  32. print(f"记录历史失败: {e}")
  33. return -1
  34. @staticmethod
  35. def get_history(game_type: Optional[str] = None, start_date: Optional[str] = None, end_date: Optional[str] = None, page: int = 1, page_size: int = 10, only_favorite: bool = False, include_deleted: bool = False) :
  36. """获取历史记录,支持分页查询
  37. Args:
  38. game_type: 类型
  39. start_date: 开始日期,格式为 YYYY-MM-DD
  40. end_date: 结束日期,格式为 YYYY-MM-DD
  41. page: 页码,从1开始
  42. page_size: 每页记录数
  43. only_favorite: 是否只查询收藏的记录
  44. include_deleted: 是否包含已删除的记录
  45. Returns:
  46. Tuple[List[Dict[str, Any]], int]: 历史记录DTO列表和总记录数
  47. """
  48. # 从数据库获取历史记录,使用分页
  49. db_records, total_count = db.get_history(game_type, start_date, end_date, page, page_size, only_favorite, include_deleted)
  50. # 转换为DTO对象
  51. history_records = []
  52. for record in db_records:
  53. try:
  54. # 尝试将记录转换为DTO
  55. history_dto = History.from_db_record(record)
  56. history_records.append(history_dto.dict())
  57. except Exception as e:
  58. print(f"转换历史记录失败: {e}")
  59. # 添加原始记录,确保不丢失数据
  60. history_records.append(record)
  61. return history_records, total_count
  62. @staticmethod
  63. def toggle_favorite(history_id: int) -> bool:
  64. """切换历史记录的收藏状态
  65. Args:
  66. history_id: 历史记录ID
  67. Returns:
  68. bool: 操作是否成功
  69. """
  70. return db.toggle_favorite(history_id)
  71. @staticmethod
  72. def toggle_delete(history_id: int) -> bool:
  73. """切换历史记录的删除状态
  74. Args:
  75. history_id: 历史记录ID
  76. Returns:
  77. bool: 操作是否成功
  78. """
  79. return db.toggle_delete(history_id)
  80. @staticmethod
  81. def soft_delete(history_id: int) -> bool:
  82. """软删除历史记录
  83. Args:
  84. history_id: 历史记录ID
  85. Returns:
  86. bool: 操作是否成功
  87. """
  88. return db.soft_delete(history_id)