Coverage for api/controllers/elasticsearch/related.py: 79%

1from __future__ import annotations 

2 

3from elasticsearch_dsl import Search 

4from elasticsearch_dsl.query import Match, Q, Term 

5from elasticsearch_dsl.response import Hit 

6 

7from api.controllers.elasticsearch.helpers import get_es_response, get_query_slice 

8from api.controllers.search_controller import ( 

9 _post_process_results, 

10 get_excluded_sources_query, 

11) 

12 

13 

14def related_media(uuid: str, index: str, filter_dead: bool) -> list[Hit]: 

15 """ 

16 Given a UUID, finds 10 related search results based on title and tags. 

17 

18 Uses Match query for title or SimpleQueryString for tags. 

19 If the item has no title and no tags, returns items by the same creator. 

20 If the item has no title, no tags or no creator, returns empty list. 

21 

22 :param uuid: The UUID of the item to find related results for. 

23 :param index: The Elasticsearch index to search (e.g. 'image') 

24 :param filter_dead: Whether dead links should be removed. 

25 :return: List of related results. 

26 """ 

27 

28 # Search the default index for the item itself as it might be sensitive. 

29 item_search = Search(index=index) 

30 # This will raise ``IndexError`` if no hits are found. This error is caught 

31 # in the viewset handler function. 

32 item_hit = item_search.query(Term(identifier=uuid)).execute().hits[0] 

33 

34 # Match related using title. 

35 title = getattr(item_hit, "title", None) 

36 tags = getattr(item_hit, "tags", None) 

37 creator = getattr(item_hit, "creator", None) 

38 

39 related_query = {"must_not": [], "must": [], "should": []} 

40 

41 if not title and not tags: 41 ↛ 42line 41 didn't jump to line 42 because the condition on line 41 was never true

42 if not creator: 

43 return [] 

44 else: 

45 # Only use `creator` query if there are no `title` and `tags` 

46 related_query["should"].append(Term(creator=creator)) 

47 else: 

48 if title: 48 ↛ 53line 48 didn't jump to line 53 because the condition on line 48 was always true

49 related_query["should"].append(Match(title=title)) 

50 

51 # Match related using tags, if the item has any. 

52 # Only use the first 10 tags 

53 if tags: 

54 tags = [tag["name"] for tag in tags[:10]] 

55 related_query["should"].append(Q("terms", tags__name__keyword=tags)) 

56 

57 # Exclude the dynamically disabled sources. 

58 if excluded_sources_query := get_excluded_sources_query(): 58 ↛ 59line 58 didn't jump to line 59 because the condition on line 58 was never true

59 related_query["must_not"].append(excluded_sources_query) 

60 # Exclude the current item and mature content. 

61 related_query["must_not"].extend( 

62 [Q("term", mature=True), Q("term", identifier=uuid)] 

63 ) 

64 

65 # Search the filtered index for related items. 

66 s = Search(index=f"{index}-filtered") 

67 s = s.query("bool", **related_query) 

68 

69 page, page_size = 1, 10 

70 start, end = get_query_slice(s, page_size, page, filter_dead) 

71 s = s[start:end] 

72 

73 response = get_es_response(s, es_query="related_media") 

74 results = _post_process_results(s, start, end, page_size, response, filter_dead) 

75 return results or []