Coverage for src/secchi/trending.py: 34%

1"""GitHub weekly trend card — shows a highly-starred new repo in the sidebar.""" 

2 

3from __future__ import annotations 

4 

5import json 

6from dataclasses import dataclass 

7from datetime import datetime, timedelta, timezone 

8from typing import Any 

9 

10import httpx 

11 

12from secchi.cache import cache_root 

13from secchi.http import HttpClientFactory 

14 

15TRENDING_CACHE = cache_root() / "trending.json" 

16MAX_RESPONSE_BYTES = 8_192 

17 

18 

19@dataclass(frozen=True) 

20class TrendingRepo: 

21 title: str 

22 description: str 

23 url: str 

24 stars: str 

25 language: str = "" 

26 

27 

28FALLBACK_TRENDING = TrendingRepo( 

29 title="tuffcli", 

30 description="Capability lifecycle manager for coding agents", 

31 url="github.com/kannandreams/tuff", 

32 stars="—", 

33 language="Rust", 

34) 

35 

36 

37def _trending_date() -> str: 

38 return (datetime.now(timezone.utc) - timedelta(days=7)).strftime("%Y-%m-%d") 

39 

40 

41async def fetch_trending() -> TrendingRepo | None: 

42 try: 

43 async with HttpClientFactory().create(timeout=3.0) as client: 

44 response = await client.get( 

45 "https://api.github.com/search/repositories", 

46 params={ 

47 "q": f"created:>{_trending_date()}", 

48 "sort": "stars", 

49 "order": "desc", 

50 "per_page": 1, 

51 }, 

52 headers={"Accept": "application/vnd.github+json"}, 

53 ) 

54 response.raise_for_status() 

55 body = response.content 

56 if len(body) > MAX_RESPONSE_BYTES: 

57 return FALLBACK_TRENDING 

58 data = json.loads(body.decode("utf-8")) 

59 items = data.get("items", []) 

60 if not items: 

61 return FALLBACK_TRENDING 

62 

63 repo = items[0] 

64 return TrendingRepo( 

65 title=repo.get("full_name", repo.get("name", "")), 

66 description=(repo.get("description", "") or "")[:72], 

67 url=repo.get("html_url", "").replace("https://", "").rstrip("/"), 

68 stars=_short_stars(repo.get("stargazers_count", 0)), 

69 language=repo.get("language", "") or "", 

70 ) 

71 except (httpx.HTTPError, UnicodeDecodeError, json.JSONDecodeError, KeyError): 

72 return FALLBACK_TRENDING 

73 

74 

75def _short_stars(count: int) -> str: 

76 if count >= 1_000_000: 

77 return f"{count / 1_000_000:.1f}M" 

78 if count >= 1_000: 

79 return f"{count / 1_000:.1f}K" 

80 return str(count) 

81 

82 

83def load_cached_trending() -> TrendingRepo | None: 

84 path = TRENDING_CACHE 

85 if not path.exists(): 

86 return None 

87 try: 

88 raw = json.loads(path.read_text()) 

89 fetched_at = _parse_datetime(raw.get("fetched_at")) 

90 today = datetime.now().astimezone().date() 

91 if fetched_at is None or fetched_at.astimezone().date() != today: 

92 return None 

93 return TrendingRepo(**raw.get("repo", {})) 

94 except (OSError, json.JSONDecodeError, TypeError, ValueError): 

95 return None 

96 

97 

98def save_cached_trending(repo: TrendingRepo) -> None: 

99 path = TRENDING_CACHE 

100 payload = { 

101 "fetched_at": datetime.now().astimezone().isoformat(), 

102 "repo": { 

103 "title": repo.title, 

104 "description": repo.description, 

105 "url": repo.url, 

106 "stars": repo.stars, 

107 "language": repo.language, 

108 }, 

109 } 

110 try: 

111 path.parent.mkdir(parents=True, exist_ok=True) 

112 path.write_text(json.dumps(payload, indent=2, sort_keys=True)) 

113 except OSError: 

114 pass 

115 

116 

117def _parse_datetime(raw: Any) -> datetime | None: 

118 if not raw: 

119 return None 

120 try: 

121 return datetime.fromisoformat(raw) 

122 except (TypeError, ValueError): 

123 return None