feat: Add endpoints for activity statistics and uncategorized activities
All checks were successful
Deploy Development / deploy (push) Successful in 48s
Build Test / pytest-backend (push) Successful in 4s
Build Test / lint-backend (push) Successful in 0s
Build Test / build-frontend (push) Successful in 16s

- Implemented a new endpoint to retrieve activity statistics for the last 30 entries, including total calories and duration by activity type.
- Added an endpoint to list activities without assigned training types, grouped by activity type.
- Removed deprecated versions of the statistics and uncategorized activities endpoints for cleaner code.
This commit is contained in:
Lars 2026-04-14 12:32:38 +02:00
parent 196b6c5cf1
commit db9952525a

View File

@ -133,6 +133,66 @@ def create_activity(e: ActivityEntry, x_profile_id: Optional[str]=Header(default
return {"id":eid,"date":e.date}
@router.get("/stats")
def activity_stats(session: dict = Depends(require_auth)):
"""Get activity statistics (last 30 entries)."""
pid = str(session["profile_id"])
with get_db() as conn:
cur = get_cursor(conn)
cur.execute("SELECT * FROM profiles WHERE id=%s", (pid,))
profile = r2d(cur.fetchone())
quality_filter = get_quality_filter_sql(profile or {})
cur.execute(
f"""
SELECT * FROM activity_log
WHERE profile_id=%s {quality_filter}
ORDER BY date DESC
LIMIT 30
""",
(pid,),
)
rows = [r2d(r) for r in cur.fetchall()]
if not rows:
return {"count": 0, "total_kcal": 0, "total_min": 0, "by_type": {}}
total_kcal = sum(float(r.get("kcal_active") or 0) for r in rows)
total_min = sum(float(r.get("duration_min") or 0) for r in rows)
by_type = {}
for r in rows:
t = r["activity_type"]
by_type.setdefault(t, {"count": 0, "kcal": 0, "min": 0})
by_type[t]["count"] += 1
by_type[t]["kcal"] += float(r.get("kcal_active") or 0)
by_type[t]["min"] += float(r.get("duration_min") or 0)
return {
"count": len(rows),
"total_kcal": round(total_kcal),
"total_min": round(total_min),
"by_type": by_type,
}
@router.get("/uncategorized")
def list_uncategorized_activities(
session: dict = Depends(require_auth),
):
"""Get activities without assigned training type, grouped by activity_type."""
pid = str(session["profile_id"])
with get_db() as conn:
cur = get_cursor(conn)
cur.execute(
"""
SELECT activity_type, COUNT(*) as count,
MIN(date) as first_date, MAX(date) as last_date
FROM activity_log
WHERE profile_id=%s AND training_type_id IS NULL
GROUP BY activity_type
ORDER BY count DESC
""",
(pid,),
)
return [r2d(r) for r in cur.fetchall()]
@router.put("/{eid}")
def update_activity(eid: str, e: ActivityEntry, x_profile_id: Optional[str]=Header(default=None), session: dict=Depends(require_auth)):
"""Update existing activity entry."""
@ -228,37 +288,6 @@ def get_activity_session(
return unit
@router.get("/stats")
def activity_stats(session: dict = Depends(require_auth)):
"""Get activity statistics (last 30 entries)."""
pid = str(session["profile_id"])
with get_db() as conn:
cur = get_cursor(conn)
cur.execute("SELECT * FROM profiles WHERE id=%s", (pid,))
profile = r2d(cur.fetchone())
quality_filter = get_quality_filter_sql(profile or {})
cur.execute(
f"""
SELECT * FROM activity_log
WHERE profile_id=%s {quality_filter}
ORDER BY date DESC
LIMIT 30
""",
(pid,),
)
rows = [r2d(r) for r in cur.fetchall()]
if not rows: return {"count":0,"total_kcal":0,"total_min":0,"by_type":{}}
total_kcal=sum(float(r.get('kcal_active') or 0) for r in rows)
total_min=sum(float(r.get('duration_min') or 0) for r in rows)
by_type={}
for r in rows:
t=r['activity_type']; by_type.setdefault(t,{'count':0,'kcal':0,'min':0})
by_type[t]['count']+=1
by_type[t]['kcal']+=float(r.get('kcal_active') or 0)
by_type[t]['min']+=float(r.get('duration_min') or 0)
return {"count":len(rows),"total_kcal":round(total_kcal),"total_min":round(total_min),"by_type":by_type}
def get_training_type_for_activity_with_cursor(cur, activity_type: str, profile_id: str | None = None):
"""
Wie get_training_type_for_activity, aber mit bestehendem Cursor (z. B. Universal-CSV-Import).
@ -312,23 +341,6 @@ def get_training_type_for_activity(activity_type: str, profile_id: str = None):
return get_training_type_for_activity_with_cursor(cur, activity_type, profile_id)
@router.get("/uncategorized")
def list_uncategorized_activities(x_profile_id: Optional[str]=Header(default=None), session: dict=Depends(require_auth)):
"""Get activities without assigned training type, grouped by activity_type."""
pid = get_pid(x_profile_id)
with get_db() as conn:
cur = get_cursor(conn)
cur.execute("""
SELECT activity_type, COUNT(*) as count,
MIN(date) as first_date, MAX(date) as last_date
FROM activity_log
WHERE profile_id=%s AND training_type_id IS NULL
GROUP BY activity_type
ORDER BY count DESC
""", (pid,))
return [r2d(r) for r in cur.fetchall()]
@router.post("/bulk-categorize")
def bulk_categorize_activities(
data: dict,