Coverage for app \ services \ report_service.py: 62%
104 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-04-28 20:58 -0400
1"""
2Report service for expert-scoped parental reports.
3Reads quiz result JSON files and computes summary stats per child.
4"""
6import json
7from collections import defaultdict
8from pathlib import Path
9from typing import Any, Dict, List, Optional
12def _get_downloads_dir() -> Path:
13 from app.settings import DOWNLOADS_DIR
14 return DOWNLOADS_DIR
17def _get_video_meta(video_id: str, downloads_dir: Path) -> Dict[str, Any]:
18 meta_path = downloads_dir / video_id / "meta.json"
19 if meta_path.exists():
20 try:
21 meta = json.loads(meta_path.read_text(encoding="utf-8"))
22 duration_seconds = meta.get("duration", 0) or 0
23 return {
24 "title": meta.get("title") or video_id,
25 "duration_minutes": round(duration_seconds / 60, 1),
26 }
27 except Exception:
28 pass
29 return {"title": video_id, "duration_minutes": 0}
32def _get_video_title(video_id: str, downloads_dir: Path) -> str:
33 return _get_video_meta(video_id, downloads_dir)["title"]
36def _load_attempts(child_id: str, downloads_dir: Path) -> List[Dict[str, Any]]:
37 results_file = downloads_dir / "quiz_results" / f"{child_id}_results.json"
38 if not results_file.exists():
39 return []
40 try:
41 data = json.loads(results_file.read_text(encoding="utf-8"))
42 return data.get("attempts", [])
43 except Exception:
44 return []
47def _compute_top_categories(attempts: List[Dict[str, Any]], window: int = 10) -> List[Dict[str, Any]]:
48 """
49 Compute top 3 question-type categories from the last `window` attempts.
50 Per-answer points: correct=1, almost=0.5, wrong=0.
51 Category score: round((points_sum / answer_count) * 100).
52 Rank by score desc, then answer_count desc.
53 """
54 recent = attempts[-window:]
55 category_points: Dict[str, float] = defaultdict(float)
56 category_counts: Dict[str, int] = defaultdict(int)
58 for attempt in recent:
59 for detail in attempt.get("details", []):
60 q_type = detail.get("question_type")
61 if not q_type:
62 continue
63 status = detail.get("status", "wrong")
64 if status == "correct":
65 points = 1.0
66 elif status == "almost":
67 points = 0.5
68 else:
69 points = 0.0
70 category_points[q_type] += points
71 category_counts[q_type] += 1
73 if not category_counts:
74 return []
76 categories = []
77 for q_type, count in category_counts.items():
78 score = round((category_points[q_type] / count) * 100)
79 categories.append({
80 "type": q_type,
81 "score": score,
82 "answer_count": count,
83 })
85 categories.sort(key=lambda c: (-c["score"], -c["answer_count"]))
86 return categories[:3]
89def get_child_report_scoped(
90 child_id: str,
91 video_id: Optional[str] = None,
92 mode: Optional[str] = None,
93) -> Dict[str, Any]:
94 """
95 Return a report filtered by optional video_id and/or interaction_mode.
96 mode='all' (or None) means no mode filter.
97 """
98 downloads_dir = _get_downloads_dir()
99 all_attempts = _load_attempts(child_id, downloads_dir)
101 # Filter by video
102 if video_id:
103 all_attempts = [a for a in all_attempts if a.get("video_id") == video_id]
105 # Filter by mode (skip filter when mode is None or 'all')
106 if mode and mode != "all":
107 all_attempts = [a for a in all_attempts if a.get("interaction_mode") == mode]
109 if not all_attempts:
110 return {
111 "success": True,
112 "child_id": child_id,
113 "overall_score": 0,
114 "total_attempts": 0,
115 "total_retries": 0,
116 "avg_retries_per_question": 0.0,
117 "top_categories": [],
118 "recent_videos": [],
119 "videos_watched": 0,
120 "total_watch_minutes": 0,
121 }
123 enriched = [
124 {**a, "video_title": _get_video_title(a.get("video_id", ""), downloads_dir)}
125 for a in all_attempts
126 ]
128 percentages = [a.get("percentage", 0) for a in enriched]
129 overall_score = round(sum(percentages) / len(percentages)) if percentages else 0
130 total_retries = sum(a.get("total_retries", 0) for a in enriched)
131 total_questions_answered = sum(a.get("total", 0) for a in enriched)
132 avg_retries_per_question = round(
133 total_retries / total_questions_answered, 2
134 ) if total_questions_answered > 0 else 0.0
136 recent_videos = []
137 for a in reversed(enriched[-6:]):
138 vid_id = a.get("video_id", "")
139 meta = _get_video_meta(vid_id, downloads_dir)
140 watch_min = round(a.get("watch_minutes", 0), 1)
141 dur_min = meta["duration_minutes"]
142 finished = dur_min > 0 and watch_min >= dur_min * 0.9
143 recent_videos.append({
144 "video_id": vid_id,
145 "video_title": a.get("video_title"),
146 "percentage": a.get("percentage", 0),
147 "timestamp": a.get("timestamp"),
148 "interaction_mode": a.get("interaction_mode"),
149 "watch_minutes": watch_min,
150 "duration_minutes": dur_min,
151 "finished": finished,
152 "manual_pauses": a.get("manual_pauses", 0),
153 })
155 return {
156 "success": True,
157 "child_id": child_id,
158 "overall_score": overall_score,
159 "total_attempts": len(enriched),
160 "total_retries": total_retries,
161 "avg_retries_per_question": avg_retries_per_question,
162 "top_categories": _compute_top_categories(all_attempts),
163 "recent_videos": recent_videos,
164 "videos_watched": len(enriched),
165 "total_watch_minutes": sum(round(a.get("watch_minutes", 0)) for a in enriched),
166 }
169def get_child_report(child_id: str, limit: int = 10) -> Dict[str, Any]:
170 """
171 Return a full report payload for one child.
172 """
173 downloads_dir = _get_downloads_dir()
174 attempts = _load_attempts(child_id, downloads_dir)
176 if not attempts:
177 return {
178 "success": True,
179 "child_id": child_id,
180 "overall_score": 0,
181 "total_attempts": 0,
182 "total_retries": 0,
183 "avg_retries_per_question": 0.0,
184 "top_categories": [],
185 "recent_videos": [],
186 }
188 # Enrich each attempt with video title
189 enriched = []
190 for attempt in attempts:
191 video_id = attempt.get("video_id", "")
192 enriched.append({
193 **attempt,
194 "video_title": _get_video_title(video_id, downloads_dir),
195 })
197 # Overall score: average percentage across all attempts
198 percentages = [a.get("percentage", 0) for a in enriched]
199 overall_score = round(sum(percentages) / len(percentages)) if percentages else 0
201 # Total attempts
202 total_attempts = len(enriched)
203 # Passive metrics
204 videos_watched = len(enriched)
205 total_watch_minutes = sum(
206 round(a.get("watch_minutes", 0)) for a in enriched
207 )
209 # Aggregate retry metrics across all attempts
210 total_retries = sum(a.get("total_retries", 0) for a in enriched)
211 total_questions_answered = sum(a.get("total", 0) for a in enriched)
212 avg_retries_per_question = round(
213 total_retries / total_questions_answered, 2
214 ) if total_questions_answered > 0 else 0.0
216 # Top categories
217 top_categories = _compute_top_categories(attempts)
219 # Recent videos: latest 4, newest first
220 recent_videos = []
221 for a in reversed(enriched[-4:]):
222 vid_id = a.get("video_id", "")
223 meta = _get_video_meta(vid_id, downloads_dir)
224 watch_min = round(a.get("watch_minutes", 0), 1)
225 dur_min = meta["duration_minutes"]
226 finished = dur_min > 0 and watch_min >= dur_min * 0.9
227 recent_videos.append({
228 "video_id": vid_id,
229 "video_title": a.get("video_title"),
230 "percentage": a.get("percentage", 0),
231 "timestamp": a.get("timestamp"),
232 "watch_minutes": watch_min,
233 "duration_minutes": dur_min,
234 "finished": finished,
235 "manual_pauses": a.get("manual_pauses", 0),
236 })
238 return {
239 "success": True,
240 "child_id": child_id,
241 "overall_score": overall_score,
242 "total_attempts": total_attempts,
243 "total_correct": sum(a.get("questions_correct", 0) for a in enriched),
244 "total_wrong": sum(a.get("questions_wrong", 0) for a in enriched),
245 "total_questions_answered": sum(a.get("total", 0) for a in enriched),
246 "total_retries": total_retries,
247 "avg_retries_per_question": avg_retries_per_question,
248 "top_categories": top_categories,
249 "recent_videos": recent_videos,
250 "videos_watched": videos_watched,
251 "total_watch_minutes": total_watch_minutes,
252 }