Coverage for src/data_api/domain/newsletters/services.py: 100%

1""" 

2Newsletters domain service. 

3 

4Handles newsletter data fetching. 

5""" 

6 

7from typing import Optional 

8 

9from thefuzz import fuzz 

10 

11from data_api.data.manager import nthudata 

12from data_api.data.nthudata import NTHUDataManager 

13from data_api.utils.search import fuzzy_matches 

14 

15JSON_PATH = "newsletters.json" 

16FUZZY_SEARCH_THRESHOLD = 60 

17 

18 

19class NewslettersService: 

20 """Service for newsletter data operations.""" 

21 

22 def __init__(self, manager: NTHUDataManager = nthudata): 

23 self.manager = manager 

24 

25 async def get_all_newsletters(self) -> tuple[Optional[str], list[dict]]: 

26 """Get all newsletters.""" 

27 return await self.manager.get(JSON_PATH) 

28 

29 @staticmethod 

30 def _matches_newsletter_name(newsletter_name: str, query: Optional[str], fuzzy: bool) -> bool: 

31 if not query: 

32 return True 

33 if fuzzy: 

34 return fuzz.partial_ratio(query, newsletter_name) >= FUZZY_SEARCH_THRESHOLD 

35 return newsletter_name == query 

36 

37 @staticmethod 

38 def _filter_newsletter_articles( 

39 articles: list[dict], title: Optional[str], fuzzy: bool 

40 ) -> list[dict]: 

41 if not title: 

42 return list(articles) 

43 if fuzzy: 

44 return fuzzy_matches(articles, title, "title", FUZZY_SEARCH_THRESHOLD) 

45 return [article for article in articles if title in (article.get("title") or "")] 

46 

47 async def get_newsletters( 

48 self, 

49 name: Optional[str] = None, 

50 title: Optional[str] = None, 

51 fuzzy: bool = True, 

52 ) -> tuple[Optional[str], list[dict]]: 

53 """Filter sources and nested articles without modifying cached data.""" 

54 commit_hash, newsletters = await self.get_all_newsletters() 

55 filtered = [] 

56 for newsletter in newsletters: 

57 if not self._matches_newsletter_name(newsletter["name"], name, fuzzy): 

58 continue 

59 

60 articles = self._filter_newsletter_articles(newsletter["articles"], title, fuzzy) 

61 if title and not articles: 

62 continue 

63 

64 filtered.append({**newsletter, "articles": articles}) 

65 return commit_hash, filtered 

66 

67 async def get_newsletter_sources( 

68 self, name: Optional[str] = None 

69 ) -> tuple[Optional[str], list[dict]]: 

70 """Get source metadata without articles, optionally matching an exact name.""" 

71 commit_hash, newsletters = await self.get_all_newsletters() 

72 sources = [ 

73 {key: newsletter[key] for key in ("name", "link", "details")} 

74 for newsletter in newsletters 

75 if not name or newsletter["name"] == name 

76 ] 

77 return commit_hash, sources 

78 

79 

80# Global service instance 

81newsletters_service = NewslettersService()