60 lines
2.6 KiB
Python
60 lines
2.6 KiB
Python
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy import select, func
|
|
from app.models.achievement import Achievement, UserAchievement
|
|
from app.models.scan import Scan
|
|
|
|
ACHIEVEMENT_DEFS = [
|
|
{"code": "first_scan", "name": "Primeiro Scan", "description": "Escaneie seu primeiro produto", "emoji": "🎯", "target": 1},
|
|
{"code": "detective", "name": "Detetive de Rótulos", "description": "Escaneie 10 produtos", "emoji": "🔍", "target": 10},
|
|
{"code": "expert", "name": "Expert", "description": "Escaneie 50 produtos", "emoji": "🧠", "target": 50},
|
|
{"code": "master", "name": "Master", "description": "Escaneie 100 produtos", "emoji": "👑", "target": 100},
|
|
{"code": "photographer", "name": "Fotógrafo", "description": "Analise 5 produtos por foto", "emoji": "📸", "target": 5},
|
|
{"code": "comparator", "name": "Comparador", "description": "Compare produtos 3 vezes", "emoji": "⚖️", "target": 3},
|
|
]
|
|
|
|
async def seed_achievements(db: AsyncSession):
|
|
for a_def in ACHIEVEMENT_DEFS:
|
|
res = await db.execute(select(Achievement).where(Achievement.code == a_def["code"]))
|
|
if not res.scalar_one_or_none():
|
|
db.add(Achievement(**a_def))
|
|
await db.commit()
|
|
|
|
async def check_achievements(user_id: int, db: AsyncSession, action: str = "scan"):
|
|
"""Check and unlock achievements after an action."""
|
|
# Get all achievements
|
|
res = await db.execute(select(Achievement))
|
|
all_achievements = {a.code: a for a in res.scalars().all()}
|
|
|
|
# Get already unlocked
|
|
unlocked_res = await db.execute(select(UserAchievement).where(UserAchievement.user_id == user_id))
|
|
unlocked_ids = {ua.achievement_id for ua in unlocked_res.scalars().all()}
|
|
|
|
# Count scans
|
|
total_res = await db.execute(select(func.count(Scan.id)).where(Scan.user_id == user_id))
|
|
total_scans = total_res.scalar() or 0
|
|
|
|
photo_res = await db.execute(
|
|
select(func.count(Scan.id)).where(Scan.user_id == user_id, Scan.barcode == "PHOTO")
|
|
)
|
|
photo_scans = photo_res.scalar() or 0
|
|
|
|
checks = {
|
|
"first_scan": total_scans >= 1,
|
|
"detective": total_scans >= 10,
|
|
"expert": total_scans >= 50,
|
|
"master": total_scans >= 100,
|
|
"photographer": photo_scans >= 5,
|
|
}
|
|
|
|
new_unlocked = []
|
|
for code, condition in checks.items():
|
|
if code in all_achievements and all_achievements[code].id not in unlocked_ids and condition:
|
|
ua = UserAchievement(user_id=user_id, achievement_id=all_achievements[code].id)
|
|
db.add(ua)
|
|
new_unlocked.append(all_achievements[code].name)
|
|
|
|
if new_unlocked:
|
|
await db.commit()
|
|
|
|
return new_unlocked
|