Spaces:
Sleeping
Sleeping
File size: 6,086 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 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 | """
Ghost Shopper Analytics Service — Assessment-level analytics.
Provides:
- Outlet comparison (avg scores across all outlets)
- Category breakdown (which category is weakest/strongest)
- Weekly trend analysis per outlet
"""
import uuid
from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncSession
CATEGORY_LABELS = {
"A": "Pelayanan Awal",
"B": "Kualitas Produk",
"C": "Kebersihan & Suasana",
"D": "Kepatuhan SOP",
"E": "Keseluruhan Pengalaman",
}
class GhostShopperAnalyticsService:
"""Analytics service for Ghost Shopper assessment data."""
async def get_outlet_comparison(self, db: AsyncSession) -> dict:
"""
Compare all outlets side by side:
- Overall avg score per outlet
- Per-category avg score per outlet
"""
# Overall scores per outlet
result = await db.execute(text("""
SELECT
o.id, o.name,
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, o.name
ORDER BY avg_score DESC
"""))
outlets = []
for row in result.fetchall():
outlets.append({
"id": str(row[0]),
"name": row[1],
"visit_count": row[2],
"avg_score": float(row[3]),
})
# Per-category scores per outlet
cat_result = await db.execute(text("""
SELECT
o.name, a.category,
ROUND(AVG(a.score)::numeric, 2) AS avg_score
FROM outlets o
JOIN visits v ON v.outlet_id = o.id
JOIN assessments a ON a.visit_id = v.id
GROUP BY o.name, a.category
ORDER BY o.name, a.category
"""))
category_matrix = {}
for row in cat_result.fetchall():
outlet_name = row[0]
if outlet_name not in category_matrix:
category_matrix[outlet_name] = {}
category_matrix[outlet_name][row[1]] = float(row[2])
return {
"status": "success",
"data": {
"outlets": outlets,
"category_matrix": category_matrix,
"category_labels": CATEGORY_LABELS,
},
}
async def get_outlet_trends(
self, db: AsyncSession, outlet_id: uuid.UUID
) -> dict | None:
"""
Weekly score trends for a specific outlet.
Returns chronological list of visits with per-category scores.
"""
# Verify outlet exists
result = await db.execute(
text("SELECT name FROM outlets WHERE id = :oid"),
{"oid": outlet_id},
)
row = result.fetchone()
if not row:
return None
outlet_name = row[0]
# Get per-visit scores over time
trend_result = await db.execute(text("""
SELECT
v.id, v.title, v.visit_date,
ROUND(AVG(a.score)::numeric, 2) AS avg_score,
ROUND(AVG(CASE WHEN a.category = 'A' THEN a.score END)::numeric, 2) AS cat_a,
ROUND(AVG(CASE WHEN a.category = 'B' THEN a.score END)::numeric, 2) AS cat_b,
ROUND(AVG(CASE WHEN a.category = 'C' THEN a.score END)::numeric, 2) AS cat_c,
ROUND(AVG(CASE WHEN a.category = 'D' THEN a.score END)::numeric, 2) AS cat_d,
ROUND(AVG(CASE WHEN a.category = 'E' THEN a.score END)::numeric, 2) AS cat_e
FROM visits v
JOIN assessments a ON a.visit_id = v.id
WHERE v.outlet_id = :oid
GROUP BY v.id, v.title, v.visit_date
ORDER BY v.visit_date ASC
"""), {"oid": outlet_id})
trends = []
for tr in trend_result.fetchall():
trends.append({
"visit_id": str(tr[0]),
"title": tr[1],
"visit_date": tr[2].isoformat() if tr[2] else None,
"avg_score": float(tr[3]) if tr[3] else 0.0,
"categories": {
"A": float(tr[4]) if tr[4] else None,
"B": float(tr[5]) if tr[5] else None,
"C": float(tr[6]) if tr[6] else None,
"D": float(tr[7]) if tr[7] else None,
"E": float(tr[8]) if tr[8] else None,
},
})
return {
"status": "success",
"data": {
"outlet_id": str(outlet_id),
"outlet_name": outlet_name,
"trends": trends,
"category_labels": CATEGORY_LABELS,
},
}
async def get_category_summary(self, db: AsyncSession) -> dict:
"""
Global category breakdown: average score per assessment category
across all outlets and visits. Identifies strengths and weaknesses.
"""
result = await db.execute(text("""
SELECT
a.category,
ROUND(AVG(a.score)::numeric, 2) AS avg_score,
COUNT(a.id) AS total_items,
ROUND(MIN(a.score)::numeric, 2) AS min_score,
ROUND(MAX(a.score)::numeric, 2) AS max_score
FROM assessments a
GROUP BY a.category
ORDER BY a.category
"""))
categories = []
for row in result.fetchall():
categories.append({
"category": row[0],
"label": CATEGORY_LABELS.get(row[0], row[0]),
"avg_score": float(row[1]),
"total_items": row[2],
"min_score": float(row[3]),
"max_score": float(row[4]),
})
return {
"status": "success",
"data": {"categories": categories},
}
# Singleton
gs_analytics_service = GhostShopperAnalyticsService()
|