Coverage for integrations / social / rating_service.py: 43.9%

1""" 

2HevolveSocial - Rating Service 

3Multi-dimensional ratings (skill, usefulness, reliability, creativity) and trust scores. 

4""" 

5import logging 

6from typing import Optional, Dict, List 

7 

8from sqlalchemy import desc, func 

9from sqlalchemy.orm import Session 

10 

11from .models import User, Rating, TrustScore, ResonanceWallet 

12 

13logger = logging.getLogger('hevolve_social') 

14 

15DIMENSIONS = ['skill', 'usefulness', 'reliability', 'creativity'] 

16TRUST_WEIGHTS = {'skill': 0.25, 'usefulness': 0.30, 'reliability': 0.30, 'creativity': 0.15} 

17 

18 

19class RatingService: 

20 

21 @staticmethod 

22 def submit_rating(db: Session, rater_id: str, rated_id: str, 

23 context_type: str, context_id: str, 

24 dimension: str, score: float, 

25 comment: str = '') -> Optional[Dict]: 

26 """Submit or update a rating.""" 

27 if dimension not in DIMENSIONS: 

28 return None 

29 if not (1.0 <= score <= 5.0): 

30 return None 

31 if rater_id == rated_id: 

32 return None 

33 

34 existing = db.query(Rating).filter_by( 

35 rater_id=rater_id, rated_id=rated_id, 

36 context_type=context_type, context_id=context_id, 

37 dimension=dimension, 

38 ).first() 

39 

40 if existing: 

41 existing.score = score 

42 existing.comment = comment 

43 rating = existing 

44 else: 

45 rating = Rating( 

46 rater_id=rater_id, rated_id=rated_id, 

47 context_type=context_type, context_id=context_id, 

48 dimension=dimension, score=score, comment=comment, 

49 ) 

50 db.add(rating) 

51 db.flush() 

52 

53 # Recalculate trust score 

54 RatingService._recalculate_trust(db, rated_id) 

55 

56 return rating.to_dict() 

57 

58 @staticmethod 

59 def _recalculate_trust(db: Session, user_id: str): 

60 """Recalculate composite trust score for a user. 

61 Ratings from higher-Signal users count more.""" 

62 trust = db.query(TrustScore).filter_by(user_id=user_id).first() 

63 if not trust: 

64 trust = TrustScore(user_id=user_id) 

65 db.add(trust) 

66 

67 total_ratings = 0 

68 dim_totals = {} 

69 dim_weights = {} 

70 

71 for dim in DIMENSIONS: 

72 ratings = db.query(Rating).filter_by( 

73 rated_id=user_id, dimension=dim 

74 ).all() 

75 

76 weighted_sum = 0.0 

77 weight_sum = 0.0 

78 for r in ratings: 

79 # Rater credibility based on their Signal 

80 rater_wallet = db.query(ResonanceWallet).filter_by(user_id=r.rater_id).first() 

81 rater_signal = rater_wallet.signal if rater_wallet else 0 

82 weight = min(rater_signal / 10.0, 3.0) 

83 weight = max(weight, 0.5) # minimum weight of 0.5 

84 weighted_sum += r.score * weight 

85 weight_sum += weight 

86 

87 if weight_sum > 0: 

88 dim_totals[dim] = weighted_sum / weight_sum 

89 else: 

90 dim_totals[dim] = 0.0 

91 dim_weights[dim] = len(ratings) 

92 total_ratings += len(ratings) 

93 

94 trust.avg_skill = dim_totals.get('skill', 0.0) 

95 trust.avg_usefulness = dim_totals.get('usefulness', 0.0) 

96 trust.avg_reliability = dim_totals.get('reliability', 0.0) 

97 trust.avg_creativity = dim_totals.get('creativity', 0.0) 

98 trust.total_ratings_received = total_ratings 

99 

100 # Composite trust = weighted average of dimensions 

101 composite = sum( 

102 dim_totals.get(dim, 0) * TRUST_WEIGHTS[dim] 

103 for dim in DIMENSIONS 

104 ) 

105 trust.composite_trust = round(composite, 3) 

106 db.flush() 

107 

108 @staticmethod 

109 def get_trust_score(db: Session, user_id: str) -> Optional[Dict]: 

110 trust = db.query(TrustScore).filter_by(user_id=user_id).first() 

111 return trust.to_dict() if trust else None 

112 

113 @staticmethod 

114 def get_ratings_received(db: Session, user_id: str, 

115 limit: int = 50, offset: int = 0) -> List[Dict]: 

116 ratings = db.query(Rating).filter_by( 

117 rated_id=user_id 

118 ).order_by(desc(Rating.created_at)).offset(offset).limit(limit).all() 

119 return [r.to_dict() for r in ratings] 

120 

121 @staticmethod 

122 def get_ratings_given(db: Session, user_id: str, 

123 limit: int = 50, offset: int = 0) -> List[Dict]: 

124 ratings = db.query(Rating).filter_by( 

125 rater_id=user_id 

126 ).order_by(desc(Rating.created_at)).offset(offset).limit(limit).all() 

127 return [r.to_dict() for r in ratings] 

128 

129 @staticmethod 

130 def get_context_ratings(db: Session, context_type: str, 

131 context_id: str) -> List[Dict]: 

132 ratings = db.query(Rating).filter_by( 

133 context_type=context_type, context_id=context_id 

134 ).all() 

135 return [r.to_dict() for r in ratings] 

136 

137 @staticmethod 

138 def get_aggregated(db: Session, user_id: str) -> Dict: 

139 """Get aggregated rating info for display.""" 

140 trust = RatingService.get_trust_score(db, user_id) 

141 if not trust: 

142 return { 

143 'dimensions': {d: 0.0 for d in DIMENSIONS}, 

144 'composite_trust': 0.0, 

145 'total_ratings': 0, 

146 } 

147 return { 

148 'dimensions': { 

149 'skill': trust.get('avg_skill', 0), 

150 'usefulness': trust.get('avg_usefulness', 0), 

151 'reliability': trust.get('avg_reliability', 0), 

152 'creativity': trust.get('avg_creativity', 0), 

153 }, 

154 'composite_trust': trust.get('composite_trust', 0), 

155 'total_ratings': trust.get('total_ratings_received', 0), 

156 }