Spaces:
Sleeping
Sleeping
File size: 4,535 Bytes
4fafc3c 24f10ad |
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 |
"""
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
def get_feedbacked_events(self, user_id: str) -> list:
"""
Get all events where a user has already given feedback (is_feedback: True)
Args:
user_id: User ID
Returns:
List of event_codes with feedback info
"""
try:
results = self.collection.find({
"user_id": user_id,
"is_feedback": True
})
feedbacked = []
for doc in results:
feedbacked.append({
"event_code": doc.get("event_code"),
"rating": doc.get("rating"),
"comment": doc.get("comment"),
"feedback_date": doc.get("feedback_date")
})
print(f"π Found {len(feedbacked)} feedbacked events for user {user_id}")
return feedbacked
except Exception as e:
print(f"β Error getting feedbacked events: {e}")
return []
|