Coverage for integrations / social / onboarding_service.py: 68.0%

1""" 

2HevolveSocial - Onboarding Service 

3Progressive 7-step onboarding with auto-advancement and rewards. 

4""" 

5import json 

6import logging 

7from datetime import datetime 

8from typing import Optional, Dict, List 

9 

10from sqlalchemy.orm import Session 

11 

12from .models import User, OnboardingProgress 

13from .resonance_engine import ResonanceService 

14 

15logger = logging.getLogger('hevolve_social') 

16 

17ONBOARDING_STEPS = [ 

18 {'key': 'welcome', 'title': 'Welcome', 'reward_type': 'pulse', 'reward_amount': 50}, 

19 {'key': 'first_follow', 'title': 'Follow Someone', 'reward_type': 'pulse', 'reward_amount': 25}, 

20 {'key': 'join_community', 'title': 'Join a Community', 'reward_type': 'pulse', 'reward_amount': 50}, 

21 {'key': 'first_vote', 'title': 'Vote on a Post', 'reward_type': 'spark', 'reward_amount': 25}, 

22 {'key': 'first_comment', 'title': 'Leave a Comment', 'reward_type': 'spark', 'reward_amount': 25}, 

23 {'key': 'first_post', 'title': 'Create a Post', 'reward_type': 'xp', 'reward_amount': 100}, 

24 {'key': 'explore_agents', 'title': 'Explore Agents', 'reward_type': 'xp', 'reward_amount': 100}, 

25 {'key': 'invite_friends', 'title': 'Invite Friends', 'reward_type': 'spark', 'reward_amount': 100}, 

26] 

27 

28 

29class OnboardingService: 

30 

31 @staticmethod 

32 def get_or_create_progress(db: Session, user_id: str) -> OnboardingProgress: 

33 progress = db.query(OnboardingProgress).filter_by(user_id=user_id).first() 

34 if not progress: 

35 progress = OnboardingProgress( 

36 user_id=user_id, 

37 steps_completed='{}', 

38 current_step='welcome', 

39 ) 

40 db.add(progress) 

41 db.flush() 

42 return progress 

43 

44 @staticmethod 

45 def get_progress(db: Session, user_id: str) -> Dict: 

46 progress = OnboardingService.get_or_create_progress(db, user_id) 

47 try: 

48 steps = json.loads(progress.steps_completed) if progress.steps_completed else {} 

49 except (json.JSONDecodeError, TypeError): 

50 steps = {} 

51 

52 total = len(ONBOARDING_STEPS) 

53 done = sum(1 for s in ONBOARDING_STEPS if steps.get(s['key'])) 

54 

55 return { 

56 'user_id': user_id, 

57 'steps_completed': steps, 

58 'current_step': progress.current_step, 

59 'completed_at': progress.completed_at.isoformat() if progress.completed_at else None, 

60 'tutorial_dismissed': progress.tutorial_dismissed, 

61 'total_steps': total, 

62 'done_steps': done, 

63 'steps': [{ 

64 'key': s['key'], 

65 'title': s['title'], 

66 'completed': bool(steps.get(s['key'])), 

67 'reward_type': s['reward_type'], 

68 'reward_amount': s['reward_amount'], 

69 } for s in ONBOARDING_STEPS], 

70 } 

71 

72 @staticmethod 

73 def complete_step(db: Session, user_id: str, step_key: str) -> Dict: 

74 """Mark an onboarding step as complete and award rewards.""" 

75 progress = OnboardingService.get_or_create_progress(db, user_id) 

76 try: 

77 steps = json.loads(progress.steps_completed) if progress.steps_completed else {} 

78 except (json.JSONDecodeError, TypeError): 

79 steps = {} 

80 

81 if steps.get(step_key): 

82 return {'already_completed': True, 'step': step_key} 

83 

84 # Find step definition 

85 step_def = next((s for s in ONBOARDING_STEPS if s['key'] == step_key), None) 

86 if not step_def: 

87 return {'error': 'Unknown step'} 

88 

89 steps[step_key] = datetime.utcnow().isoformat() 

90 progress.steps_completed = json.dumps(steps) 

91 

92 # Track specific timestamps 

93 timestamp_map = { 

94 'first_post': 'first_post_at', 

95 'first_comment': 'first_comment_at', 

96 'first_vote': 'first_vote_at', 

97 'first_follow': 'first_follow_at', 

98 'join_community': 'first_community_join_at', 

99 } 

100 attr = timestamp_map.get(step_key) 

101 if attr and hasattr(progress, attr) and not getattr(progress, attr): 

102 setattr(progress, attr, datetime.utcnow()) 

103 

104 # Award reward 

105 reward_result = {} 

106 rtype = step_def['reward_type'] 

107 ramount = step_def['reward_amount'] 

108 if rtype == 'pulse': 

109 ResonanceService.award_pulse(db, user_id, ramount, 'onboarding', step_key, 

110 f'Onboarding: {step_def["title"]}') 

111 reward_result = {'pulse': ramount} 

112 elif rtype == 'spark': 

113 ResonanceService.award_spark(db, user_id, ramount, 'onboarding', step_key, 

114 f'Onboarding: {step_def["title"]}') 

115 reward_result = {'spark': ramount} 

116 elif rtype == 'xp': 

117 ResonanceService.award_xp(db, user_id, ramount, 'onboarding', step_key, 

118 f'Onboarding: {step_def["title"]}') 

119 reward_result = {'xp': ramount} 

120 

121 # Advance current step 

122 all_done = all(steps.get(s['key']) for s in ONBOARDING_STEPS) 

123 if all_done: 

124 progress.completed_at = datetime.utcnow() 

125 progress.current_step = 'complete' 

126 # Bonus for completing all steps 

127 ResonanceService.award_spark(db, user_id, 50, 'onboarding', 'complete', 

128 'Onboarding complete bonus') 

129 reward_result['completion_bonus'] = {'spark': 50} 

130 else: 

131 # Find next incomplete step 

132 for s in ONBOARDING_STEPS: 

133 if not steps.get(s['key']): 

134 progress.current_step = s['key'] 

135 break 

136 

137 return { 

138 'step': step_key, 

139 'completed': True, 

140 'rewards': reward_result, 

141 'all_complete': all_done, 

142 'next_step': progress.current_step, 

143 } 

144 

145 @staticmethod 

146 def dismiss(db: Session, user_id: str) -> bool: 

147 progress = OnboardingService.get_or_create_progress(db, user_id) 

148 progress.tutorial_dismissed = True 

149 return True 

150 

151 @staticmethod 

152 def get_suggestion(db: Session, user_id: str) -> Optional[Dict]: 

153 """Get the next suggested action for onboarding.""" 

154 progress = OnboardingService.get_or_create_progress(db, user_id) 

155 if progress.completed_at or progress.tutorial_dismissed: 

156 return None 

157 

158 try: 

159 steps = json.loads(progress.steps_completed) if progress.steps_completed else {} 

160 except (json.JSONDecodeError, TypeError): 

161 steps = {} 

162 

163 for s in ONBOARDING_STEPS: 

164 if not steps.get(s['key']): 

165 return { 

166 'step': s['key'], 

167 'title': s['title'], 

168 'reward_type': s['reward_type'], 

169 'reward_amount': s['reward_amount'], 

170 } 

171 return None 

172 

173 @staticmethod 

174 def auto_advance(db: Session, user_id: str, action: str): 

175 """Auto-advance onboarding based on natural user actions. 

176 Called by service hooks after relevant actions.""" 

177 action_to_step = { 

178 'follow': 'first_follow', 

179 'join_community': 'join_community', 

180 'vote': 'first_vote', 

181 'comment': 'first_comment', 

182 'post': 'first_post', 

183 'view_agents': 'explore_agents', 

184 'share_referral': 'invite_friends', 

185 } 

186 step_key = action_to_step.get(action) 

187 if step_key: 

188 progress = db.query(OnboardingProgress).filter_by(user_id=user_id).first() 

189 if progress and not progress.completed_at: 

190 try: 

191 steps = json.loads(progress.steps_completed) if progress.steps_completed else {} 

192 except (json.JSONDecodeError, TypeError): 

193 steps = {} 

194 if not steps.get(step_key): 

195 OnboardingService.complete_step(db, user_id, step_key)