224 lines
7.0 KiB
Python
224 lines
7.0 KiB
Python
import logging
|
|
|
|
from sqlalchemy import inspect, select, text
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.db.base import engine
|
|
from app.db.models import FitnessProfile
|
|
from app.fitness.calculators import DEFAULT_NEAT_KCAL, compute_targets, macro_targets
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
TDEE_V2_BACKFILL = "fitness_tdee_v2_backfill"
|
|
MACROS_GKG_BACKFILL = "fitness_macros_gkg_v1"
|
|
|
|
|
|
def _table_exists(table: str) -> bool:
|
|
return table in inspect(engine).get_table_names()
|
|
|
|
|
|
def _add_column_if_missing(table: str, column: str, ddl: str) -> None:
|
|
inspector = inspect(engine)
|
|
if table not in inspector.get_table_names():
|
|
return
|
|
columns = {col["name"] for col in inspector.get_columns(table)}
|
|
if column in columns:
|
|
return
|
|
with engine.begin() as conn:
|
|
conn.execute(text(ddl))
|
|
|
|
|
|
def _ensure_schema_migrations_table() -> None:
|
|
with engine.begin() as conn:
|
|
conn.execute(
|
|
text(
|
|
"CREATE TABLE IF NOT EXISTS _schema_migrations ("
|
|
"name TEXT PRIMARY KEY, "
|
|
"applied_at DATETIME DEFAULT CURRENT_TIMESTAMP)"
|
|
)
|
|
)
|
|
|
|
|
|
def _migration_applied(name: str) -> bool:
|
|
_ensure_schema_migrations_table()
|
|
with engine.begin() as conn:
|
|
row = conn.execute(
|
|
text("SELECT 1 FROM _schema_migrations WHERE name = :name"),
|
|
{"name": name},
|
|
).fetchone()
|
|
return row is not None
|
|
|
|
|
|
def _mark_migration_applied(name: str) -> None:
|
|
with engine.begin() as conn:
|
|
conn.execute(
|
|
text("INSERT INTO _schema_migrations (name) VALUES (:name)"),
|
|
{"name": name},
|
|
)
|
|
|
|
|
|
def _profile_targets(row: FitnessProfile) -> dict[str, float]:
|
|
neat = row.neat_base_kcal if row.neat_base_kcal is not None else DEFAULT_NEAT_KCAL
|
|
return compute_targets(
|
|
{
|
|
"sex": row.sex,
|
|
"age": row.age,
|
|
"height_cm": row.height_cm,
|
|
"weight_kg": row.weight_kg,
|
|
"goal": row.goal,
|
|
"neat_base_kcal": neat,
|
|
}
|
|
)
|
|
|
|
|
|
def backfill_tdee_targets(*, force: bool = False) -> int:
|
|
"""Recalculate stored calorie/macro targets for all profiles (PAL → BMR+NEAT)."""
|
|
if not _table_exists("fitness_profiles"):
|
|
return 0
|
|
_ensure_schema_migrations_table()
|
|
if not force and _migration_applied(TDEE_V2_BACKFILL):
|
|
return 0
|
|
|
|
with engine.begin() as conn:
|
|
conn.execute(
|
|
text(
|
|
"UPDATE fitness_profiles "
|
|
"SET neat_base_kcal = :neat "
|
|
"WHERE neat_base_kcal IS NULL"
|
|
),
|
|
{"neat": DEFAULT_NEAT_KCAL},
|
|
)
|
|
|
|
updated = 0
|
|
with Session(engine) as db:
|
|
rows = db.scalars(select(FitnessProfile)).all()
|
|
for row in rows:
|
|
if row.neat_base_kcal is None:
|
|
row.neat_base_kcal = DEFAULT_NEAT_KCAL
|
|
targets = _profile_targets(row)
|
|
row.calorie_target = targets["calorie_target"]
|
|
row.protein_g = targets["protein_g"]
|
|
row.fat_g = targets["fat_g"]
|
|
row.carbs_g = targets["carbs_g"]
|
|
row.water_l = targets["water_l"]
|
|
updated += 1
|
|
db.commit()
|
|
|
|
if not force or not _migration_applied(TDEE_V2_BACKFILL):
|
|
_mark_migration_applied(TDEE_V2_BACKFILL)
|
|
|
|
logger.info("TDEE v2 backfill: recalculated %s fitness profile(s)", updated)
|
|
return updated
|
|
|
|
|
|
def backfill_macros_gkg(*, force: bool = False) -> int:
|
|
"""Recalculate stored BJU from weight (protein/fat g/kg, carbs = remainder)."""
|
|
if not _table_exists("fitness_profiles"):
|
|
return 0
|
|
_ensure_schema_migrations_table()
|
|
if not force and _migration_applied(MACROS_GKG_BACKFILL):
|
|
return 0
|
|
|
|
updated = 0
|
|
with Session(engine) as db:
|
|
rows = db.scalars(select(FitnessProfile)).all()
|
|
for row in rows:
|
|
macros = macro_targets(row.calorie_target, row.weight_kg, row.goal)
|
|
row.protein_g = macros["protein_g"]
|
|
row.fat_g = macros["fat_g"]
|
|
row.carbs_g = macros["carbs_g"]
|
|
updated += 1
|
|
db.commit()
|
|
|
|
_mark_migration_applied(MACROS_GKG_BACKFILL)
|
|
logger.info("Macros g/kg backfill: updated %s fitness profile(s)", updated)
|
|
return updated
|
|
|
|
|
|
def run_fitness_migrations() -> None:
|
|
inspector = inspect(engine)
|
|
|
|
if "fitness_profiles" in inspector.get_table_names():
|
|
_add_column_if_missing(
|
|
"fitness_profiles",
|
|
"baseline_steps",
|
|
"ALTER TABLE fitness_profiles ADD COLUMN baseline_steps INTEGER",
|
|
)
|
|
_add_column_if_missing(
|
|
"fitness_profiles",
|
|
"baseline_workout_kcal",
|
|
"ALTER TABLE fitness_profiles ADD COLUMN baseline_workout_kcal FLOAT",
|
|
)
|
|
_add_column_if_missing(
|
|
"fitness_profiles",
|
|
"neat_base_kcal",
|
|
"ALTER TABLE fitness_profiles ADD COLUMN neat_base_kcal FLOAT DEFAULT 200.0",
|
|
)
|
|
|
|
if "workout_logs" in inspector.get_table_names():
|
|
_add_column_if_missing(
|
|
"workout_logs",
|
|
"active_calories",
|
|
"ALTER TABLE workout_logs ADD COLUMN active_calories FLOAT",
|
|
)
|
|
_add_column_if_missing(
|
|
"workout_logs",
|
|
"total_calories",
|
|
"ALTER TABLE workout_logs ADD COLUMN total_calories FLOAT",
|
|
)
|
|
_add_column_if_missing(
|
|
"workout_logs",
|
|
"steps",
|
|
"ALTER TABLE workout_logs ADD COLUMN steps INTEGER",
|
|
)
|
|
|
|
if "step_logs" not in inspector.get_table_names():
|
|
with engine.begin() as conn:
|
|
conn.execute(
|
|
text(
|
|
"CREATE TABLE step_logs ("
|
|
"id INTEGER PRIMARY KEY AUTOINCREMENT, "
|
|
"logged_at DATETIME DEFAULT CURRENT_TIMESTAMP, "
|
|
"steps INTEGER DEFAULT 0, "
|
|
"active_calories FLOAT, "
|
|
"source VARCHAR(32) DEFAULT 'manual', "
|
|
"notes TEXT DEFAULT ''"
|
|
")"
|
|
)
|
|
)
|
|
|
|
if "body_metrics" in inspector.get_table_names():
|
|
_add_column_if_missing(
|
|
"body_metrics",
|
|
"neck_cm",
|
|
"ALTER TABLE body_metrics ADD COLUMN neck_cm FLOAT",
|
|
)
|
|
_add_column_if_missing(
|
|
"body_metrics",
|
|
"hip_cm",
|
|
"ALTER TABLE body_metrics ADD COLUMN hip_cm FLOAT",
|
|
)
|
|
_add_column_if_missing(
|
|
"body_metrics",
|
|
"body_fat_method",
|
|
"ALTER TABLE body_metrics ADD COLUMN body_fat_method VARCHAR(16)",
|
|
)
|
|
_add_column_if_missing(
|
|
"body_metrics",
|
|
"whr",
|
|
"ALTER TABLE body_metrics ADD COLUMN whr FLOAT",
|
|
)
|
|
_add_column_if_missing(
|
|
"body_metrics",
|
|
"lbm_kg",
|
|
"ALTER TABLE body_metrics ADD COLUMN lbm_kg FLOAT",
|
|
)
|
|
_add_column_if_missing(
|
|
"body_metrics",
|
|
"ffmi",
|
|
"ALTER TABLE body_metrics ADD COLUMN ffmi FLOAT",
|
|
)
|
|
|
|
backfill_tdee_targets()
|
|
backfill_macros_gkg()
|