Coverage for src/secchi/utils.py: 18%
194 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-08-04 22:15 +0000
1"""Utility helpers — formatting, GitHub fetching, repo derivation."""
3from __future__ import annotations
5import asyncio
6from datetime import datetime, timedelta, timezone
7from urllib.parse import urlparse
9import httpx
11from secchi.config import get_env_token
12from secchi.http import HttpClientFactory
13from secchi.models import GitHubIssueEvent, GitHubStats
16def shorten_number(n: int) -> str:
17 """Format a large number with K, M, B suffixes."""
18 if n >= 1_000_000_000:
19 return f"{n / 1_000_000_000:.1f}B"
20 if n >= 1_000_000:
21 return f"{n / 1_000_000:.1f}M"
22 if n >= 1_000:
23 return f"{n / 1_000:.1f}K"
24 return str(n)
27def format_age(dt: datetime | None) -> str:
28 """Human 'time ago' string from a datetime (tz-naive treated as UTC)."""
29 if dt is None:
30 return "—"
31 if dt.tzinfo is None:
32 dt = dt.replace(tzinfo=timezone.utc)
33 days = (datetime.now(timezone.utc) - dt).days
34 if days < 0:
35 days = 0
36 if days == 0:
37 return "today"
38 if days == 1:
39 return "1 day ago"
40 if days < 14:
41 return f"{days} days ago"
42 if days < 60:
43 return f"{days // 7} weeks ago"
44 if days < 365:
45 return f"{days // 30} months ago"
46 years = days // 365
47 return "1 year ago" if years == 1 else f"{years} years ago"
50def format_age_short(dt: datetime | None) -> str:
51 """Compact 'time ago' — '2d', '3w', '5mo', '2y' — for narrow panels."""
52 if dt is None:
53 return "—"
54 if dt.tzinfo is None:
55 dt = dt.replace(tzinfo=timezone.utc)
56 days = max((datetime.now(timezone.utc) - dt).days, 0)
57 if days == 0:
58 return "today"
59 if days < 14:
60 return f"{days}d"
61 if days < 60:
62 return f"{days // 7}w"
63 if days < 365:
64 return f"{days // 30}mo"
65 return f"{days // 365}y"
68def format_pct_delta(pct: float | None) -> tuple[str, str]:
69 """Return (text, rich_color) for a percentage change, or ('—', 'dim')."""
70 if pct is None:
71 return "—", "dim"
72 arrow = "↑" if pct >= 0 else "↓"
73 color = "green" if pct >= 0 else "red"
74 return f"{arrow} {abs(pct):.1f}%", color
77def shorten_bytes(n: int | None) -> str:
78 """Format a byte count with B/KB/MB/GB suffixes."""
79 if not n:
80 return "—"
81 value = float(n)
82 for unit in ("B", "KB", "MB", "GB"):
83 if value < 1024 or unit == "GB":
84 if unit == "B":
85 return f"{int(value)} B"
86 return f"{value:.1f} {unit}"
87 value /= 1024
88 return f"{value:.1f} GB"
91def derive_github_repo(urls: list[str]) -> tuple[str, str] | None:
92 """Try to extract (owner, repo) from a list of URLs.
94 Returns None if no GitHub URL is found.
95 """
96 for url in urls:
97 if not url:
98 continue
99 parsed = urlparse(url)
100 host = parsed.hostname or ""
101 if "github.com" not in host:
102 continue
103 path = parsed.path.strip("/")
104 if path.endswith(".git"):
105 path = path[: -len(".git")]
106 parts = path.split("/")
107 if len(parts) >= 2 and parts[0] and parts[1]:
108 return parts[0], parts[1]
109 return None
112def _gh_headers() -> dict[str, str]:
113 token = get_env_token("SECCHI_GITHUB_TOKEN")
114 headers: dict[str, str] = {"Accept": "application/vnd.github+json"}
115 if token:
116 headers["Authorization"] = f"Bearer {token}"
117 return headers
120def _parse_gh_time(raw: str | None) -> datetime | None:
121 if not raw:
122 return None
123 try:
124 return datetime.fromisoformat(raw.replace("Z", "+00:00"))
125 except (ValueError, TypeError):
126 return None
129async def fetch_github_release_notes(
130 owner: str,
131 repo: str,
132 tag: str | None = None,
133 client: httpx.AsyncClient | None = None,
134) -> str:
135 """Fetch the latest release notes from GitHub.
137 Uses SECCHI_GITHUB_TOKEN env var for higher rate limits.
138 """
139 url = f"https://api.github.com/repos/{owner}/{repo}/releases"
140 params: dict[str, str] = {"per_page": "5"}
142 if client is None:
143 async with HttpClientFactory().create() as owned_client:
144 return await fetch_github_release_notes(owner, repo, tag, owned_client)
145 try:
146 resp = await client.get(url, headers=_gh_headers(), params=params)
147 resp.raise_for_status()
148 releases = resp.json()
149 except httpx.HTTPError:
150 return ""
152 if not releases:
153 return ""
155 release = releases[0]
156 if tag and tag in [r.get("tag_name", "") for r in releases]:
157 release = next(r for r in releases if r.get("tag_name") == tag)
159 body = release.get("body", "")
160 max_chars = 3000
161 return body[:max_chars] + ("..." if len(body) > max_chars else "")
164async def fetch_github_stats(
165 owner: str, repo: str, client: httpx.AsyncClient | None = None
166) -> GitHubStats:
167 """Fetch GitHub stars, forks, issues, and timestamps for a repo."""
168 url = f"https://api.github.com/repos/{owner}/{repo}"
169 own_client = client is None
170 own_client = client is None
171 client = client or HttpClientFactory().create()
172 try:
173 resp = await client.get(url, headers=_gh_headers())
174 resp.raise_for_status()
175 data = resp.json()
176 return GitHubStats(
177 stars=data.get("stargazers_count", 0),
178 forks=data.get("forks_count", 0),
179 open_issues=data.get("open_issues_count", 0),
180 created_at=_parse_gh_time(data.get("created_at")),
181 pushed_at=_parse_gh_time(data.get("pushed_at")),
182 resolved=True,
183 )
184 except httpx.HTTPError:
185 return GitHubStats(resolved=False)
186 finally:
187 if own_client:
188 await client.aclose()
191async def fetch_github_repo_signals(
192 owner: str, repo: str, client: httpx.AsyncClient
193) -> tuple[bool, bool]:
194 """Return (has_ci, has_readme).
196 has_ci: .github/workflows contents resolves to a non-empty list.
197 has_readme: /readme resolves 200. Both best-effort; errors => False.
198 """
199 has_ci = False
200 has_readme = False
201 try:
202 resp = await client.get(
203 f"https://api.github.com/repos/{owner}/{repo}/contents/.github/workflows",
204 headers=_gh_headers(),
205 )
206 if resp.status_code == 200:
207 body = resp.json()
208 has_ci = isinstance(body, list) and len(body) > 0
209 except httpx.HTTPError:
210 pass
212 try:
213 resp = await client.get(
214 f"https://api.github.com/repos/{owner}/{repo}/readme",
215 headers=_gh_headers(),
216 )
217 has_readme = resp.status_code == 200
218 except httpx.HTTPError:
219 pass
221 return has_ci, has_readme
224async def fetch_github_issue_events(
225 owner: str,
226 repo: str,
227 client: httpx.AsyncClient,
228 since_days: int = 90,
229 max_pages: int = 3,
230) -> list[GitHubIssueEvent]:
231 """Fetch recent issues AND pull requests, merged.
233 Uses /issues?state=all&since=<now-since_days>. `since` filters on updated_at,
234 which is always >= created_at and >= closed_at, so nothing created/closed in
235 the window is missed. PRs are distinguished by the 'pull_request' key.
236 """
237 since = (datetime.now(timezone.utc) - timedelta(days=since_days)).strftime(
238 "%Y-%m-%dT%H:%M:%SZ"
239 )
240 events: list[GitHubIssueEvent] = []
241 for page in range(1, max_pages + 1):
242 try:
243 resp = await client.get(
244 f"https://api.github.com/repos/{owner}/{repo}/issues",
245 headers=_gh_headers(),
246 params={
247 "state": "all",
248 "since": since,
249 "per_page": "100",
250 "page": str(page),
251 "sort": "updated",
252 "direction": "desc",
253 },
254 )
255 resp.raise_for_status()
256 batch = resp.json()
257 except httpx.HTTPError:
258 break
260 if not batch:
261 break
263 for item in batch:
264 created = _parse_gh_time(item.get("created_at"))
265 if created is None:
266 continue
267 events.append(
268 GitHubIssueEvent(
269 number=item.get("number", 0),
270 title=item.get("title", ""),
271 is_pull_request="pull_request" in item,
272 created_at=created,
273 closed_at=_parse_gh_time(item.get("closed_at")),
274 url=item.get("html_url", ""),
275 )
276 )
278 if len(batch) < 100:
279 break
281 return events
284async def fetch_github_stats_for_package(
285 homepage: str,
286 repository_url: str,
287 client: httpx.AsyncClient | None = None,
288) -> GitHubStats:
289 """Fetch GitHub stats by deriving repo from package URLs."""
290 repo = derive_github_repo([repository_url, homepage])
291 if repo:
292 return await fetch_github_stats(repo[0], repo[1], client=client)
293 return GitHubStats()
296async def fetch_github_extended_stats_for_package(
297 homepage: str,
298 repository_url: str,
299 client: httpx.AsyncClient | None = None,
300) -> tuple[GitHubStats, list[GitHubIssueEvent]]:
301 """Derive the repo once, then fetch stats + signals + issue events in parallel."""
302 repo = derive_github_repo([repository_url, homepage])
303 if not repo:
304 return GitHubStats(), []
306 owner, name = repo
307 if client is None:
308 async with HttpClientFactory().create() as owned_client:
309 return await fetch_github_extended_stats_for_package(
310 homepage, repository_url, client=owned_client
311 )
312 stats, signals, events = await asyncio.gather(
313 fetch_github_stats(owner, name, client=client),
314 fetch_github_repo_signals(owner, name, client),
315 fetch_github_issue_events(owner, name, client),
316 )
317 has_ci, has_readme = signals
318 stats.has_ci = has_ci
319 stats.has_readme = has_readme
320 return stats, events
323async def fetch_release_notes_for_package(
324 homepage: str,
325 repository_url: str,
326 version: str,
327 client: httpx.AsyncClient | None = None,
328) -> str:
329 """Fetch release notes by deriving GitHub repo from package URLs."""
330 repo = derive_github_repo([repository_url, homepage])
331 if repo:
332 owner, name = repo
333 tag = f"v{version}" if not version.startswith("v") else version
334 notes = await fetch_github_release_notes(owner, name, tag, client=client)
335 if notes:
336 return notes
337 return await fetch_github_release_notes(owner, name, client=client)
338 return ""