File size: 2,738 Bytes
f7cf4db | 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 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 | import datetime
import logging
from firebase_admin import firestore
from firebase_manager import firebase_manager
logger = logging.getLogger("HamoraServer")
class AnalyticsManager:
"""
Manages AI Assessment Telemetry and Weekly Reports.
"""
def _get_db(self):
return firebase_manager.get_db()
def update_weekly_analytics(self, uid: str, assessment: dict):
"""
Updates the weekly analytics document for a user using atomic operations.
Runs silently.
"""
if not uid or not assessment:
return
db = self._get_db()
if not db:
logger.error("❌ [ANALYTICS] Firestore DB not available.")
return
try:
# 1. Parse assessment data
primary_skill = assessment.get("primary_skill", "Unknown")
sub_skill = assessment.get("sub_skill", "")
mastery_score = assessment.get("mastery_score", 0)
parent_note = assessment.get("parent_note", "")
# Ensure score is an integer
try:
mastery_score = int(mastery_score)
except (ValueError, TypeError):
mastery_score = 0
# 2. Compute week_id (e.g., week_11_2026)
today = datetime.date.today()
year, week_num, _ = today.isocalendar()
week_id = f"week_{week_num:02d}_{year}"
# 3. Construct the document reference
# Path: users/{uid}/analytics/{week_id}
doc_ref = db.collection("users").document(uid).collection("analytics").document(week_id)
# 4. Construct payload with FieldValue operations for atomic updates
# Create the nested skill payload
skill_payload = {
"sum_scores": firestore.Increment(mastery_score),
"count": firestore.Increment(1)
}
if sub_skill:
skill_payload["sub_skills"] = firestore.ArrayUnion([sub_skill])
# Combine into final payload
payload = {
"total_exercises": firestore.Increment(1),
f"skills_data.{primary_skill}": skill_payload,
"last_updated": firestore.SERVER_TIMESTAMP
}
if parent_note:
payload["parent_notes"] = firestore.ArrayUnion([parent_note])
# 5. Set with merge=True
doc_ref.set(payload, merge=True)
logger.info(f"📊 [ANALYTICS] Successfully updated telemetry for {uid} in {week_id}.")
except Exception as e:
logger.error(f"❌ [ANALYTICS] Error updating analytics for {uid}: {e}")
# Global instance
analytics_manager = AnalyticsManager()
|