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

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 contextlib 

11import json 

12import os 

13import time 

14from collections.abc import Callable, Iterator 

15from datetime import UTC, datetime 

16from pathlib import Path 

17 

18from secchi.models import HistorySnapshot 

19 

20_LOCK_STALE_SECONDS = 30 

21_LOCK_TIMEOUT_SECONDS = 5 

22_LOCK_POLL_SECONDS = 0.05 

23 

24 

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

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

27 if root is not None: 

28 return root / "history.json" 

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

30 root = Path(base) 

31 else: 

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

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

34 

35 

36@contextlib.contextmanager 

37def _locked(path: Path) -> Iterator[None]: 

38 """Best-effort advisory lock so two secchi processes don't lose each 

39 other's snapshots in a read-modify-write race on history.json. 

40 

41 Uses atomic lock-file creation (``O_CREAT | O_EXCL``), which behaves 

42 identically on POSIX and Windows — no ``fcntl``/``msvcrt`` split needed. 

43 A lock older than ``_LOCK_STALE_SECONDS`` is assumed abandoned by a 

44 crashed holder and reclaimed. If the lock can't be acquired within 

45 ``_LOCK_TIMEOUT_SECONDS``, the operation proceeds unlocked rather than 

46 hanging or failing — history is a best-effort local cache, never a 

47 reason to block a run. 

48 """ 

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

50 lock_path = path.with_name(f"{path.name}.lock") 

51 deadline = time.monotonic() + _LOCK_TIMEOUT_SECONDS 

52 fd: int | None = None 

53 try: 

54 while fd is None: 

55 try: 

56 fd = os.open(lock_path, os.O_CREAT | os.O_EXCL | os.O_WRONLY) 

57 except OSError: 

58 try: 

59 age = time.time() - lock_path.stat().st_mtime 

60 except OSError: 

61 age = 0.0 

62 if age > _LOCK_STALE_SECONDS: 

63 with contextlib.suppress(OSError): 

64 lock_path.unlink() 

65 continue 

66 if time.monotonic() >= deadline: 

67 break 

68 time.sleep(_LOCK_POLL_SECONDS) 

69 yield 

70 finally: 

71 if fd is not None: 

72 with contextlib.suppress(OSError): 

73 os.close(fd) 

74 with contextlib.suppress(OSError): 

75 lock_path.unlink() 

76 

77 

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

79 path = path or history_file_path() 

80 if not path.exists(): 

81 return {} 

82 try: 

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

84 except (json.JSONDecodeError, OSError): 

85 return {} 

86 

87 

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

89 path = path or history_file_path() 

90 tmp_path = path.with_name(f"{path.name}.{os.getpid()}.tmp") 

91 try: 

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

93 tmp_path.write_text(json.dumps(data, indent=2)) 

94 os.replace(tmp_path, path) 

95 except OSError: 

96 with contextlib.suppress(OSError): 

97 tmp_path.unlink() 

98 

99 

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

101 snapshots: list[HistorySnapshot] = [] 

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

103 ts = raw.get("timestamp") 

104 try: 

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

106 except (ValueError, TypeError): 

107 timestamp = None 

108 if timestamp is None: 

109 continue 

110 snapshots.append( 

111 HistorySnapshot( 

112 timestamp=timestamp, 

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

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

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

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

117 ) 

118 ) 

119 return snapshots 

120 

121 

122def append_snapshot( 

123 key: str, 

124 snapshot: HistorySnapshot, 

125 max_keep: int = 420, 

126 *, 

127 path: Path | None = None, 

128) -> None: 

129 path = path or history_file_path() 

130 with _locked(path): 

131 data = _load_all(path) 

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

133 entries.append( 

134 { 

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

136 "stars": snapshot.stars, 

137 "open_issues": snapshot.open_issues, 

138 "health_score": snapshot.health_score, 

139 "reverse_dependency_count": snapshot.reverse_dependency_count, 

140 } 

141 ) 

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

143 _save_all(data, path) 

144 

145 

146def find_baseline( 

147 snapshots: list[HistorySnapshot], 

148 min_age_days: int = 6, 

149 max_age_days: int = 10, 

150 *, 

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

152) -> HistorySnapshot | None: 

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

154 current_time = (now or (lambda: datetime.now(UTC)))() 

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

156 for snap in snapshots: 

157 ts = snap.timestamp 

158 if ts.tzinfo is None: 

159 ts = ts.replace(tzinfo=UTC) 

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

161 if min_age_days <= age_days <= max_age_days: 

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

163 if not candidates: 

164 return None 

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

166 return candidates[0][1] 

167 

168 

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

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

171 if baseline is None: 

172 return None 

173 return current - baseline