""" Placeholder Resolver for AI Prompts Provides a registry of placeholder functions that resolve to actual user data. Used for prompt templates and preview functionality. """ import re from datetime import datetime, timedelta from typing import Dict, List, Optional, Callable from db import get_db, get_cursor, r2d # ── Helper Functions ────────────────────────────────────────────────────────── def get_profile_data(profile_id: str) -> Dict: """Load profile data for a user.""" with get_db() as conn: cur = get_cursor(conn) cur.execute("SELECT * FROM profiles WHERE id=%s", (profile_id,)) return r2d(cur.fetchone()) if cur.rowcount > 0 else {} def get_latest_weight(profile_id: str) -> Optional[str]: """Get latest weight entry.""" with get_db() as conn: cur = get_cursor(conn) cur.execute( "SELECT weight FROM weight_log WHERE profile_id=%s ORDER BY date DESC LIMIT 1", (profile_id,) ) row = cur.fetchone() return f"{row['weight']:.1f} kg" if row else "nicht verfügbar" def get_weight_trend(profile_id: str, days: int = 28) -> str: """Calculate weight trend description.""" with get_db() as conn: cur = get_cursor(conn) cutoff = (datetime.now() - timedelta(days=days)).strftime('%Y-%m-%d') cur.execute( """SELECT weight, date FROM weight_log WHERE profile_id=%s AND date >= %s ORDER BY date""", (profile_id, cutoff) ) rows = [r2d(r) for r in cur.fetchall()] if len(rows) < 2: return "nicht genug Daten" first = rows[0]['weight'] last = rows[-1]['weight'] delta = last - first if abs(delta) < 0.3: return "stabil" elif delta > 0: return f"steigend (+{delta:.1f} kg in {days} Tagen)" else: return f"sinkend ({delta:.1f} kg in {days} Tagen)" def get_latest_bf(profile_id: str) -> Optional[str]: """Get latest body fat percentage from caliper.""" with get_db() as conn: cur = get_cursor(conn) cur.execute( """SELECT body_fat_pct FROM caliper_log WHERE profile_id=%s AND body_fat_pct IS NOT NULL ORDER BY date DESC LIMIT 1""", (profile_id,) ) row = cur.fetchone() return f"{row['body_fat_pct']:.1f}%" if row else "nicht verfügbar" def get_nutrition_avg(profile_id: str, field: str, days: int = 30) -> str: """Calculate average nutrition value.""" with get_db() as conn: cur = get_cursor(conn) cutoff = (datetime.now() - timedelta(days=days)).strftime('%Y-%m-%d') # Map field names to actual column names field_map = { 'protein': 'protein_g', 'fat': 'fat_g', 'carb': 'carbs_g', 'kcal': 'kcal' } db_field = field_map.get(field, field) cur.execute( f"""SELECT AVG({db_field}) as avg FROM nutrition_log WHERE profile_id=%s AND date >= %s AND {db_field} IS NOT NULL""", (profile_id, cutoff) ) row = cur.fetchone() if row and row['avg']: if field == 'kcal': return f"{int(row['avg'])} kcal/Tag (Ø {days} Tage)" else: return f"{int(row['avg'])}g/Tag (Ø {days} Tage)" return "nicht verfügbar" def get_caliper_summary(profile_id: str) -> str: """Get latest caliper measurements summary.""" with get_db() as conn: cur = get_cursor(conn) cur.execute( """SELECT body_fat_pct, sf_method, date FROM caliper_log WHERE profile_id=%s AND body_fat_pct IS NOT NULL ORDER BY date DESC LIMIT 1""", (profile_id,) ) row = r2d(cur.fetchone()) if cur.rowcount > 0 else None if not row: return "keine Caliper-Messungen" method = row.get('sf_method', 'unbekannt') return f"{row['body_fat_pct']:.1f}% ({method} am {row['date']})" def get_circ_summary(profile_id: str) -> str: """Get latest circumference measurements summary with age annotations. For each measurement point, fetches the most recent value (even if from different dates). Annotates each value with measurement age for AI context. """ with get_db() as conn: cur = get_cursor(conn) # Define all circumference points with their labels fields = [ ('c_neck', 'Nacken'), ('c_chest', 'Brust'), ('c_waist', 'Taille'), ('c_belly', 'Bauch'), ('c_hip', 'Hüfte'), ('c_thigh', 'Oberschenkel'), ('c_calf', 'Wade'), ('c_arm', 'Arm') ] parts = [] today = datetime.now().date() # Get latest value for each field individually for field_name, label in fields: cur.execute( f"""SELECT {field_name}, date, CURRENT_DATE - date AS age_days FROM circumference_log WHERE profile_id=%s AND {field_name} IS NOT NULL ORDER BY date DESC LIMIT 1""", (profile_id,) ) row = r2d(cur.fetchone()) if cur.rowcount > 0 else None if row: value = row[field_name] age_days = row['age_days'] # Format age annotation if age_days == 0: age_str = "heute" elif age_days == 1: age_str = "gestern" elif age_days <= 7: age_str = f"vor {age_days} Tagen" elif age_days <= 30: weeks = age_days // 7 age_str = f"vor {weeks} Woche{'n' if weeks > 1 else ''}" else: months = age_days // 30 age_str = f"vor {months} Monat{'en' if months > 1 else ''}" parts.append(f"{label} {value:.1f}cm ({age_str})") return ', '.join(parts) if parts else "keine Umfangsmessungen" def get_goal_weight(profile_id: str) -> str: """Get goal weight from profile.""" profile = get_profile_data(profile_id) goal = profile.get('goal_weight') return f"{goal:.1f}" if goal else "nicht gesetzt" def get_goal_bf_pct(profile_id: str) -> str: """Get goal body fat percentage from profile.""" profile = get_profile_data(profile_id) goal = profile.get('goal_bf_pct') return f"{goal:.1f}" if goal else "nicht gesetzt" def get_nutrition_days(profile_id: str, days: int = 30) -> str: """Get number of days with nutrition data.""" with get_db() as conn: cur = get_cursor(conn) cutoff = (datetime.now() - timedelta(days=days)).strftime('%Y-%m-%d') cur.execute( """SELECT COUNT(DISTINCT date) as days FROM nutrition_log WHERE profile_id=%s AND date >= %s""", (profile_id, cutoff) ) row = cur.fetchone() return str(row['days']) if row else "0" def get_protein_ziel_low(profile_id: str) -> str: """Calculate lower protein target based on current weight (1.6g/kg).""" with get_db() as conn: cur = get_cursor(conn) cur.execute( """SELECT weight FROM weight_log WHERE profile_id=%s ORDER BY date DESC LIMIT 1""", (profile_id,) ) row = cur.fetchone() if row: return f"{int(float(row['weight']) * 1.6)}" return "nicht verfügbar" def get_protein_ziel_high(profile_id: str) -> str: """Calculate upper protein target based on current weight (2.2g/kg).""" with get_db() as conn: cur = get_cursor(conn) cur.execute( """SELECT weight FROM weight_log WHERE profile_id=%s ORDER BY date DESC LIMIT 1""", (profile_id,) ) row = cur.fetchone() if row: return f"{int(float(row['weight']) * 2.2)}" return "nicht verfügbar" def get_activity_summary(profile_id: str, days: int = 14) -> str: """Get activity summary for recent period.""" with get_db() as conn: cur = get_cursor(conn) cutoff = (datetime.now() - timedelta(days=days)).strftime('%Y-%m-%d') cur.execute( """SELECT COUNT(*) as count, SUM(duration_min) as total_min, SUM(kcal_active) as total_kcal FROM activity_log WHERE profile_id=%s AND date >= %s""", (profile_id, cutoff) ) row = r2d(cur.fetchone()) if row['count'] == 0: return f"Keine Aktivitäten in den letzten {days} Tagen" avg_min = int(row['total_min'] / row['count']) if row['total_min'] else 0 return f"{row['count']} Einheiten in {days} Tagen (Ø {avg_min} min/Einheit, {int(row['total_kcal'] or 0)} kcal gesamt)" def calculate_age(dob) -> str: """Calculate age from date of birth (accepts date object or string).""" if not dob: return "unbekannt" try: # Handle both datetime.date objects and strings if isinstance(dob, str): birth = datetime.strptime(dob, '%Y-%m-%d').date() else: birth = dob # Already a date object from PostgreSQL today = datetime.now().date() age = today.year - birth.year - ((today.month, today.day) < (birth.month, birth.day)) return str(age) except Exception as e: return "unbekannt" def get_activity_detail(profile_id: str, days: int = 14) -> str: """Get detailed activity log for analysis.""" with get_db() as conn: cur = get_cursor(conn) cutoff = (datetime.now() - timedelta(days=days)).strftime('%Y-%m-%d') cur.execute( """SELECT date, activity_type, duration_min, kcal_active, hr_avg FROM activity_log WHERE profile_id=%s AND date >= %s ORDER BY date DESC LIMIT 50""", (profile_id, cutoff) ) rows = [r2d(r) for r in cur.fetchall()] if not rows: return f"Keine Aktivitäten in den letzten {days} Tagen" # Format as readable list lines = [] for r in rows: hr_str = f" HF={r['hr_avg']}" if r.get('hr_avg') else "" lines.append( f"{r['date']}: {r['activity_type']} ({r['duration_min']}min, {r.get('kcal_active', 0)}kcal{hr_str})" ) return '\n'.join(lines[:20]) # Max 20 entries to avoid token bloat def get_trainingstyp_verteilung(profile_id: str, days: int = 14) -> str: """Get training type distribution.""" with get_db() as conn: cur = get_cursor(conn) cutoff = (datetime.now() - timedelta(days=days)).strftime('%Y-%m-%d') cur.execute( """SELECT training_category, COUNT(*) as count FROM activity_log WHERE profile_id=%s AND date >= %s AND training_category IS NOT NULL GROUP BY training_category ORDER BY count DESC""", (profile_id, cutoff) ) rows = [r2d(r) for r in cur.fetchall()] if not rows: return "Keine kategorisierten Trainings" total = sum(r['count'] for r in rows) parts = [f"{r['training_category']}: {int(r['count']/total*100)}%" for r in rows[:3]] return ", ".join(parts) def get_sleep_avg_duration(profile_id: str, days: int = 7) -> str: """Calculate average sleep duration in hours.""" with get_db() as conn: cur = get_cursor(conn) cutoff = (datetime.now() - timedelta(days=days)).strftime('%Y-%m-%d') cur.execute( """SELECT sleep_segments FROM sleep_log WHERE profile_id=%s AND date >= %s ORDER BY date DESC""", (profile_id, cutoff) ) rows = cur.fetchall() if not rows: return "nicht verfügbar" total_minutes = 0 for row in rows: segments = row['sleep_segments'] if segments: # Sum duration_min from all segments for seg in segments: total_minutes += seg.get('duration_min', 0) if total_minutes == 0: return "nicht verfügbar" avg_hours = total_minutes / len(rows) / 60 return f"{avg_hours:.1f}h" def get_sleep_avg_quality(profile_id: str, days: int = 7) -> str: """Calculate average sleep quality (Deep+REM %).""" with get_db() as conn: cur = get_cursor(conn) cutoff = (datetime.now() - timedelta(days=days)).strftime('%Y-%m-%d') cur.execute( """SELECT sleep_segments FROM sleep_log WHERE profile_id=%s AND date >= %s ORDER BY date DESC""", (profile_id, cutoff) ) rows = cur.fetchall() if not rows: return "nicht verfügbar" total_quality = 0 count = 0 for row in rows: segments = row['sleep_segments'] if segments: # Note: segments use 'phase' key (not 'stage'), stored lowercase (deep, rem, light, awake) deep_rem_min = sum(s.get('duration_min', 0) for s in segments if s.get('phase') in ['deep', 'rem']) total_min = sum(s.get('duration_min', 0) for s in segments) if total_min > 0: quality_pct = (deep_rem_min / total_min) * 100 total_quality += quality_pct count += 1 if count == 0: return "nicht verfügbar" avg_quality = total_quality / count return f"{avg_quality:.0f}% (Deep+REM)" def get_rest_days_count(profile_id: str, days: int = 30) -> str: """Count rest days in the given period.""" with get_db() as conn: cur = get_cursor(conn) cutoff = (datetime.now() - timedelta(days=days)).strftime('%Y-%m-%d') cur.execute( """SELECT COUNT(DISTINCT date) as count FROM rest_days WHERE profile_id=%s AND date >= %s""", (profile_id, cutoff) ) row = cur.fetchone() count = row['count'] if row else 0 return f"{count} Ruhetage" def get_vitals_avg_hr(profile_id: str, days: int = 7) -> str: """Calculate average resting heart rate.""" with get_db() as conn: cur = get_cursor(conn) cutoff = (datetime.now() - timedelta(days=days)).strftime('%Y-%m-%d') cur.execute( """SELECT AVG(resting_hr) as avg FROM vitals_baseline WHERE profile_id=%s AND date >= %s AND resting_hr IS NOT NULL""", (profile_id, cutoff) ) row = cur.fetchone() if row and row['avg']: return f"{int(row['avg'])} bpm" return "nicht verfügbar" def get_vitals_avg_hrv(profile_id: str, days: int = 7) -> str: """Calculate average heart rate variability.""" with get_db() as conn: cur = get_cursor(conn) cutoff = (datetime.now() - timedelta(days=days)).strftime('%Y-%m-%d') cur.execute( """SELECT AVG(hrv) as avg FROM vitals_baseline WHERE profile_id=%s AND date >= %s AND hrv IS NOT NULL""", (profile_id, cutoff) ) row = cur.fetchone() if row and row['avg']: return f"{int(row['avg'])} ms" return "nicht verfügbar" def get_vitals_vo2_max(profile_id: str) -> str: """Get latest VO2 Max value.""" with get_db() as conn: cur = get_cursor(conn) cur.execute( """SELECT vo2_max FROM vitals_baseline WHERE profile_id=%s AND vo2_max IS NOT NULL ORDER BY date DESC LIMIT 1""", (profile_id,) ) row = cur.fetchone() if row and row['vo2_max']: return f"{row['vo2_max']:.1f} ml/kg/min" return "nicht verfügbar" # ── Phase 0b Calculation Engine Integration ────────────────────────────────── def _safe_int(func_name: str, profile_id: str) -> str: """ Safely call calculation function and return integer value or fallback. Args: func_name: Name of the calculation function (e.g., 'goal_progress_score') profile_id: Profile ID Returns: String representation of integer value or 'nicht verfügbar' """ import traceback try: # Import calculations dynamically to avoid circular imports from calculations import scores, body_metrics, nutrition_metrics, activity_metrics, recovery_metrics, correlation_metrics # Map function names to actual functions func_map = { 'goal_progress_score': scores.calculate_goal_progress_score, 'body_progress_score': body_metrics.calculate_body_progress_score, 'nutrition_score': nutrition_metrics.calculate_nutrition_score, 'activity_score': activity_metrics.calculate_activity_score, 'recovery_score_v2': recovery_metrics.calculate_recovery_score_v2, 'data_quality_score': scores.calculate_data_quality_score, 'top_goal_progress_pct': lambda pid: scores.get_top_priority_goal(pid)['progress_pct'] if scores.get_top_priority_goal(pid) else None, 'top_focus_area_progress': lambda pid: scores.get_top_focus_area(pid)['progress'] if scores.get_top_focus_area(pid) else None, 'focus_cat_körper_progress': lambda pid: scores.calculate_category_progress(pid, 'körper'), 'focus_cat_ernährung_progress': lambda pid: scores.calculate_category_progress(pid, 'ernährung'), 'focus_cat_aktivität_progress': lambda pid: scores.calculate_category_progress(pid, 'aktivität'), 'focus_cat_recovery_progress': lambda pid: scores.calculate_category_progress(pid, 'recovery'), 'focus_cat_vitalwerte_progress': lambda pid: scores.calculate_category_progress(pid, 'vitalwerte'), 'focus_cat_mental_progress': lambda pid: scores.calculate_category_progress(pid, 'mental'), 'focus_cat_lebensstil_progress': lambda pid: scores.calculate_category_progress(pid, 'lebensstil'), 'training_minutes_week': activity_metrics.calculate_training_minutes_week, 'training_frequency_7d': activity_metrics.calculate_training_frequency_7d, 'quality_sessions_pct': activity_metrics.calculate_quality_sessions_pct, 'ability_balance_strength': activity_metrics.calculate_ability_balance_strength, 'ability_balance_endurance': activity_metrics.calculate_ability_balance_endurance, 'ability_balance_mental': activity_metrics.calculate_ability_balance_mental, 'ability_balance_coordination': activity_metrics.calculate_ability_balance_coordination, 'ability_balance_mobility': activity_metrics.calculate_ability_balance_mobility, 'proxy_internal_load_7d': activity_metrics.calculate_proxy_internal_load_7d, 'strain_score': activity_metrics.calculate_strain_score, 'rest_day_compliance': activity_metrics.calculate_rest_day_compliance, 'protein_adequacy_28d': nutrition_metrics.calculate_protein_adequacy_28d, 'macro_consistency_score': nutrition_metrics.calculate_macro_consistency_score, 'recent_load_balance_3d': recovery_metrics.calculate_recent_load_balance_3d, 'sleep_quality_7d': recovery_metrics.calculate_sleep_quality_7d, } func = func_map.get(func_name) if not func: return 'nicht verfügbar' result = func(profile_id) return str(int(result)) if result is not None else 'nicht verfügbar' except Exception as e: print(f"[ERROR] _safe_int({func_name}, {profile_id}): {type(e).__name__}: {e}") traceback.print_exc() return 'nicht verfügbar' def _safe_float(func_name: str, profile_id: str, decimals: int = 1) -> str: """ Safely call calculation function and return float value or fallback. Args: func_name: Name of the calculation function profile_id: Profile ID decimals: Number of decimal places Returns: String representation of float value or 'nicht verfügbar' """ import traceback try: from calculations import body_metrics, nutrition_metrics, activity_metrics, recovery_metrics, scores func_map = { 'weight_7d_median': body_metrics.calculate_weight_7d_median, 'weight_28d_slope': body_metrics.calculate_weight_28d_slope, 'weight_90d_slope': body_metrics.calculate_weight_90d_slope, 'fm_28d_change': body_metrics.calculate_fm_28d_change, 'lbm_28d_change': body_metrics.calculate_lbm_28d_change, 'waist_28d_delta': body_metrics.calculate_waist_28d_delta, 'hip_28d_delta': body_metrics.calculate_hip_28d_delta, 'chest_28d_delta': body_metrics.calculate_chest_28d_delta, 'arm_28d_delta': body_metrics.calculate_arm_28d_delta, 'thigh_28d_delta': body_metrics.calculate_thigh_28d_delta, 'waist_hip_ratio': body_metrics.calculate_waist_hip_ratio, 'energy_balance_7d': nutrition_metrics.calculate_energy_balance_7d, 'protein_g_per_kg': nutrition_metrics.calculate_protein_g_per_kg, 'monotony_score': activity_metrics.calculate_monotony_score, 'vo2max_trend_28d': activity_metrics.calculate_vo2max_trend_28d, 'hrv_vs_baseline_pct': recovery_metrics.calculate_hrv_vs_baseline_pct, 'rhr_vs_baseline_pct': recovery_metrics.calculate_rhr_vs_baseline_pct, 'sleep_avg_duration_7d': recovery_metrics.calculate_sleep_avg_duration_7d, 'sleep_debt_hours': recovery_metrics.calculate_sleep_debt_hours, 'sleep_regularity_proxy': recovery_metrics.calculate_sleep_regularity_proxy, 'focus_cat_körper_weight': lambda pid: scores.calculate_category_weight(pid, 'körper'), 'focus_cat_ernährung_weight': lambda pid: scores.calculate_category_weight(pid, 'ernährung'), 'focus_cat_aktivität_weight': lambda pid: scores.calculate_category_weight(pid, 'aktivität'), 'focus_cat_recovery_weight': lambda pid: scores.calculate_category_weight(pid, 'recovery'), 'focus_cat_vitalwerte_weight': lambda pid: scores.calculate_category_weight(pid, 'vitalwerte'), 'focus_cat_mental_weight': lambda pid: scores.calculate_category_weight(pid, 'mental'), 'focus_cat_lebensstil_weight': lambda pid: scores.calculate_category_weight(pid, 'lebensstil'), } func = func_map.get(func_name) if not func: return 'nicht verfügbar' result = func(profile_id) return f"{result:.{decimals}f}" if result is not None else 'nicht verfügbar' except Exception as e: print(f"[ERROR] _safe_float({func_name}, {profile_id}): {type(e).__name__}: {e}") traceback.print_exc() return 'nicht verfügbar' def _safe_str(func_name: str, profile_id: str) -> str: """ Safely call calculation function and return string value or fallback. """ import traceback try: from calculations import body_metrics, nutrition_metrics, activity_metrics, scores, correlation_metrics func_map = { 'top_goal_name': lambda pid: (scores.get_top_priority_goal(pid).get('name') or scores.get_top_priority_goal(pid).get('goal_type')) if scores.get_top_priority_goal(pid) else None, 'top_goal_status': lambda pid: scores.get_top_priority_goal(pid)['status'] if scores.get_top_priority_goal(pid) else None, 'top_focus_area_name': lambda pid: scores.get_top_focus_area(pid)['label'] if scores.get_top_focus_area(pid) else None, 'recomposition_quadrant': body_metrics.calculate_recomposition_quadrant, 'energy_deficit_surplus': nutrition_metrics.calculate_energy_deficit_surplus, 'protein_days_in_target': nutrition_metrics.calculate_protein_days_in_target, 'intake_volatility': nutrition_metrics.calculate_intake_volatility, 'active_goals_md': lambda pid: _format_goals_as_markdown(pid), 'focus_areas_weighted_md': lambda pid: _format_focus_areas_as_markdown(pid), 'top_3_focus_areas': lambda pid: _format_top_focus_areas(pid), 'top_3_goals_behind_schedule': lambda pid: _format_goals_behind(pid), 'top_3_goals_on_track': lambda pid: _format_goals_on_track(pid), } func = func_map.get(func_name) if not func: return 'nicht verfügbar' result = func(profile_id) return str(result) if result is not None else 'nicht verfügbar' except Exception as e: print(f"[ERROR] _safe_str({func_name}, {profile_id}): {type(e).__name__}: {e}") traceback.print_exc() return 'nicht verfügbar' def _safe_json(func_name: str, profile_id: str) -> str: """ Safely call calculation function and return JSON string or fallback. """ import traceback try: import json from calculations import scores, correlation_metrics func_map = { 'correlation_energy_weight_lag': lambda pid: correlation_metrics.calculate_lag_correlation(pid, 'energy', 'weight'), 'correlation_protein_lbm': lambda pid: correlation_metrics.calculate_lag_correlation(pid, 'protein', 'lbm'), 'correlation_load_hrv': lambda pid: correlation_metrics.calculate_lag_correlation(pid, 'training_load', 'hrv'), 'correlation_load_rhr': lambda pid: correlation_metrics.calculate_lag_correlation(pid, 'training_load', 'rhr'), 'correlation_sleep_recovery': correlation_metrics.calculate_correlation_sleep_recovery, 'plateau_detected': correlation_metrics.calculate_plateau_detected, 'top_drivers': correlation_metrics.calculate_top_drivers, 'active_goals_json': lambda pid: _get_active_goals_json(pid), 'focus_areas_weighted_json': lambda pid: _get_focus_areas_weighted_json(pid), 'focus_area_weights_json': lambda pid: json.dumps(scores.get_user_focus_weights(pid), ensure_ascii=False), } func = func_map.get(func_name) if not func: return '{}' result = func(profile_id) if result is None: return '{}' # If already string, return it; otherwise convert to JSON if isinstance(result, str): return result else: return json.dumps(result, ensure_ascii=False) except Exception as e: print(f"[ERROR] _safe_json({func_name}, {profile_id}): {type(e).__name__}: {e}") traceback.print_exc() return '{}' def _get_active_goals_json(profile_id: str) -> str: """Get active goals as JSON string""" import json try: from goal_utils import get_active_goals goals = get_active_goals(profile_id) return json.dumps(goals, default=str) except Exception: return '[]' def _get_focus_areas_weighted_json(profile_id: str) -> str: """Get focus areas with weights as JSON string""" import json try: from calculations.scores import get_user_focus_weights from goal_utils import get_db, get_cursor weights = get_user_focus_weights(profile_id) # Get focus area details with get_db() as conn: cur = get_cursor(conn) cur.execute(""" SELECT key, name_de, name_en, category FROM focus_area_definitions WHERE is_active = true """) definitions = {row['key']: row for row in cur.fetchall()} # Build weighted list result = [] for area_key, weight in weights.items(): if weight > 0 and area_key in definitions: area = definitions[area_key] result.append({ 'key': area_key, 'name': area['name_de'], 'category': area['category'], 'weight': weight }) # Sort by weight descending result.sort(key=lambda x: x['weight'], reverse=True) return json.dumps(result, default=str) except Exception: return '[]' def _format_goals_as_markdown(profile_id: str) -> str: """Format goals as markdown table""" try: from goal_utils import get_active_goals goals = get_active_goals(profile_id) if not goals: return 'Keine Ziele definiert' # Build markdown table lines = ['| Ziel | Aktuell | Ziel | Progress | Status |', '|------|---------|------|----------|--------|'] for goal in goals: name = goal.get('name') or goal.get('goal_type', 'Unbekannt') current = goal.get('current_value') target = goal.get('target_value') start = goal.get('start_value') # Calculate progress if possible progress_str = '-' if None not in [current, target, start]: try: current_f = float(current) target_f = float(target) start_f = float(start) if target_f == start_f: progress_pct = 100 if current_f == target_f else 0 else: progress_pct = ((current_f - start_f) / (target_f - start_f)) * 100 progress_pct = max(0, min(100, progress_pct)) progress_str = f"{int(progress_pct)}%" except (ValueError, ZeroDivisionError): progress_str = '-' current_str = f"{current}" if current is not None else '-' target_str = f"{target}" if target is not None else '-' status = '🎯' if goal.get('is_primary') else '○' lines.append(f"| {name} | {current_str} | {target_str} | {progress_str} | {status} |") return '\n'.join(lines) except Exception: return 'Keine Ziele definiert' def _format_focus_areas_as_markdown(profile_id: str) -> str: """Format focus areas as markdown""" try: import json weighted_json = _get_focus_areas_weighted_json(profile_id) areas = json.loads(weighted_json) if not areas: return 'Keine Focus Areas aktiv' # Build markdown list lines = [] for area in areas: name = area.get('name', 'Unbekannt') weight = area.get('weight', 0) lines.append(f"- **{name}**: {weight}%") return '\n'.join(lines) except Exception: return 'Keine Focus Areas aktiv' def _format_top_focus_areas(profile_id: str, n: int = 3) -> str: """Format top N focus areas as text""" try: import json weighted_json = _get_focus_areas_weighted_json(profile_id) areas = json.loads(weighted_json) if not areas: return 'Keine Focus Areas definiert' # Sort by weight descending and take top N sorted_areas = sorted(areas, key=lambda x: x.get('weight', 0), reverse=True)[:n] lines = [] for i, area in enumerate(sorted_areas, 1): name = area.get('name', 'Unbekannt') weight = area.get('weight', 0) lines.append(f"{i}. {name} ({weight}%)") return ', '.join(lines) except Exception: return 'nicht verfügbar' def _format_goals_behind(profile_id: str, n: int = 3) -> str: """ Format top N goals behind schedule (based on time deviation). Compares actual progress vs. expected progress based on elapsed time. Negative deviation = behind schedule. """ try: from goal_utils import get_active_goals from datetime import date goals = get_active_goals(profile_id) if not goals: return 'Keine Ziele definiert' today = date.today() goals_with_deviation = [] print(f"[DEBUG] _format_goals_behind: Processing {len(goals)} goals") for g in goals: goal_name = g.get('name') or g.get('goal_type', 'Unknown') current = g.get('current_value') target = g.get('target_value') start = g.get('start_value') start_date = g.get('start_date') target_date = g.get('target_date') print(f"[DEBUG] Goal '{goal_name}': current={current}, target={target}, start={start}, start_date={start_date}, target_date={target_date}") # Skip if missing required values if None in [current, target, start]: print(f"[DEBUG] → Skipped: Missing current/target/start") continue # Skip if no target_date (can't calculate time-based progress) if not target_date: print(f"[DEBUG] → Skipped: No target_date") continue try: current = float(current) target = float(target) start = float(start) # Calculate actual progress percentage if target == start: actual_progress_pct = 100 if current == target else 0 else: actual_progress_pct = ((current - start) / (target - start)) * 100 actual_progress_pct = max(0, min(100, actual_progress_pct)) # Calculate expected progress based on time if start_date: # Use start_date if available start_dt = start_date if isinstance(start_date, date) else date.fromisoformat(str(start_date)) else: # Fallback: assume start date = created_at date created_at = g.get('created_at') if created_at: start_dt = date.fromisoformat(str(created_at).split('T')[0]) print(f"[DEBUG] → Using created_at as start_date: {start_dt}") else: print(f"[DEBUG] → Skipped: No start_date and no created_at") continue # Can't calculate without start date target_dt = target_date if isinstance(target_date, date) else date.fromisoformat(str(target_date)) # Calculate time progress total_days = (target_dt - start_dt).days elapsed_days = (today - start_dt).days if total_days <= 0: print(f"[DEBUG] → Skipped: Invalid date range (total_days={total_days})") continue # Invalid date range expected_progress_pct = (elapsed_days / total_days) * 100 expected_progress_pct = max(0, min(100, expected_progress_pct)) # Calculate deviation (negative = behind schedule) deviation = actual_progress_pct - expected_progress_pct g['_actual_progress'] = int(actual_progress_pct) g['_expected_progress'] = int(expected_progress_pct) g['_deviation'] = int(deviation) goals_with_deviation.append(g) print(f"[DEBUG] → Added: actual={int(actual_progress_pct)}%, expected={int(expected_progress_pct)}%, deviation={int(deviation)}%") except (ValueError, ZeroDivisionError, TypeError) as e: print(f"[DEBUG] → Exception: {type(e).__name__}: {e}") continue # Also process goals WITHOUT target_date (simple progress) goals_without_date = [] print(f"[DEBUG] Processing goals without target_date for simple progress") for g in goals: if g.get('target_date'): continue # Already processed above goal_name = g.get('name') or g.get('goal_type', 'Unknown') current = g.get('current_value') target = g.get('target_value') start = g.get('start_value') if None in [current, target, start]: print(f"[DEBUG] Goal '{goal_name}' (no date): Skipped - missing values") continue try: current = float(current) target = float(target) start = float(start) if target == start: progress_pct = 100 if current == target else 0 else: progress_pct = ((current - start) / (target - start)) * 100 progress_pct = max(0, min(100, progress_pct)) g['_simple_progress'] = int(progress_pct) goals_without_date.append(g) print(f"[DEBUG] Goal '{goal_name}' (no date): Added with {int(progress_pct)}% progress") except (ValueError, ZeroDivisionError, TypeError) as e: print(f"[DEBUG] Goal '{goal_name}' (no date): Exception - {e}") continue # Combine: Goals with negative deviation + Goals without date with low progress behind_with_date = [g for g in goals_with_deviation if g['_deviation'] < 0] behind_without_date = [g for g in goals_without_date if g['_simple_progress'] < 50] print(f"[DEBUG] Behind with date: {len(behind_with_date)}, Behind without date: {len(behind_without_date)}") # Create combined list with sort keys combined = [] for g in behind_with_date: combined.append({ 'goal': g, 'sort_key': g['_deviation'], # Negative deviation (worst first) 'has_date': True }) for g in behind_without_date: # Map progress to deviation-like scale: 0% = -100, 50% = -50 combined.append({ 'goal': g, 'sort_key': g['_simple_progress'] - 100, # Convert to negative scale 'has_date': False }) if not combined: return 'Alle Ziele im Zeitplan oder erreicht' # Sort by sort_key (most negative first) sorted_combined = sorted(combined, key=lambda x: x['sort_key'])[:n] lines = [] for item in sorted_combined: g = item['goal'] name = g.get('name') or g.get('goal_type', 'Unbekannt') if item['has_date']: actual = g['_actual_progress'] expected = g['_expected_progress'] deviation = g['_deviation'] lines.append(f"{name} ({actual}% statt {expected}%, {deviation}%)") else: progress = g['_simple_progress'] lines.append(f"{name} ({progress}% erreicht)") return ', '.join(lines) except Exception as e: print(f"[ERROR] _format_goals_behind: {e}") import traceback traceback.print_exc() return 'nicht verfügbar' def _format_goals_on_track(profile_id: str, n: int = 3) -> str: """ Format top N goals ahead of schedule (based on time deviation). Compares actual progress vs. expected progress based on elapsed time. Positive deviation = ahead of schedule / on track. """ try: from goal_utils import get_active_goals from datetime import date goals = get_active_goals(profile_id) if not goals: return 'Keine Ziele definiert' today = date.today() goals_with_deviation = [] print(f"[DEBUG] _format_goals_on_track: Processing {len(goals)} goals") for g in goals: goal_name = g.get('name') or g.get('goal_type', 'Unknown') current = g.get('current_value') target = g.get('target_value') start = g.get('start_value') start_date = g.get('start_date') target_date = g.get('target_date') print(f"[DEBUG] Goal '{goal_name}': current={current}, target={target}, start={start}, start_date={start_date}, target_date={target_date}") # Skip if missing required values if None in [current, target, start]: print(f"[DEBUG] → Skipped: Missing current/target/start") continue # Skip if no target_date if not target_date: print(f"[DEBUG] → Skipped: No target_date") continue try: current = float(current) target = float(target) start = float(start) # Calculate actual progress percentage if target == start: actual_progress_pct = 100 if current == target else 0 else: actual_progress_pct = ((current - start) / (target - start)) * 100 actual_progress_pct = max(0, min(100, actual_progress_pct)) # Calculate expected progress based on time if start_date: start_dt = start_date if isinstance(start_date, date) else date.fromisoformat(str(start_date)) else: created_at = g.get('created_at') if created_at: start_dt = date.fromisoformat(str(created_at).split('T')[0]) else: continue target_dt = target_date if isinstance(target_date, date) else date.fromisoformat(str(target_date)) # Calculate time progress total_days = (target_dt - start_dt).days elapsed_days = (today - start_dt).days if total_days <= 0: continue expected_progress_pct = (elapsed_days / total_days) * 100 expected_progress_pct = max(0, min(100, expected_progress_pct)) # Calculate deviation (positive = ahead of schedule) deviation = actual_progress_pct - expected_progress_pct g['_actual_progress'] = int(actual_progress_pct) g['_expected_progress'] = int(expected_progress_pct) g['_deviation'] = int(deviation) goals_with_deviation.append(g) print(f"[DEBUG] → Added: actual={int(actual_progress_pct)}%, expected={int(expected_progress_pct)}%, deviation={int(deviation)}%") except (ValueError, ZeroDivisionError, TypeError) as e: print(f"[DEBUG] → Exception: {type(e).__name__}: {e}") continue # Also process goals WITHOUT target_date (simple progress) goals_without_date = [] print(f"[DEBUG] Processing goals without target_date for simple progress") for g in goals: if g.get('target_date'): continue # Already processed above goal_name = g.get('name') or g.get('goal_type', 'Unknown') current = g.get('current_value') target = g.get('target_value') start = g.get('start_value') if None in [current, target, start]: print(f"[DEBUG] Goal '{goal_name}' (no date): Skipped - missing values") continue try: current = float(current) target = float(target) start = float(start) if target == start: progress_pct = 100 if current == target else 0 else: progress_pct = ((current - start) / (target - start)) * 100 progress_pct = max(0, min(100, progress_pct)) g['_simple_progress'] = int(progress_pct) goals_without_date.append(g) print(f"[DEBUG] Goal '{goal_name}' (no date): Added with {int(progress_pct)}% progress") except (ValueError, ZeroDivisionError, TypeError) as e: print(f"[DEBUG] Goal '{goal_name}' (no date): Exception - {e}") continue # Combine: Goals with positive deviation + Goals without date with high progress ahead_with_date = [g for g in goals_with_deviation if g['_deviation'] >= 0] ahead_without_date = [g for g in goals_without_date if g['_simple_progress'] >= 50] print(f"[DEBUG] Ahead with date: {len(ahead_with_date)}, Ahead without date: {len(ahead_without_date)}") # Create combined list with sort keys combined = [] for g in ahead_with_date: combined.append({ 'goal': g, 'sort_key': g['_deviation'], # Positive deviation (best first) 'has_date': True }) for g in ahead_without_date: # Map progress to deviation-like scale: 50% = 0, 100% = +50 combined.append({ 'goal': g, 'sort_key': g['_simple_progress'] - 50, # Convert to positive scale 'has_date': False }) if not combined: return 'Keine Ziele im Zeitplan' # Sort by sort_key descending (most positive first) sorted_combined = sorted(combined, key=lambda x: x['sort_key'], reverse=True)[:n] lines = [] for item in sorted_combined: g = item['goal'] name = g.get('name') or g.get('goal_type', 'Unbekannt') if item['has_date']: actual = g['_actual_progress'] deviation = g['_deviation'] lines.append(f"{name} ({actual}%, +{deviation}% voraus)") else: progress = g['_simple_progress'] lines.append(f"{name} ({progress}% erreicht)") return ', '.join(lines) except Exception as e: print(f"[ERROR] _format_goals_on_track: {e}") import traceback traceback.print_exc() return 'nicht verfügbar' # ── Placeholder Registry ────────────────────────────────────────────────────── PLACEHOLDER_MAP: Dict[str, Callable[[str], str]] = { # Profil '{{name}}': lambda pid: get_profile_data(pid).get('name', 'Nutzer'), '{{age}}': lambda pid: calculate_age(get_profile_data(pid).get('dob')), '{{height}}': lambda pid: str(get_profile_data(pid).get('height', 'unbekannt')), '{{geschlecht}}': lambda pid: 'männlich' if get_profile_data(pid).get('sex') == 'm' else 'weiblich', # Körper '{{weight_aktuell}}': get_latest_weight, '{{weight_trend}}': get_weight_trend, '{{kf_aktuell}}': get_latest_bf, '{{bmi}}': lambda pid: calculate_bmi(pid), '{{caliper_summary}}': get_caliper_summary, '{{circ_summary}}': get_circ_summary, '{{goal_weight}}': get_goal_weight, '{{goal_bf_pct}}': get_goal_bf_pct, # Ernährung '{{kcal_avg}}': lambda pid: get_nutrition_avg(pid, 'kcal', 30), '{{protein_avg}}': lambda pid: get_nutrition_avg(pid, 'protein', 30), '{{carb_avg}}': lambda pid: get_nutrition_avg(pid, 'carb', 30), '{{fat_avg}}': lambda pid: get_nutrition_avg(pid, 'fat', 30), '{{nutrition_days}}': lambda pid: get_nutrition_days(pid, 30), '{{protein_ziel_low}}': get_protein_ziel_low, '{{protein_ziel_high}}': get_protein_ziel_high, # Training '{{activity_summary}}': get_activity_summary, '{{activity_detail}}': get_activity_detail, '{{trainingstyp_verteilung}}': get_trainingstyp_verteilung, # Schlaf & Erholung '{{sleep_avg_duration}}': lambda pid: get_sleep_avg_duration(pid, 7), '{{sleep_avg_quality}}': lambda pid: get_sleep_avg_quality(pid, 7), '{{rest_days_count}}': lambda pid: get_rest_days_count(pid, 30), # Vitalwerte '{{vitals_avg_hr}}': lambda pid: get_vitals_avg_hr(pid, 7), '{{vitals_avg_hrv}}': lambda pid: get_vitals_avg_hrv(pid, 7), '{{vitals_vo2_max}}': get_vitals_vo2_max, # Zeitraum '{{datum_heute}}': lambda pid: datetime.now().strftime('%d.%m.%Y'), '{{zeitraum_7d}}': lambda pid: 'letzte 7 Tage', '{{zeitraum_30d}}': lambda pid: 'letzte 30 Tage', '{{zeitraum_90d}}': lambda pid: 'letzte 90 Tage', # ======================================================================== # PHASE 0b: Goal-Aware Placeholders (Dynamic Focus Areas v2.0) # ======================================================================== # --- Meta Scores (Ebene 1: Aggregierte Scores) --- '{{goal_progress_score}}': lambda pid: _safe_int('goal_progress_score', pid), '{{body_progress_score}}': lambda pid: _safe_int('body_progress_score', pid), '{{nutrition_score}}': lambda pid: _safe_int('nutrition_score', pid), '{{activity_score}}': lambda pid: _safe_int('activity_score', pid), '{{recovery_score}}': lambda pid: _safe_int('recovery_score_v2', pid), '{{data_quality_score}}': lambda pid: _safe_int('data_quality_score', pid), # --- Top-Weighted Goals/Focus Areas (Ebene 2: statt Primary) --- '{{top_goal_name}}': lambda pid: _safe_str('top_goal_name', pid), '{{top_goal_progress_pct}}': lambda pid: _safe_str('top_goal_progress_pct', pid), '{{top_goal_status}}': lambda pid: _safe_str('top_goal_status', pid), '{{top_focus_area_name}}': lambda pid: _safe_str('top_focus_area_name', pid), '{{top_focus_area_progress}}': lambda pid: _safe_int('top_focus_area_progress', pid), # --- Category Scores (Ebene 3: 7 Kategorien) --- '{{focus_cat_körper_progress}}': lambda pid: _safe_int('focus_cat_körper_progress', pid), '{{focus_cat_körper_weight}}': lambda pid: _safe_float('focus_cat_körper_weight', pid), '{{focus_cat_ernährung_progress}}': lambda pid: _safe_int('focus_cat_ernährung_progress', pid), '{{focus_cat_ernährung_weight}}': lambda pid: _safe_float('focus_cat_ernährung_weight', pid), '{{focus_cat_aktivität_progress}}': lambda pid: _safe_int('focus_cat_aktivität_progress', pid), '{{focus_cat_aktivität_weight}}': lambda pid: _safe_float('focus_cat_aktivität_weight', pid), '{{focus_cat_recovery_progress}}': lambda pid: _safe_int('focus_cat_recovery_progress', pid), '{{focus_cat_recovery_weight}}': lambda pid: _safe_float('focus_cat_recovery_weight', pid), '{{focus_cat_vitalwerte_progress}}': lambda pid: _safe_int('focus_cat_vitalwerte_progress', pid), '{{focus_cat_vitalwerte_weight}}': lambda pid: _safe_float('focus_cat_vitalwerte_weight', pid), '{{focus_cat_mental_progress}}': lambda pid: _safe_int('focus_cat_mental_progress', pid), '{{focus_cat_mental_weight}}': lambda pid: _safe_float('focus_cat_mental_weight', pid), '{{focus_cat_lebensstil_progress}}': lambda pid: _safe_int('focus_cat_lebensstil_progress', pid), '{{focus_cat_lebensstil_weight}}': lambda pid: _safe_float('focus_cat_lebensstil_weight', pid), # --- Body Metrics (Ebene 4: Einzelmetriken K1-K5) --- '{{weight_7d_median}}': lambda pid: _safe_float('weight_7d_median', pid), '{{weight_28d_slope}}': lambda pid: _safe_float('weight_28d_slope', pid, decimals=4), '{{weight_90d_slope}}': lambda pid: _safe_float('weight_90d_slope', pid, decimals=4), '{{fm_28d_change}}': lambda pid: _safe_float('fm_28d_change', pid), '{{lbm_28d_change}}': lambda pid: _safe_float('lbm_28d_change', pid), '{{waist_28d_delta}}': lambda pid: _safe_float('waist_28d_delta', pid), '{{hip_28d_delta}}': lambda pid: _safe_float('hip_28d_delta', pid), '{{chest_28d_delta}}': lambda pid: _safe_float('chest_28d_delta', pid), '{{arm_28d_delta}}': lambda pid: _safe_float('arm_28d_delta', pid), '{{thigh_28d_delta}}': lambda pid: _safe_float('thigh_28d_delta', pid), '{{waist_hip_ratio}}': lambda pid: _safe_float('waist_hip_ratio', pid, decimals=3), '{{recomposition_quadrant}}': lambda pid: _safe_str('recomposition_quadrant', pid), # --- Nutrition Metrics (E1-E5) --- '{{energy_balance_7d}}': lambda pid: _safe_float('energy_balance_7d', pid, decimals=0), '{{energy_deficit_surplus}}': lambda pid: _safe_str('energy_deficit_surplus', pid), '{{protein_g_per_kg}}': lambda pid: _safe_float('protein_g_per_kg', pid), '{{protein_days_in_target}}': lambda pid: _safe_str('protein_days_in_target', pid), '{{protein_adequacy_28d}}': lambda pid: _safe_int('protein_adequacy_28d', pid), '{{macro_consistency_score}}': lambda pid: _safe_int('macro_consistency_score', pid), '{{intake_volatility}}': lambda pid: _safe_str('intake_volatility', pid), # --- Activity Metrics (A1-A8) --- '{{training_minutes_week}}': lambda pid: _safe_int('training_minutes_week', pid), '{{training_frequency_7d}}': lambda pid: _safe_int('training_frequency_7d', pid), '{{quality_sessions_pct}}': lambda pid: _safe_int('quality_sessions_pct', pid), '{{ability_balance_strength}}': lambda pid: _safe_int('ability_balance_strength', pid), '{{ability_balance_endurance}}': lambda pid: _safe_int('ability_balance_endurance', pid), '{{ability_balance_mental}}': lambda pid: _safe_int('ability_balance_mental', pid), '{{ability_balance_coordination}}': lambda pid: _safe_int('ability_balance_coordination', pid), '{{ability_balance_mobility}}': lambda pid: _safe_int('ability_balance_mobility', pid), '{{proxy_internal_load_7d}}': lambda pid: _safe_int('proxy_internal_load_7d', pid), '{{monotony_score}}': lambda pid: _safe_float('monotony_score', pid), '{{strain_score}}': lambda pid: _safe_int('strain_score', pid), '{{rest_day_compliance}}': lambda pid: _safe_int('rest_day_compliance', pid), '{{vo2max_trend_28d}}': lambda pid: _safe_float('vo2max_trend_28d', pid), # --- Recovery Metrics (Recovery Score v2) --- '{{hrv_vs_baseline_pct}}': lambda pid: _safe_float('hrv_vs_baseline_pct', pid), '{{rhr_vs_baseline_pct}}': lambda pid: _safe_float('rhr_vs_baseline_pct', pid), '{{sleep_avg_duration_7d}}': lambda pid: _safe_float('sleep_avg_duration_7d', pid), '{{sleep_debt_hours}}': lambda pid: _safe_float('sleep_debt_hours', pid), '{{sleep_regularity_proxy}}': lambda pid: _safe_float('sleep_regularity_proxy', pid), '{{recent_load_balance_3d}}': lambda pid: _safe_int('recent_load_balance_3d', pid), '{{sleep_quality_7d}}': lambda pid: _safe_int('sleep_quality_7d', pid), # --- Correlation Metrics (C1-C7) --- '{{correlation_energy_weight_lag}}': lambda pid: _safe_json('correlation_energy_weight_lag', pid), '{{correlation_protein_lbm}}': lambda pid: _safe_json('correlation_protein_lbm', pid), '{{correlation_load_hrv}}': lambda pid: _safe_json('correlation_load_hrv', pid), '{{correlation_load_rhr}}': lambda pid: _safe_json('correlation_load_rhr', pid), '{{correlation_sleep_recovery}}': lambda pid: _safe_json('correlation_sleep_recovery', pid), '{{plateau_detected}}': lambda pid: _safe_json('plateau_detected', pid), '{{top_drivers}}': lambda pid: _safe_json('top_drivers', pid), # --- JSON/Markdown Structured Data (Ebene 5) --- '{{active_goals_json}}': lambda pid: _safe_json('active_goals_json', pid), '{{active_goals_md}}': lambda pid: _safe_str('active_goals_md', pid), '{{focus_areas_weighted_json}}': lambda pid: _safe_json('focus_areas_weighted_json', pid), '{{focus_areas_weighted_md}}': lambda pid: _safe_str('focus_areas_weighted_md', pid), '{{focus_area_weights_json}}': lambda pid: _safe_json('focus_area_weights_json', pid), '{{top_3_focus_areas}}': lambda pid: _safe_str('top_3_focus_areas', pid), '{{top_3_goals_behind_schedule}}': lambda pid: _safe_str('top_3_goals_behind_schedule', pid), '{{top_3_goals_on_track}}': lambda pid: _safe_str('top_3_goals_on_track', pid), } def calculate_bmi(profile_id: str) -> str: """Calculate BMI from latest weight and profile height.""" profile = get_profile_data(profile_id) if not profile.get('height'): return "nicht verfügbar" with get_db() as conn: cur = get_cursor(conn) cur.execute( "SELECT weight FROM weight_log WHERE profile_id=%s ORDER BY date DESC LIMIT 1", (profile_id,) ) row = cur.fetchone() if not row: return "nicht verfügbar" height_m = profile['height'] / 100 bmi = row['weight'] / (height_m ** 2) return f"{bmi:.1f}" # ── Public API ──────────────────────────────────────────────────────────────── def resolve_placeholders(template: str, profile_id: str) -> str: """ Replace all {{placeholders}} in template with actual user data. Args: template: Prompt template with placeholders profile_id: User profile ID Returns: Resolved template with placeholders replaced by values """ result = template for placeholder, resolver in PLACEHOLDER_MAP.items(): if placeholder in result: try: value = resolver(profile_id) result = result.replace(placeholder, str(value)) except Exception as e: # On error, replace with error message result = result.replace(placeholder, f"[Fehler: {placeholder}]") return result def get_unknown_placeholders(template: str) -> List[str]: """ Find all placeholders in template that are not in PLACEHOLDER_MAP. Args: template: Prompt template Returns: List of unknown placeholder names (without {{}}) """ # Find all {{...}} patterns found = re.findall(r'\{\{(\w+)\}\}', template) # Filter to only unknown ones known_names = {p.strip('{}') for p in PLACEHOLDER_MAP.keys()} unknown = [p for p in found if p not in known_names] return list(set(unknown)) # Remove duplicates def get_available_placeholders(categories: Optional[List[str]] = None) -> Dict[str, List[str]]: """ Get available placeholders, optionally filtered by categories. Args: categories: Optional list of categories to filter (körper, ernährung, training, etc.) Returns: Dict mapping category to list of placeholders """ placeholder_categories = { 'profil': [ '{{name}}', '{{age}}', '{{height}}', '{{geschlecht}}' ], 'körper': [ '{{weight_aktuell}}', '{{weight_trend}}', '{{kf_aktuell}}', '{{bmi}}' ], 'ernährung': [ '{{kcal_avg}}', '{{protein_avg}}', '{{carb_avg}}', '{{fat_avg}}' ], 'training': [ '{{activity_summary}}', '{{trainingstyp_verteilung}}' ], 'zeitraum': [ '{{datum_heute}}', '{{zeitraum_7d}}', '{{zeitraum_30d}}', '{{zeitraum_90d}}' ] } if not categories: return placeholder_categories # Filter to requested categories return {k: v for k, v in placeholder_categories.items() if k in categories} def get_placeholder_example_values(profile_id: str) -> Dict[str, str]: """ Get example values for all placeholders using real user data. Args: profile_id: User profile ID Returns: Dict mapping placeholder to example value """ examples = {} for placeholder, resolver in PLACEHOLDER_MAP.items(): try: examples[placeholder] = resolver(profile_id) except Exception as e: examples[placeholder] = f"[Fehler: {str(e)}]" return examples def get_placeholder_catalog(profile_id: str) -> Dict[str, List[Dict[str, str]]]: """ Get grouped placeholder catalog with descriptions and example values. Args: profile_id: User profile ID Returns: Dict mapping category to list of {key, description, example} """ # Placeholder definitions with descriptions placeholders = { 'Profil': [ ('name', 'Name des Nutzers'), ('age', 'Alter in Jahren'), ('height', 'Körpergröße in cm'), ('geschlecht', 'Geschlecht'), ], 'Körper': [ ('weight_aktuell', 'Aktuelles Gewicht in kg'), ('weight_trend', 'Gewichtstrend (7d/30d)'), ('kf_aktuell', 'Aktueller Körperfettanteil in %'), ('bmi', 'Body Mass Index'), ('weight_7d_median', 'Gewicht 7d Median (kg)'), ('weight_28d_slope', 'Gewichtstrend 28d (kg/Tag)'), ('fm_28d_change', 'Fettmasse Änderung 28d (kg)'), ('lbm_28d_change', 'Magermasse Änderung 28d (kg)'), ('waist_28d_delta', 'Taillenumfang Änderung 28d (cm)'), ('waist_hip_ratio', 'Taille/Hüfte-Verhältnis'), ('recomposition_quadrant', 'Rekomposition-Status'), ], 'Ernährung': [ ('kcal_avg', 'Durchschn. Kalorien (30d)'), ('protein_avg', 'Durchschn. Protein in g (30d)'), ('carb_avg', 'Durchschn. Kohlenhydrate in g (30d)'), ('fat_avg', 'Durchschn. Fett in g (30d)'), ('energy_balance_7d', 'Energiebilanz 7d (kcal/Tag)'), ('protein_g_per_kg', 'Protein g/kg Körpergewicht'), ('protein_adequacy_28d', 'Protein Adequacy Score (0-100)'), ('macro_consistency_score', 'Makro-Konsistenz Score (0-100)'), ], 'Training': [ ('activity_summary', 'Aktivitäts-Zusammenfassung (7d)'), ('trainingstyp_verteilung', 'Verteilung nach Trainingstypen'), ('training_minutes_week', 'Trainingsminuten pro Woche'), ('training_frequency_7d', 'Trainingshäufigkeit 7d'), ('quality_sessions_pct', 'Qualitätssessions (%)'), ('ability_balance_strength', 'Ability Balance - Kraft (0-100)'), ('ability_balance_endurance', 'Ability Balance - Ausdauer (0-100)'), ('proxy_internal_load_7d', 'Proxy Load 7d'), ('rest_day_compliance', 'Ruhetags-Compliance (%)'), ], 'Schlaf & Erholung': [ ('sleep_avg_duration', 'Durchschn. Schlafdauer (7d)'), ('sleep_avg_quality', 'Durchschn. Schlafqualität (7d)'), ('rest_days_count', 'Anzahl Ruhetage (30d)'), ('sleep_avg_duration_7d', 'Schlaf 7d (Stunden)'), ('sleep_debt_hours', 'Schlafschuld (Stunden)'), ('sleep_regularity_proxy', 'Schlaf-Regelmäßigkeit (Min Abweichung)'), ('sleep_quality_7d', 'Schlafqualität 7d (0-100)'), ], 'Vitalwerte': [ ('vitals_avg_hr', 'Durchschn. Ruhepuls (7d)'), ('vitals_avg_hrv', 'Durchschn. HRV (7d)'), ('vitals_vo2_max', 'Aktueller VO2 Max'), ('hrv_vs_baseline_pct', 'HRV vs. Baseline (%)'), ('rhr_vs_baseline_pct', 'RHR vs. Baseline (%)'), ('vo2max_trend_28d', 'VO2max Trend 28d'), ], 'Scores (Phase 0b)': [ ('goal_progress_score', 'Goal Progress Score (0-100)'), ('body_progress_score', 'Body Progress Score (0-100)'), ('nutrition_score', 'Nutrition Score (0-100)'), ('activity_score', 'Activity Score (0-100)'), ('recovery_score', 'Recovery Score (0-100)'), ('data_quality_score', 'Data Quality Score (0-100)'), ], 'Focus Areas': [ ('top_focus_area_name', 'Top Focus Area Name'), ('top_focus_area_progress', 'Top Focus Area Progress (%)'), ('focus_cat_körper_progress', 'Kategorie Körper - Progress (%)'), ('focus_cat_körper_weight', 'Kategorie Körper - Gewichtung (%)'), ('focus_cat_ernährung_progress', 'Kategorie Ernährung - Progress (%)'), ('focus_cat_ernährung_weight', 'Kategorie Ernährung - Gewichtung (%)'), ('focus_cat_aktivität_progress', 'Kategorie Aktivität - Progress (%)'), ('focus_cat_aktivität_weight', 'Kategorie Aktivität - Gewichtung (%)'), ], 'Zeitraum': [ ('datum_heute', 'Heutiges Datum'), ('zeitraum_7d', '7-Tage-Zeitraum'), ('zeitraum_30d', '30-Tage-Zeitraum'), ], } catalog = {} for category, items in placeholders.items(): catalog[category] = [] for key, description in items: placeholder = f'{{{{{key}}}}}' # Get example value if resolver exists resolver = PLACEHOLDER_MAP.get(placeholder) if resolver: try: example = resolver(profile_id) except Exception: example = '[Nicht verfügbar]' else: example = '[Nicht implementiert]' catalog[category].append({ 'key': key, 'description': description, 'example': str(example) }) return catalog