Spaces:
Paused
Paused
| from __future__ import annotations | |
| import json | |
| import os | |
| import re | |
| from datetime import date, datetime, time, timedelta | |
| from typing import Any | |
| from sqlalchemy import func | |
| from sqlalchemy.orm import Session | |
| from app import models | |
| from app.scoring import FILLER_WORDS, compute_score, compute_wpm, count_fillers, tokenize | |
| ALLOWED_TREND_METRICS = { | |
| "avg_wpm", | |
| "avg_filler_rate", | |
| "avg_overall_score", | |
| "avg_speech_score", | |
| "avg_delivery_score", | |
| "session_count", | |
| "total_duration_seconds", | |
| "best_overall_score", | |
| } | |
| GOAL_METRICS = { | |
| "wpm", | |
| "filler_rate", | |
| "overall_score", | |
| "practice_days_per_week", | |
| "session_duration_minutes", | |
| "eye_contact_score", | |
| "delivery_score", | |
| } | |
| def _parse_target_keywords(raw: str | None) -> list[str]: | |
| if not raw: | |
| return [] | |
| try: | |
| parsed = json.loads(raw) | |
| if isinstance(parsed, list): | |
| return [str(item) for item in parsed if str(item).strip()] | |
| except json.JSONDecodeError: | |
| pass | |
| return [] | |
| def prepare_user_analytics(db: Session, user_id: int, limit: int = 100) -> None: | |
| missing_sessions = ( | |
| db.query(models.Session.id) | |
| .outerjoin(models.SessionMetrics, models.SessionMetrics.session_id == models.Session.id) | |
| .filter(models.Session.user_id == user_id, models.SessionMetrics.id.is_(None)) | |
| .order_by(models.Session.created_at.desc()) | |
| .limit(limit) | |
| .all() | |
| ) | |
| for (session_id,) in missing_sessions: | |
| compute_session_metrics(db, session_id) | |
| def compute_session_metrics(db: Session, session_id: int) -> models.SessionMetrics: | |
| session = db.query(models.Session).filter(models.Session.id == session_id).first() | |
| if session is None: | |
| raise ValueError(f"Session {session_id} not found") | |
| segments = _segment_payload(session) | |
| metrics = _metric_payload(session) | |
| duration_ms = _duration_ms(session) | |
| duration_seconds = duration_ms / 1000 if duration_ms > 0 else 0 | |
| word_count = sum(len(tokenize(segment["text"])) for segment in segments) | |
| words = tokenize(" ".join(segment["text"] for segment in segments)) | |
| unique_word_count = len(set(words)) | |
| vocabulary_density = round(unique_word_count / word_count, 3) if word_count else 0 | |
| filler_breakdown = _filler_breakdown(segments) | |
| filler_count = sum(filler_breakdown.values()) or int(session.filler_count or 0) | |
| filler_rate = round(filler_count / (duration_seconds / 60), 2) if duration_seconds > 0 else 0 | |
| pause_count = len([point for point in metrics if point.get("pause_ms", 0) > 0]) | |
| wpm = session.wpm_avg or compute_wpm(word_count, duration_ms) | |
| score_data = session.score.data if session.score and isinstance(session.score.data, dict) else None | |
| if score_data is None: | |
| score_data = compute_score( | |
| segments, | |
| metrics, | |
| duration_ms, | |
| _parse_target_keywords(session.target_keywords_json), | |
| ) | |
| eye_contact = _metric_avg(metrics, "eye_contact") | |
| if (eye_contact is None or eye_contact == 0) and session.eye_contact_pct: | |
| eye_contact = session.eye_contact_pct | |
| smile = _metric_avg(metrics, "smile") | |
| loudness = min(1.0, _metric_avg(metrics, "rms") / 100) if metrics else None | |
| head_stability = _head_stability(metrics) | |
| existing = ( | |
| db.query(models.SessionMetrics) | |
| .filter(models.SessionMetrics.session_id == session_id) | |
| .first() | |
| ) | |
| snapshot = existing or models.SessionMetrics(session_id=session_id, user_id=session.user_id) | |
| snapshot.user_id = session.user_id | |
| snapshot.recorded_at = session.created_at or datetime.utcnow() | |
| snapshot.duration_seconds = duration_seconds | |
| snapshot.word_count = word_count | |
| snapshot.wpm = round(float(wpm or 0), 1) | |
| snapshot.filler_count = filler_count | |
| snapshot.filler_rate = filler_rate | |
| snapshot.pause_count = pause_count | |
| snapshot.unique_word_count = unique_word_count | |
| snapshot.vocabulary_density = vocabulary_density | |
| snapshot.eye_contact_score = _nullable_round(eye_contact) | |
| snapshot.smile_score = _nullable_round(smile) | |
| snapshot.head_stability_score = _nullable_round(head_stability) | |
| snapshot.loudness_score = _nullable_round(loudness) | |
| snapshot.speech_score = _normalize_score(score_data.get("speech", 0), 40) | |
| snapshot.delivery_score = _normalize_score(score_data.get("delivery", 0), 30) | |
| snapshot.overall_score = round(float(score_data.get("total", 0) or 0), 1) | |
| snapshot.filler_breakdown = filler_breakdown | |
| snapshot.wpm_timeline = _wpm_timeline(session) | |
| if existing is None: | |
| db.add(snapshot) | |
| db.commit() | |
| db.refresh(snapshot) | |
| update_daily_aggregate(db, session.user_id, snapshot.recorded_at.date()) | |
| return snapshot | |
| def update_daily_aggregate(db: Session, user_id: int, practice_date: date) -> models.DailyAggregate: | |
| start = datetime.combine(practice_date, time.min) | |
| end = start + timedelta(days=1) | |
| rows = ( | |
| db.query(models.SessionMetrics) | |
| .filter( | |
| models.SessionMetrics.user_id == user_id, | |
| models.SessionMetrics.recorded_at >= start, | |
| models.SessionMetrics.recorded_at < end, | |
| ) | |
| .all() | |
| ) | |
| aggregate = ( | |
| db.query(models.DailyAggregate) | |
| .filter( | |
| models.DailyAggregate.user_id == user_id, | |
| models.DailyAggregate.practice_date == practice_date, | |
| ) | |
| .first() | |
| ) | |
| if aggregate is None: | |
| aggregate = models.DailyAggregate(user_id=user_id, practice_date=practice_date) | |
| db.add(aggregate) | |
| aggregate.session_count = len(rows) | |
| aggregate.total_duration_seconds = round(sum(row.duration_seconds for row in rows), 1) | |
| aggregate.avg_wpm = _avg([row.wpm for row in rows]) | |
| aggregate.avg_filler_rate = _avg([row.filler_rate for row in rows]) | |
| aggregate.avg_overall_score = _avg([row.overall_score for row in rows]) | |
| aggregate.avg_speech_score = _avg([row.speech_score for row in rows]) | |
| aggregate.avg_delivery_score = _avg([row.delivery_score for row in rows]) | |
| aggregate.best_overall_score = round(max([row.overall_score for row in rows] or [0]), 1) | |
| db.commit() | |
| db.refresh(aggregate) | |
| return aggregate | |
| def get_overview_stats(db: Session, user_id: int) -> dict[str, Any]: | |
| total_sessions = db.query(func.count(models.SessionMetrics.id)).filter(models.SessionMetrics.user_id == user_id).scalar() or 0 | |
| total_seconds = db.query(func.coalesce(func.sum(models.SessionMetrics.duration_seconds), 0)).filter(models.SessionMetrics.user_id == user_id).scalar() or 0 | |
| start = datetime.utcnow() - timedelta(days=30) | |
| recent = ( | |
| db.query(models.SessionMetrics) | |
| .filter(models.SessionMetrics.user_id == user_id, models.SessionMetrics.recorded_at >= start) | |
| .all() | |
| ) | |
| practice_days = ( | |
| db.query(func.count(func.distinct(models.DailyAggregate.practice_date))) | |
| .filter( | |
| models.DailyAggregate.user_id == user_id, | |
| models.DailyAggregate.practice_date >= start.date(), | |
| models.DailyAggregate.session_count > 0, | |
| ) | |
| .scalar() | |
| or 0 | |
| ) | |
| practice_dates = [ | |
| value[0].isoformat() | |
| for value in db.query(models.DailyAggregate.practice_date) | |
| .filter(models.DailyAggregate.user_id == user_id, models.DailyAggregate.session_count > 0) | |
| .order_by(models.DailyAggregate.practice_date.asc()) | |
| .all() | |
| ] | |
| streaks = calculate_streak(practice_dates) | |
| curriculum_completed = ( | |
| db.query(func.count(models.CurriculumDayCompletion.id)) | |
| .filter(models.CurriculumDayCompletion.user_id == user_id) | |
| .scalar() | |
| or 0 | |
| ) | |
| today_agg = ( | |
| db.query(models.DailyAggregate) | |
| .filter( | |
| models.DailyAggregate.user_id == user_id, | |
| models.DailyAggregate.practice_date == date.today(), | |
| ) | |
| .first() | |
| ) | |
| today_minutes = round((today_agg.total_duration_seconds / 60), 1) if today_agg else 0.0 | |
| return { | |
| "total_sessions": total_sessions, | |
| "total_practice_minutes": round(float(total_seconds) / 60, 1), | |
| "today_practice_minutes": today_minutes, | |
| "current_streak": streaks["current_streak"], | |
| "longest_streak": streaks["longest_streak"], | |
| "curriculum_completed_days": curriculum_completed, | |
| "recent_30d": { | |
| "session_count": len(recent), | |
| "avg_overall_score": _avg([row.overall_score for row in recent]), | |
| "avg_wpm": _avg([row.wpm for row in recent]), | |
| "avg_filler_rate": _avg([row.filler_rate for row in recent]), | |
| "avg_delivery_score": _avg([row.delivery_score for row in recent]), | |
| "practice_days": practice_days, | |
| }, | |
| } | |
| def get_trend_data(db: Session, user_id: int, metric: str, days: int = 90) -> list[dict[str, Any]]: | |
| if metric not in ALLOWED_TREND_METRICS: | |
| raise ValueError(f"Invalid metric: {metric}") | |
| bounded_days = min(max(days, 7), 365) | |
| start = date.today() - timedelta(days=bounded_days) | |
| rows = ( | |
| db.query(models.DailyAggregate) | |
| .filter(models.DailyAggregate.user_id == user_id, models.DailyAggregate.practice_date >= start) | |
| .order_by(models.DailyAggregate.practice_date.asc()) | |
| .limit(365) | |
| .all() | |
| ) | |
| return [ | |
| {"date": row.practice_date.isoformat(), "value": round(float(getattr(row, metric) or 0), 1)} | |
| for row in rows | |
| ] | |
| def get_session_history(db: Session, user_id: int, limit: int = 20, offset: int = 0) -> list[dict[str, Any]]: | |
| rows = ( | |
| db.query(models.SessionMetrics, models.Session) | |
| .join(models.Session, models.Session.id == models.SessionMetrics.session_id) | |
| .filter(models.SessionMetrics.user_id == user_id) | |
| .order_by(models.SessionMetrics.recorded_at.desc()) | |
| .limit(min(limit, 100)) | |
| .offset(offset) | |
| .all() | |
| ) | |
| return [_session_metrics_to_dict(metric_row, sess_row) for metric_row, sess_row in rows] | |
| def get_activity_heatmap(db: Session, user_id: int, days: int = 365) -> list[dict[str, Any]]: | |
| bounded_days = min(max(days, 30), 365) | |
| start = date.today() - timedelta(days=bounded_days) | |
| rows = ( | |
| db.query(models.DailyAggregate) | |
| .filter(models.DailyAggregate.user_id == user_id, models.DailyAggregate.practice_date >= start) | |
| .order_by(models.DailyAggregate.practice_date.asc()) | |
| .limit(365) | |
| .all() | |
| ) | |
| return [ | |
| { | |
| "date": row.practice_date.isoformat(), | |
| "count": row.session_count, | |
| "minutes": round(row.total_duration_seconds / 60, 1), | |
| } | |
| for row in rows | |
| ] | |
| def get_filler_breakdown(db: Session, user_id: int, last_n: int = 20) -> dict[str, int]: | |
| rows = ( | |
| db.query(models.SessionMetrics.filler_breakdown) | |
| .filter(models.SessionMetrics.user_id == user_id) | |
| .order_by(models.SessionMetrics.recorded_at.desc()) | |
| .limit(min(last_n, 100)) | |
| .all() | |
| ) | |
| combined: dict[str, int] = {} | |
| for (breakdown,) in rows: | |
| if not isinstance(breakdown, dict): | |
| continue | |
| for word, count in breakdown.items(): | |
| combined[str(word)] = combined.get(str(word), 0) + int(count or 0) | |
| return dict(sorted(combined.items(), key=lambda item: item[1], reverse=True)) | |
| def _serialize_comparison_session(snapshot: models.SessionMetrics, session: models.Session | None) -> dict[str, Any]: | |
| segments = session.transcript_segments if session else [] | |
| full_text = " ".join(seg.text for seg in segments).strip() | |
| score_data = session.score.data if session and session.score and isinstance(session.score.data, dict) else {} | |
| video_available = bool( | |
| session and session.video_path and (session.video_path.startswith("s3://") or os.path.exists(session.video_path)) | |
| ) | |
| return { | |
| "id": snapshot.session_id, | |
| "date": snapshot.recorded_at.isoformat(), | |
| "title": (session.title if session else None) or f"Session #{snapshot.session_id}", | |
| "mode": (session.mode if session else None) or "Practice", | |
| "curriculum_day": session.curriculum_day if session else None, | |
| "duration_seconds": round(snapshot.duration_seconds, 1), | |
| "word_count": snapshot.word_count, | |
| "wpm": snapshot.wpm, | |
| "filler_count": snapshot.filler_count, | |
| "filler_rate": snapshot.filler_rate, | |
| "pause_count": snapshot.pause_count, | |
| "eye_contact_pct": round((snapshot.eye_contact_score or 0) * 100, 1), | |
| "vocabulary_density": snapshot.vocabulary_density, | |
| "overall_score": snapshot.overall_score, | |
| "speech_score": snapshot.speech_score, | |
| "delivery_score": snapshot.delivery_score, | |
| "filler_breakdown": snapshot.filler_breakdown or {}, | |
| "video_available": video_available, | |
| "opening_excerpt": full_text[:280] + ("..." if len(full_text) > 280 else ""), | |
| "top_fixes": score_data.get("topFixes", []), | |
| } | |
| def compare_sessions(db: Session, user_id: int, session_a: int, session_b: int) -> dict[str, Any]: | |
| rows = ( | |
| db.query(models.SessionMetrics) | |
| .filter( | |
| models.SessionMetrics.user_id == user_id, | |
| models.SessionMetrics.session_id.in_([session_a, session_b]), | |
| ) | |
| .all() | |
| ) | |
| by_id = {row.session_id: row for row in rows} | |
| if session_a not in by_id or session_b not in by_id: | |
| raise ValueError("One or both sessions not found") | |
| left = by_id[session_a] | |
| right = by_id[session_b] | |
| sess_rows = ( | |
| db.query(models.Session) | |
| .filter(models.Session.user_id == user_id, models.Session.id.in_([session_a, session_b])) | |
| .all() | |
| ) | |
| sess_by_id = {s.id: s for s in sess_rows} | |
| def diff(attr: str) -> dict[str, float]: | |
| a_value = float(getattr(left, attr) or 0) | |
| b_value = float(getattr(right, attr) or 0) | |
| return {"a": round(a_value, 2), "b": round(b_value, 2), "delta": round(b_value - a_value, 2)} | |
| comparison = { | |
| "overall_score": diff("overall_score"), | |
| "speech_score": diff("speech_score"), | |
| "delivery_score": diff("delivery_score"), | |
| "wpm": diff("wpm"), | |
| "filler_rate": diff("filler_rate"), | |
| "eye_contact_score": diff("eye_contact_score"), | |
| "vocabulary_density": diff("vocabulary_density"), | |
| } | |
| insights: list[str] = [] | |
| score_delta = comparison["overall_score"]["delta"] | |
| filler_delta = comparison["filler_rate"]["delta"] | |
| eye_delta = round(comparison["eye_contact_score"]["delta"] * 100, 1) | |
| if score_delta > 0: | |
| insights.append(f"Overall presenting score climbed +{score_delta:.1f} points from Session A to Session B.") | |
| elif score_delta < 0: | |
| insights.append(f"Overall score dipped {score_delta:.1f} points — check pace and filler habits between takes.") | |
| if filler_delta < -0.3: | |
| insights.append(f"Filler rate dropped by {abs(filler_delta):.1f} fillers/min — cleaner silent pauses!") | |
| elif filler_delta > 0.5: | |
| insights.append(f"Filler rate increased by +{filler_delta:.1f}/min on Session B; slow down transitions.") | |
| if eye_delta >= 5: | |
| insights.append(f"Camera lens eye contact jumped +{eye_delta:.0f}% between takes.") | |
| if 125 <= right.wpm <= 160 and not (125 <= left.wpm <= 160): | |
| insights.append(f"Speaking speed moved into the optimal 125–160 WPM keynote window ({right.wpm:.0f} WPM).") | |
| if not insights: | |
| insights.append("Consistent performance across both takes — push difficulty with a Cue-Card or Contrarian challenge next.") | |
| return { | |
| "session_a": _serialize_comparison_session(left, sess_by_id.get(session_a)), | |
| "session_b": _serialize_comparison_session(right, sess_by_id.get(session_b)), | |
| "comparison": comparison, | |
| "insights": insights, | |
| } | |
| def get_weekly_review(db: Session, user_id: int, weeks_ago: int = 0) -> dict[str, Any]: | |
| bounded_weeks = max(0, min(weeks_ago, 12)) | |
| today = date.today() | |
| # Rolling 7-day window anchored at today (or shifted by weeks_ago * 7) | |
| period_end = today - timedelta(days=bounded_weeks * 7) | |
| period_start = period_end - timedelta(days=6) | |
| prev_end = period_start - timedelta(days=1) | |
| prev_start = prev_end - timedelta(days=6) | |
| def _window_rows(start_d: date, end_d: date) -> list[models.SessionMetrics]: | |
| dt_start = datetime.combine(start_d, time.min) | |
| dt_end = datetime.combine(end_d + timedelta(days=1), time.min) | |
| return ( | |
| db.query(models.SessionMetrics) | |
| .filter( | |
| models.SessionMetrics.user_id == user_id, | |
| models.SessionMetrics.recorded_at >= dt_start, | |
| models.SessionMetrics.recorded_at < dt_end, | |
| ) | |
| .order_by(models.SessionMetrics.recorded_at.asc()) | |
| .all() | |
| ) | |
| current_rows = _window_rows(period_start, period_end) | |
| prev_rows = _window_rows(prev_start, prev_end) | |
| def _summarize(rows: list[models.SessionMetrics]) -> dict[str, Any]: | |
| if not rows: | |
| return { | |
| "session_count": 0, | |
| "active_days": 0, | |
| "total_minutes": 0.0, | |
| "avg_overall_score": 0.0, | |
| "avg_speech_score": 0.0, | |
| "avg_delivery_score": 0.0, | |
| "avg_wpm": 0.0, | |
| "avg_filler_rate": 0.0, | |
| "avg_eye_contact_pct": 0.0, | |
| "best_session_id": None, | |
| "best_score": 0.0, | |
| } | |
| active_days = len({r.recorded_at.date().isoformat() for r in rows}) | |
| eye_vals = [(r.eye_contact_score or 0) * 100 for r in rows if r.eye_contact_score is not None] | |
| best_row = max(rows, key=lambda r: r.overall_score) | |
| return { | |
| "session_count": len(rows), | |
| "active_days": active_days, | |
| "total_minutes": round(sum(r.duration_seconds for r in rows) / 60, 1), | |
| "avg_overall_score": _avg([r.overall_score for r in rows]), | |
| "avg_speech_score": _avg([r.speech_score for r in rows]), | |
| "avg_delivery_score": _avg([r.delivery_score for r in rows]), | |
| "avg_wpm": _avg([r.wpm for r in rows]), | |
| "avg_filler_rate": _avg([r.filler_rate for r in rows]), | |
| "avg_eye_contact_pct": _avg(eye_vals) if eye_vals else 0.0, | |
| "best_session_id": best_row.session_id, | |
| "best_score": round(best_row.overall_score, 1), | |
| } | |
| curr = _summarize(current_rows) | |
| prev = _summarize(prev_rows) | |
| deltas = { | |
| "overall_score": round(curr["avg_overall_score"] - prev["avg_overall_score"], 1) if prev["session_count"] else 0.0, | |
| "filler_rate": round(curr["avg_filler_rate"] - prev["avg_filler_rate"], 2) if prev["session_count"] else 0.0, | |
| "wpm": round(curr["avg_wpm"] - prev["avg_wpm"], 1) if prev["session_count"] else 0.0, | |
| "eye_contact_pct": round(curr["avg_eye_contact_pct"] - prev["avg_eye_contact_pct"], 1) if prev["session_count"] else 0.0, | |
| "minutes": round(curr["total_minutes"] - prev["total_minutes"], 1), | |
| } | |
| filler_totals: dict[str, int] = {} | |
| for r in current_rows: | |
| if isinstance(r.filler_breakdown, dict): | |
| for word, count in r.filler_breakdown.items(): | |
| filler_totals[str(word)] = filler_totals.get(str(word), 0) + int(count or 0) | |
| top_fillers = [ | |
| {"word": w, "count": c} | |
| for w, c in sorted(filler_totals.items(), key=lambda x: x[1], reverse=True)[:5] | |
| ] | |
| wins: list[str] = [] | |
| focus_items: list[str] = [] | |
| if curr["session_count"] == 0: | |
| wins.append("Ready for a fresh training week — record one 60-second take today to ignite your streak.") | |
| focus_items.append("Complete your recommended 30-Day Curriculum challenge in the Studio.") | |
| else: | |
| if curr["active_days"] >= 4: | |
| wins.append(f"Strong consistency: practiced across {curr['active_days']} of 7 days ({curr['total_minutes']} mins total).") | |
| else: | |
| wins.append(f"Logged {curr['session_count']} practice take{'s' if curr['session_count'] != 1 else ''} with a best score of {curr['best_score']}/100.") | |
| if 125 <= curr["avg_wpm"] <= 160: | |
| wins.append(f"Average speaking speed ({curr['avg_wpm']} WPM) stayed right inside the 125–160 WPM keynote window.") | |
| elif curr["avg_wpm"] > 160: | |
| focus_items.append(f"Pace averaged {curr['avg_wpm']} WPM — practice 1-second breath pauses between technical clauses.") | |
| elif curr["avg_wpm"] > 0: | |
| focus_items.append(f"Pace averaged {curr['avg_wpm']} WPM — tighten sentence openings toward 135 WPM.") | |
| if curr["avg_filler_rate"] <= 2.0: | |
| wins.append(f"Crisp filler discipline at {curr['avg_filler_rate']}/min.") | |
| else: | |
| top_w = top_fillers[0]["word"] if top_fillers else "um/like" | |
| focus_items.append(f"Cut filler rate from {curr['avg_filler_rate']}/min (<2.0/min target), especially '{top_w}'.") | |
| if curr["avg_eye_contact_pct"] >= 70: | |
| wins.append(f"Confident camera connection at {curr['avg_eye_contact_pct']:.0f}% lens eye contact.") | |
| else: | |
| focus_items.append(f"Lift lens eye contact from {curr['avg_eye_contact_pct']:.0f}% toward 75%+ using Cue-Card teleprompter mode.") | |
| if not focus_items: | |
| focus_items.append("Level up difficulty: record a 2-Minute Deep-Dive Tech News Explainer without reading word-for-word.") | |
| reflection = ( | |
| db.query(models.WeeklyReflection) | |
| .filter( | |
| models.WeeklyReflection.user_id == user_id, | |
| models.WeeklyReflection.week_start_date == period_start, | |
| ) | |
| .first() | |
| ) | |
| first_vs_last_comparison = None | |
| if len(current_rows) >= 2: | |
| first_vs_last_comparison = { | |
| "first_session_id": current_rows[0].session_id, | |
| "latest_session_id": current_rows[-1].session_id, | |
| "score_change": round(current_rows[-1].overall_score - current_rows[0].overall_score, 1), | |
| } | |
| return { | |
| "week_start_date": period_start.isoformat(), | |
| "week_end_date": period_end.isoformat(), | |
| "weeks_ago": bounded_weeks, | |
| "current_week": curr, | |
| "previous_week": prev, | |
| "deltas": deltas, | |
| "top_fillers": top_fillers, | |
| "wins": wins, | |
| "recommended_focus": focus_items, | |
| "first_vs_latest": first_vs_last_comparison, | |
| "reflection": { | |
| "wins_note": reflection.wins_note if reflection else "", | |
| "focus_next_week": reflection.focus_next_week if reflection else "", | |
| }, | |
| } | |
| def save_weekly_reflection( | |
| db: Session, | |
| user_id: int, | |
| week_start_date: date, | |
| wins_note: str | None, | |
| focus_next_week: str | None, | |
| ) -> dict[str, Any]: | |
| row = ( | |
| db.query(models.WeeklyReflection) | |
| .filter( | |
| models.WeeklyReflection.user_id == user_id, | |
| models.WeeklyReflection.week_start_date == week_start_date, | |
| ) | |
| .first() | |
| ) | |
| if row is None: | |
| row = models.WeeklyReflection( | |
| user_id=user_id, | |
| week_start_date=week_start_date, | |
| wins_note=wins_note or "", | |
| focus_next_week=focus_next_week or "", | |
| ) | |
| db.add(row) | |
| else: | |
| if wins_note is not None: | |
| row.wins_note = wins_note | |
| if focus_next_week is not None: | |
| row.focus_next_week = focus_next_week | |
| db.commit() | |
| db.refresh(row) | |
| return { | |
| "week_start_date": row.week_start_date.isoformat(), | |
| "wins_note": row.wins_note or "", | |
| "focus_next_week": row.focus_next_week or "", | |
| } | |
| def get_user_goals(db: Session, user_id: int) -> list[dict[str, Any]]: | |
| rows = ( | |
| db.query(models.UserGoal) | |
| .filter(models.UserGoal.user_id == user_id) | |
| .order_by(models.UserGoal.is_active.desc(), models.UserGoal.created_at.asc()) | |
| .all() | |
| ) | |
| current_values = get_goal_current_values(db, user_id) | |
| return [serialize_goal(row, current_values.get(row.metric_name)) for row in rows] | |
| def get_active_goal_context(db: Session, user_id: int) -> list[dict[str, Any]]: | |
| rows = ( | |
| db.query(models.UserGoal) | |
| .filter(models.UserGoal.user_id == user_id, models.UserGoal.is_active.is_(True)) | |
| .order_by(models.UserGoal.created_at.asc()) | |
| .all() | |
| ) | |
| current_values = get_goal_current_values(db, user_id) | |
| return [serialize_goal(row, current_values.get(row.metric_name)) for row in rows] | |
| def get_goal_current_values(db: Session, user_id: int) -> dict[str, float | None]: | |
| latest = ( | |
| db.query(models.SessionMetrics) | |
| .filter(models.SessionMetrics.user_id == user_id) | |
| .order_by(models.SessionMetrics.recorded_at.desc()) | |
| .first() | |
| ) | |
| week_start = date.today() - timedelta(days=6) | |
| practice_days = ( | |
| db.query(func.count(models.DailyAggregate.id)) | |
| .filter( | |
| models.DailyAggregate.user_id == user_id, | |
| models.DailyAggregate.practice_date >= week_start, | |
| models.DailyAggregate.session_count > 0, | |
| ) | |
| .scalar() | |
| or 0 | |
| ) | |
| if latest is None: | |
| return { | |
| metric: (float(practice_days) if metric == "practice_days_per_week" else None) | |
| for metric in GOAL_METRICS | |
| } | |
| return { | |
| "wpm": float(latest.wpm), | |
| "filler_rate": float(latest.filler_rate), | |
| "overall_score": float(latest.overall_score), | |
| "practice_days_per_week": float(practice_days), | |
| "session_duration_minutes": round(float(latest.duration_seconds) / 60, 1), | |
| "eye_contact_score": float(latest.eye_contact_score) if latest.eye_contact_score is not None else None, | |
| "delivery_score": float(latest.delivery_score), | |
| } | |
| def serialize_goal(goal: models.UserGoal, current_value: float | None) -> dict[str, Any]: | |
| achieved = False | |
| if current_value is not None: | |
| if goal.comparison == "gte": | |
| achieved = current_value >= goal.target_value | |
| elif goal.comparison == "lte": | |
| achieved = current_value <= goal.target_value | |
| elif goal.comparison == "between" and goal.target_upper is not None: | |
| achieved = goal.target_value <= current_value <= goal.target_upper | |
| return { | |
| "id": goal.id, | |
| "metric_name": goal.metric_name, | |
| "comparison": goal.comparison, | |
| "target_value": goal.target_value, | |
| "target_upper": goal.target_upper, | |
| "is_active": goal.is_active, | |
| "current_value": current_value, | |
| "achieved": achieved, | |
| "created_at": goal.created_at, | |
| "updated_at": goal.updated_at, | |
| } | |
| def calculate_streak(practice_dates: list[str]) -> dict[str, int]: | |
| if not practice_dates: | |
| return {"current_streak": 0, "longest_streak": 0} | |
| parsed = sorted({datetime.strptime(value, "%Y-%m-%d").date() for value in practice_dates}) | |
| longest = 1 | |
| current_run = 1 | |
| for index in range(1, len(parsed)): | |
| gap = (parsed[index] - parsed[index - 1]).days | |
| if gap == 1: | |
| current_run += 1 | |
| elif gap > 1: | |
| current_run = 1 | |
| longest = max(longest, current_run) | |
| days_since_last = (date.today() - parsed[-1]).days | |
| return {"current_streak": current_run if days_since_last <= 1 else 0, "longest_streak": longest} | |
| def _session_metrics_to_dict(row: models.SessionMetrics, sess: models.Session | None = None) -> dict[str, Any]: | |
| return { | |
| "session_id": row.session_id, | |
| "title": (sess.title if sess else None) or f"Session #{row.session_id}", | |
| "mode": (sess.mode if sess else None) or "Practice", | |
| "curriculum_day": sess.curriculum_day if sess else None, | |
| "video_available": bool(sess and sess.video_path), | |
| "recorded_at": row.recorded_at.isoformat(), | |
| "duration_seconds": row.duration_seconds, | |
| "wpm": row.wpm, | |
| "filler_count": row.filler_count, | |
| "filler_rate": row.filler_rate, | |
| "speech_score": row.speech_score, | |
| "delivery_score": row.delivery_score, | |
| "overall_score": row.overall_score, | |
| "eye_contact_score": row.eye_contact_score, | |
| "vocabulary_density": row.vocabulary_density, | |
| } | |
| def _segment_payload(session: models.Session) -> list[dict[str, Any]]: | |
| return [ | |
| {"start_ms": segment.start_ms, "end_ms": segment.end_ms, "text": segment.text} | |
| for segment in session.transcript_segments | |
| ] | |
| def _metric_payload(session: models.Session) -> list[dict[str, Any]]: | |
| return [ | |
| { | |
| "t": point.t, | |
| "wpm": point.wpm, | |
| "rms": point.rms, | |
| "eye_contact": point.eye_contact, | |
| "smile": point.smile, | |
| "yaw": point.yaw, | |
| "pitch": point.pitch, | |
| "roll": point.roll, | |
| "filler_count": point.filler_count, | |
| "pause_ms": point.pause_ms, | |
| } | |
| for point in session.metrics | |
| ] | |
| def _duration_ms(session: models.Session) -> int: | |
| if session.duration_ms: | |
| return session.duration_ms | |
| if session.transcript_segments: | |
| return max(segment.end_ms for segment in session.transcript_segments) | |
| return 0 | |
| def _filler_breakdown(segments: list[dict[str, Any]]) -> dict[str, int]: | |
| text = " ".join(str(segment.get("text", "")).lower() for segment in segments) | |
| counts: dict[str, int] = {} | |
| for filler in FILLER_WORDS: | |
| escaped = re.escape(filler).replace(" ", r"\s+") | |
| pattern = re.compile(rf"\b{escaped}\b") | |
| count = len(pattern.findall(text)) | |
| if count: | |
| counts[filler] = count | |
| if not counts: | |
| total = count_fillers(segments) | |
| if total: | |
| counts["fillers"] = total | |
| return counts | |
| def _wpm_timeline(session: models.Session) -> list[dict[str, float]]: | |
| buckets: dict[int, dict[str, float]] = {} | |
| for segment in session.transcript_segments: | |
| minute = max(segment.start_ms, 0) // 60000 | |
| duration_seconds = max((segment.end_ms - segment.start_ms) / 1000, 0) | |
| words = len(tokenize(segment.text)) | |
| bucket = buckets.setdefault(minute, {"words": 0, "seconds": 0}) | |
| bucket["words"] += words | |
| bucket["seconds"] += duration_seconds | |
| return [ | |
| {"minute": minute, "wpm": round(bucket["words"] / (bucket["seconds"] / 60), 1) if bucket["seconds"] else 0} | |
| for minute, bucket in sorted(buckets.items()) | |
| ] | |
| def _metric_avg(metrics: list[dict[str, Any]], key: str) -> float | None: | |
| values = [float(point.get(key) or 0) for point in metrics if point.get(key) is not None] | |
| if not values: | |
| return None | |
| return sum(values) / len(values) | |
| def _head_stability(metrics: list[dict[str, Any]]) -> float | None: | |
| if not metrics: | |
| return None | |
| variance_total = sum(_variance([float(point.get(axis) or 0) for point in metrics]) for axis in ("yaw", "pitch", "roll")) | |
| return max(0.0, min(1.0, 1 - variance_total / 20)) | |
| def _variance(values: list[float]) -> float: | |
| if not values: | |
| return 0 | |
| average = sum(values) / len(values) | |
| return sum((value - average) ** 2 for value in values) / len(values) | |
| def _avg(values: list[float]) -> float: | |
| clean = [float(value or 0) for value in values] | |
| return round(sum(clean) / len(clean), 1) if clean else 0.0 | |
| def _normalize_score(value: Any, max_value: float) -> float: | |
| if max_value <= 0: | |
| return 0 | |
| return round(max(0, min(100, (float(value or 0) / max_value) * 100)), 1) | |
| def _nullable_round(value: float | None) -> float | None: | |
| if value is None: | |
| return None | |
| return round(float(value), 3) | |