Coverage for app \ services \ quiz_scoring_service.py: 78%

1# Quiz Scoring Service-saves quiz results to JSON files 

2 

3import json 

4import re 

5from datetime import datetime 

6from pathlib import Path 

7 

8 

9def get_downloads_dir(): 

10 """Import DOWNLOADS_DIR from settings""" 

11 from app.settings import DOWNLOADS_DIR 

12 return DOWNLOADS_DIR 

13 

14 

15def save_quiz_result(child_id: str, video_id: str, score_data: dict, session_id: str = None) -> dict: 

16 DOWNLOADS_DIR = get_downloads_dir() 

17 child_id = re.sub(r'[^a-zA-Z0-9_-]', '', child_id) 

18 

19 results_dir = DOWNLOADS_DIR / "quiz_results" 

20 results_dir.mkdir(exist_ok=True) 

21 results_file = results_dir / f"{child_id}_results.json" 

22 

23 if results_file.exists(): 

24 try: 

25 data = json.loads(results_file.read_text(encoding="utf-8")) 

26 except Exception: 

27 data = {"child_id": child_id, "attempts": []} 

28 else: 

29 data = {"child_id": child_id, "attempts": []} 

30 

31 attempt = { 

32 "session_id": session_id, 

33 "video_id": video_id, 

34 "timestamp": datetime.now().isoformat(), 

35 "total": score_data.get("total", 0), 

36 "questions_total": score_data.get("total", 0), 

37 "questions_correct": score_data.get("correct", 0), 

38 "questions_wrong": score_data.get("wrong", 0), 

39 "percentage": score_data.get("percentage", 0), 

40 "total_retries": score_data.get("total_retries", 0), 

41 "avg_retries_per_question": score_data.get("avg_retries_per_question", 0.0), 

42 "watch_minutes": score_data.get("watch_minutes", 0), 

43 "manual_pauses": score_data.get("manual_pauses", 0), 

44 "details": score_data.get("details", []) 

45 } 

46 

47 # If session_id provided, update existing attempt instead of appending 

48 if session_id: 

49 existing_index = next( 

50 (i for i, a in enumerate(data["attempts"]) if a.get("session_id") == session_id), 

51 None 

52 ) 

53 if existing_index is not None: 

54 existing = data["attempts"][existing_index] 

55 if score_data.get("checkpoint_only"): 

56 # Add watch time and update pause count 

57 existing["watch_minutes"] = round( 

58 existing.get("watch_minutes", 0) + score_data.get("watch_minutes", 0), 2 

59 ) 

60 existing["manual_pauses"] = score_data.get("manual_pauses", existing.get("manual_pauses", 0)) 

61 data["attempts"][existing_index] = existing 

62 else: 

63 # Full save: accumulate watch_minutes, replace other fields 

64 attempt["watch_minutes"] = round( 

65 existing.get("watch_minutes", 0) + score_data.get("watch_minutes", 0), 2 

66 ) 

67 attempt["timestamp"] = existing.get("timestamp", attempt["timestamp"]) 

68 data["attempts"][existing_index] = attempt 

69 else: 

70 data["attempts"].append(attempt) 

71 else: 

72 data["attempts"].append(attempt) 

73 

74 try: 

75 results_file.write_text( 

76 json.dumps(data, indent=2, ensure_ascii=False), 

77 encoding="utf-8" 

78 ) 

79 return {"success": True, "message": "Score saved!", "file": str(results_file)} 

80 except Exception as e: 

81 return {"success": False, "message": f"Error saving: {str(e)}"} 

82 

83 

84def get_child_scores(child_id: str) -> dict: 

85 """ 

86 Get all quiz attempts for a child. 

87  

88 Returns their full history and summary stats. 

89 """ 

90 DOWNLOADS_DIR = get_downloads_dir() 

91 results_file = DOWNLOADS_DIR / "quiz_results" / f"{child_id}_results.json" 

92 

93 if not results_file.exists(): 

94 return { 

95 "success": False, 

96 "message": "No scores found for this child" 

97 } 

98 

99 try: 

100 data = json.loads(results_file.read_text(encoding="utf-8")) 

101 

102 # Calculate summary 

103 attempts = data.get("attempts", []) 

104 total_correct = sum(a.get("questions_correct", 0) for a in attempts) 

105 total_questions = sum(a.get("questions_total", 0) for a in attempts) 

106 

107 return { 

108 "success": True, 

109 "child_id": child_id, 

110 "total_attempts": len(attempts), 

111 "total_correct": total_correct, 

112 "total_questions": total_questions, 

113 "overall_percentage": round(total_correct / total_questions * 100, 1) if total_questions > 0 else 0, 

114 "attempts": attempts 

115 } 

116 except Exception as e: 

117 return { 

118 "success": False, 

119 "message": f"Error reading scores: {str(e)}" 

120 }