185 lines
8.7 KiB
Python
185 lines
8.7 KiB
Python
class Exercises:
|
|
def __init__(self, db_connection_method):
|
|
self.execute = db_connection_method
|
|
|
|
def get(self, query):
|
|
if not query:
|
|
exercises = self.execute("SELECT exercise_id, name FROM exercise ORDER BY name ASC;")
|
|
for ex in exercises:
|
|
ex['attributes'] = self.get_exercise_attributes(ex['exercise_id'])
|
|
return exercises
|
|
|
|
# Check for category:value syntax
|
|
if ':' in query:
|
|
category_part, value_part = query.split(':', 1)
|
|
category_part = f"%{category_part.strip().lower()}%"
|
|
value_part = f"%{value_part.strip().lower()}%"
|
|
|
|
query = """
|
|
SELECT DISTINCT e.exercise_id, e.name
|
|
FROM exercise e
|
|
JOIN exercise_to_attribute eta ON e.exercise_id = eta.exercise_id
|
|
JOIN exercise_attribute attr ON eta.attribute_id = attr.attribute_id
|
|
JOIN exercise_attribute_category cat ON attr.category_id = cat.category_id
|
|
WHERE LOWER(cat.name) LIKE LOWER(%s) AND LOWER(attr.name) LIKE LOWER(%s)
|
|
ORDER BY e.name ASC;
|
|
"""
|
|
exercises = self.execute(query, [category_part, value_part])
|
|
else:
|
|
# Fallback: search in name OR attribute name
|
|
search_term = query.strip().lower()
|
|
search_query = f"%{search_term}%"
|
|
query = """
|
|
SELECT DISTINCT e.exercise_id, e.name
|
|
FROM exercise e
|
|
LEFT JOIN exercise_to_attribute eta ON e.exercise_id = eta.exercise_id
|
|
LEFT JOIN exercise_attribute attr ON eta.attribute_id = attr.attribute_id
|
|
WHERE LOWER(e.name) LIKE LOWER(%s) OR LOWER(attr.name) LIKE LOWER(%s)
|
|
ORDER BY e.name ASC;
|
|
"""
|
|
exercises = self.execute(query, [search_query, search_query])
|
|
|
|
for ex in exercises:
|
|
ex['attributes'] = self.get_exercise_attributes(ex['exercise_id'])
|
|
|
|
return exercises
|
|
|
|
def get_exercise(self, exercise_id):
|
|
exercise = self.execute("SELECT exercise_id, name FROM exercise WHERE exercise_id=%s;", [exercise_id], one=True)
|
|
if exercise:
|
|
exercise['attributes'] = self.get_exercise_attributes(exercise_id)
|
|
return exercise
|
|
|
|
def get_exercise_attributes(self, exercise_id):
|
|
query = """
|
|
SELECT cat.name as category_name, attr.attribute_id, attr.name as attribute_name
|
|
FROM exercise_to_attribute eta
|
|
JOIN exercise_attribute attr ON eta.attribute_id = attr.attribute_id
|
|
JOIN exercise_attribute_category cat ON attr.category_id = cat.category_id
|
|
WHERE eta.exercise_id = %s
|
|
ORDER BY cat.name, attr.name
|
|
"""
|
|
return self.execute(query, [exercise_id])
|
|
|
|
def get_all_attribute_categories(self):
|
|
return self.execute("SELECT category_id, name FROM exercise_attribute_category ORDER BY name")
|
|
|
|
def get_attributes_by_category(self):
|
|
# Returns a dict: { category_name: [ {id, name}, ... ] }
|
|
categories = self.get_all_attribute_categories()
|
|
all_attrs = self.execute("SELECT attribute_id, name, category_id FROM exercise_attribute ORDER BY name")
|
|
|
|
result = {}
|
|
for cat in categories:
|
|
result[cat['name']] = [a for a in all_attrs if a['category_id'] == cat['category_id']]
|
|
return result
|
|
|
|
def update_exercise(self, exercise_id, name, attribute_ids=None):
|
|
self.execute("UPDATE exercise SET name = %s WHERE exercise_id = %s;", [name, exercise_id], commit=True)
|
|
|
|
# Update attributes: simple delete and re-insert for now
|
|
self.execute("DELETE FROM exercise_to_attribute WHERE exercise_id = %s", [exercise_id], commit=True)
|
|
|
|
if attribute_ids:
|
|
for attr_id in attribute_ids:
|
|
if attr_id:
|
|
self.execute("INSERT INTO exercise_to_attribute (exercise_id, attribute_id) VALUES (%s, %s)",
|
|
[exercise_id, attr_id], commit=True)
|
|
|
|
return self.get_exercise(exercise_id)
|
|
|
|
def delete_exercise(self, exercise_id):
|
|
self.execute('DELETE FROM exercise WHERE exercise_id=%s', [
|
|
exercise_id], commit=True)
|
|
|
|
def add_exercise(self, name, attribute_ids=None):
|
|
result = self.execute('INSERT INTO exercise (name) VALUES (%s) RETURNING exercise_id', [name], commit=True, one=True)
|
|
exercise_id = result['exercise_id']
|
|
|
|
if attribute_ids:
|
|
for attr_id in attribute_ids:
|
|
if attr_id:
|
|
self.execute("INSERT INTO exercise_to_attribute (exercise_id, attribute_id) VALUES (%s, %s)",
|
|
[exercise_id, attr_id], commit=True)
|
|
|
|
return self.get_exercise(exercise_id)
|
|
|
|
def get_workout_attribute_distribution(self, workout_id, category_name):
|
|
query = """
|
|
SELECT attr.name as attribute_name, COUNT(*) as count
|
|
FROM topset t
|
|
JOIN exercise e ON t.exercise_id = e.exercise_id
|
|
JOIN exercise_to_attribute eta ON e.exercise_id = eta.exercise_id
|
|
JOIN exercise_attribute attr ON eta.attribute_id = attr.attribute_id
|
|
JOIN exercise_attribute_category cat ON attr.category_id = cat.category_id
|
|
WHERE t.workout_id = %s AND cat.name = %s
|
|
GROUP BY attr.name
|
|
ORDER BY count DESC
|
|
"""
|
|
distribution = self.execute(query, [workout_id, category_name])
|
|
|
|
# Calculate percentages and SVG parameters
|
|
total_counts = sum(item['count'] for item in distribution)
|
|
accumulated_percentage = 0
|
|
|
|
# Color palette for segments
|
|
colors = [
|
|
"#3b82f6", # blue-500
|
|
"#06b6d4", # cyan-500
|
|
"#8b5cf6", # violet-500
|
|
"#ec4899", # pink-500
|
|
"#f59e0b", # amber-500
|
|
"#10b981", # emerald-500
|
|
"#6366f1", # indigo-500
|
|
"#f43f5e", # rose-500
|
|
"#84cc16", # lime-500
|
|
"#0ea5e9", # sky-500
|
|
]
|
|
|
|
if total_counts > 0:
|
|
for i, item in enumerate(distribution):
|
|
percentage = (item['count'] / total_counts) * 100
|
|
item['percentage'] = round(percentage)
|
|
item['dasharray'] = f"{percentage} 100"
|
|
item['dashoffset'] = -accumulated_percentage
|
|
item['color'] = colors[i % len(colors)]
|
|
accumulated_percentage += percentage
|
|
|
|
return distribution
|
|
|
|
# Category Management
|
|
def add_category(self, name):
|
|
result = self.execute('INSERT INTO exercise_attribute_category (name) VALUES (%s) RETURNING category_id, name', [name], commit=True, one=True)
|
|
return result
|
|
|
|
def update_category(self, category_id, name):
|
|
self.execute('UPDATE exercise_attribute_category SET name = %s WHERE category_id = %s', [name, category_id], commit=True)
|
|
return {"category_id": category_id, "name": name}
|
|
|
|
def delete_category(self, category_id):
|
|
# First delete all attributes in this category
|
|
attributes = self.execute('SELECT attribute_id FROM exercise_attribute WHERE category_id = %s', [category_id])
|
|
for attr in attributes:
|
|
self.delete_attribute(attr['attribute_id'])
|
|
|
|
self.execute('DELETE FROM exercise_attribute_category WHERE category_id = %s', [category_id], commit=True)
|
|
|
|
# Attribute Management
|
|
def add_attribute(self, name, category_id):
|
|
result = self.execute('INSERT INTO exercise_attribute (name, category_id) VALUES (%s, %s) RETURNING attribute_id, name, category_id', [name, category_id], commit=True, one=True)
|
|
return result
|
|
|
|
def update_attribute(self, attribute_id, name, category_id=None):
|
|
if category_id:
|
|
self.execute('UPDATE exercise_attribute SET name = %s, category_id = %s WHERE attribute_id = %s', [name, category_id, attribute_id], commit=True)
|
|
else:
|
|
self.execute('UPDATE exercise_attribute SET name = %s WHERE attribute_id = %s', [name, attribute_id], commit=True)
|
|
return self.execute('SELECT attribute_id, name, category_id FROM exercise_attribute WHERE attribute_id = %s', [attribute_id], one=True)
|
|
|
|
def delete_attribute(self, attribute_id):
|
|
# Remove from all exercises first
|
|
self.execute('DELETE FROM exercise_to_attribute WHERE attribute_id = %s', [attribute_id], commit=True)
|
|
# Delete the attribute
|
|
self.execute('DELETE FROM exercise_attribute WHERE attribute_id = %s', [attribute_id], commit=True)
|
|
|