Coverage for functions \ flipdare \ search \ factory \ content_search_factory.py: 85%

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# 

12 

13 

14from typing import Any, override 

15from flipdare.app_log import LOG 

16from flipdare.constants import IS_DEBUG 

17from flipdare.core.tokenizer import Tokenizer 

18from flipdare.generated.shared.search.search_obj_type import SearchObjType 

19from flipdare.search.core.search_score import SearchScore 

20from flipdare.search.doc._search_document import SearchDocument 

21from flipdare.search.doc.general_document import GeneralDocument 

22from flipdare.search.factory._search_document_factory import SearchDocumentFactory 

23from flipdare.util.time_util import TypesenseTime 

24from flipdare.wrapper import ContentWrapper, GroupWrapper, PersistedGuard, UserWrapper 

25 

26__all__ = ["ContentSearchFactory"] 

27 

28 

29class ContentSearchFactory(SearchDocumentFactory): 

30 

31 def __init__( 

32 self, 

33 content: ContentWrapper, 

34 description: str, # we separate this so we know it exists! 

35 owner: UserWrapper | GroupWrapper, 

36 tokenizer: Tokenizer | None = None, 

37 ) -> None: 

38 self.content = content 

39 self.owner = owner 

40 self.description = description 

41 

42 super().__init__(tokenizer=tokenizer) 

43 

44 @property 

45 @override 

46 def obj_type(self) -> SearchObjType: 

47 if PersistedGuard.is_user(self.owner): 

48 return SearchObjType.USER 

49 return SearchObjType.GROUP 

50 

51 @override 

52 def get_documents(self) -> list[SearchDocument[Any]] | None: 

53 content = self.content 

54 owner: UserWrapper | GroupWrapper = self.owner 

55 

56 obj_id = content.doc_id 

57 

58 views: int 

59 creator_id: str 

60 if PersistedGuard.is_user(owner): 

61 creator_id = owner.doc_id 

62 score = SearchScore.score_user(owner) 

63 views = owner.views 

64 elif PersistedGuard.is_group(owner): 

65 creator_id = owner.uid 

66 score = SearchScore.score_group(owner) 

67 views = owner.views 

68 else: 

69 # for some reason type checking complains .. 

70 # should never get here because owner is a union of UserWrapper and GroupWrapper 

71 msg = f"Owner must be either UserWrapper or GroupWrapper (got {type(owner)})" 

72 raise TypeError(msg) 

73 

74 values = content.searchable_values 

75 token_result = self.tokenizer.create_tokens(values) 

76 

77 created_at = TypesenseTime.from_firestore(content.created_at_db) 

78 updated_at = TypesenseTime.from_firestore(content.updated_at_db) 

79 

80 if IS_DEBUG: 

81 LOG().debug( 

82 f"Creating search document for content {obj_id} with values: {values}\n" 

83 f"Token result: {token_result}\n" 

84 f"ObjId={obj_id} CreatorId={creator_id} Score={score}, Views{views}", 

85 ) 

86 

87 return [ 

88 GeneralDocument.create( 

89 obj_id=obj_id, 

90 uid=creator_id, 

91 obj_type=self.obj_type, 

92 keywords=values, 

93 score=score, 

94 views=views, 

95 created_at=created_at, 

96 updated_at=updated_at, 

97 tags=token_result.tokens, 

98 tag_score=token_result.token_score.score * score, 

99 ), 

100 ]