Spaces:
Sleeping
Sleeping
| """ | |
| Outlet Controller — CRUD operations for coffee-shop outlets. | |
| Functions: | |
| get_all_outlets — List all outlets with visit count and avg score | |
| get_outlet_by_id — Get outlet detail with visits and score breakdown | |
| """ | |
| import uuid | |
| from sqlalchemy import text | |
| from sqlalchemy.ext.asyncio import AsyncSession | |
| from models.outlet_model import Outlet, OutletResponse | |
| async def get_all_outlets(db: AsyncSession) -> dict: | |
| """ | |
| List all outlets with aggregated visit count and average assessment score. | |
| Returns sorted by average score (descending). | |
| """ | |
| result = await db.execute(text(""" | |
| SELECT | |
| o.id, o.name, o.location, o.description, | |
| o.created_at, o.updated_at, | |
| COUNT(DISTINCT v.id) AS visit_count, | |
| COALESCE(ROUND(AVG(a.score)::numeric, 2), 0) AS avg_score | |
| FROM outlets o | |
| LEFT JOIN visits v ON v.outlet_id = o.id | |
| LEFT JOIN assessments a ON a.visit_id = v.id | |
| GROUP BY o.id | |
| ORDER BY avg_score DESC, o.name | |
| """)) | |
| rows = result.fetchall() | |
| data = [] | |
| for row in rows: | |
| data.append({ | |
| "id": str(row[0]), | |
| "name": row[1], | |
| "location": row[2], | |
| "description": row[3], | |
| "created_at": row[4].isoformat() if row[4] else None, | |
| "updated_at": row[5].isoformat() if row[5] else None, | |
| "visit_count": row[6], | |
| "avg_score": float(row[7]), | |
| }) | |
| return {"status": "success", "data": data} | |
| async def get_outlet_by_id(db: AsyncSession, outlet_id: uuid.UUID) -> dict | None: | |
| """ | |
| Get outlet detail including: | |
| - Outlet info | |
| - Visit list with dates and scores | |
| - Category breakdown (avg score per A–E) | |
| """ | |
| # Check outlet exists | |
| result = await db.execute( | |
| text("SELECT id, name, location, description, created_at, updated_at FROM outlets WHERE id = :oid"), | |
| {"oid": outlet_id}, | |
| ) | |
| row = result.fetchone() | |
| if not row: | |
| return None | |
| outlet_data = { | |
| "id": str(row[0]), | |
| "name": row[1], | |
| "location": row[2], | |
| "description": row[3], | |
| "created_at": row[4].isoformat() if row[4] else None, | |
| "updated_at": row[5].isoformat() if row[5] else None, | |
| } | |
| # Get visits for this outlet with avg scores | |
| visit_result = await db.execute(text(""" | |
| SELECT | |
| v.id, v.title, v.visit_date, v.ghost_shopper, v.visit_type, | |
| v.comments, v.suggestions, | |
| COALESCE(ROUND(AVG(a.score)::numeric, 2), 0) AS avg_score | |
| FROM visits v | |
| LEFT JOIN assessments a ON a.visit_id = v.id | |
| WHERE v.outlet_id = :oid | |
| GROUP BY v.id | |
| ORDER BY v.visit_date DESC | |
| """), {"oid": outlet_id}) | |
| visits = [] | |
| for vr in visit_result.fetchall(): | |
| visits.append({ | |
| "id": str(vr[0]), | |
| "title": vr[1], | |
| "visit_date": vr[2].isoformat() if vr[2] else None, | |
| "ghost_shopper": vr[3], | |
| "visit_type": vr[4], | |
| "comments": vr[5], | |
| "suggestions": vr[6], | |
| "avg_score": float(vr[7]), | |
| }) | |
| # Get category breakdown (A–E average scores) | |
| cat_result = await db.execute(text(""" | |
| SELECT | |
| a.category, | |
| ROUND(AVG(a.score)::numeric, 2) AS avg_score, | |
| COUNT(a.id) AS item_count | |
| FROM assessments a | |
| JOIN visits v ON v.id = a.visit_id | |
| WHERE v.outlet_id = :oid | |
| GROUP BY a.category | |
| ORDER BY a.category | |
| """), {"oid": outlet_id}) | |
| categories = {} | |
| category_labels = { | |
| "A": "Pelayanan Awal", | |
| "B": "Kualitas Produk", | |
| "C": "Kebersihan & Suasana", | |
| "D": "Kepatuhan SOP", | |
| "E": "Keseluruhan Pengalaman", | |
| } | |
| for cr in cat_result.fetchall(): | |
| categories[cr[0]] = { | |
| "label": category_labels.get(cr[0], cr[0]), | |
| "avg_score": float(cr[1]), | |
| "item_count": cr[2], | |
| } | |
| outlet_data["visits"] = visits | |
| outlet_data["visit_count"] = len(visits) | |
| outlet_data["categories"] = categories | |
| return {"status": "success", "data": outlet_data} | |