Spaces:
Sleeping
Sleeping
File size: 4,181 Bytes
b84ea83 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 | """
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}
|