1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
|
#!/usr/bin/env python3
"""Aggregate course events -> course/insights.json (counts + capped question text, no sids/IPs)."""
import json, os, glob, re
from collections import Counter
from datetime import datetime, timezone
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
DATA = os.path.join(ROOT, 'data', 'course')
OUT = os.path.join(ROOT, 'course', 'insights.json')
def norm_q(q):
q = re.sub(r'https?://\S+', '', q or '')
return re.sub(r'\s+', ' ', q).strip().lower()[:200]
def main():
opens, completes, fails, cmds, questions = Counter(), Counter(), Counter(), Counter(), Counter()
projects = Counter()
files = sorted(glob.glob(os.path.join(DATA, 'events-*.jsonl')))[-2:]
for path in files:
with open(path, encoding='utf-8', errors='ignore') as f:
for line in f:
try: e = json.loads(line)
except Exception: continue
t, d = e.get('type'), e.get('detail') or {}
lesson = (d.get('lesson') or '')[:40]
if t == 'lesson_open' and lesson: opens[lesson] += 1
elif t == 'lesson_complete' and lesson: completes[lesson] += 1
elif t == 'exercise_fail':
if lesson: fails[lesson] += 1
c = (d.get('command') or '').strip().split(' ')[0][:40]
if c: cmds[c] += 1
elif t == 'question':
q = norm_q(d.get('q'))
if len(q) >= 8: questions[q] += 1
elif t in ('project_open', 'project_view'):
p = (d.get('project') or '')[:40]
if p: projects[p] += 1
lessons = {}
for lid in set(opens) | set(completes):
o, c = opens[lid], completes[lid]
lessons[lid] = {'opens': o, 'completes': c, 'failures': fails.get(lid, 0),
'completion_rate': round(c / o, 2) if o else 0.0}
payload = {'generated': datetime.now(timezone.utc).strftime('%Y-%m-%dT%H:%M:%SZ'),
'window_files': [os.path.basename(p) for p in files],
'lessons': dict(sorted(lessons.items())),
'top_failed_commands': [{'cmd': c, 'n': n} for c, n in cmds.most_common(20)],
'top_questions': [{'q': q, 'n': n} for q, n in questions.most_common(20) if n >= 2],
'projects': dict(projects),
'totals': {'opens': sum(opens.values()), 'completes': sum(completes.values()), 'failures': sum(fails.values())}}
os.makedirs(os.path.dirname(OUT), exist_ok=True)
tmp = OUT + '.tmp'
with open(tmp, 'w', encoding='utf-8') as f: json.dump(payload, f, ensure_ascii=False, indent=1)
os.replace(tmp, OUT); os.chmod(OUT, 0o644)
print(json.dumps({'lessons': len(lessons), 'questions': len(payload['top_questions']), 'totals': payload['totals']}))
if __name__ == '__main__':
main()
|
