Lars/mitai-jinkendo
1
0
Fork 0
Code Issues 23 Pull Requests Actions Packages Projects Releases 2 Wiki Activity

refactor: split goals.py into 5 modular routers
All checks were successful
Deploy Development / deploy (push) Successful in 50s
Details
Build Test / lint-backend (push) Successful in 0s
Details
Build Test / build-frontend (push) Successful in 14s
Details

Browse Source 
Code Splitting Results:
- goals.py: 1339 → 655 lines (-684 lines, -51%)
- Created 4 new routers:
  * goal_types.py (426 lines) - Goal Type Definitions CRUD
  * goal_progress.py (155 lines) - Progress tracking
  * training_phases.py (107 lines) - Training phases
  * fitness_tests.py (94 lines) - Fitness tests

Benefits:
✅ Improved maintainability (smaller, focused files)
✅ Better context window efficiency for AI tools
✅ Clearer separation of concerns
✅ Easier testing and debugging

All routers registered in main.py.
Backward compatible - no API changes.
This commit is contained in:
Lars 2026-03-28 06:31:31 +01:00
parent 448f6ad4f4
commit 12d516c881
6 changed files with 801 additions and 698 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
backend/main.py
View File

@ -24,6 +24,7 @@ from routers import admin_activity_mappings, sleep, rest_days
from routers import vitals_baseline, blood_pressure # v9d Phase 2d Refactored
from routers import evaluation # v9d/v9e Training Type Profiles (#15)
from routers import goals, focus_areas # v9e/v9g Goal System v2.0 (Dynamic Focus Areas)
from routers import goal_types, goal_progress, training_phases, fitness_tests # v9h Goal System (Split routers)
# ── App Configuration ─────────────────────────────────────────────────────────
DATA_DIR = Path(os.getenv("DATA_DIR", "./data"))
@ -98,7 +99,11 @@ app.include_router(rest_days.router) # /api/rest-days/* (v9d Phase 2a
app.include_router(vitals_baseline.router) # /api/vitals/baseline/* (v9d Phase 2d Refactored)
app.include_router(blood_pressure.router) # /api/blood-pressure/* (v9d Phase 2d Refactored)
app.include_router(evaluation.router) # /api/evaluation/* (v9d/v9e Training Profiles #15)
app.include_router(goals.router) # /api/goals/* (v9e Goal System Strategic + Tactical)
app.include_router(goals.router) # /api/goals/* (v9h Goal System Core CRUD + Focus Areas)
app.include_router(goal_types.router) # /api/goals/goal-types/* (v9h Goal Type Definitions)
app.include_router(goal_progress.router) # /api/goals/{goal_id}/progress/* (v9h Progress Tracking)
app.include_router(training_phases.router) # /api/goals/phases/* (v9h Training Phases)
app.include_router(fitness_tests.router) # /api/goals/tests/* (v9h Fitness Tests)
app.include_router(focus_areas.router) # /api/focus-areas/* (v9g Focus Area System v2.0 - Dynamic)
# ── Health Check ──────────────────────────────────────────────────────────────

94
backend/routers/fitness_tests.py Normal file
View File

@ -0,0 +1,94 @@
"""
Fitness Tests Router - Fitness Test Recording & Norm Tracking
Endpoints for managing fitness tests:
- List fitness tests
- Record fitness test results
- Calculate norm categories
Part of v9h Goal System.
"""
from fastapi import APIRouter, Depends, HTTPException
from pydantic import BaseModel
from typing import Optional
from datetime import date
from db import get_db, get_cursor, r2d
from auth import require_auth
router = APIRouter(prefix="/api/goals", tags=["fitness-tests"])
# ============================================================================
# Pydantic Models
# ============================================================================
class FitnessTestCreate(BaseModel):
"""Record fitness test result"""
test_type: str
result_value: float
result_unit: str
test_date: date
test_conditions: Optional[str] = None
# ============================================================================
# Endpoints
# ============================================================================
@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 _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

155
backend/routers/goal_progress.py Normal file
View File

@ -0,0 +1,155 @@
"""
Goal Progress Router - Progress Tracking for Goals
Endpoints for logging and managing goal progress:
- Get progress history
- Create manual progress entries
- Delete progress entries
Part of v9h Goal System.
"""
from fastapi import APIRouter, Depends, HTTPException
from pydantic import BaseModel
from typing import Optional
from datetime import date
from db import get_db, get_cursor, r2d
from auth import require_auth
router = APIRouter(prefix="/api/goals", tags=["goal-progress"])
# ============================================================================
# Pydantic Models
# ============================================================================
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
# ============================================================================
# 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"}

426
backend/routers/goal_types.py Normal file
View File

@ -0,0 +1,426 @@
"""
Goal Types Router - Custom Goal Type Definitions
Endpoints for managing goal type definitions (admin-only):
- CRUD for goal type definitions
- Schema info for building custom types
Part of v9h Goal System.
"""
from fastapi import APIRouter, Depends, HTTPException
from pydantic import BaseModel
from typing import Optional
import traceback
from db import get_db, get_cursor, r2d
from auth import require_auth
router = APIRouter(prefix="/api/goals", tags=["goal-types"])
# ============================================================================
# Pydantic Models
# ============================================================================
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
# ============================================================================
# Endpoints
# ============================================================================
@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, filter_conditions, 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"}

710
backend/routers/goals.py
View File

@ -1,20 +1,23 @@
"""
Goals Router - Goal System (Strategic + Tactical)
Goals Router - Core Goal CRUD & Focus Areas (Streamlined v2.0)
Endpoints for managing:
- Strategic goal modes (weight_loss, strength, etc.)
- Strategic focus areas (weighted multi-goal system)
- Tactical goal targets (concrete values with deadlines)
- Training phase detection
- Fitness tests
- Grouped goal views
Part of v9e Goal System implementation.
Part of v9h Goal System (Phase 0a).
NOTE: Code split complete! Related endpoints moved to:
- goal_types.py → Goal Type Definitions (Admin CRUD)
- goal_progress.py → Progress tracking
- training_phases.py → Training phase management
- fitness_tests.py → Fitness test recording
"""
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 datetime import date, timedelta
from db import get_db, get_cursor, r2d
from auth import require_auth
@ -69,62 +72,6 @@ class GoalUpdate(BaseModel):
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
# ============================================================================
@ -372,7 +319,7 @@ def update_focus_areas(data: FocusAreasUpdate, session: dict = Depends(require_a
}
# ============================================================================
# Tactical Layer: Concrete Goals
# Tactical Layer: Concrete Goals - Core CRUD
# ============================================================================
@router.get("/list")
@ -572,131 +519,10 @@ def delete_goal(goal_id: str, session: dict = Depends(require_auth)):
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.
Get all goals grouped by category.
Returns structure:
{
@ -777,134 +603,6 @@ def get_goals_grouped(session: dict = Depends(require_auth)):
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
# ============================================================================
@ -955,385 +653,3 @@ def _update_goal_progress(conn, profile_id: str, goal: dict):
# 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, filter_conditions, 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"}

107
backend/routers/training_phases.py Normal file
View File

@ -0,0 +1,107 @@
"""
Training Phases Router - Training Phase Detection & Management
Endpoints for managing training phases:
- List training phases
- Create manual training phases
- Update phase status (accept/reject auto-detected phases)
Part of v9h Goal System.
"""
from fastapi import APIRouter, Depends, HTTPException
from pydantic import BaseModel
from typing import Optional
from datetime import date
from db import get_db, get_cursor, r2d
from auth import require_auth
router = APIRouter(prefix="/api/goals", tags=["training-phases"])
# ============================================================================
# Pydantic Models
# ============================================================================
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
# ============================================================================
# Endpoints
# ============================================================================
@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"}