Coverage for functions \ flipdare \ search \ core \ search_score.py: 92%
98 statements
« prev ^ index » next coverage.py v7.13.0, created at 2026-05-08 12:22 +1000
« prev ^ index » next coverage.py v7.13.0, created at 2026-05-08 12:22 +1000
1#!/usr/bin/env python
2# Copyright (c) 2026 Flipdare Pty Ltd. All rights reserved.
3#
4# This file is part of Flipdare's proprietary software and contains
5# confidential and copyrighted material. Unauthorised copying,
6# modification, distribution, or use of this file is strictly
7# prohibited without prior written permission from Flipdare Pty Ltd.
8#
9# This software includes third-party components licensed under MIT,
10# BSD, and Apache 2.0 licences. See THIRD_PARTY_NOTICES for details.
11#
14import math
15from dataclasses import dataclass
17from flipdare.app_log import LOG
18from flipdare.constants import DEF_ERROR_SCORE, IS_DEBUG
19from flipdare.firestore.context.dare_context import DareContext
20from flipdare.firestore.context.friend_context import FriendContext
21from flipdare.wrapper import DareWrapper, GroupWrapper, PersistedGuard, UserWrapper
23__all__ = ["SearchScore"]
26@dataclass
27class ScoreStats:
28 views: int = 1
29 bookmarks: int = 1
30 likes: int = 1
31 dislikes: int = 0
32 flags: int = 0
33 requested: int = 0
34 completed: int = 0
37class SearchScore:
39 def __init__(self, stats: ScoreStats) -> None:
40 self._stats = stats
42 @property
43 def stats(self) -> ScoreStats:
44 return self._stats
46 @property
47 def weighted_score(self) -> float:
48 """Calculate normalized score (0-100) from engagement metrics using log-based normalization."""
49 stats = self.stats
51 views = stats.views
52 bookmarks = stats.bookmarks
53 likes = stats.likes
54 dislikes = stats.dislikes
55 flags = stats.flags
56 requested = stats.requested
57 completed = stats.completed
59 # Engagement score (views typically highest volume)
60 engagement = math.log1p(views) * 1.0
62 # Quality indicators (higher value per action)
63 quality = math.log1p(bookmarks) * 3.0 + math.log1p(likes) * 2.0
65 # Negative indicators
66 negative = math.log1p(dislikes) * 2.0 + math.log1p(flags) * 5.0 # Flags are serious
68 # Dare activity (if applicable)
69 dare_activity = math.log1p(completed) * 3.0 + math.log1p(requested) * 1.0
71 # Combined score
72 raw_score = engagement + quality + dare_activity - negative
74 # Normalize to 0-100 range
75 # Adjust max_expected based on your typical values
76 max_expected = 50 # Tune this based on your data
77 normalized = (raw_score / max_expected) * 100
79 return max(0.0, min(100.0, normalized))
81 @staticmethod
82 def score_dare(
83 dare: DareWrapper,
84 from_user: UserWrapper,
85 to_obj: UserWrapper | GroupWrapper,
86 ) -> float:
87 stats = dare.view_stats
88 score_stats = ScoreStats(
89 views=stats.views,
90 bookmarks=stats.bookmarks,
91 likes=stats.likes,
92 dislikes=stats.dislikes,
93 flags=stats.flags,
94 )
95 dare_score = SearchScore(score_stats).weighted_score
97 from_score = SearchScore.score_user(from_user)
98 to_score = 0.0
99 if PersistedGuard.is_user(to_obj):
100 to_score = SearchScore.score_user(to_obj)
101 elif PersistedGuard.is_group(to_obj):
102 to_score = SearchScore.score_group(to_obj)
103 else:
104 LOG().error(f"Unexpected 'to' object type for scoring dare: {type(to_obj)}")
106 # Add bonus from user scores (scaled appropriately)
107 dare_score += max(from_score, to_score) * 0.1
108 final_score = max(0.0, min(100.0, dare_score))
109 if IS_DEBUG:
110 msg = f"Score={final_score} (Dare={dare_score}, user={from_score}, to_obj={to_score})"
111 LOG().debug(msg)
113 return final_score
115 @staticmethod
116 def score_dare_context(dare_context: DareContext) -> float:
117 try:
118 return SearchScore.score_dare(
119 dare=dare_context.dare,
120 from_user=dare_context.from_user,
121 to_obj=dare_context.to_obj,
122 )
123 except Exception as e:
124 LOG().error(f"Error scoring dare context: {e}")
125 return DEF_ERROR_SCORE
127 @staticmethod
128 def score_user(user: UserWrapper) -> float:
129 view_stats = user.view_stats
130 dare_stats = user.dare_stats
132 score_stats = ScoreStats(
133 views=view_stats.views,
134 bookmarks=view_stats.bookmarks,
135 likes=view_stats.likes,
136 dislikes=view_stats.dislikes,
137 flags=view_stats.flags + dare_stats.flags, # Combine flags from both stats
138 requested=dare_stats.requested,
139 completed=dare_stats.completed,
140 )
142 user_score = SearchScore(score_stats).weighted_score
143 if IS_DEBUG:
144 msg = (
145 f"User score={user_score} (views={view_stats.views}, bookmarks={view_stats.bookmarks}, "
146 f"likes={view_stats.likes}, dislikes={view_stats.dislikes}, flags={score_stats.flags}, "
147 f"requested={dare_stats.requested}, completed={dare_stats.completed})"
148 )
149 LOG().debug(msg)
151 return user_score
153 @staticmethod
154 def score_group(group: GroupWrapper) -> float:
155 stats = group.view_stats
156 score_stats = ScoreStats(
157 views=stats.views,
158 bookmarks=stats.bookmarks,
159 likes=stats.likes,
160 dislikes=stats.dislikes,
161 flags=stats.flags,
162 )
163 group_score = SearchScore(score_stats).weighted_score
164 if IS_DEBUG:
165 msg = (
166 f"Group score={group_score} (views={stats.views}, bookmarks={stats.bookmarks}, "
167 f"likes={stats.likes}, dislikes={stats.dislikes}, flags={score_stats.flags})"
168 )
169 LOG().debug(msg)
171 return group_score
173 @staticmethod
174 def score_friend(friend_context: FriendContext) -> tuple[float, int]:
175 from_view_stats = friend_context.from_user.view_stats
176 from_dare_stats = friend_context.from_user.dare_stats
177 to_view_stats = friend_context.to_user.view_stats
178 to_dare_stats = friend_context.to_user.dare_stats
179 total_views = from_view_stats.views + to_view_stats.views
181 score_stats = ScoreStats(
182 views=total_views,
183 bookmarks=from_view_stats.bookmarks + to_view_stats.bookmarks,
184 likes=from_view_stats.likes + to_view_stats.likes,
185 dislikes=from_view_stats.dislikes + to_view_stats.dislikes,
186 flags=from_view_stats.flags
187 + to_view_stats.flags
188 + from_dare_stats.flags
189 + to_dare_stats.flags, # Combine flags from both stats
190 requested=from_dare_stats.requested + to_dare_stats.requested,
191 completed=from_dare_stats.completed + to_dare_stats.completed,
192 )
193 score = SearchScore(score_stats).weighted_score
194 if IS_DEBUG:
195 msg = (
196 f"Friend score={score} (total_views={total_views}, bookmarks={score_stats.bookmarks}, "
197 f"likes={score_stats.likes}, dislikes={score_stats.dislikes}, flags={score_stats.flags}, "
198 f"requested={score_stats.requested}, completed={score_stats.completed})"
199 )
200 LOG().debug(msg)
202 return score, total_views