forked from retoor/devplacepy
refactor: split utils.py into package (ref.md 3.8)
This commit is contained in:
@@ -0,0 +1,215 @@
|
||||
# retoor <retoor@molodetz.nl>
|
||||
|
||||
from devplacepy.database import (
|
||||
get_table,
|
||||
get_user_stars,
|
||||
record_activity,
|
||||
record_unique_activity,
|
||||
)
|
||||
from devplacepy.services.background import background
|
||||
from devplacepy.utils.authcache import clear_user_cache
|
||||
from devplacepy.utils.badges import award_badge, notify_badge, LEVEL_BADGES
|
||||
from devplacepy.utils.notifications import create_notification
|
||||
|
||||
LEVEL_XP = 100
|
||||
|
||||
XP_POST = 10
|
||||
XP_COMMENT = 2
|
||||
XP_PROJECT = 15
|
||||
XP_GIST = 5
|
||||
XP_UPVOTE = 5
|
||||
XP_FOLLOW = 5
|
||||
|
||||
|
||||
def level_for_xp(xp: int) -> int:
|
||||
return 1 + max(0, xp) // LEVEL_XP
|
||||
|
||||
|
||||
def award_xp(user_uid: str, amount: int) -> dict:
|
||||
users = get_table("users")
|
||||
user = users.find_one(uid=user_uid)
|
||||
if not user or amount <= 0:
|
||||
return {
|
||||
"xp": user.get("xp", 0) if user else 0,
|
||||
"level": user.get("level", 1) if user else 1,
|
||||
"leveled_up": False,
|
||||
}
|
||||
current_xp = user.get("xp", 0) or 0
|
||||
current_level = user.get("level", 1) or 1
|
||||
new_xp = max(0, current_xp + amount)
|
||||
new_level = level_for_xp(new_xp)
|
||||
users.update({"uid": user_uid, "xp": new_xp, "level": new_level}, ["uid"])
|
||||
clear_user_cache(user_uid)
|
||||
leveled_up = new_level > current_level
|
||||
from devplacepy.services.audit import record as audit
|
||||
|
||||
audit.record_system(
|
||||
"reward.xp.grant",
|
||||
actor_kind="user",
|
||||
actor_uid=user_uid,
|
||||
new_value=amount,
|
||||
metadata={"amount": amount, "total_xp": new_xp},
|
||||
summary=f"user earned {amount} XP",
|
||||
links=[audit.actor(user_uid)],
|
||||
)
|
||||
if leveled_up:
|
||||
audit.record_system(
|
||||
"reward.level.up",
|
||||
actor_kind="user",
|
||||
actor_uid=user_uid,
|
||||
old_value=current_level,
|
||||
new_value=new_level,
|
||||
metadata={"level": new_level},
|
||||
summary=f"user reached level {new_level}",
|
||||
links=[audit.actor(user_uid)],
|
||||
)
|
||||
create_notification(
|
||||
user_uid,
|
||||
"level",
|
||||
f"You reached level {new_level}",
|
||||
user_uid,
|
||||
f"/profile/{user['username']}",
|
||||
)
|
||||
for level in range(current_level + 1, new_level + 1):
|
||||
badge_name = LEVEL_BADGES.get(level)
|
||||
if badge_name and award_badge(user_uid, badge_name):
|
||||
notify_badge(user_uid, badge_name)
|
||||
return {"xp": new_xp, "level": new_level, "leveled_up": leveled_up}
|
||||
|
||||
|
||||
_COUNT_MILESTONES = [
|
||||
("Prolific", "posts", 10),
|
||||
("Wordsmith", "posts", 50),
|
||||
("Veteran", "posts", 100),
|
||||
("Conversationalist", "comments", 25),
|
||||
("Debater", "comments", 100),
|
||||
("Builder", "projects", 5),
|
||||
("Architect", "projects", 10),
|
||||
("Snippet Collector", "gists", 5),
|
||||
("Rising Star", "stars", 25),
|
||||
("Star Author", "stars", 100),
|
||||
("Superstar", "stars", 500),
|
||||
("Popular", "followers", 10),
|
||||
("Influencer", "followers", 50),
|
||||
("Celebrity", "followers", 100),
|
||||
("Connector", "following", 10),
|
||||
("On Fire", "streak", 7),
|
||||
("Dedicated", "streak", 30),
|
||||
("Unstoppable", "streak", 100),
|
||||
]
|
||||
|
||||
|
||||
def _milestone_metrics(user_uid: str):
|
||||
cache: dict[str, int] = {}
|
||||
|
||||
def resolve(key: str) -> int:
|
||||
if key in cache:
|
||||
return cache[key]
|
||||
if key == "posts":
|
||||
value = get_table("posts").count(user_uid=user_uid, deleted_at=None)
|
||||
elif key == "comments":
|
||||
value = get_table("comments").count(user_uid=user_uid, deleted_at=None)
|
||||
elif key == "projects":
|
||||
value = get_table("projects").count(user_uid=user_uid, deleted_at=None)
|
||||
elif key == "gists":
|
||||
value = get_table("gists").count(user_uid=user_uid, deleted_at=None)
|
||||
elif key == "stars":
|
||||
value = get_user_stars(user_uid)
|
||||
elif key == "followers":
|
||||
value = get_table("follows").count(following_uid=user_uid, deleted_at=None)
|
||||
elif key == "following":
|
||||
value = get_table("follows").count(follower_uid=user_uid, deleted_at=None)
|
||||
elif key == "streak":
|
||||
from devplacepy.database import get_streaks
|
||||
|
||||
value = get_streaks(user_uid)["current"]
|
||||
else:
|
||||
value = 0
|
||||
cache[key] = value
|
||||
return value
|
||||
|
||||
return resolve
|
||||
|
||||
|
||||
def check_milestone_badges(user_uid: str) -> list:
|
||||
held = {row["badge_name"] for row in get_table("badges").find(user_uid=user_uid)}
|
||||
pending = [m for m in _COUNT_MILESTONES if m[0] not in held]
|
||||
if not pending:
|
||||
return []
|
||||
metric = _milestone_metrics(user_uid)
|
||||
awarded = []
|
||||
for badge_name, metric_key, threshold in pending:
|
||||
if metric(metric_key) >= threshold and award_badge(user_uid, badge_name):
|
||||
awarded.append(badge_name)
|
||||
if metric_key == "streak":
|
||||
from devplacepy.services.audit import record as audit
|
||||
|
||||
audit.record_system(
|
||||
"reward.streak.milestone",
|
||||
actor_kind="user",
|
||||
actor_uid=user_uid,
|
||||
metadata={"streak": metric("streak")},
|
||||
summary=f"user reached a {metric('streak')}-day streak",
|
||||
links=[audit.actor(user_uid)],
|
||||
)
|
||||
for badge_name in awarded:
|
||||
notify_badge(user_uid, badge_name)
|
||||
return awarded
|
||||
|
||||
|
||||
def award_rewards(user_uid: str, amount: int, first_badge: str | None = None) -> None:
|
||||
background.submit(_apply_rewards, user_uid, amount, first_badge)
|
||||
|
||||
|
||||
def _apply_rewards(user_uid: str, amount: int, first_badge: str | None = None) -> None:
|
||||
if first_badge:
|
||||
award_badge(user_uid, first_badge)
|
||||
award_xp(user_uid, amount)
|
||||
check_milestone_badges(user_uid)
|
||||
|
||||
|
||||
ACHIEVEMENTS = {
|
||||
"docs.read": [(1, "Curious"), (5, "Studious"), (15, "Scholar")],
|
||||
"devii": [(1, "AI Curious"), (25, "AI Whisperer")],
|
||||
"fork": [(1, "First Fork"), (5, "Forker")],
|
||||
"zip": [(1, "Archivist")],
|
||||
"seo": [(1, "SEO Auditor")],
|
||||
"deepsearch": [(1, "Researcher"), (10, "Deep Diver")],
|
||||
"container": [(1, "Container Captain")],
|
||||
"message": [(1, "Messenger"), (100, "Chatterbox")],
|
||||
"bookmark": [(1, "Bookmarker"), (25, "Curator")],
|
||||
"reaction": [(1, "First Reaction"), (50, "Cheerleader")],
|
||||
"vote": [(1, "First Star Given"), (50, "Supporter"), (250, "Patron")],
|
||||
"follow": [(1, "Friendly")],
|
||||
"upload": [(1, "First Upload")],
|
||||
"project_file": [(1, "File Creator")],
|
||||
"issue": [(1, "Bug Reporter")],
|
||||
"poll": [(1, "Pollster")],
|
||||
"profile": [(1, "Profiled")],
|
||||
"harvest": [(1, "Green Thumb"), (50, "Master Farmer")],
|
||||
"water": [(1, "Good Neighbor")],
|
||||
"harvest_stolen": [(1, "Cat Burglar")],
|
||||
"got_stolen_from": [(1, "Robbed")],
|
||||
}
|
||||
|
||||
UNIQUE_ACTIONS = {"docs.read"}
|
||||
|
||||
|
||||
def track_action(user_uid: str, action: str, target: str | None = None) -> None:
|
||||
if not user_uid or action not in ACHIEVEMENTS:
|
||||
return
|
||||
background.submit(_apply_achievements, user_uid, action, target)
|
||||
|
||||
|
||||
def _apply_achievements(user_uid: str, action: str, target: str | None) -> None:
|
||||
if action in UNIQUE_ACTIONS:
|
||||
if target is None:
|
||||
return
|
||||
count = record_unique_activity(user_uid, action, target)
|
||||
else:
|
||||
count = record_activity(user_uid, action)
|
||||
if not count:
|
||||
return
|
||||
for threshold, badge_name in ACHIEVEMENTS.get(action, []):
|
||||
if count >= threshold and award_badge(user_uid, badge_name):
|
||||
notify_badge(user_uid, badge_name)
|
||||
Reference in New Issue
Block a user