""" Goals Router - Goal System (Strategic + Tactical) Endpoints for managing: - Strategic goal modes (weight_loss, strength, etc.) - Tactical goal targets (concrete values with deadlines) - Training phase detection - Fitness tests Part of v9e Goal System implementation. """ from fastapi import APIRouter, Depends, HTTPException from pydantic import BaseModel from typing import Optional, List from datetime import date, datetime, timedelta from decimal import Decimal import traceback from db import get_db, get_cursor, r2d from auth import require_auth from goal_utils import get_current_value_for_goal router = APIRouter(prefix="/api/goals", tags=["goals"]) # ============================================================================ # Pydantic Models # ============================================================================ class GoalModeUpdate(BaseModel): """Update strategic goal mode (deprecated - use FocusAreasUpdate)""" goal_mode: str # weight_loss, strength, endurance, recomposition, health class FocusAreasUpdate(BaseModel): """Update focus area weights (v2.0)""" weight_loss_pct: int muscle_gain_pct: int strength_pct: int endurance_pct: int flexibility_pct: int health_pct: int class FocusContribution(BaseModel): """Focus area contribution (v2.0)""" focus_area_id: str contribution_weight: float = 100.0 # 0-100% class GoalCreate(BaseModel): """Create or update a concrete goal""" goal_type: str # weight, body_fat, lean_mass, vo2max, strength, flexibility, bp, rhr is_primary: bool = False # Kept for backward compatibility target_value: float unit: str # kg, %, ml/kg/min, bpm, mmHg, cm, reps target_date: Optional[date] = None category: Optional[str] = 'other' # body, training, nutrition, recovery, health, other priority: Optional[int] = 2 # 1=high, 2=medium, 3=low name: Optional[str] = None description: Optional[str] = None focus_contributions: Optional[List[FocusContribution]] = [] # v2.0: Many-to-Many class GoalUpdate(BaseModel): """Update existing goal""" target_value: Optional[float] = None target_date: Optional[date] = None status: Optional[str] = None # active, reached, abandoned, expired is_primary: Optional[bool] = None # Kept for backward compatibility category: Optional[str] = None # body, training, nutrition, recovery, health, other priority: Optional[int] = None # 1=high, 2=medium, 3=low name: Optional[str] = None description: Optional[str] = None focus_contributions: Optional[List[FocusContribution]] = None # v2.0: Many-to-Many class TrainingPhaseCreate(BaseModel): """Create training phase (manual or auto-detected)""" phase_type: str # calorie_deficit, calorie_surplus, deload, maintenance, periodization start_date: date end_date: Optional[date] = None notes: Optional[str] = None class FitnessTestCreate(BaseModel): """Record fitness test result""" test_type: str result_value: float result_unit: str test_date: date test_conditions: Optional[str] = None class GoalProgressCreate(BaseModel): """Log progress for a goal""" date: date value: float note: Optional[str] = None class GoalProgressUpdate(BaseModel): """Update progress entry""" value: Optional[float] = None note: Optional[str] = None class GoalTypeCreate(BaseModel): """Create custom goal type definition""" type_key: str label_de: str label_en: Optional[str] = None unit: str icon: Optional[str] = None category: Optional[str] = 'custom' source_table: Optional[str] = None source_column: Optional[str] = None aggregation_method: Optional[str] = 'latest' calculation_formula: Optional[str] = None filter_conditions: Optional[dict] = None description: Optional[str] = None class GoalTypeUpdate(BaseModel): """Update goal type definition""" label_de: Optional[str] = None label_en: Optional[str] = None unit: Optional[str] = None icon: Optional[str] = None category: Optional[str] = None source_table: Optional[str] = None source_column: Optional[str] = None aggregation_method: Optional[str] = None calculation_formula: Optional[str] = None filter_conditions: Optional[dict] = None description: Optional[str] = None is_active: Optional[bool] = None # ============================================================================ # Strategic Layer: Goal Modes # ============================================================================ @router.get("/mode") def get_goal_mode(session: dict = Depends(require_auth)): """Get user's current strategic goal mode""" pid = session['profile_id'] with get_db() as conn: cur = get_cursor(conn) cur.execute( "SELECT goal_mode FROM profiles WHERE id = %s", (pid,) ) row = cur.fetchone() if not row: raise HTTPException(status_code=404, detail="Profil nicht gefunden") return { "goal_mode": row['goal_mode'] or 'health', "description": _get_goal_mode_description(row['goal_mode'] or 'health') } @router.put("/mode") def update_goal_mode(data: GoalModeUpdate, session: dict = Depends(require_auth)): """Update user's strategic goal mode""" pid = session['profile_id'] # Validate goal mode valid_modes = ['weight_loss', 'strength', 'endurance', 'recomposition', 'health'] if data.goal_mode not in valid_modes: raise HTTPException( status_code=400, detail=f"Ungültiger Goal Mode. Erlaubt: {', '.join(valid_modes)}" ) with get_db() as conn: cur = get_cursor(conn) cur.execute( "UPDATE profiles SET goal_mode = %s WHERE id = %s", (data.goal_mode, pid) ) return { "goal_mode": data.goal_mode, "description": _get_goal_mode_description(data.goal_mode) } def _get_goal_mode_description(mode: str) -> str: """Get description for goal mode""" descriptions = { 'weight_loss': 'Gewichtsreduktion (Kaloriendefizit, Fettabbau)', 'strength': 'Kraftaufbau (Muskelwachstum, progressive Belastung)', 'endurance': 'Ausdauer (VO2Max, aerobe Kapazität)', 'recomposition': 'Körperkomposition (gleichzeitig Fett ab- und Muskeln aufbauen)', 'health': 'Allgemeine Gesundheit (ausgewogen, präventiv)' } return descriptions.get(mode, 'Unbekannt') # ============================================================================ # Focus Areas (v2.0): Weighted Multi-Goal System # ============================================================================ @router.get("/focus-areas") def get_focus_areas(session: dict = Depends(require_auth)): """ Get current focus area weights. Returns custom weights if set, otherwise derives from goal_mode. """ pid = session['profile_id'] with get_db() as conn: cur = get_cursor(conn) # Try to get custom focus areas (user_focus_preferences after Migration 031) try: cur.execute(""" SELECT weight_loss_pct, muscle_gain_pct, strength_pct, endurance_pct, flexibility_pct, health_pct, created_at, updated_at FROM user_focus_preferences WHERE profile_id = %s LIMIT 1 """, (pid,)) row = cur.fetchone() except Exception as e: # Migration 031 not applied yet, try old table name print(f"[WARNING] user_focus_preferences not found, trying old focus_areas: {e}") try: cur.execute(""" SELECT weight_loss_pct, muscle_gain_pct, strength_pct, endurance_pct, flexibility_pct, health_pct, created_at, updated_at FROM focus_areas WHERE profile_id = %s AND active = true LIMIT 1 """, (pid,)) row = cur.fetchone() except: row = None if row: return { "custom": True, "weight_loss_pct": row['weight_loss_pct'], "muscle_gain_pct": row['muscle_gain_pct'], "strength_pct": row['strength_pct'], "endurance_pct": row['endurance_pct'], "flexibility_pct": row['flexibility_pct'], "health_pct": row['health_pct'], "updated_at": row['updated_at'] } # Fallback: Derive from goal_mode cur.execute("SELECT goal_mode FROM profiles WHERE id = %s", (pid,)) profile = cur.fetchone() if not profile or not profile['goal_mode']: # Default balanced health return { "custom": False, "weight_loss_pct": 0, "muscle_gain_pct": 0, "strength_pct": 10, "endurance_pct": 20, "flexibility_pct": 15, "health_pct": 55, "source": "default" } # Derive from goal_mode (using same logic as migration) mode = profile['goal_mode'] mode_mappings = { 'weight_loss': { 'weight_loss_pct': 60, 'muscle_gain_pct': 0, 'strength_pct': 10, 'endurance_pct': 20, 'flexibility_pct': 5, 'health_pct': 5 }, 'strength': { 'weight_loss_pct': 0, 'muscle_gain_pct': 40, 'strength_pct': 50, 'endurance_pct': 10, 'flexibility_pct': 0, 'health_pct': 0 }, 'endurance': { 'weight_loss_pct': 0, 'muscle_gain_pct': 0, 'strength_pct': 0, 'endurance_pct': 70, 'flexibility_pct': 10, 'health_pct': 20 }, 'recomposition': { 'weight_loss_pct': 30, 'muscle_gain_pct': 30, 'strength_pct': 25, 'endurance_pct': 10, 'flexibility_pct': 5, 'health_pct': 0 }, 'health': { 'weight_loss_pct': 0, 'muscle_gain_pct': 0, 'strength_pct': 10, 'endurance_pct': 20, 'flexibility_pct': 15, 'health_pct': 55 } } mapping = mode_mappings.get(mode, mode_mappings['health']) mapping['custom'] = False mapping['source'] = f"goal_mode:{mode}" return mapping @router.put("/focus-areas") def update_focus_areas(data: FocusAreasUpdate, session: dict = Depends(require_auth)): """ Update focus area weights (upsert). Validates that sum = 100 and all values are 0-100. """ pid = session['profile_id'] # Validate sum = 100 total = ( data.weight_loss_pct + data.muscle_gain_pct + data.strength_pct + data.endurance_pct + data.flexibility_pct + data.health_pct ) if total != 100: raise HTTPException( status_code=400, detail=f"Summe muss 100% sein (aktuell: {total}%)" ) # Validate range 0-100 values = [ data.weight_loss_pct, data.muscle_gain_pct, data.strength_pct, data.endurance_pct, data.flexibility_pct, data.health_pct ] if any(v < 0 or v > 100 for v in values): raise HTTPException( status_code=400, detail="Alle Werte müssen zwischen 0 und 100 liegen" ) with get_db() as conn: cur = get_cursor(conn) # Deactivate old focus_areas cur.execute( "UPDATE focus_areas SET active = false WHERE profile_id = %s", (pid,) ) # Insert new focus_areas cur.execute(""" INSERT INTO focus_areas ( profile_id, weight_loss_pct, muscle_gain_pct, strength_pct, endurance_pct, flexibility_pct, health_pct ) VALUES (%s, %s, %s, %s, %s, %s, %s) RETURNING id """, ( pid, data.weight_loss_pct, data.muscle_gain_pct, data.strength_pct, data.endurance_pct, data.flexibility_pct, data.health_pct )) return { "message": "Fokus-Bereiche aktualisiert", "weight_loss_pct": data.weight_loss_pct, "muscle_gain_pct": data.muscle_gain_pct, "strength_pct": data.strength_pct, "endurance_pct": data.endurance_pct, "flexibility_pct": data.flexibility_pct, "health_pct": data.health_pct } # ============================================================================ # Tactical Layer: Concrete Goals # ============================================================================ @router.get("/list") def list_goals(session: dict = Depends(require_auth)): """List all goals for current user""" pid = session['profile_id'] try: with get_db() as conn: cur = get_cursor(conn) cur.execute(""" SELECT id, goal_type, is_primary, status, target_value, current_value, start_value, unit, start_date, target_date, reached_date, name, description, progress_pct, projection_date, on_track, created_at, updated_at FROM goals WHERE profile_id = %s ORDER BY is_primary DESC, created_at DESC """, (pid,)) goals = [r2d(row) for row in cur.fetchall()] print(f"[DEBUG] Loaded {len(goals)} goals for profile {pid}") # Update current values for each goal for goal in goals: try: _update_goal_progress(conn, pid, goal) except Exception as e: print(f"[ERROR] Failed to update progress for goal {goal.get('id')}: {e}") # Continue with other goals even if one fails return goals except Exception as e: print(f"[ERROR] list_goals failed: {e}") import traceback traceback.print_exc() raise HTTPException( status_code=500, detail=f"Fehler beim Laden der Ziele: {str(e)}" ) @router.post("/create") def create_goal(data: GoalCreate, session: dict = Depends(require_auth)): """Create new goal""" pid = session['profile_id'] with get_db() as conn: cur = get_cursor(conn) # If this is set as primary, unset other primary goals if data.is_primary: cur.execute( "UPDATE goals SET is_primary = false WHERE profile_id = %s", (pid,) ) # Get current value for this goal type current_value = _get_current_value_for_goal_type(conn, pid, data.goal_type) # Insert goal cur.execute(""" INSERT INTO goals ( profile_id, goal_type, is_primary, target_value, current_value, start_value, unit, target_date, category, priority, name, description ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) RETURNING id """, ( pid, data.goal_type, data.is_primary, data.target_value, current_value, current_value, data.unit, data.target_date, data.category, data.priority, data.name, data.description )) goal_id = cur.fetchone()['id'] # v2.0: Insert focus area contributions if data.focus_contributions: for contrib in data.focus_contributions: cur.execute(""" INSERT INTO goal_focus_contributions (goal_id, focus_area_id, contribution_weight) VALUES (%s, %s, %s) ON CONFLICT (goal_id, focus_area_id) DO UPDATE SET contribution_weight = EXCLUDED.contribution_weight """, (goal_id, contrib.focus_area_id, contrib.contribution_weight)) return {"id": goal_id, "message": "Ziel erstellt"} @router.put("/{goal_id}") def update_goal(goal_id: str, data: GoalUpdate, session: dict = Depends(require_auth)): """Update existing goal""" pid = session['profile_id'] with get_db() as conn: cur = get_cursor(conn) # Verify ownership cur.execute( "SELECT id FROM goals WHERE id = %s AND profile_id = %s", (goal_id, pid) ) if not cur.fetchone(): raise HTTPException(status_code=404, detail="Ziel nicht gefunden") # If setting this goal as primary, unset all other primary goals if data.is_primary is True: cur.execute( "UPDATE goals SET is_primary = false WHERE profile_id = %s AND id != %s", (pid, goal_id) ) # Build update query dynamically updates = [] params = [] if data.target_value is not None: updates.append("target_value = %s") params.append(data.target_value) if data.target_date is not None: updates.append("target_date = %s") params.append(data.target_date) if data.status is not None: updates.append("status = %s") params.append(data.status) if data.status == 'reached': updates.append("reached_date = CURRENT_DATE") if data.is_primary is not None: updates.append("is_primary = %s") params.append(data.is_primary) if data.category is not None: updates.append("category = %s") params.append(data.category) if data.priority is not None: updates.append("priority = %s") params.append(data.priority) if data.name is not None: updates.append("name = %s") params.append(data.name) if data.description is not None: updates.append("description = %s") params.append(data.description) # Handle focus_contributions separately (can be updated even if no other changes) if data.focus_contributions is not None: # Delete existing contributions cur.execute( "DELETE FROM goal_focus_contributions WHERE goal_id = %s", (goal_id,) ) # Insert new contributions for contrib in data.focus_contributions: cur.execute(""" INSERT INTO goal_focus_contributions (goal_id, focus_area_id, contribution_weight) VALUES (%s, %s, %s) """, (goal_id, contrib.focus_area_id, contrib.contribution_weight)) if not updates and data.focus_contributions is None: raise HTTPException(status_code=400, detail="Keine Änderungen angegeben") if updates: updates.append("updated_at = NOW()") params.extend([goal_id, pid]) cur.execute( f"UPDATE goals SET {', '.join(updates)} WHERE id = %s AND profile_id = %s", tuple(params) ) return {"message": "Ziel aktualisiert"} @router.delete("/{goal_id}") def delete_goal(goal_id: str, session: dict = Depends(require_auth)): """Delete goal""" pid = session['profile_id'] with get_db() as conn: cur = get_cursor(conn) cur.execute( "DELETE FROM goals WHERE id = %s AND profile_id = %s", (goal_id, pid) ) if cur.rowcount == 0: raise HTTPException(status_code=404, detail="Ziel nicht gefunden") return {"message": "Ziel gelöscht"} # ============================================================================ # Goal Progress Endpoints # ============================================================================ @router.get("/{goal_id}/progress") def get_goal_progress(goal_id: str, session: dict = Depends(require_auth)): """Get progress history for a goal""" pid = session['profile_id'] with get_db() as conn: cur = get_cursor(conn) # Verify ownership cur.execute( "SELECT id FROM goals WHERE id = %s AND profile_id = %s", (goal_id, pid) ) if not cur.fetchone(): raise HTTPException(status_code=404, detail="Ziel nicht gefunden") # Get progress entries cur.execute(""" SELECT id, date, value, note, source, created_at FROM goal_progress_log WHERE goal_id = %s ORDER BY date DESC """, (goal_id,)) entries = cur.fetchall() return [r2d(e) for e in entries] @router.post("/{goal_id}/progress") def create_goal_progress(goal_id: str, data: GoalProgressCreate, session: dict = Depends(require_auth)): """Log new progress for a goal""" pid = session['profile_id'] with get_db() as conn: cur = get_cursor(conn) # Verify ownership and check if manual entry is allowed cur.execute(""" SELECT g.id, g.unit, gt.source_table FROM goals g LEFT JOIN goal_type_definitions gt ON g.goal_type = gt.type_key WHERE g.id = %s AND g.profile_id = %s """, (goal_id, pid)) goal = cur.fetchone() if not goal: raise HTTPException(status_code=404, detail="Ziel nicht gefunden") # Prevent manual entries for goals with automatic data sources if goal['source_table']: raise HTTPException( status_code=400, detail=f"Manuelle Einträge nicht erlaubt für automatisch erfasste Ziele. " f"Bitte nutze die entsprechende Erfassungsseite (z.B. Gewicht, Aktivität)." ) # Insert progress entry try: cur.execute(""" INSERT INTO goal_progress_log (goal_id, profile_id, date, value, note, source) VALUES (%s, %s, %s, %s, %s, 'manual') RETURNING id """, (goal_id, pid, data.date, data.value, data.note)) progress_id = cur.fetchone()['id'] # Trigger will auto-update goals.current_value return { "id": progress_id, "message": f"Fortschritt erfasst: {data.value} {goal['unit']}" } except Exception as e: if "unique_progress_per_day" in str(e): raise HTTPException( status_code=400, detail=f"Für {data.date} existiert bereits ein Eintrag. Bitte bearbeite den existierenden Eintrag." ) raise HTTPException(status_code=500, detail=f"Fehler beim Speichern: {str(e)}") @router.delete("/{goal_id}/progress/{progress_id}") def delete_goal_progress(goal_id: str, progress_id: str, session: dict = Depends(require_auth)): """Delete progress entry""" pid = session['profile_id'] with get_db() as conn: cur = get_cursor(conn) # Verify ownership cur.execute( "SELECT id FROM goals WHERE id = %s AND profile_id = %s", (goal_id, pid) ) if not cur.fetchone(): raise HTTPException(status_code=404, detail="Ziel nicht gefunden") # Delete progress entry cur.execute( "DELETE FROM goal_progress_log WHERE id = %s AND goal_id = %s AND profile_id = %s", (progress_id, goal_id, pid) ) if cur.rowcount == 0: raise HTTPException(status_code=404, detail="Progress-Eintrag nicht gefunden") # After deletion, recalculate current_value from remaining entries cur.execute(""" UPDATE goals SET current_value = ( SELECT value FROM goal_progress_log WHERE goal_id = %s ORDER BY date DESC LIMIT 1 ) WHERE id = %s """, (goal_id, goal_id)) return {"message": "Progress-Eintrag gelöscht"} @router.get("/grouped") def get_goals_grouped(session: dict = Depends(require_auth)): """ Get goals grouped by category, sorted by priority. Returns structure: { "body": [{"id": "...", "goal_type": "weight", "priority": 1, ...}, ...], "training": [...], "nutrition": [...], "recovery": [...], "health": [...], "other": [...] } """ pid = session['profile_id'] with get_db() as conn: cur = get_cursor(conn) # Get all active goals with type definitions cur.execute(""" SELECT g.id, g.goal_type, g.target_value, g.current_value, g.start_value, g.unit, g.target_date, g.status, g.is_primary, g.category, g.priority, g.name, g.description, g.progress_pct, g.on_track, g.projection_date, g.created_at, g.updated_at, gt.label_de, gt.icon, gt.category as type_category, gt.source_table, gt.source_column FROM goals g LEFT JOIN goal_type_definitions gt ON g.goal_type = gt.type_key WHERE g.profile_id = %s ORDER BY g.category, g.priority ASC, g.created_at DESC """, (pid,)) goals = cur.fetchall() # v2.0: Load focus_contributions for each goal goal_ids = [g['id'] for g in goals] focus_map = {} # goal_id → [contributions] if goal_ids: try: placeholders = ','.join(['%s'] * len(goal_ids)) cur.execute(f""" SELECT gfc.goal_id, gfc.contribution_weight, fa.id as focus_area_id, fa.key, fa.name_de, fa.icon, fa.category FROM goal_focus_contributions gfc JOIN focus_area_definitions fa ON gfc.focus_area_id = fa.id WHERE gfc.goal_id IN ({placeholders}) ORDER BY gfc.contribution_weight DESC """, tuple(goal_ids)) for row in cur.fetchall(): gid = row['goal_id'] if gid not in focus_map: focus_map[gid] = [] focus_map[gid].append({ 'focus_area_id': row['focus_area_id'], 'key': row['key'], 'name_de': row['name_de'], 'icon': row['icon'], 'category': row['category'], 'contribution_weight': float(row['contribution_weight']) }) except Exception as e: # Migration 031 not yet applied - focus_contributions tables don't exist print(f"[WARNING] Could not load focus_contributions: {e}") # Continue without focus_contributions (backward compatible) # Group by category and attach focus_contributions grouped = {} for goal in goals: cat = goal['category'] or 'other' if cat not in grouped: grouped[cat] = [] goal_dict = r2d(goal) goal_dict['focus_contributions'] = focus_map.get(goal['id'], []) grouped[cat].append(goal_dict) return grouped # ============================================================================ # Training Phases # ============================================================================ @router.get("/phases") def list_training_phases(session: dict = Depends(require_auth)): """List training phases""" pid = session['profile_id'] with get_db() as conn: cur = get_cursor(conn) cur.execute(""" SELECT id, phase_type, detected_automatically, confidence_score, status, start_date, end_date, duration_days, detection_params, notes, created_at FROM training_phases WHERE profile_id = %s ORDER BY start_date DESC """, (pid,)) return [r2d(row) for row in cur.fetchall()] @router.post("/phases") def create_training_phase(data: TrainingPhaseCreate, session: dict = Depends(require_auth)): """Create training phase (manual)""" pid = session['profile_id'] with get_db() as conn: cur = get_cursor(conn) duration = None if data.end_date: duration = (data.end_date - data.start_date).days cur.execute(""" INSERT INTO training_phases ( profile_id, phase_type, detected_automatically, status, start_date, end_date, duration_days, notes ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s) RETURNING id """, ( pid, data.phase_type, False, 'active', data.start_date, data.end_date, duration, data.notes )) phase_id = cur.fetchone()['id'] return {"id": phase_id, "message": "Trainingsphase erstellt"} @router.put("/phases/{phase_id}/status") def update_phase_status( phase_id: str, status: str, session: dict = Depends(require_auth) ): """Update training phase status (accept/reject auto-detected phases)""" pid = session['profile_id'] valid_statuses = ['suggested', 'accepted', 'active', 'completed', 'rejected'] if status not in valid_statuses: raise HTTPException( status_code=400, detail=f"Ungültiger Status. Erlaubt: {', '.join(valid_statuses)}" ) with get_db() as conn: cur = get_cursor(conn) cur.execute( "UPDATE training_phases SET status = %s WHERE id = %s AND profile_id = %s", (status, phase_id, pid) ) if cur.rowcount == 0: raise HTTPException(status_code=404, detail="Trainingsphase nicht gefunden") return {"message": "Status aktualisiert"} # ============================================================================ # Fitness Tests # ============================================================================ @router.get("/tests") def list_fitness_tests(session: dict = Depends(require_auth)): """List all fitness tests""" pid = session['profile_id'] with get_db() as conn: cur = get_cursor(conn) cur.execute(""" SELECT id, test_type, result_value, result_unit, test_date, test_conditions, norm_category, created_at FROM fitness_tests WHERE profile_id = %s ORDER BY test_date DESC """, (pid,)) return [r2d(row) for row in cur.fetchall()] @router.post("/tests") def create_fitness_test(data: FitnessTestCreate, session: dict = Depends(require_auth)): """Record fitness test result""" pid = session['profile_id'] with get_db() as conn: cur = get_cursor(conn) # Calculate norm category (simplified for now) norm_category = _calculate_norm_category( data.test_type, data.result_value, data.result_unit ) cur.execute(""" INSERT INTO fitness_tests ( profile_id, test_type, result_value, result_unit, test_date, test_conditions, norm_category ) VALUES (%s, %s, %s, %s, %s, %s, %s) RETURNING id """, ( pid, data.test_type, data.result_value, data.result_unit, data.test_date, data.test_conditions, norm_category )) test_id = cur.fetchone()['id'] return {"id": test_id, "norm_category": norm_category} # ============================================================================ # Helper Functions # ============================================================================ def _get_current_value_for_goal_type(conn, profile_id: str, goal_type: str) -> Optional[float]: """ Get current value for a goal type. DEPRECATED: This function now delegates to the universal fetcher in goal_utils.py. Phase 1.5: All goal types are now defined in goal_type_definitions table. Args: conn: Database connection profile_id: User's profile ID goal_type: Goal type key (e.g., 'weight', 'meditation_minutes') Returns: Current value or None """ # Delegate to universal fetcher (Phase 1.5) return get_current_value_for_goal(conn, profile_id, goal_type) def _update_goal_progress(conn, profile_id: str, goal: dict): """Update goal progress (modifies goal dict in-place)""" # Get current value current = _get_current_value_for_goal_type(conn, profile_id, goal['goal_type']) if current is not None and goal['start_value'] is not None and goal['target_value'] is not None: goal['current_value'] = current # Calculate progress percentage total_delta = float(goal['target_value']) - float(goal['start_value']) current_delta = current - float(goal['start_value']) if total_delta != 0: progress_pct = (current_delta / total_delta) * 100 goal['progress_pct'] = round(progress_pct, 2) # Simple linear projection if goal['start_date'] and current_delta != 0: days_elapsed = (date.today() - goal['start_date']).days if days_elapsed > 0: days_per_unit = days_elapsed / current_delta remaining_units = float(goal['target_value']) - current remaining_days = int(days_per_unit * remaining_units) goal['projection_date'] = date.today() + timedelta(days=remaining_days) # Check if on track if goal['target_date'] and goal['projection_date']: goal['on_track'] = goal['projection_date'] <= goal['target_date'] def _calculate_norm_category(test_type: str, value: float, unit: str) -> Optional[str]: """ Calculate norm category for fitness test (Simplified - would need age/gender-specific norms) """ # Placeholder - should use proper norm tables return None # ============================================================================ # Goal Type Definitions (Phase 1.5 - Flexible Goal System) # ============================================================================ @router.get("/schema-info") def get_schema_info(session: dict = Depends(require_auth)): """ Get available tables and columns for goal type creation. Admin-only endpoint for building custom goal types. Returns structure with descriptions for UX guidance. """ pid = session['profile_id'] # Check admin role with get_db() as conn: cur = get_cursor(conn) cur.execute("SELECT role FROM profiles WHERE id = %s", (pid,)) profile = cur.fetchone() if not profile or profile['role'] != 'admin': raise HTTPException(status_code=403, detail="Admin-Zugriff erforderlich") # Define relevant tables with descriptions # Only include tables that make sense for goal tracking schema = { "weight_log": { "description": "Gewichtsverlauf", "columns": { "weight": {"type": "DECIMAL", "description": "Körpergewicht in kg"} } }, "caliper_log": { "description": "Caliper-Messungen (Hautfalten)", "columns": { "body_fat_pct": {"type": "DECIMAL", "description": "Körperfettanteil in %"}, "sum_mm": {"type": "DECIMAL", "description": "Summe Hautfalten in mm"} } }, "circumference_log": { "description": "Umfangsmessungen", "columns": { "c_neck": {"type": "DECIMAL", "description": "Nackenumfang in cm"}, "c_chest": {"type": "DECIMAL", "description": "Brustumfang in cm"}, "c_waist": {"type": "DECIMAL", "description": "Taillenumfang in cm"}, "c_hips": {"type": "DECIMAL", "description": "Hüftumfang in cm"}, "c_thigh_l": {"type": "DECIMAL", "description": "Oberschenkel links in cm"}, "c_thigh_r": {"type": "DECIMAL", "description": "Oberschenkel rechts in cm"}, "c_calf_l": {"type": "DECIMAL", "description": "Wade links in cm"}, "c_calf_r": {"type": "DECIMAL", "description": "Wade rechts in cm"}, "c_bicep_l": {"type": "DECIMAL", "description": "Bizeps links in cm"}, "c_bicep_r": {"type": "DECIMAL", "description": "Bizeps rechts in cm"} } }, "activity_log": { "description": "Trainingseinheiten", "columns": { "id": {"type": "UUID", "description": "ID (für Zählung von Einheiten)"}, "duration_minutes": {"type": "INTEGER", "description": "Trainingsdauer in Minuten"}, "perceived_exertion": {"type": "INTEGER", "description": "Belastungsempfinden (1-10)"}, "quality_rating": {"type": "INTEGER", "description": "Qualitätsbewertung (1-10)"} } }, "nutrition_log": { "description": "Ernährungstagebuch", "columns": { "calories": {"type": "INTEGER", "description": "Kalorien in kcal"}, "protein_g": {"type": "DECIMAL", "description": "Protein in g"}, "carbs_g": {"type": "DECIMAL", "description": "Kohlenhydrate in g"}, "fat_g": {"type": "DECIMAL", "description": "Fett in g"} } }, "sleep_log": { "description": "Schlafprotokoll", "columns": { "total_minutes": {"type": "INTEGER", "description": "Gesamtschlafdauer in Minuten"} } }, "vitals_baseline": { "description": "Vitalwerte (morgens)", "columns": { "resting_hr": {"type": "INTEGER", "description": "Ruhepuls in bpm"}, "hrv_rmssd": {"type": "INTEGER", "description": "Herzratenvariabilität (RMSSD) in ms"}, "vo2_max": {"type": "DECIMAL", "description": "VO2 Max in ml/kg/min"}, "spo2": {"type": "INTEGER", "description": "Sauerstoffsättigung in %"}, "respiratory_rate": {"type": "INTEGER", "description": "Atemfrequenz pro Minute"} } }, "blood_pressure_log": { "description": "Blutdruckmessungen", "columns": { "systolic": {"type": "INTEGER", "description": "Systolischer Blutdruck in mmHg"}, "diastolic": {"type": "INTEGER", "description": "Diastolischer Blutdruck in mmHg"}, "pulse": {"type": "INTEGER", "description": "Puls in bpm"} } }, "rest_days": { "description": "Ruhetage", "columns": { "id": {"type": "UUID", "description": "ID (für Zählung von Ruhetagen)"} } } } return schema @router.get("/goal-types") def list_goal_type_definitions(session: dict = Depends(require_auth)): """ Get all active goal type definitions. Public endpoint - returns all available goal types for dropdown. """ try: with get_db() as conn: cur = get_cursor(conn) cur.execute(""" SELECT id, type_key, label_de, label_en, unit, icon, category, source_table, source_column, aggregation_method, calculation_formula, description, is_system, is_active, created_at, updated_at FROM goal_type_definitions WHERE is_active = true ORDER BY CASE WHEN is_system = true THEN 0 ELSE 1 END, label_de """) results = [r2d(row) for row in cur.fetchall()] print(f"[DEBUG] Loaded {len(results)} goal types") return results except Exception as e: print(f"[ERROR] list_goal_type_definitions failed: {e}") print(traceback.format_exc()) raise HTTPException( status_code=500, detail=f"Fehler beim Laden der Goal Types: {str(e)}" ) @router.post("/goal-types") def create_goal_type_definition( data: GoalTypeCreate, session: dict = Depends(require_auth) ): """ Create custom goal type definition. Admin-only endpoint for creating new goal types. Users with admin role can define custom metrics. """ pid = session['profile_id'] # Check admin role with get_db() as conn: cur = get_cursor(conn) cur.execute("SELECT role FROM profiles WHERE id = %s", (pid,)) profile = cur.fetchone() if not profile or profile['role'] != 'admin': raise HTTPException( status_code=403, detail="Admin-Zugriff erforderlich" ) # Validate type_key is unique cur.execute( "SELECT id FROM goal_type_definitions WHERE type_key = %s", (data.type_key,) ) if cur.fetchone(): raise HTTPException( status_code=400, detail=f"Goal Type '{data.type_key}' existiert bereits" ) # Insert new goal type import json as json_lib filter_json = json_lib.dumps(data.filter_conditions) if data.filter_conditions else None cur.execute(""" INSERT INTO goal_type_definitions ( type_key, label_de, label_en, unit, icon, category, source_table, source_column, aggregation_method, calculation_formula, filter_conditions, description, is_active, is_system ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) RETURNING id """, ( data.type_key, data.label_de, data.label_en, data.unit, data.icon, data.category, data.source_table, data.source_column, data.aggregation_method, data.calculation_formula, filter_json, data.description, True, False # is_active=True, is_system=False )) goal_type_id = cur.fetchone()['id'] return { "id": goal_type_id, "message": f"Goal Type '{data.label_de}' erstellt" } @router.put("/goal-types/{goal_type_id}") def update_goal_type_definition( goal_type_id: str, data: GoalTypeUpdate, session: dict = Depends(require_auth) ): """ Update goal type definition. Admin-only. System goal types can be updated but not deleted. """ pid = session['profile_id'] with get_db() as conn: cur = get_cursor(conn) # Check admin role cur.execute("SELECT role FROM profiles WHERE id = %s", (pid,)) profile = cur.fetchone() if not profile or profile['role'] != 'admin': raise HTTPException( status_code=403, detail="Admin-Zugriff erforderlich" ) # Check goal type exists cur.execute( "SELECT id FROM goal_type_definitions WHERE id = %s", (goal_type_id,) ) if not cur.fetchone(): raise HTTPException(status_code=404, detail="Goal Type nicht gefunden") # Build update query updates = [] params = [] if data.label_de is not None: updates.append("label_de = %s") params.append(data.label_de) if data.label_en is not None: updates.append("label_en = %s") params.append(data.label_en) if data.unit is not None: updates.append("unit = %s") params.append(data.unit) if data.icon is not None: updates.append("icon = %s") params.append(data.icon) if data.category is not None: updates.append("category = %s") params.append(data.category) if data.source_table is not None: updates.append("source_table = %s") params.append(data.source_table) if data.source_column is not None: updates.append("source_column = %s") params.append(data.source_column) if data.aggregation_method is not None: updates.append("aggregation_method = %s") params.append(data.aggregation_method) if data.calculation_formula is not None: updates.append("calculation_formula = %s") params.append(data.calculation_formula) if data.filter_conditions is not None: import json as json_lib filter_json = json_lib.dumps(data.filter_conditions) if data.filter_conditions else None updates.append("filter_conditions = %s") params.append(filter_json) if data.description is not None: updates.append("description = %s") params.append(data.description) if data.is_active is not None: updates.append("is_active = %s") params.append(data.is_active) if not updates: raise HTTPException(status_code=400, detail="Keine Änderungen angegeben") updates.append("updated_at = NOW()") params.append(goal_type_id) cur.execute( f"UPDATE goal_type_definitions SET {', '.join(updates)} WHERE id = %s", tuple(params) ) return {"message": "Goal Type aktualisiert"} @router.delete("/goal-types/{goal_type_id}") def delete_goal_type_definition( goal_type_id: str, session: dict = Depends(require_auth) ): """ Delete (deactivate) goal type definition. Admin-only. System goal types cannot be deleted, only deactivated. Custom goal types can be fully deleted if no goals reference them. """ pid = session['profile_id'] with get_db() as conn: cur = get_cursor(conn) # Check admin role cur.execute("SELECT role FROM profiles WHERE id = %s", (pid,)) profile = cur.fetchone() if not profile or profile['role'] != 'admin': raise HTTPException( status_code=403, detail="Admin-Zugriff erforderlich" ) # Get goal type info cur.execute( "SELECT id, type_key, is_system FROM goal_type_definitions WHERE id = %s", (goal_type_id,) ) goal_type = cur.fetchone() if not goal_type: raise HTTPException(status_code=404, detail="Goal Type nicht gefunden") # Check if any goals use this type cur.execute( "SELECT COUNT(*) as count FROM goals WHERE goal_type = %s", (goal_type['type_key'],) ) count = cur.fetchone()['count'] if count > 0: # Deactivate instead of delete cur.execute( "UPDATE goal_type_definitions SET is_active = false WHERE id = %s", (goal_type_id,) ) return { "message": f"Goal Type deaktiviert ({count} Ziele nutzen diesen Typ)" } else: if goal_type['is_system']: # System types: only deactivate cur.execute( "UPDATE goal_type_definitions SET is_active = false WHERE id = %s", (goal_type_id,) ) return {"message": "System Goal Type deaktiviert"} else: # Custom types: delete cur.execute( "DELETE FROM goal_type_definitions WHERE id = %s", (goal_type_id,) ) return {"message": "Goal Type gelöscht"}