Barcode Betty
76 lines
2.6 KiB
Python
76 lines
2.6 KiB
Python
"""Alert service — price and shrinkflation alerts for users.
|
|
|
|
Alerts are generated by StickerShock and ShrinkRay services and written to the DB.
|
|
This service reads them for the API gateway.
|
|
"""
|
|
|
|
from uuid import UUID
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy.orm import selectinload
|
|
|
|
|
|
class AlertService:
|
|
def __init__(self, db: AsyncSession) -> None:
|
|
self.db = db
|
|
|
|
async def list_alerts(self, user_id: UUID) -> list[dict]:
|
|
"""List shrinkflation events for products the user has purchased."""
|
|
from cartsnitch_api.models import Purchase, PurchaseItem, ShrinkflationEvent
|
|
|
|
# Get product IDs from user's purchases
|
|
items_result = await self.db.execute(
|
|
select(PurchaseItem.normalized_product_id)
|
|
.join(Purchase)
|
|
.where(
|
|
Purchase.user_id == user_id,
|
|
PurchaseItem.normalized_product_id.isnot(None),
|
|
)
|
|
.distinct()
|
|
)
|
|
product_ids = [row[0] for row in items_result.all()]
|
|
|
|
if not product_ids:
|
|
return []
|
|
|
|
result = await self.db.execute(
|
|
select(ShrinkflationEvent)
|
|
.where(ShrinkflationEvent.normalized_product_id.in_(product_ids))
|
|
.options(selectinload(ShrinkflationEvent.normalized_product))
|
|
.order_by(ShrinkflationEvent.detected_date.desc())
|
|
)
|
|
events = result.scalars().all()
|
|
|
|
return [
|
|
{
|
|
"id": e.id,
|
|
"alert_type": "shrinkflation",
|
|
"product_id": e.normalized_product_id,
|
|
"product_name": e.normalized_product.canonical_name,
|
|
"message": (
|
|
f"Size changed from {e.old_size}{e.old_unit} to {e.new_size}{e.new_unit}"
|
|
),
|
|
"triggered_at": e.detected_date,
|
|
"read": False,
|
|
}
|
|
for e in events
|
|
]
|
|
|
|
async def get_settings(self, user_id: UUID) -> dict:
|
|
# Alert settings would be stored in a user_settings table.
|
|
# For now, return defaults since the table doesn't exist yet in common lib.
|
|
return {
|
|
"price_increase_threshold_pct": 5.0,
|
|
"shrinkflation_enabled": True,
|
|
"email_notifications": False,
|
|
}
|
|
|
|
async def update_settings(self, user_id: UUID, **fields) -> dict:
|
|
# Would update user_settings table. Return merged defaults for now.
|
|
current = await self.get_settings(user_id)
|
|
for k, v in fields.items():
|
|
if v is not None and k in current:
|
|
current[k] = v
|
|
return current
|