Spaces:
Sleeping
Sleeping
File size: 3,467 Bytes
4fafc3c |
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 |
"""
Feedback Tracking Service
Tracks which events users have already given feedback for
"""
from typing import Optional, Dict
from pymongo.collection import Collection
from datetime import datetime
class FeedbackTrackingService:
"""
Track feedback status per user per event
Prevents redundant "check purchase history" calls
"""
def __init__(self, mongo_collection: Collection):
self.collection = mongo_collection
self._ensure_indexes()
def _ensure_indexes(self):
"""Create indexes for fast lookup"""
try:
# Compound index for quick lookup
self.collection.create_index([("user_id", 1), ("event_code", 1)], unique=True)
self.collection.create_index("user_id")
print("✓ Feedback tracking indexes created")
except Exception as e:
print(f"Feedback tracking indexes exist: {e}")
def has_given_feedback(self, user_id: str, event_code: str) -> bool:
"""
Check if user has already given feedback for this event
Args:
user_id: User ID
event_code: Event code
Returns:
True if feedback already given, False otherwise
"""
result = self.collection.find_one({
"user_id": user_id,
"event_code": event_code,
"is_feedback": True
})
return result is not None
def mark_feedback_given(self, user_id: str, event_code: str, rating: int, comment: str = "") -> bool:
"""
Mark that user has given feedback for this event
Args:
user_id: User ID
event_code: Event code
rating: Rating given (1-5)
comment: Feedback comment
Returns:
True if saved successfully
"""
try:
self.collection.update_one(
{
"user_id": user_id,
"event_code": event_code
},
{
"$set": {
"is_feedback": True,
"rating": rating,
"comment": comment,
"feedback_date": datetime.utcnow(),
"updated_at": datetime.utcnow()
},
"$setOnInsert": {
"created_at": datetime.utcnow()
}
},
upsert=True
)
print(f"✅ Marked feedback: {user_id} → {event_code} (rating: {rating})")
return True
except Exception as e:
print(f"❌ Error marking feedback: {e}")
return False
def get_pending_events(self, user_id: str, purchased_events: list) -> list:
"""
Filter purchased events to only those without feedback
Args:
user_id: User ID
purchased_events: List of events user has purchased
Returns:
List of events that need feedback
"""
pending = []
for event in purchased_events:
event_code = event.get("eventCode")
if event_code and not self.has_given_feedback(user_id, event_code):
pending.append(event)
return pending
|