399 lines
18 KiB
Python
399 lines
18 KiB
Python
import os
|
|
import json
|
|
from flask import Blueprint, render_template, request, redirect, url_for, current_app, flash, jsonify
|
|
from extensions import db
|
|
from flask_login import login_required, current_user
|
|
from jinja2_fragments import render_block
|
|
from urllib.parse import parse_qs, urlencode
|
|
from flask_htmx import HTMX
|
|
|
|
programs_bp = Blueprint('programs', __name__, url_prefix='/programs')
|
|
htmx = HTMX()
|
|
|
|
@programs_bp.route('/create', methods=['GET', 'POST'])
|
|
@login_required
|
|
def create_program():
|
|
if request.method == 'POST':
|
|
program_name = request.form.get('program_name', '').strip()
|
|
description = request.form.get('description', '').strip()
|
|
sessions_data = []
|
|
i = 0
|
|
while True:
|
|
session_order_key = f'session_order_{i}'
|
|
if session_order_key not in request.form:
|
|
break
|
|
|
|
session_order = request.form.get(session_order_key)
|
|
session_name = request.form.get(f'session_name_{i}', '').strip()
|
|
exercise_ids_str = request.form.getlist(f'exercises_{i}')
|
|
sets_list = request.form.getlist(f'sets_{i}')
|
|
reps_list = request.form.getlist(f'reps_{i}')
|
|
|
|
if not exercise_ids_str or not session_order:
|
|
flash(f"Error processing session {i+1}: Missing exercises.", "error")
|
|
return redirect(url_for('programs.create_program'))
|
|
|
|
try:
|
|
exercise_data = []
|
|
for idx, eid in enumerate(exercise_ids_str):
|
|
exercise_data.append({
|
|
'id': int(eid),
|
|
'sets': int(sets_list[idx]) if idx < len(sets_list) and sets_list[idx] else None,
|
|
'rep_range': reps_list[idx] if idx < len(reps_list) else None,
|
|
'order': idx + 1
|
|
})
|
|
|
|
sessions_data.append({
|
|
'order': int(session_order),
|
|
'name': session_name if session_name else None,
|
|
'exercises': exercise_data
|
|
})
|
|
except ValueError:
|
|
flash(f"Error processing session {i+1}: Invalid data.", "error")
|
|
return redirect(url_for('programs.create_program'))
|
|
|
|
i += 1
|
|
|
|
if not program_name:
|
|
flash("Program Name is required.", "error")
|
|
return redirect(url_for('programs.create_program'))
|
|
if not sessions_data:
|
|
flash("At least one session must be added.", "error")
|
|
return redirect(url_for('programs.create_program'))
|
|
|
|
try:
|
|
program_result = db.execute(
|
|
"INSERT INTO workout_program (name, description) VALUES (%s, %s) RETURNING program_id",
|
|
[program_name, description if description else None],
|
|
commit=True, one=True
|
|
)
|
|
new_program_id = program_result['program_id']
|
|
|
|
for session in sessions_data:
|
|
exercise_ids = sorted([ex['id'] for ex in session['exercises']])
|
|
tag_filter = "?" + "&".join(f"exercise_id={eid}" for eid in exercise_ids)
|
|
tag_name = session['name'] if session['name'] else f"Program Session {session['order']} Exercises"
|
|
|
|
existing_tag = db.execute(
|
|
"SELECT tag_id FROM tag WHERE filter = %s AND person_id IS NULL",
|
|
[tag_filter], one=True
|
|
)
|
|
|
|
if existing_tag:
|
|
session_tag_id = existing_tag['tag_id']
|
|
else:
|
|
new_tag_result = db.execute(
|
|
"INSERT INTO tag (name, filter, person_id) VALUES (%s, %s, NULL) RETURNING tag_id",
|
|
[tag_name, tag_filter], commit=True, one=True
|
|
)
|
|
session_tag_id = new_tag_result['tag_id']
|
|
|
|
session_record = db.execute(
|
|
"INSERT INTO program_session (program_id, session_order, session_name, tag_id) "
|
|
"VALUES (%s, %s, %s, %s) RETURNING session_id",
|
|
[new_program_id, session['order'], session['name'], session_tag_id],
|
|
commit=True, one=True
|
|
)
|
|
session_id = session_record['session_id']
|
|
|
|
for ex in session['exercises']:
|
|
db.execute(
|
|
"INSERT INTO program_session_exercise (session_id, exercise_id, sets, rep_range, exercise_order) "
|
|
"VALUES (%s, %s, %s, %s, %s)",
|
|
[session_id, ex['id'], ex['sets'], ex['rep_range'], ex['order']],
|
|
commit=True
|
|
)
|
|
|
|
flash(f"Workout Program '{program_name}' created successfully!", "success")
|
|
return redirect(url_for('programs.view_program', program_id=new_program_id))
|
|
|
|
except Exception as e:
|
|
print(f"Error creating program: {e}")
|
|
flash(f"Database error creating program: {e}", "error")
|
|
return redirect(url_for('programs.create_program'))
|
|
else:
|
|
exercises = db.execute("SELECT exercise_id, name FROM exercise ORDER BY name")
|
|
return render_template('program_create.html', exercises=exercises if exercises else [])
|
|
|
|
@programs_bp.route('/', methods=['GET'])
|
|
@login_required
|
|
def list_programs():
|
|
programs = db.execute("SELECT program_id, name, description FROM workout_program ORDER BY created_at DESC")
|
|
if programs is None:
|
|
programs = []
|
|
|
|
# Enrich programs with sessions and exercises for preview
|
|
for program in programs:
|
|
sessions = db.execute(
|
|
"SELECT session_id, session_order, session_name FROM program_session WHERE program_id = %s ORDER BY session_order",
|
|
[program['program_id']]
|
|
)
|
|
for session in sessions:
|
|
exercises = db.execute(
|
|
"""SELECT e.name
|
|
FROM program_session_exercise pse
|
|
JOIN exercise e ON pse.exercise_id = e.exercise_id
|
|
WHERE pse.session_id = %s
|
|
ORDER BY pse.exercise_order""",
|
|
[session['session_id']]
|
|
)
|
|
session['exercises'] = exercises
|
|
program['sessions'] = sessions
|
|
|
|
htmx_req = request.headers.get('HX-Request')
|
|
if htmx_req:
|
|
return render_block(current_app.jinja_env, 'program_list.html', 'content', programs=programs)
|
|
return render_template('program_list.html', programs=programs)
|
|
|
|
@programs_bp.route('/import', methods=['GET', 'POST'])
|
|
@login_required
|
|
def import_program():
|
|
htmx_req = request.headers.get('HX-Request')
|
|
if request.method == 'POST':
|
|
if 'file' not in request.files:
|
|
flash("No file part", "error")
|
|
return redirect(request.url)
|
|
|
|
file = request.files['file']
|
|
if file.filename == '':
|
|
flash("No selected file", "error")
|
|
return redirect(request.url)
|
|
|
|
try:
|
|
data = json.load(file)
|
|
program_name = data.get('program_name', 'Imported Program')
|
|
description = data.get('description', '')
|
|
|
|
program_result = db.execute(
|
|
"INSERT INTO workout_program (name, description) VALUES (%s, %s) RETURNING program_id",
|
|
[program_name, description],
|
|
commit=True, one=True
|
|
)
|
|
new_program_id = program_result['program_id']
|
|
|
|
for session_data in data.get('sessions', []):
|
|
order = session_data.get('order')
|
|
name = session_data.get('name')
|
|
exercises_list = session_data.get('exercises', [])
|
|
|
|
exercise_ids = sorted([int(ex['id']) for ex in exercises_list])
|
|
tag_filter = "?" + "&".join(f"exercise_id={eid}" for eid in exercise_ids)
|
|
tag_name = name if name else f"Session {order} Exercises"
|
|
|
|
existing_tag = db.execute(
|
|
"SELECT tag_id FROM tag WHERE filter = %s AND person_id IS NULL",
|
|
[tag_filter], one=True
|
|
)
|
|
|
|
if existing_tag:
|
|
tag_id = existing_tag['tag_id']
|
|
else:
|
|
new_tag_result = db.execute(
|
|
"INSERT INTO tag (name, filter, person_id) VALUES (%s, %s, NULL) RETURNING tag_id",
|
|
[tag_name, tag_filter], commit=True, one=True
|
|
)
|
|
tag_id = new_tag_result['tag_id']
|
|
|
|
session_result = db.execute(
|
|
"INSERT INTO program_session (program_id, session_order, session_name, tag_id) "
|
|
"VALUES (%s, %s, %s, %s) RETURNING session_id",
|
|
[new_program_id, order, name, tag_id],
|
|
commit=True, one=True
|
|
)
|
|
session_id = session_result['session_id']
|
|
|
|
for ex in exercises_list:
|
|
db.execute(
|
|
"INSERT INTO program_session_exercise (session_id, exercise_id, sets, rep_range, exercise_order) "
|
|
"VALUES (%s, %s, %s, %s, %s)",
|
|
[session_id, ex['id'], ex.get('sets'), ex.get('rep_range'), ex.get('order')],
|
|
commit=True
|
|
)
|
|
|
|
flash(f"Program '{program_name}' imported successfully!", "success")
|
|
return redirect(url_for('programs.view_program', program_id=new_program_id))
|
|
|
|
except Exception as e:
|
|
flash(f"Error importing program: {e}", "error")
|
|
return redirect(url_for('programs.list_programs'))
|
|
|
|
if htmx_req:
|
|
return render_block(current_app.jinja_env, 'program_import.html', 'content')
|
|
return render_template('program_import.html')
|
|
|
|
@programs_bp.route('/<int:program_id>/delete', methods=['DELETE'])
|
|
@login_required
|
|
def delete_program(program_id):
|
|
try:
|
|
db.execute("DELETE FROM workout_program WHERE program_id = %s", [program_id], commit=True)
|
|
return "", 200
|
|
except Exception as e:
|
|
return str(e), 500
|
|
|
|
@programs_bp.route('/<int:program_id>', methods=['GET'])
|
|
@login_required
|
|
def view_program(program_id):
|
|
program = db.execute("SELECT * FROM workout_program WHERE program_id = %s", [program_id], one=True)
|
|
if not program:
|
|
flash("Program not found.", "error")
|
|
return redirect(url_for('programs.list_programs'))
|
|
|
|
sessions = db.execute("SELECT * FROM program_session WHERE program_id = %s ORDER BY session_order", [program_id])
|
|
|
|
for session in sessions:
|
|
exercises = db.execute(
|
|
"""SELECT e.exercise_id, e.name, pse.sets, pse.rep_range, pse.exercise_order
|
|
FROM program_session_exercise pse
|
|
JOIN exercise e ON pse.exercise_id = e.exercise_id
|
|
WHERE pse.session_id = %s
|
|
ORDER BY pse.exercise_order""",
|
|
[session['session_id']]
|
|
)
|
|
|
|
if not exercises:
|
|
tag = db.execute("SELECT filter FROM tag WHERE tag_id = %s", [session['tag_id']], one=True)
|
|
if tag and tag['filter']:
|
|
from urllib.parse import parse_qs
|
|
qs = parse_qs(tag['filter'].lstrip('?'))
|
|
exercise_ids = qs.get('exercise_id', [])
|
|
if exercise_ids:
|
|
exercises = db.execute(
|
|
f"SELECT exercise_id, name FROM exercise WHERE exercise_id IN ({','.join(['%s']*len(exercise_ids))})",
|
|
exercise_ids
|
|
)
|
|
session['exercises'] = exercises
|
|
|
|
htmx_req = request.headers.get('HX-Request')
|
|
if htmx_req:
|
|
return render_block(current_app.jinja_env, 'program_view.html', 'content', program=program, sessions=sessions)
|
|
return render_template('program_view.html', program=program, sessions=sessions)
|
|
|
|
@programs_bp.route('/<int:program_id>/edit', methods=['GET', 'POST'])
|
|
@login_required
|
|
def edit_program(program_id):
|
|
program = db.execute("SELECT * FROM workout_program WHERE program_id = %s", [program_id], one=True)
|
|
if not program:
|
|
flash("Program not found.", "error")
|
|
return redirect(url_for('programs.list_programs'))
|
|
|
|
if request.method == 'POST':
|
|
program_name = request.form.get('program_name', '').strip()
|
|
description = request.form.get('description', '').strip()
|
|
sessions_data = []
|
|
i = 0
|
|
while True:
|
|
session_order_key = f'session_order_{i}'
|
|
if session_order_key not in request.form:
|
|
break
|
|
|
|
session_order = request.form.get(session_order_key)
|
|
session_name = request.form.get(f'session_name_{i}', '').strip()
|
|
exercise_ids_str = request.form.getlist(f'exercises_{i}')
|
|
sets_list = request.form.getlist(f'sets_{i}')
|
|
reps_list = request.form.getlist(f'reps_{i}')
|
|
|
|
if not exercise_ids_str or not session_order:
|
|
flash(f"Error processing session {i+1}: Missing exercises.", "error")
|
|
return redirect(url_for('programs.edit_program', program_id=program_id))
|
|
|
|
try:
|
|
exercise_data = []
|
|
for idx, eid in enumerate(exercise_ids_str):
|
|
exercise_data.append({
|
|
'id': int(eid),
|
|
'sets': int(sets_list[idx]) if idx < len(sets_list) and sets_list[idx] else None,
|
|
'rep_range': reps_list[idx] if idx < len(reps_list) else None,
|
|
'order': idx + 1
|
|
})
|
|
|
|
sessions_data.append({
|
|
'order': int(session_order),
|
|
'name': session_name if session_name else None,
|
|
'exercises': exercise_data
|
|
})
|
|
except ValueError:
|
|
flash(f"Error processing session {i+1}: Invalid data.", "error")
|
|
return redirect(url_for('programs.edit_program', program_id=program_id))
|
|
|
|
i += 1
|
|
|
|
if not program_name:
|
|
flash("Program Name is required.", "error")
|
|
return redirect(url_for('programs.edit_program', program_id=program_id))
|
|
|
|
try:
|
|
# Update Program
|
|
db.execute(
|
|
"UPDATE workout_program SET name = %s, description = %s WHERE program_id = %s",
|
|
[program_name, description if description else None, program_id],
|
|
commit=True
|
|
)
|
|
|
|
# Delete existing sessions (metadata will be deleted via CASCADE)
|
|
db.execute("DELETE FROM program_session WHERE program_id = %s", [program_id], commit=True)
|
|
|
|
# Re-insert Sessions
|
|
for session in sessions_data:
|
|
exercise_ids = sorted([ex['id'] for ex in session['exercises']])
|
|
tag_filter = "?" + "&".join(f"exercise_id={eid}" for eid in exercise_ids)
|
|
tag_name = session['name'] if session['name'] else f"Program Session {session['order']} Exercises"
|
|
|
|
existing_tag = db.execute(
|
|
"SELECT tag_id FROM tag WHERE filter = %s AND person_id IS NULL",
|
|
[tag_filter], one=True
|
|
)
|
|
|
|
if existing_tag:
|
|
session_tag_id = existing_tag['tag_id']
|
|
else:
|
|
new_tag_result = db.execute(
|
|
"INSERT INTO tag (name, filter, person_id) VALUES (%s, %s, NULL) RETURNING tag_id",
|
|
[tag_name, tag_filter], commit=True, one=True
|
|
)
|
|
session_tag_id = new_tag_result['tag_id']
|
|
|
|
session_record = db.execute(
|
|
"INSERT INTO program_session (program_id, session_order, session_name, tag_id) "
|
|
"VALUES (%s, %s, %s, %s) RETURNING session_id",
|
|
[program_id, session['order'], session['name'], session_tag_id],
|
|
commit=True, one=True
|
|
)
|
|
session_id = session_record['session_id']
|
|
|
|
for ex in session['exercises']:
|
|
db.execute(
|
|
"INSERT INTO program_session_exercise (session_id, exercise_id, sets, rep_range, exercise_order) "
|
|
"VALUES (%s, %s, %s, %s, %s)",
|
|
[session_id, ex['id'], ex['sets'], ex['rep_range'], ex['order']],
|
|
commit=True
|
|
)
|
|
|
|
flash(f"Program '{program_name}' updated successfully!", "success")
|
|
return redirect(url_for('programs.view_program', program_id=program_id))
|
|
|
|
except Exception as e:
|
|
print(f"Error updating program: {e}")
|
|
flash(f"Database error updating program: {e}", "error")
|
|
return redirect(url_for('programs.edit_program', program_id=program_id))
|
|
|
|
# GET Request
|
|
sessions = db.execute("SELECT * FROM program_session WHERE program_id = %s ORDER BY session_order", [program_id])
|
|
for session in sessions:
|
|
exercises = db.execute(
|
|
"""SELECT e.exercise_id, e.name, pse.sets, pse.rep_range, pse.exercise_order
|
|
FROM program_session_exercise pse
|
|
JOIN exercise e ON pse.exercise_id = e.exercise_id
|
|
WHERE pse.session_id = %s
|
|
ORDER BY pse.exercise_order""",
|
|
[session['session_id']]
|
|
)
|
|
session['exercises'] = exercises
|
|
|
|
all_exercises = db.execute("SELECT exercise_id, name FROM exercise ORDER BY name")
|
|
|
|
htmx_req = request.headers.get('HX-Request')
|
|
if htmx_req:
|
|
return render_block(current_app.jinja_env, 'program_edit.html', 'content',
|
|
program=program, sessions=sessions, exercises=all_exercises)
|
|
return render_template('program_edit.html', program=program, sessions=sessions, exercises=all_exercises)
|