Skip to content
135 changes: 133 additions & 2 deletions src/claude_monitor/cli/main.py
Original file line number Diff line number Diff line change
Expand Up @@ -385,6 +385,32 @@ def _run_table_view(
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

tz = TimezoneHandler(args.timezone)

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

start_dt = _parse_date(getattr(args, "start_date", None))
end_dt = _parse_date(getattr(args, "end_date", None))

# 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 +421,114 @@ def _run_table_view(
# 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()

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
)

if historical_data:
print_themed(
f"Loaded {len(historical_data)} days from history",
style="info",
)

# 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",
)

# 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"
)

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
30 changes: 30 additions & 0 deletions src/claude_monitor/core/settings.py
Original file line number Diff line number Diff line change
Expand Up @@ -138,6 +138,14 @@ 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 +178,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 +253,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 +377,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
14 changes: 10 additions & 4 deletions src/claude_monitor/data/aggregator.py
Original file line number Diff line number Diff line change
Expand Up @@ -266,9 +266,15 @@ 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 +294,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