diff --git a/daemon/codex-usage-daemon.py b/daemon/codex-usage-daemon.py index 5e809f7..becd846 100755 --- a/daemon/codex-usage-daemon.py +++ b/daemon/codex-usage-daemon.py @@ -934,44 +934,25 @@ def claude_plan_usage_snapshot() -> UsageSnapshot | None: except (OSError, json.JSONDecodeError): return None - session = data.get("session") - weekly = data.get("weekly") - if not isinstance(session, dict) and not isinstance(weekly, dict): - return None + session = data.get("session") if isinstance(data.get("session"), dict) else None + weekly = data.get("weekly") if isinstance(data.get("weekly"), dict) else None - session_used = ( - float(session.get("used_pct") or 0) - if isinstance(session, dict) - else 0.0 - ) - weekly_used = ( - float(weekly.get("used_pct") or 0) - if isinstance(weekly, dict) - else 0.0 - ) - session_reset = ( - int(session.get("reset_mins") or -1) - if isinstance(session, dict) - else -1 - ) - weekly_reset = ( - int(weekly.get("reset_mins") or -1) - if isinstance(weekly, dict) - else -1 - ) + # Require both windows so a partial payload can't override another source + # with a phantom 100%-remaining for the missing one. + if session is None or weekly is None: + return None - status_parts = [] - if isinstance(session, dict): - status_parts.append(f"5h {round(session_used)}% used") - if isinstance(weekly, dict): - status_parts.append(f"7d {round(weekly_used)}% used") + session_used = float(session.get("used_pct") or 0) + weekly_used = float(weekly.get("used_pct") or 0) + session_reset = int(session.get("reset_mins") or -1) + weekly_reset = int(weekly.get("reset_mins") or -1) return UsageSnapshot( session_pct=remaining_pct_from_used(session_used), session_reset_mins=session_reset, weekly_pct=remaining_pct_from_used(weekly_used), weekly_reset_mins=weekly_reset, - status=" · ".join(status_parts) or "Claude plan usage", + status=f"5h {round(session_used)}% used · 7d {round(weekly_used)}% used", ok=True, ) diff --git a/scripts/claude-plan-usage-cache.py b/scripts/claude-plan-usage-cache.py index 159d4cd..8812312 100755 --- a/scripts/claude-plan-usage-cache.py +++ b/scripts/claude-plan-usage-cache.py @@ -31,45 +31,87 @@ def minutes_from_duration(hours: int, minutes: int) -> int: return max(0, hours * 60 + minutes) +def _parse_duration_reset(text: str) -> int | None: + """Parse 'Resets in 3 hr 17 min', 'Resets in 32 min', or 'Resets in 2 hr'.""" + match = re.search( + r"Resets in\s+(?:(\d+)\s*hr)?\s*(?:(\d+)\s*min)?", + text, + re.IGNORECASE, + ) + if not match or (match.group(1) is None and match.group(2) is None): + return None + hours = int(match.group(1) or 0) + minutes = int(match.group(2) or 0) + return minutes_from_duration(hours, minutes) + + +def _parse_weekday_reset(text: str) -> int | None: + """Parse 'Resets Fri 7:00 PM' (legacy format).""" + match = re.search( + r"Resets\s+([A-Za-z]{3})\s+(\d{1,2}):(\d{2})\s*([AP]M)", + text, + re.IGNORECASE, + ) + if not match: + return None + hour = int(match.group(2)) + minute = int(match.group(3)) + meridiem = match.group(4).lower() + if meridiem == "pm" and hour != 12: + hour += 12 + elif meridiem == "am" and hour == 12: + hour = 0 + return minutes_until_weekday_time(match.group(1), hour, minute) + + def parse_usage_text(text: str) -> dict: compact = "\n".join(line.strip() for line in text.splitlines() if line.strip()) - session_match = re.search( - r"Current session\s+Resets in\s+(\d+)\s+hr\s+(\d+)\s+min.*?(\d+)%\s+used", + + # Find each section's text block: the label up to the next blank line / next + # known section. We use lookahead-free non-greedy match against the other + # known section headers and a sentinel "% used" boundary. + sentinels = r"(?:Weekly limits|All models|Claude Design|Additional features|Last updated|$)" + session_block = re.search( + rf"Current session\s+(.*?{sentinels})", compact, re.IGNORECASE | re.DOTALL, ) - weekly_match = re.search( - r"All models\s+Resets\s+([A-Za-z]{3})\s+(\d{1,2}):(\d{2})\s*([AP]M).*?(\d+)%\s+used", + weekly_block = re.search( + rf"All models\s+(.*?{sentinels})", compact, re.IGNORECASE | re.DOTALL, ) - if not session_match and not weekly_match: - raise ValueError("Claude usage text did not include recognizable plan usage") - usage: dict = { "source": "claude_usage_page_text", "captured_at": datetime.now().astimezone().isoformat(), } - if session_match: - usage["session"] = { - "used_pct": int(session_match.group(3)), - "reset_mins": minutes_from_duration( - int(session_match.group(1)), int(session_match.group(2)) - ), - } - if weekly_match: - hour = int(weekly_match.group(2)) - minute = int(weekly_match.group(3)) - meridiem = weekly_match.group(4).lower() - if meridiem == "pm" and hour != 12: - hour += 12 - elif meridiem == "am" and hour == 12: - hour = 0 - usage["weekly"] = { - "used_pct": int(weekly_match.group(5)), - "reset_mins": minutes_until_weekday_time(weekly_match.group(1), hour, minute), - } + + def percent_used(block_text: str) -> int | None: + m = re.search(r"(\d+)\s*%\s*used", block_text, re.IGNORECASE) + return int(m.group(1)) if m else None + + if session_block: + block = session_block.group(1) + used = percent_used(block) + reset = _parse_duration_reset(block) + if reset is None: + reset = _parse_weekday_reset(block) + if used is not None and reset is not None: + usage["session"] = {"used_pct": used, "reset_mins": reset} + + if weekly_block: + block = weekly_block.group(1) + used = percent_used(block) + reset = _parse_duration_reset(block) + if reset is None: + reset = _parse_weekday_reset(block) + if used is not None and reset is not None: + usage["weekly"] = {"used_pct": used, "reset_mins": reset} + + if "session" not in usage and "weekly" not in usage: + raise ValueError("Claude usage text did not include recognizable plan usage") + return usage