Coverage for src/secchi/history.py: 43%

1"""Local snapshot cache — produces real week-over-week deltas. 

2 

3GitHub stars and open-issue counts have no cheap point-in-time historical API, 

4so we persist a small rolling cache of snapshots per package. Deltas degrade to 

5None (rendered "—") until a baseline of the right age exists — never fabricated. 

6""" 

7 

8from __future__ import annotations 

9 

10import json 

11import os 

12from collections.abc import Callable 

13from datetime import datetime, timezone 

14from pathlib import Path 

15 

16from secchi.models import HistorySnapshot 

17 

18 

19def history_file_path(root: Path | None = None) -> Path: 

20 """XDG_CACHE_HOME/secchi/history.json, else ~/.cache/secchi/history.json.""" 

21 if root is not None: 

22 return root / "history.json" 

23 if base := os.environ.get("XDG_CACHE_HOME", ""): 

24 root = Path(base) 

25 else: 

26 root = Path.home() / ".cache" 

27 return root / "secchi" / "history.json" 

28 

29 

30def _load_all(path: Path | None = None) -> dict[str, list[dict]]: 

31 path = path or history_file_path() 

32 if not path.exists(): 

33 return {} 

34 try: 

35 return json.loads(path.read_text()) 

36 except (json.JSONDecodeError, OSError): 

37 return {} 

38 

39 

40def _save_all(data: dict[str, list[dict]], path: Path | None = None) -> None: 

41 path = path or history_file_path() 

42 try: 

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

44 path.write_text(json.dumps(data, indent=2)) 

45 except OSError: 

46 pass 

47 

48 

49def load_snapshots(key: str, *, path: Path | None = None) -> list[HistorySnapshot]: 

50 snapshots: list[HistorySnapshot] = [] 

51 for raw in _load_all(path).get(key, []): 

52 ts = raw.get("timestamp") 

53 try: 

54 timestamp = datetime.fromisoformat(ts) if ts else None 

55 except (ValueError, TypeError): 

56 timestamp = None 

57 if timestamp is None: 

58 continue 

59 snapshots.append( 

60 HistorySnapshot( 

61 timestamp=timestamp, 

62 stars=raw.get("stars", 0), 

63 open_issues=raw.get("open_issues", 0), 

64 health_score=raw.get("health_score"), 

65 reverse_dependency_count=raw.get("reverse_dependency_count"), 

66 ) 

67 ) 

68 return snapshots 

69 

70 

71def append_snapshot( 

72 key: str, 

73 snapshot: HistorySnapshot, 

74 max_keep: int = 420, 

75 *, 

76 path: Path | None = None, 

77) -> None: 

78 data = _load_all(path) 

79 entries = data.get(key, []) 

80 entries.append( 

81 { 

82 "timestamp": snapshot.timestamp.isoformat(), 

83 "stars": snapshot.stars, 

84 "open_issues": snapshot.open_issues, 

85 "health_score": snapshot.health_score, 

86 "reverse_dependency_count": snapshot.reverse_dependency_count, 

87 } 

88 ) 

89 data[key] = entries[-max_keep:] 

90 _save_all(data, path) 

91 

92 

93def find_baseline( 

94 snapshots: list[HistorySnapshot], 

95 min_age_days: int = 6, 

96 max_age_days: int = 10, 

97 *, 

98 now: Callable[[], datetime] | None = None, 

99) -> HistorySnapshot | None: 

100 """Closest snapshot whose age falls in [min_age_days, max_age_days].""" 

101 current_time = (now or (lambda: datetime.now(timezone.utc)))() 

102 candidates: list[tuple[float, HistorySnapshot]] = [] 

103 for snap in snapshots: 

104 ts = snap.timestamp 

105 if ts.tzinfo is None: 

106 ts = ts.replace(tzinfo=timezone.utc) 

107 age_days = (current_time - ts).total_seconds() / 86400 

108 if min_age_days <= age_days <= max_age_days: 

109 candidates.append((abs(age_days - 7), snap)) 

110 if not candidates: 

111 return None 

112 candidates.sort(key=lambda c: c[0]) 

113 return candidates[0][1] 

114 

115 

116def compute_delta(current: int, baseline: int | None) -> int | None: 

117 """current - baseline, or None if no baseline — never fabricate.""" 

118 if baseline is None: 

119 return None 

120 return current - baseline