"""Service for seeding the database from YAML config files. Reads config/exercises.yaml and config/user_programs.yaml to populate the exercise library, warmups, workout days, and user programs. """ import os from pathlib import Path from typing import Optional import bcrypt import structlog import yaml from sqlmodel import Session, select from app.models.exercise import Exercise from app.models.user import User from app.models.user_exercise_program import UserExerciseProgram from app.models.warmup import Warmup from app.models.workout_day import WorkoutDay logger = structlog.get_logger(__name__) # Default workout day definitions WORKOUT_DAYS = [ {"name": "Push", "day_number": 1, "description": "Chest, shoulders, triceps"}, {"name": "Pull", "day_number": 2, "description": "Back, biceps, traps"}, {"name": "Lower", "day_number": 3, "description": "Quads, hamstrings, glutes, calves"}, {"name": "Full Body", "day_number": 4, "description": "Compound full-body movements"}, ] class SeedService: """Seeds the database from YAML configuration files. Args: session: An active SQLModel Session. config_dir: Path to the config directory containing YAML files. """ def __init__(self, session: Session, config_dir: Optional[Path] = None) -> None: self._session = session self._config_dir = config_dir or Path(__file__).resolve().parent.parent.parent / "config" def _load_yaml(self, filename: str) -> dict: """Load and parse a YAML file from the config directory. Args: filename: Name of the YAML file to load. Returns: Parsed YAML content as a dictionary. """ filepath = self._config_dir / filename with open(filepath, "r") as f: return yaml.safe_load(f) def seed_workout_days(self) -> None: """Seed the 4 workout day records if they don't already exist.""" existing = self._session.exec(select(WorkoutDay)).first() if existing: logger.info("seed_skipped", table="workout_days", reason="already seeded") return for day_data in WORKOUT_DAYS: day = WorkoutDay(**day_data) self._session.add(day) self._session.commit() logger.info("seed_complete", table="workout_days", count=len(WORKOUT_DAYS)) def seed_exercises(self) -> None: """Seed exercises from config/exercises.yaml if table is empty.""" existing = self._session.exec(select(Exercise)).first() if existing: logger.info("seed_skipped", table="exercises", reason="already seeded") return data = self._load_yaml("exercises.yaml") exercises = data.get("exercises", []) for ex in exercises: exercise = Exercise( name=ex["name"], muscle_group=ex["muscle_group"], workout_day=ex["workout_day"], sets=ex.get("sets", 3), tempo=ex.get("tempo", ""), form_cues=ex.get("form_cues", "").strip(), ) self._session.add(exercise) self._session.commit() logger.info("seed_complete", table="exercises", count=len(exercises)) def seed_warmups(self) -> None: """Seed warmups from config/exercises.yaml if table is empty.""" existing = self._session.exec(select(Warmup)).first() if existing: logger.info("seed_skipped", table="warmups", reason="already seeded") return data = self._load_yaml("exercises.yaml") warmups = data.get("warmup", []) for i, wu in enumerate(warmups, start=1): warmup = Warmup( name=wu["name"], type=wu.get("type", ""), reps=wu.get("reps", ""), form_cues=wu.get("form_cues", "").strip(), sort_order=i, ) self._session.add(warmup) self._session.commit() logger.info("seed_complete", table="warmups", count=len(warmups)) def seed_admin(self) -> None: """Create admin user from environment variables if not exists. Reads ADMIN_USERNAME and ADMIN_PASSWORD from env, hashes the password with bcrypt, and creates the admin user. """ admin_username = os.environ.get("ADMIN_USERNAME", "admin") admin_password = os.environ.get("ADMIN_PASSWORD", "") existing = self._session.exec( select(User).where(User.username == admin_username) ).first() if existing: logger.info("seed_skipped", table="users", reason="admin already exists") return if not admin_password: logger.warning("seed_skipped", table="users", reason="ADMIN_PASSWORD not set") return # Hash password with bcrypt password_hash = bcrypt.hashpw( admin_password.encode("utf-8"), bcrypt.gensalt(), ).decode("utf-8") admin = User( username=admin_username, password_hash=password_hash, display_name="Admin", is_admin=True, ) self._session.add(admin) self._session.commit() logger.info("seed_complete", table="users", user="admin") def seed_user_programs(self) -> None: """Seed user profiles and their exercise programs from user_programs.yaml. Creates non-admin user profiles and links them to exercises with week 1/4 rep and weight targets. """ data = self._load_yaml("user_programs.yaml") programs = data.get("programs", []) for program in programs: username = program["user"].lower().replace(" ", "_") display_name = program["user"] profile = program.get("profile", {}) # Check if user already exists existing_user = self._session.exec( select(User).where(User.username == username) ).first() if existing_user: user = existing_user logger.info("seed_skipped", table="users", user=username, reason="already exists") else: user = User( username=username, password_hash="", # Non-admin users don't log in initially display_name=display_name, height=profile.get("height", ""), weight=profile.get("weight", ""), goals=profile.get("goals", ""), is_admin=False, ) self._session.add(user) self._session.commit() self._session.refresh(user) logger.info("seed_complete", table="users", user=username) # Link exercises to user for ex_data in program.get("exercises", []): exercise = self._session.exec( select(Exercise).where(Exercise.name == ex_data["name"]) ).first() if exercise is None: logger.warning("seed_exercise_not_found", name=ex_data["name"]) continue # Check if program already exists existing_program = self._session.exec( select(UserExerciseProgram).where( UserExerciseProgram.user_id == user.id, UserExerciseProgram.exercise_id == exercise.id, ) ).first() if existing_program: continue uep = UserExerciseProgram( user_id=user.id, exercise_id=exercise.id, wk1_reps=str(ex_data.get("wk1_reps", "")), wk4_reps=str(ex_data.get("wk4_reps", "")), wk1_weight=str(ex_data.get("wk1_weight", "")), wk4_weight=str(ex_data.get("wk4_weight", "")), ) self._session.add(uep) self._session.commit() logger.info("seed_complete", table="user_exercise_programs", user=username) def seed_all(self) -> None: """Run all seed operations in the correct order. Order matters: workout_days and exercises must exist before user_programs can reference them. """ logger.info("seed_all_started") self.seed_workout_days() self.seed_exercises() self.seed_warmups() self.seed_admin() self.seed_user_programs() logger.info("seed_all_complete")