Read savings.jsonl and return a SavingsSummary.
(path: Path | None = None)
| 105 | |
| 106 | |
| 107 | def build_savings_summary(path: Path | None = None) -> SavingsSummary: |
| 108 | """Read savings.jsonl and return a SavingsSummary.""" |
| 109 | if path is None: |
| 110 | path = _get_stats_file() |
| 111 | now = datetime.now(timezone.utc) |
| 112 | today = now.date() |
| 113 | seven_days_ago = (now - timedelta(days=7)).date() |
| 114 | |
| 115 | buckets = { |
| 116 | "Today": BucketStats(), |
| 117 | "Last 7 days": BucketStats(), |
| 118 | "All time": BucketStats(), |
| 119 | } |
| 120 | call_type_counts: defaultdict[str, int] = defaultdict(int) |
| 121 | |
| 122 | with path.open() as f: |
| 123 | for line in f: |
| 124 | try: |
| 125 | record = json.loads(line) |
| 126 | except json.JSONDecodeError: |
| 127 | logger.warning("Skipping malformed JSON line in stats file") |
| 128 | continue |
| 129 | snippet_chars = record["snippet_chars"] |
| 130 | file_chars = record["file_chars"] |
| 131 | call_type = record["call"] |
| 132 | call_type_counts[call_type] += 1 |
| 133 | dt = datetime.fromtimestamp(record["ts"], tz=timezone.utc) |
| 134 | in_today = dt.date() == today |
| 135 | in_last_7 = dt.date() > seven_days_ago |
| 136 | buckets["All time"].add(snippet_chars, file_chars) |
| 137 | if in_last_7: |
| 138 | buckets["Last 7 days"].add(snippet_chars, file_chars) |
| 139 | if in_today: |
| 140 | buckets["Today"].add(snippet_chars, file_chars) |
| 141 | |
| 142 | return SavingsSummary(buckets=buckets, call_type_counts=dict(call_type_counts)) |
| 143 | |
| 144 | |
| 145 | def _format_token_count(tokens: int) -> str: |