Coverage for src/secchi/spotlight.py: 36%

1"""Product-controlled Spotlight feed for the sidebar promo card.""" 

2 

3from __future__ import annotations 

4 

5import json 

6import os 

7from dataclasses import dataclass 

8from datetime import datetime 

9from pathlib import Path 

10from typing import Any 

11 

12import httpx 

13 

14from secchi.cache import cache_root 

15from secchi.http import HttpClientFactory 

16 

17SPOTLIGHT_URL = "https://kannandreams.github.io/secchi-spotlight/spotlight.json" 

18DISABLE_ENV = "SECCHI_DISABLE_SPOTLIGHT" 

19MAX_RESPONSE_BYTES = 8_192 

20 

21 

22@dataclass(frozen=True) 

23class Spotlight: 

24 title: str 

25 description: str 

26 url: str 

27 accent: str = "" 

28 expires_at: datetime | None = None 

29 stars: int | None = None 

30 

31 @property 

32 def project_stage(self) -> str: 

33 """Editorial context for discovery; never controls Spotlight visibility.""" 

34 if self.stars is None: 

35 return "Spotlight project" 

36 if self.stars < 50: 

37 return "Early project" 

38 if self.stars < 500: 

39 return "Growing project" 

40 return "Established project" 

41 

42 

43FALLBACK_SPOTLIGHT = Spotlight( 

44 title="tuffcli", 

45 description="Capability lifecycle manager for coding agents.", 

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

47 accent="blue", 

48 stars=1, 

49) 

50 

51 

52def spotlight_disabled() -> bool: 

53 raw = os.environ.get(DISABLE_ENV, "") 

54 return raw.lower() in {"1", "true", "yes", "on"} 

55 

56 

57def spotlight_cache_path() -> Path: 

58 return cache_root() / "spotlight.json" 

59 

60 

61def load_cached_spotlight() -> Spotlight | None: 

62 path = spotlight_cache_path() 

63 if not path.exists(): 

64 return None 

65 try: 

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

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

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

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

70 return None 

71 return _decode_spotlight(raw.get("spotlight", {})) 

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

73 return None 

74 

75 

76def save_cached_spotlight(spotlight: Spotlight, fetched_at: datetime) -> None: 

77 path = spotlight_cache_path() 

78 payload = { 

79 "fetched_at": fetched_at.astimezone().isoformat(), 

80 "spotlight": { 

81 "title": spotlight.title, 

82 "description": spotlight.description, 

83 "url": spotlight.url, 

84 "accent": spotlight.accent, 

85 "expires_at": spotlight.expires_at.isoformat() 

86 if spotlight.expires_at 

87 else "", 

88 "stars": spotlight.stars, 

89 }, 

90 } 

91 try: 

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

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

94 except OSError: 

95 pass 

96 

97 

98async def fetch_spotlight() -> Spotlight | None: 

99 if spotlight_disabled(): 

100 return None 

101 

102 cached = load_cached_spotlight() 

103 if cached is not None: 

104 return cached 

105 

106 try: 

107 async with HttpClientFactory().create(timeout=2.0) as client: 

108 response = await client.get(SPOTLIGHT_URL) 

109 response.raise_for_status() 

110 body = response.content 

111 if len(body) > MAX_RESPONSE_BYTES: 

112 return FALLBACK_SPOTLIGHT 

113 spotlight = _decode_spotlight(json.loads(body.decode("utf-8"))) 

114 save_cached_spotlight(spotlight, datetime.now().astimezone()) 

115 return spotlight 

116 except (httpx.HTTPError, UnicodeDecodeError, json.JSONDecodeError, ValueError): 

117 return FALLBACK_SPOTLIGHT 

118 

119 

120def _decode_spotlight(raw: dict[str, Any]) -> Spotlight: 

121 title = _clean_text(raw.get("title", ""), max_len=28) 

122 description = _clean_text(raw.get("description", ""), max_len=72) 

123 url = _clean_url(raw.get("url", "")) 

124 accent = _clean_text(raw.get("accent", ""), max_len=16) 

125 stars = _parse_stars(raw.get("stars")) 

126 expires_at = _parse_datetime(raw.get("expires_at")) 

127 if not title or not description or not url: 

128 raise ValueError("Spotlight requires title, description, and url") 

129 if expires_at and expires_at.astimezone() < datetime.now().astimezone(): 

130 raise ValueError("Spotlight has expired") 

131 return Spotlight( 

132 title=title, 

133 description=description, 

134 url=url, 

135 accent=accent, 

136 expires_at=expires_at, 

137 stars=stars, 

138 ) 

139 

140 

141def _clean_text(value: Any, *, max_len: int) -> str: 

142 text = str(value or "").replace("\n", " ").strip() 

143 return text[:max_len] 

144 

145 

146def _clean_url(value: Any) -> str: 

147 url = str(value or "").replace("\n", "").strip() 

148 if url.startswith("https://"): 

149 url = url[len("https://") :] 

150 elif url.startswith("http://"): 

151 url = url[len("http://") :] 

152 if not url or " " in url: 

153 raise ValueError("Invalid Spotlight URL") 

154 return url[:80].rstrip("/") 

155 

156 

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

158 if not raw: 

159 return None 

160 try: 

161 return datetime.fromisoformat(str(raw).replace("Z", "+00:00")) 

162 except ValueError: 

163 return None 

164 

165 

166def _parse_stars(raw: Any) -> int | None: 

167 if raw is None or isinstance(raw, bool): 

168 return None 

169 try: 

170 return max(0, int(raw)) 

171 except (TypeError, ValueError): 

172 return None