Spaces:
Running
Running
File size: 5,780 Bytes
d951fea 07fa724 335b256 a4cc435 294884c d951fea 07fa724 d951fea 07fa724 d951fea 335b256 4bb967d 335b256 4bb967d a4cc435 4bb967d a4cc435 4bb967d 335b256 07fa724 d951fea 07fa724 d951fea 07fa724 d951fea 07fa724 d951fea 07fa724 335b256 07fa724 d951fea 07fa724 d951fea 07fa724 a4cc435 335b256 a4cc435 294884c a4cc435 294884c 335b256 a4cc435 335b256 294884c 07fa724 294884c 07fa724 4bb967d 07fa724 d951fea |
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 |
from pymongo import MongoClient
from datetime import datetime, timedelta
from typing import Dict, List, Union
from gamification.objects import Points,PointMultipliersWeekendWarrior,PlatformEngagement
from gamification.logic import create_points_func
from concurrent.futures import ThreadPoolExecutor
executor = ThreadPoolExecutor(max_workers=5)
def get_current_date() -> str:
# Get the current date and return it in YYYY-MM-DD format
return datetime.now().date().isoformat()
def get_another_date(day: int) -> str:
# Get the current date and add specified days to it
tomorrow = datetime.now() + timedelta(days=day)
return tomorrow.date().isoformat()
def check_date_streak(start_date: str, end_date: str) -> bool:
# Convert ISO strings to datetime objects
start_date = datetime.fromisoformat(start_date)
end_date = datetime.fromisoformat(end_date)
# Compare the dates to check if they are consecutive (1 day apart)
if end_date - start_date == timedelta(days=0) or end_date - start_date == timedelta(days=1):
return True
else:
return False
def save_in_streaks_history(db_uri,document):
client = MongoClient(db_uri)
document.pop("_id",None)
print(document)
db = client["crayonics"]
collection = db["StreaksHistory"]
# Check for existing user
found_user = collection.find_one({"user_id": document['user_id']})
current_date = get_current_date()
# try:
if found_user is None:
# New user case
print("added new streak reset record")
document["streaks_records"] = [{"date_of_streak_reset":current_date,"streak_dates":document["streak_dates"]}]
document.pop("streak_dates")
streakPoints= Points(userId=document['user_id'],platformEngagement=PlatformEngagement(daily_check_in=5))
create_points_func(document=streakPoints)
collection.insert_one(document)
return True
else:
print("added another streak reset record")
current_record={"date_of_streaks_reset":current_date,"streak_dates":document["streak_dates"]}
today = datetime.date(datetime.now())
if (today.weekday()>=5):
streakPoints= Points(userId=document['user_id'],platformEngagement=PlatformEngagement(daily_check_in=5),weekendWarrior=PointMultipliersWeekendWarrior())
else:
streakPoints= Points(userId=document['user_id'],platformEngagement=PlatformEngagement(daily_check_in=5))
wasCreated= create_points_func(document=streakPoints)
dates = found_user["streaks_records"] + [current_record]
collection.update_one(
{"user_id": document.get("user_id")},
{"$set": {"streaks_records": dates}}
)
return True
# except Exception as e:
# print(f"Error in creating a streaks reset record: {str(e)}")
# print(document,found_user)
# return False
# finally:
# client.close()
def streaks_manager(db_uri: str, document: Dict) -> Union[bool, str]:
"""
Manage user streaks in MongoDB.
Args:
db_uri: MongoDB connection string
document: Dictionary containing user data with 'user_id' key
Returns:
Union[bool, str]: True if successful, "User Already Exists" if streak is reset
"""
client = MongoClient(db_uri)
try:
db = client["crayonics"]
collection = db["Streaks"]
# Check for existing user
found_user = collection.find_one({"user_id": document.get('user_id')})
current_date = get_current_date()
document['streak_dates'] = [current_date]
if found_user is None:
# New user case
collection.insert_one(document)
return True
else:
is_a_streak = check_date_streak(
start_date=found_user['streak_dates'][-1],
end_date=current_date
)
if is_a_streak:
print("its a streak guys")
# Extend existing streak
if not ((current_date)== (found_user["streak_dates"][-1])):
dates = found_user["streak_dates"] + [current_date]
print("an actual streak")
today = datetime.date(datetime.now())
if (today.weekday()>=5):
streakPoints= Points(userId=document['user_id'],platformEngagement=PlatformEngagement(daily_check_in=5),weekendWarrior=PointMultipliersWeekendWarrior())
executor.submit(create_points_func,document=streakPoints)
# create_points_func(document=streakPoints)
else:
streakPoints= Points(userId=document['user_id'],platformEngagement=PlatformEngagement(daily_check_in=5))
executor.submit(create_points_func,document=streakPoints)
# create_points_func(document=streakPoints)
else:
dates=found_user["streak_dates"]
result = collection.update_one(
{"user_id": document.get("user_id")},
{"$set": {"streak_dates": dates}}
)
print(result)
return True
else:
save_in_streaks_history(db_uri=db_uri,document=found_user)
# Reset streak if not consecutive
collection.find_one_and_replace(
{"user_id": document.get('user_id')},
document
)
return "User Already Exists"
except Exception as e:
print(f"Error: {str(e)}")
raise
finally:
client.close()
|