Skip to content
105 changes: 103 additions & 2 deletions src/claude_monitor/cli/main.py
Original file line number Diff line number Diff line change
Expand Up @@ -385,6 +385,29 @@
logger = logging.getLogger(__name__)

try:
# Parse date filters early so they can be used for both current and historical data
from datetime import datetime, timedelta
from claude_monitor.utils.time_utils import TimezoneHandler

Check failure on line 390 in src/claude_monitor/cli/main.py

View workflow job for this annotation

GitHub Actions / Lint with Ruff (3.12)

Ruff (I001)

src/claude_monitor/cli/main.py:389:9: I001 Import block is un-sorted or un-formatted

Check failure on line 390 in src/claude_monitor/cli/main.py

View workflow job for this annotation

GitHub Actions / Lint with Ruff (3.9)

Ruff (I001)

src/claude_monitor/cli/main.py:389:9: I001 Import block is un-sorted or un-formatted

Check failure on line 390 in src/claude_monitor/cli/main.py

View workflow job for this annotation

GitHub Actions / Lint with Ruff (3.11)

Ruff (I001)

src/claude_monitor/cli/main.py:389:9: I001 Import block is un-sorted or un-formatted

Check failure on line 391 in src/claude_monitor/cli/main.py

View workflow job for this annotation

GitHub Actions / Lint with Ruff (3.12)

Ruff (W293)

src/claude_monitor/cli/main.py:391:1: W293 Blank line contains whitespace

Check failure on line 391 in src/claude_monitor/cli/main.py

View workflow job for this annotation

GitHub Actions / Lint with Ruff (3.9)

Ruff (W293)

src/claude_monitor/cli/main.py:391:1: W293 Blank line contains whitespace

Check failure on line 391 in src/claude_monitor/cli/main.py

View workflow job for this annotation

GitHub Actions / Lint with Ruff (3.11)

Ruff (W293)

src/claude_monitor/cli/main.py:391:1: W293 Blank line contains whitespace
tz = TimezoneHandler(args.timezone)

Check failure on line 393 in src/claude_monitor/cli/main.py

View workflow job for this annotation

GitHub Actions / Lint with Ruff (3.12)

Ruff (W293)

src/claude_monitor/cli/main.py:393:1: W293 Blank line contains whitespace

Check failure on line 393 in src/claude_monitor/cli/main.py

View workflow job for this annotation

GitHub Actions / Lint with Ruff (3.9)

Ruff (W293)

src/claude_monitor/cli/main.py:393:1: W293 Blank line contains whitespace

Check failure on line 393 in src/claude_monitor/cli/main.py

View workflow job for this annotation

GitHub Actions / Lint with Ruff (3.11)

Ruff (W293)

src/claude_monitor/cli/main.py:393:1: W293 Blank line contains whitespace
def _parse_date(date_str: Optional[str]):
if not date_str:
return None
for fmt in ("%Y-%m-%d", "%Y.%m.%d", "%Y/%m/%d"):
try:
return tz.ensure_timezone(datetime.strptime(date_str, fmt))
except ValueError:
continue
print_themed(f"Invalid date format: {date_str}. Use YYYY-MM-DD.", style="warning")
return None

Check failure on line 404 in src/claude_monitor/cli/main.py

View workflow job for this annotation

GitHub Actions / Lint with Ruff (3.12)

Ruff (W293)

src/claude_monitor/cli/main.py:404:1: W293 Blank line contains whitespace

Check failure on line 404 in src/claude_monitor/cli/main.py

View workflow job for this annotation

GitHub Actions / Lint with Ruff (3.9)

Ruff (W293)

src/claude_monitor/cli/main.py:404:1: W293 Blank line contains whitespace

Check failure on line 404 in src/claude_monitor/cli/main.py

View workflow job for this annotation

GitHub Actions / Lint with Ruff (3.11)

Ruff (W293)

src/claude_monitor/cli/main.py:404:1: W293 Blank line contains whitespace
start_dt = _parse_date(getattr(args, "start_date", None))
end_dt = _parse_date(getattr(args, "end_date", None))

Check failure on line 407 in src/claude_monitor/cli/main.py

View workflow job for this annotation

GitHub Actions / Lint with Ruff (3.12)

Ruff (W293)

src/claude_monitor/cli/main.py:407:1: W293 Blank line contains whitespace

Check failure on line 407 in src/claude_monitor/cli/main.py

View workflow job for this annotation

GitHub Actions / Lint with Ruff (3.9)

Ruff (W293)

src/claude_monitor/cli/main.py:407:1: W293 Blank line contains whitespace

Check failure on line 407 in src/claude_monitor/cli/main.py

View workflow job for this annotation

GitHub Actions / Lint with Ruff (3.11)

Ruff (W293)

src/claude_monitor/cli/main.py:407:1: W293 Blank line contains whitespace
# Make end date inclusive by adding one day (entries use precise timestamps)
end_dt_inclusive = end_dt + timedelta(days=1) if end_dt else None

# Create aggregator with appropriate mode
aggregator = UsageAggregator(
data_path=str(data_path),
Expand All @@ -395,9 +418,87 @@
# Create table controller
controller = TableViewsController(console=console)

# Get aggregated data
# Get aggregated data with date filters
logger.info(f"Loading {view_mode} usage data...")
aggregated_data = aggregator.aggregate()
aggregated_data = aggregator.aggregate(start_dt, end_dt_inclusive)

# Initialize history manager for daily and monthly views
history_mode = getattr(args, "history", "auto")
if history_mode != "off":
from claude_monitor.data.history_manager import HistoryManager
history_manager = HistoryManager()

Check failure on line 430 in src/claude_monitor/cli/main.py

View workflow job for this annotation

GitHub Actions / Lint with Ruff (3.12)

Ruff (W293)

src/claude_monitor/cli/main.py:430:1: W293 Blank line contains whitespace

Check failure on line 430 in src/claude_monitor/cli/main.py

View workflow job for this annotation

GitHub Actions / Lint with Ruff (3.9)

Ruff (W293)

src/claude_monitor/cli/main.py:430:1: W293 Blank line contains whitespace

Check failure on line 430 in src/claude_monitor/cli/main.py

View workflow job for this annotation

GitHub Actions / Lint with Ruff (3.11)

Ruff (W293)

src/claude_monitor/cli/main.py:430:1: W293 Blank line contains whitespace
if view_mode == "daily":
# Load historical data if in auto or readonly mode
if history_mode in ["auto", "readonly"]:
# Load historical data using the same date filters
historical_data = history_manager.load_historical_daily_data(
start_date=start_dt,
end_date=end_dt_inclusive
)

Check failure on line 439 in src/claude_monitor/cli/main.py

View workflow job for this annotation

GitHub Actions / Lint with Ruff (3.12)

Ruff (W293)

src/claude_monitor/cli/main.py:439:1: W293 Blank line contains whitespace

Check failure on line 439 in src/claude_monitor/cli/main.py

View workflow job for this annotation

GitHub Actions / Lint with Ruff (3.9)

Ruff (W293)

src/claude_monitor/cli/main.py:439:1: W293 Blank line contains whitespace

Check failure on line 439 in src/claude_monitor/cli/main.py

View workflow job for this annotation

GitHub Actions / Lint with Ruff (3.11)

Ruff (W293)

src/claude_monitor/cli/main.py:439:1: W293 Blank line contains whitespace
if historical_data:
print_themed(f"Loaded {len(historical_data)} days from history", style="info")

Check failure on line 442 in src/claude_monitor/cli/main.py

View workflow job for this annotation

GitHub Actions / Lint with Ruff (3.12)

Ruff (W293)

src/claude_monitor/cli/main.py:442:1: W293 Blank line contains whitespace

Check failure on line 442 in src/claude_monitor/cli/main.py

View workflow job for this annotation

GitHub Actions / Lint with Ruff (3.9)

Ruff (W293)

src/claude_monitor/cli/main.py:442:1: W293 Blank line contains whitespace

Check failure on line 442 in src/claude_monitor/cli/main.py

View workflow job for this annotation

GitHub Actions / Lint with Ruff (3.11)

Ruff (W293)

src/claude_monitor/cli/main.py:442:1: W293 Blank line contains whitespace
# Merge with current data
aggregated_data = history_manager.merge_with_current_data(
aggregated_data, historical_data
)
print_themed(f"Displaying {len(aggregated_data)} total days", style="info")

Check failure on line 448 in src/claude_monitor/cli/main.py

View workflow job for this annotation

GitHub Actions / Lint with Ruff (3.12)

Ruff (W293)

src/claude_monitor/cli/main.py:448:1: W293 Blank line contains whitespace

Check failure on line 448 in src/claude_monitor/cli/main.py

View workflow job for this annotation

GitHub Actions / Lint with Ruff (3.9)

Ruff (W293)

src/claude_monitor/cli/main.py:448:1: W293 Blank line contains whitespace

Check failure on line 448 in src/claude_monitor/cli/main.py

View workflow job for this annotation

GitHub Actions / Lint with Ruff (3.11)

Ruff (W293)

src/claude_monitor/cli/main.py:448:1: W293 Blank line contains whitespace
# Save current data to history if in auto or writeonly mode
if aggregated_data and history_mode in ["auto", "writeonly"]:
saved_count = history_manager.save_daily_data(aggregated_data)
if saved_count > 0:
print_themed(f"Saved {saved_count} days to history", style="success")

Check failure on line 454 in src/claude_monitor/cli/main.py

View workflow job for this annotation

GitHub Actions / Lint with Ruff (3.12)

Ruff (W293)

src/claude_monitor/cli/main.py:454:1: W293 Blank line contains whitespace

Check failure on line 454 in src/claude_monitor/cli/main.py

View workflow job for this annotation

GitHub Actions / Lint with Ruff (3.9)

Ruff (W293)

src/claude_monitor/cli/main.py:454:1: W293 Blank line contains whitespace

Check failure on line 454 in src/claude_monitor/cli/main.py

View workflow job for this annotation

GitHub Actions / Lint with Ruff (3.11)

Ruff (W293)

src/claude_monitor/cli/main.py:454:1: W293 Blank line contains whitespace
elif view_mode == "monthly":
# For monthly view, always work with daily data and aggregate to monthly
if history_mode in ["auto", "readonly"]:
# Get current daily data first
daily_aggregator = UsageAggregator(
data_path=str(data_path),
aggregation_mode="daily",
timezone=args.timezone,
)
current_daily = daily_aggregator.aggregate(start_dt, end_dt_inclusive)

# Load historical daily data
daily_historical = history_manager.load_historical_daily_data(
start_date=start_dt,
end_date=end_dt_inclusive
)

# Save current daily data to history if in auto mode
if history_mode == "auto" and current_daily:
saved = history_manager.save_daily_data(current_daily)
if saved > 0:
print_themed(f"Saved {saved} days to history", style="success")

# Merge current and historical daily data
all_daily = []
if current_daily and daily_historical:
all_daily = history_manager.merge_with_current_data(
current_daily, daily_historical
)
print_themed(f"Merged {len(current_daily)} current + {len(daily_historical)} historical days", style="info")
elif current_daily:
all_daily = current_daily
print_themed(f"Using {len(current_daily)} current days", style="info")
elif daily_historical:
all_daily = daily_historical
print_themed(f"Using {len(daily_historical)} historical days", style="info")

# Always aggregate daily data into monthly
if all_daily:
monthly_from_daily = history_manager.aggregate_monthly_from_daily(all_daily)

if monthly_from_daily:
# Replace the initial aggregated_data with the one from daily
aggregated_data = monthly_from_daily
print_themed(f"Displaying {len(aggregated_data)} months aggregated from {len(all_daily)} days", style="info")
else:
print_themed("No monthly data could be aggregated from daily data", style="warning")

if not aggregated_data:
print_themed(f"No usage data found for {view_mode} view", style="warning")
Expand Down
32 changes: 32 additions & 0 deletions src/claude_monitor/core/settings.py
Original file line number Diff line number Diff line change
Expand Up @@ -138,6 +138,16 @@ def _get_system_time_format() -> str:
description="Display theme (light, dark, classic, auto)",
)

start_date: Optional[str] = Field(
default=None,
description="Start date for filtering data (YYYY-MM-DD format)"
)

end_date: Optional[str] = Field(
default=None,
description="End date for filtering data (YYYY-MM-DD format)"
)

custom_limit_tokens: Optional[int] = Field(
default=None, gt=0, description="Token limit for custom plan"
)
Expand Down Expand Up @@ -170,6 +180,11 @@ def _get_system_time_format() -> str:

clear: bool = Field(default=False, description="Clear saved configuration")

history: Literal["auto", "off", "readonly", "writeonly"] = Field(
default="auto",
description="History mode: auto (save+load), off (disable), readonly (load only), writeonly (save only)"
)

@field_validator("plan", mode="before")
@classmethod
def validate_plan(cls, v: Any) -> str:
Expand Down Expand Up @@ -240,6 +255,20 @@ def validate_log_level(cls, v: str) -> str:
raise ValueError(f"Invalid log level: {v}")
return v_upper

@field_validator("history", mode="before")
@classmethod
def validate_history(cls, v: Any) -> str:
"""Validate and normalize history mode value."""
if isinstance(v, str):
v_lower = v.lower()
valid_modes = ["auto", "off", "readonly", "writeonly"]
if v_lower in valid_modes:
return v_lower
raise ValueError(
f"Invalid history mode: {v}. Must be one of: {', '.join(valid_modes)}"
)
return v

@classmethod
def settings_customise_sources(
cls,
Expand Down Expand Up @@ -350,5 +379,8 @@ def to_namespace(self) -> argparse.Namespace:
args.log_level = self.log_level
args.log_file = str(self.log_file) if self.log_file else None
args.version = self.version
args.start_date = self.start_date
args.end_date = self.end_date
args.history = self.history

return args
16 changes: 12 additions & 4 deletions src/claude_monitor/data/aggregator.py
Original file line number Diff line number Diff line change
Expand Up @@ -266,9 +266,17 @@ def calculate_totals(self, aggregated_data: List[Dict[str, Any]]) -> Dict[str, A
"entries_count": total_stats.count,
}

def aggregate(self) -> List[Dict[str, Any]]:
def aggregate(
self,
start_date: Optional[datetime] = None,
end_date: Optional[datetime] = None
) -> List[Dict[str, Any]]:
"""Main aggregation method that reads data and returns aggregated results.

Args:
start_date: Optional start date filter
end_date: Optional end date filter

Returns:
List of aggregated data based on aggregation_mode
"""
Expand All @@ -288,10 +296,10 @@ def aggregate(self) -> List[Dict[str, Any]]:
if entry.timestamp.tzinfo is None:
entry.timestamp = self.timezone_handler.ensure_timezone(entry.timestamp)

# Aggregate based on mode
# Aggregate based on mode with date filters
if self.aggregation_mode == "daily":
return self.aggregate_daily(entries)
return self.aggregate_daily(entries, start_date, end_date)
elif self.aggregation_mode == "monthly":
return self.aggregate_monthly(entries)
return self.aggregate_monthly(entries, start_date, end_date)
else:
raise ValueError(f"Invalid aggregation mode: {self.aggregation_mode}")
Loading
Loading