Coverage for integrations / social / karma_engine.py: 96.9%

1""" 

2HevolveSocial - Karma Engine 

3Combines upvote karma + task completion karma for agent reputation. 

4""" 

5from sqlalchemy import func 

6from sqlalchemy.orm import Session 

7 

8from .models import User, Post, Comment, TaskRequest, AgentSkillBadge 

9 

10 

11def recalculate_karma(db: Session, user: User) -> int: 

12 """Recalculate and update a user's total karma.""" 

13 # Upvote karma: sum of (upvotes - downvotes) on all posts + comments 

14 post_karma = db.query(func.coalesce(func.sum(Post.score), 0)).filter( 

15 Post.author_id == user.id, Post.is_deleted == False).scalar() 

16 comment_karma = db.query(func.coalesce(func.sum(Comment.score), 0)).filter( 

17 Comment.author_id == user.id, Comment.is_deleted == False).scalar() 

18 upvote_karma = int(post_karma) + int(comment_karma) 

19 

20 # Task karma (agents only): completed tasks * 10 + success_rate bonus 

21 task_karma = 0 

22 if user.user_type == 'agent': 

23 completed = db.query(func.count(TaskRequest.id)).filter( 

24 TaskRequest.assignee_id == user.id, 

25 TaskRequest.status == 'completed' 

26 ).scalar() 

27 task_karma = int(completed) * 10 

28 

29 # Bonus from skill success rates 

30 avg_success = db.query(func.avg(AgentSkillBadge.success_rate)).filter( 

31 AgentSkillBadge.user_id == user.id).scalar() 

32 if avg_success: 

33 task_karma += int(float(avg_success) * 50) 

34 

35 user.karma_score = upvote_karma + task_karma 

36 user.task_karma = task_karma 

37 db.flush() 

38 return user.karma_score 

39 

40 

41def get_karma_breakdown(db: Session, user: User) -> dict: 

42 """Detailed karma breakdown for profile display.""" 

43 post_karma = db.query(func.coalesce(func.sum(Post.score), 0)).filter( 

44 Post.author_id == user.id, Post.is_deleted == False).scalar() 

45 comment_karma = db.query(func.coalesce(func.sum(Comment.score), 0)).filter( 

46 Comment.author_id == user.id, Comment.is_deleted == False).scalar() 

47 

48 completed_tasks = db.query(func.count(TaskRequest.id)).filter( 

49 TaskRequest.assignee_id == user.id, 

50 TaskRequest.status == 'completed' 

51 ).scalar() 

52 

53 return { 

54 'total': user.karma_score, 

55 'post_karma': int(post_karma), 

56 'comment_karma': int(comment_karma), 

57 'task_karma': user.task_karma, 

58 'completed_tasks': int(completed_tasks), 

59 } 

60 

61 

62def compute_badge_level(proficiency: float, success_rate: float, usage_count: int) -> str: 

63 """Determine skill badge level from performance metrics.""" 

64 score = (proficiency * 0.3 + success_rate * 0.4 + min(usage_count / 100, 1.0) * 0.3) 

65 if score >= 0.9: 

66 return 'platinum' 

67 elif score >= 0.7: 

68 return 'gold' 

69 elif score >= 0.4: 

70 return 'silver' 

71 return 'bronze'