tegwick
397b607442
Complete period lifecycle management system including: - PeriodManager class with full lifecycle operations - Period status management (open/calculating/closed) with validation - Period overlap detection and conflict resolution - Comprehensive cost calculation and aggregation engine - Loss carried forward calculations between periods - Period closure validation with audit trails - Current period detection and auto-creation utilities CLI Integration: - Complete period command suite (create, list, show, calculate, status, close, current) - Professional CLI output with detailed formatting - Comprehensive error handling and validation - Date filtering and status filtering capabilities Testing: - 25 core PeriodManager tests covering all functionality - 24 CLI command tests ensuring proper integration - Edge case testing for complex scenarios - 49 total tests passing with comprehensive coverage Database Integration: - Utilizes existing cost_periods schema from FinanceModels - Full SQLite integration with proper constraints - Performance-optimized indexes and queries - Seamless integration with existing cost tracking system 🤖 Generated with [Claude Code](https://claude.ai/code) Co-Authored-By: Claude <noreply@anthropic.com>
911 lines
34 KiB
Python
911 lines
34 KiB
Python
"""
|
||
CLI commands for cost tracking and reporting.
|
||
|
||
This module provides command-line interface for cost management operations
|
||
including report generation, cost item management, and period calculations.
|
||
"""
|
||
|
||
import click
|
||
import sys
|
||
from datetime import date, datetime, timedelta
|
||
from decimal import Decimal
|
||
from pathlib import Path
|
||
from typing import Optional
|
||
|
||
from .cost_manager import CostItemManager, CostItem
|
||
from .report_generator import CostReportGenerator, ReportConfig
|
||
from .session_tracker import SessionCostTracker
|
||
from .period_manager import PeriodManager, PeriodStatus
|
||
from ..config_manager import ConfigurationManager
|
||
|
||
|
||
@click.group(name='cost')
|
||
def cost_commands():
|
||
"""Cost tracking and financial reporting commands."""
|
||
pass
|
||
|
||
|
||
@cost_commands.group(name='report')
|
||
def cost_report():
|
||
"""Generate cost reports and financial summaries."""
|
||
pass
|
||
|
||
|
||
@cost_report.command('generate')
|
||
@click.option('--period',
|
||
help='Period in YYYY-MM format (e.g., 2025-01)')
|
||
@click.option('--format', 'report_format',
|
||
type=click.Choice(['summary', 'detailed', 'audit']),
|
||
default='summary',
|
||
help='Report format')
|
||
@click.option('--output', 'output_path',
|
||
help='Output file path (optional)')
|
||
@click.option('--database', 'db_path',
|
||
help='Database path (defaults to config)')
|
||
def generate_report(period: Optional[str], report_format: str,
|
||
output_path: Optional[str], db_path: Optional[str]):
|
||
"""Generate cost report for specified period."""
|
||
try:
|
||
# Get database path
|
||
if not db_path:
|
||
config_manager = ConfigurationManager()
|
||
config = config_manager.get_current_config()
|
||
db_path = config.get('database_path')
|
||
|
||
if not db_path:
|
||
click.echo("Error: No database path specified. Use --database or configure database_path.", err=True)
|
||
sys.exit(1)
|
||
|
||
# Parse period
|
||
if period:
|
||
try:
|
||
year, month = map(int, period.split('-'))
|
||
if not (1 <= month <= 12):
|
||
raise ValueError("Month must be between 1 and 12")
|
||
except ValueError:
|
||
click.echo("Error: Period must be in YYYY-MM format (e.g., 2025-01)", err=True)
|
||
sys.exit(1)
|
||
else:
|
||
# Default to current month
|
||
now = date.today()
|
||
year, month = now.year, now.month
|
||
|
||
# Generate report
|
||
generator = CostReportGenerator(db_path)
|
||
report_content = generator.generate_period_report(year, month, report_format)
|
||
|
||
# Output report
|
||
if output_path:
|
||
generator.save_report(report_content, output_path)
|
||
click.echo(f"Report saved to: {output_path}")
|
||
else:
|
||
click.echo(report_content)
|
||
|
||
except Exception as e:
|
||
click.echo(f"Error generating report: {e}", err=True)
|
||
sys.exit(1)
|
||
|
||
|
||
@cost_report.command('template')
|
||
@click.option('--show', is_flag=True, help='Show template structure')
|
||
@click.option('--format', 'report_format',
|
||
type=click.Choice(['summary', 'detailed', 'audit']),
|
||
default='summary',
|
||
help='Template format to display')
|
||
def show_template(show: bool, report_format: str):
|
||
"""Display cost report template structure."""
|
||
if show:
|
||
template_info = {
|
||
'summary': {
|
||
'description': 'High-level overview with category totals',
|
||
'sections': ['Overview', 'Cost Breakdown by Category', 'Top Cost Items'],
|
||
'frontmatter': ['report_type', 'period_start', 'period_end', 'total_costs', 'currency'],
|
||
'contentmatter': ['cost_data.total', 'cost_data.categories', 'cost_data.active_items']
|
||
},
|
||
'detailed': {
|
||
'description': 'Complete breakdown with all cost items',
|
||
'sections': ['Executive Summary', 'Category Sections (with item tables)'],
|
||
'frontmatter': ['report_type', 'period_start', 'period_end', 'categories_count'],
|
||
'contentmatter': ['cost_data.summary', 'cost_data.categories', 'cost_data.items']
|
||
},
|
||
'audit': {
|
||
'description': 'Full audit trail with transaction history',
|
||
'sections': ['Audit Summary', 'Cost Verification', 'Transaction History', 'Audit Trail'],
|
||
'frontmatter': ['report_type', 'audit_trail', 'transactions_count'],
|
||
'contentmatter': ['audit_data.period_summary', 'audit_data.transactions']
|
||
}
|
||
}
|
||
|
||
info = template_info[report_format]
|
||
click.echo(f"## {report_format.title()} Report Template")
|
||
click.echo(f"**Description**: {info['description']}")
|
||
click.echo()
|
||
click.echo("**Sections**:")
|
||
for section in info['sections']:
|
||
click.echo(f"- {section}")
|
||
click.echo()
|
||
click.echo("**Key Frontmatter Fields**:")
|
||
for field in info['frontmatter']:
|
||
click.echo(f"- {field}")
|
||
click.echo()
|
||
click.echo("**Key Contentmatter Fields**:")
|
||
for field in info['contentmatter']:
|
||
click.echo(f"- {field}")
|
||
else:
|
||
click.echo("Use --show to display template structure")
|
||
|
||
|
||
@cost_commands.group(name='item')
|
||
def cost_item():
|
||
"""Manage cost items (create, update, list)."""
|
||
pass
|
||
|
||
|
||
@cost_item.command('add')
|
||
@click.argument('name')
|
||
@click.option('--category', required=True, help='Category name')
|
||
@click.option('--amount', type=float, required=True, help='Cost amount in EUR')
|
||
@click.option('--type', 'cost_type',
|
||
type=click.Choice(['monthly', 'one_time']),
|
||
required=True,
|
||
help='Cost type')
|
||
@click.option('--start-date',
|
||
help='Start date (YYYY-MM-DD, defaults to today)')
|
||
@click.option('--description', help='Optional description')
|
||
@click.option('--database', 'db_path', help='Database path (defaults to config)')
|
||
def add_cost_item(name: str, category: str, amount: float, cost_type: str,
|
||
start_date: Optional[str], description: Optional[str],
|
||
db_path: Optional[str]):
|
||
"""Add a new cost item."""
|
||
try:
|
||
# Get database path
|
||
if not db_path:
|
||
config_manager = ConfigurationManager()
|
||
config = config_manager.get_current_config()
|
||
db_path = config.get('database_path')
|
||
|
||
if not db_path:
|
||
click.echo("Error: No database path specified.", err=True)
|
||
sys.exit(1)
|
||
|
||
# Parse start date
|
||
if start_date:
|
||
try:
|
||
start_date_obj = datetime.strptime(start_date, '%Y-%m-%d').date()
|
||
except ValueError:
|
||
click.echo("Error: Start date must be in YYYY-MM-DD format", err=True)
|
||
sys.exit(1)
|
||
else:
|
||
start_date_obj = date.today()
|
||
|
||
# Get cost manager
|
||
cost_manager = CostItemManager(db_path)
|
||
|
||
# Find category by name
|
||
category_obj = cost_manager.get_category_by_name(category)
|
||
if not category_obj:
|
||
click.echo(f"Error: Category '{category}' not found.", err=True)
|
||
click.echo("Available categories:")
|
||
for cat in cost_manager.list_categories():
|
||
click.echo(f" - {cat['name']}")
|
||
sys.exit(1)
|
||
|
||
# Create cost item
|
||
cost_item = CostItem(
|
||
category_id=category_obj['id'],
|
||
name=name,
|
||
description=description,
|
||
cost_type=cost_type,
|
||
amount_eur=Decimal(str(amount)),
|
||
starting_from_date=start_date_obj
|
||
)
|
||
|
||
cost_id = cost_manager.create_cost_item(cost_item)
|
||
click.echo(f"✅ Created cost item '{name}' (ID: {cost_id})")
|
||
|
||
except Exception as e:
|
||
click.echo(f"Error creating cost item: {e}", err=True)
|
||
sys.exit(1)
|
||
|
||
|
||
@cost_item.command('list')
|
||
@click.option('--category', help='Filter by category name')
|
||
@click.option('--type', 'cost_type',
|
||
type=click.Choice(['monthly', 'one_time']),
|
||
help='Filter by cost type')
|
||
@click.option('--active/--all', default=True, help='Show only active items (default)')
|
||
@click.option('--database', 'db_path', help='Database path (defaults to config)')
|
||
def list_cost_items(category: Optional[str], cost_type: Optional[str],
|
||
active: bool, db_path: Optional[str]):
|
||
"""List cost items with optional filtering."""
|
||
try:
|
||
# Get database path
|
||
if not db_path:
|
||
config_manager = ConfigurationManager()
|
||
config = config_manager.get_current_config()
|
||
db_path = config.get('database_path')
|
||
|
||
if not db_path:
|
||
click.echo("Error: No database path specified.", err=True)
|
||
sys.exit(1)
|
||
|
||
cost_manager = CostItemManager(db_path)
|
||
|
||
# Get category ID if specified
|
||
category_id = None
|
||
if category:
|
||
category_obj = cost_manager.get_category_by_name(category)
|
||
if not category_obj:
|
||
click.echo(f"Error: Category '{category}' not found.", err=True)
|
||
sys.exit(1)
|
||
category_id = category_obj['id']
|
||
|
||
# List cost items
|
||
items = cost_manager.list_cost_items(
|
||
active_only=active,
|
||
category_id=category_id,
|
||
cost_type=cost_type
|
||
)
|
||
|
||
if not items:
|
||
click.echo("No cost items found.")
|
||
return
|
||
|
||
# Display items
|
||
click.echo(f"{'ID':<4} {'Name':<25} {'Category':<20} {'Type':<8} {'Amount':<10} {'Status'}")
|
||
click.echo("-" * 80)
|
||
|
||
for item in items:
|
||
status = "Active" if item['is_active'] else "Inactive"
|
||
click.echo(
|
||
f"{item['id']:<4} {item['name'][:24]:<25} "
|
||
f"{(item['category_name'] or 'N/A')[:19]:<20} "
|
||
f"{item['cost_type']:<8} €{float(item['amount_eur']):<9.2f} {status}"
|
||
)
|
||
|
||
click.echo(f"\nTotal: {len(items)} items")
|
||
|
||
except Exception as e:
|
||
click.echo(f"Error listing cost items: {e}", err=True)
|
||
sys.exit(1)
|
||
|
||
|
||
@cost_commands.group(name='category')
|
||
def cost_category():
|
||
"""Manage cost categories."""
|
||
pass
|
||
|
||
|
||
@cost_category.command('list')
|
||
@click.option('--database', 'db_path', help='Database path (defaults to config)')
|
||
def list_categories(db_path: Optional[str]):
|
||
"""List all cost categories."""
|
||
try:
|
||
# Get database path
|
||
if not db_path:
|
||
config_manager = ConfigurationManager()
|
||
config = config_manager.get_current_config()
|
||
db_path = config.get('database_path')
|
||
|
||
if not db_path:
|
||
click.echo("Error: No database path specified.", err=True)
|
||
sys.exit(1)
|
||
|
||
cost_manager = CostItemManager(db_path)
|
||
categories = cost_manager.list_categories()
|
||
|
||
if not categories:
|
||
click.echo("No categories found.")
|
||
return
|
||
|
||
click.echo(f"{'ID':<4} {'Name':<25} {'Description'}")
|
||
click.echo("-" * 70)
|
||
|
||
for category in categories:
|
||
description = category['description'] or ''
|
||
click.echo(f"{category['id']:<4} {category['name'][:24]:<25} {description[:40]}")
|
||
|
||
click.echo(f"\nTotal: {len(categories)} categories")
|
||
|
||
except Exception as e:
|
||
click.echo(f"Error listing categories: {e}", err=True)
|
||
sys.exit(1)
|
||
|
||
|
||
@cost_category.command('add')
|
||
@click.argument('name')
|
||
@click.option('--description', help='Category description')
|
||
@click.option('--database', 'db_path', help='Database path (defaults to config)')
|
||
def add_category(name: str, description: Optional[str], db_path: Optional[str]):
|
||
"""Add a new cost category."""
|
||
try:
|
||
# Get database path
|
||
if not db_path:
|
||
config_manager = ConfigurationManager()
|
||
config = config_manager.get_current_config()
|
||
db_path = config.get('database_path')
|
||
|
||
if not db_path:
|
||
click.echo("Error: No database path specified.", err=True)
|
||
sys.exit(1)
|
||
|
||
cost_manager = CostItemManager(db_path)
|
||
category_id = cost_manager.create_category(name, description)
|
||
|
||
click.echo(f"✅ Created category '{name}' (ID: {category_id})")
|
||
|
||
except Exception as e:
|
||
click.echo(f"Error creating category: {e}", err=True)
|
||
sys.exit(1)
|
||
|
||
|
||
@cost_commands.command('calculate')
|
||
@click.option('--period', help='Period in YYYY-MM format (defaults to current month)')
|
||
@click.option('--database', 'db_path', help='Database path (defaults to config)')
|
||
def calculate_costs(period: Optional[str], db_path: Optional[str]):
|
||
"""Calculate costs for a specific period."""
|
||
try:
|
||
# Get database path
|
||
if not db_path:
|
||
config_manager = ConfigurationManager()
|
||
config = config_manager.get_current_config()
|
||
db_path = config.get('database_path')
|
||
|
||
if not db_path:
|
||
click.echo("Error: No database path specified.", err=True)
|
||
sys.exit(1)
|
||
|
||
# Parse period
|
||
if period:
|
||
try:
|
||
year, month = map(int, period.split('-'))
|
||
if not (1 <= month <= 12):
|
||
raise ValueError("Month must be between 1 and 12")
|
||
except ValueError:
|
||
click.echo("Error: Period must be in YYYY-MM format", err=True)
|
||
sys.exit(1)
|
||
else:
|
||
now = date.today()
|
||
year, month = now.year, now.month
|
||
|
||
# Calculate period dates
|
||
from calendar import monthrange
|
||
period_start = date(year, month, 1)
|
||
_, last_day = monthrange(year, month)
|
||
period_end = date(year, month, last_day)
|
||
|
||
# Calculate costs
|
||
cost_manager = CostItemManager(db_path)
|
||
calculations = cost_manager.calculate_period_costs(period_start, period_end)
|
||
|
||
# Display results
|
||
click.echo(f"Cost Calculation - {period_start.strftime('%B %Y')}")
|
||
click.echo("=" * 50)
|
||
click.echo(f"Period: {period_start} to {period_end}")
|
||
click.echo(f"Monthly Recurring: €{calculations['total_monthly']:.2f}")
|
||
click.echo(f"One-time Expenses: €{calculations['total_one_time']:.2f}")
|
||
click.echo(f"Total Period Cost: €{calculations['total_period']:.2f}")
|
||
click.echo(f"Active Cost Items: {calculations['active_cost_items']}")
|
||
|
||
if calculations['category_breakdown']:
|
||
click.echo("\nCategory Breakdown:")
|
||
for category, breakdown in calculations['category_breakdown'].items():
|
||
if breakdown['total'] > 0:
|
||
click.echo(f" {category}: €{breakdown['total']:.2f}")
|
||
|
||
except Exception as e:
|
||
click.echo(f"Error calculating costs: {e}", err=True)
|
||
sys.exit(1)
|
||
|
||
|
||
@cost_commands.group(name='session')
|
||
def cost_session():
|
||
"""Track Claude session costs for issue implementation."""
|
||
pass
|
||
|
||
|
||
@cost_session.command('track')
|
||
@click.argument('issue_id', type=int)
|
||
@click.argument('issue_title')
|
||
@click.option('--input-tokens', type=int, required=True, help='Number of input tokens')
|
||
@click.option('--output-tokens', type=int, required=True, help='Number of output tokens')
|
||
@click.option('--model', default='claude-sonnet-4', help='Claude model used')
|
||
@click.option('--summary', help='Implementation summary')
|
||
@click.option('--save-note/--no-save-note', default=True, help='Save cost note to file')
|
||
@click.option('--output-dir', default='cost_notes', help='Directory for cost notes')
|
||
@click.option('--database', 'db_path', help='Database path (defaults to config)')
|
||
def track_session(issue_id: int, issue_title: str, input_tokens: int, output_tokens: int,
|
||
model: str, summary: Optional[str], save_note: bool, output_dir: str,
|
||
db_path: Optional[str]):
|
||
"""Track Claude session cost for issue implementation."""
|
||
try:
|
||
# Get database path
|
||
if not db_path:
|
||
config_manager = ConfigurationManager()
|
||
config = config_manager.get_current_config()
|
||
db_path = config.get('database_path')
|
||
|
||
if not db_path:
|
||
click.echo("Error: No database path specified.", err=True)
|
||
sys.exit(1)
|
||
|
||
# Initialize tracker
|
||
tracker = SessionCostTracker(db_path)
|
||
|
||
# Track the session
|
||
result = tracker.track_issue_completion(
|
||
issue_id=issue_id,
|
||
issue_title=issue_title,
|
||
input_tokens=input_tokens,
|
||
output_tokens=output_tokens,
|
||
model=model,
|
||
implementation_summary=summary,
|
||
save_note=save_note,
|
||
output_dir=output_dir
|
||
)
|
||
|
||
if result['tracking_successful']:
|
||
session_cost = result['session_cost']
|
||
|
||
click.echo(f"✅ Issue #{issue_id} cost tracking completed")
|
||
click.echo(f"📊 Session Cost: €{session_cost['total_cost_eur']:.4f} (${session_cost['total_cost_usd']:.4f} USD)")
|
||
click.echo(f"🔤 Token Usage: {session_cost['total_tokens']:,} tokens")
|
||
click.echo(f"🤖 Model: {session_cost['model']}")
|
||
|
||
if result['saved_path']:
|
||
click.echo(f"📝 Cost note saved: {result['saved_path']}")
|
||
|
||
else:
|
||
click.echo("❌ Failed to track session cost", err=True)
|
||
sys.exit(1)
|
||
|
||
except Exception as e:
|
||
click.echo(f"Error tracking session: {e}", err=True)
|
||
sys.exit(1)
|
||
|
||
|
||
@cost_session.command('estimate')
|
||
@click.option('--input-tokens', type=int, required=True, help='Number of input tokens')
|
||
@click.option('--output-tokens', type=int, required=True, help='Number of output tokens')
|
||
@click.option('--model', default='claude-sonnet-4', help='Claude model used')
|
||
def estimate_cost(input_tokens: int, output_tokens: int, model: str):
|
||
"""Estimate Claude session cost without tracking."""
|
||
try:
|
||
# Create temporary tracker for estimation
|
||
tracker = SessionCostTracker("/tmp/dummy.db") # DB not needed for estimation
|
||
|
||
session_cost = tracker.estimate_session_cost(input_tokens, output_tokens, model)
|
||
|
||
click.echo(f"💰 Cost Estimate - {session_cost['model']}")
|
||
click.echo(f"Input: {session_cost['input_tokens']:8,} tokens × ${session_cost['pricing_rates']['input_per_million']:>5.2f}/M = ${session_cost['input_cost_usd']:.4f}")
|
||
click.echo(f"Output: {session_cost['output_tokens']:8,} tokens × ${session_cost['pricing_rates']['output_per_million']:>5.2f}/M = ${session_cost['output_cost_usd']:.4f}")
|
||
click.echo(f"{'='*60}")
|
||
click.echo(f"Total: {session_cost['total_tokens']:8,} tokens = ${session_cost['total_cost_usd']:.4f} USD")
|
||
click.echo(f" = €{session_cost['total_cost_eur']:.4f} EUR")
|
||
|
||
except Exception as e:
|
||
click.echo(f"Error estimating cost: {e}", err=True)
|
||
sys.exit(1)
|
||
|
||
|
||
@cost_session.command('summary')
|
||
@click.option('--issue-ids', help='Comma-separated list of issue IDs to filter by')
|
||
@click.option('--database', 'db_path', help='Database path (defaults to config)')
|
||
def session_summary(issue_ids: Optional[str], db_path: Optional[str]):
|
||
"""Show summary of Claude session costs."""
|
||
try:
|
||
# Get database path
|
||
if not db_path:
|
||
config_manager = ConfigurationManager()
|
||
config = config_manager.get_current_config()
|
||
db_path = config.get('database_path')
|
||
|
||
if not db_path:
|
||
click.echo("Error: No database path specified.", err=True)
|
||
sys.exit(1)
|
||
|
||
# Parse issue IDs if provided
|
||
issue_id_list = None
|
||
if issue_ids:
|
||
try:
|
||
issue_id_list = [int(id.strip()) for id in issue_ids.split(',')]
|
||
except ValueError:
|
||
click.echo("Error: Invalid issue ID format", err=True)
|
||
sys.exit(1)
|
||
|
||
# Get summary
|
||
tracker = SessionCostTracker(db_path)
|
||
summary = tracker.get_issue_costs_summary(issue_id_list)
|
||
|
||
if summary['issue_count'] == 0:
|
||
click.echo("No Claude session costs found.")
|
||
return
|
||
|
||
click.echo(f"🤖 Claude Session Cost Summary")
|
||
click.echo("=" * 40)
|
||
click.echo(f"Issues: {summary['issue_count']}")
|
||
click.echo(f"Total Cost: €{summary['total_costs']:.4f} {summary['currency']}")
|
||
|
||
if summary['items']:
|
||
click.echo(f"\nDetailed Breakdown:")
|
||
click.echo(f"{'Issue':<8} {'Date':<12} {'Amount':<12} {'Description'}")
|
||
click.echo("-" * 60)
|
||
|
||
for item in summary['items']:
|
||
# Extract issue ID from name
|
||
issue_num = "N/A"
|
||
if "Issue #" in item['name']:
|
||
try:
|
||
issue_num = f"#{item['name'].split('Issue #')[1].split()[0]}"
|
||
except:
|
||
pass
|
||
|
||
click.echo(
|
||
f"{issue_num:<8} {item['starting_from_date']:<12} "
|
||
f"€{float(item['amount_eur']):<11.4f} {item['name'][:30]}"
|
||
)
|
||
|
||
except Exception as e:
|
||
click.echo(f"Error getting session summary: {e}", err=True)
|
||
sys.exit(1)
|
||
|
||
|
||
@cost_commands.group(name='period')
|
||
def cost_period():
|
||
"""Manage calculation periods and lifecycle."""
|
||
pass
|
||
|
||
|
||
@cost_period.command('create')
|
||
@click.option('--start-date', required=True, help='Period start date (YYYY-MM-DD)')
|
||
@click.option('--end-date', required=True, help='Period end date (YYYY-MM-DD)')
|
||
@click.option('--type', 'period_type', default='monthly', help='Period type (monthly, quarterly, yearly)')
|
||
@click.option('--loss-forward', type=float, help='Loss carried forward from previous period')
|
||
@click.option('--database', 'db_path', help='Database path (defaults to config)')
|
||
def create_period(start_date: str, end_date: str, period_type: str,
|
||
loss_forward: Optional[float], db_path: Optional[str]):
|
||
"""Create a new calculation period."""
|
||
try:
|
||
# Get database path
|
||
if not db_path:
|
||
config_manager = ConfigurationManager()
|
||
config = config_manager.get_current_config()
|
||
db_path = config.get('database_path')
|
||
|
||
if not db_path:
|
||
click.echo("Error: No database path specified.", err=True)
|
||
sys.exit(1)
|
||
|
||
# Parse dates
|
||
try:
|
||
start_date_obj = datetime.strptime(start_date, '%Y-%m-%d').date()
|
||
end_date_obj = datetime.strptime(end_date, '%Y-%m-%d').date()
|
||
except ValueError:
|
||
click.echo("Error: Dates must be in YYYY-MM-DD format", err=True)
|
||
sys.exit(1)
|
||
|
||
# Initialize period manager
|
||
period_manager = PeriodManager(db_path)
|
||
|
||
# Convert loss forward to Decimal
|
||
loss_forward_decimal = None
|
||
if loss_forward is not None:
|
||
loss_forward_decimal = Decimal(str(loss_forward))
|
||
|
||
# Create period
|
||
period_id = period_manager.create_period(
|
||
period_start=start_date_obj,
|
||
period_end=end_date_obj,
|
||
period_type=period_type,
|
||
loss_carried_forward=loss_forward_decimal
|
||
)
|
||
|
||
if period_id:
|
||
click.echo(f"✅ Created period #{period_id}")
|
||
click.echo(f"📅 Period: {start_date} to {end_date}")
|
||
click.echo(f"📊 Type: {period_type}")
|
||
|
||
if loss_forward:
|
||
click.echo(f"💸 Loss carried forward: €{loss_forward:.4f}")
|
||
else:
|
||
click.echo("❌ Failed to create period", err=True)
|
||
sys.exit(1)
|
||
|
||
except ValueError as e:
|
||
click.echo(f"Error: {e}", err=True)
|
||
sys.exit(1)
|
||
except Exception as e:
|
||
click.echo(f"Error creating period: {e}", err=True)
|
||
sys.exit(1)
|
||
|
||
|
||
@cost_period.command('list')
|
||
@click.option('--status', type=click.Choice(['open', 'calculating', 'closed']),
|
||
help='Filter by status')
|
||
@click.option('--start-from', help='Show periods starting from date (YYYY-MM-DD)')
|
||
@click.option('--end-before', help='Show periods ending before date (YYYY-MM-DD)')
|
||
@click.option('--database', 'db_path', help='Database path (defaults to config)')
|
||
def list_periods(status: Optional[str], start_from: Optional[str],
|
||
end_before: Optional[str], db_path: Optional[str]):
|
||
"""List calculation periods with optional filtering."""
|
||
try:
|
||
# Get database path
|
||
if not db_path:
|
||
config_manager = ConfigurationManager()
|
||
config = config_manager.get_current_config()
|
||
db_path = config.get('database_path')
|
||
|
||
if not db_path:
|
||
click.echo("Error: No database path specified.", err=True)
|
||
sys.exit(1)
|
||
|
||
# Parse date filters
|
||
start_date_obj = None
|
||
end_date_obj = None
|
||
|
||
if start_from:
|
||
try:
|
||
start_date_obj = datetime.strptime(start_from, '%Y-%m-%d').date()
|
||
except ValueError:
|
||
click.echo("Error: Start date must be in YYYY-MM-DD format", err=True)
|
||
sys.exit(1)
|
||
|
||
if end_before:
|
||
try:
|
||
end_date_obj = datetime.strptime(end_before, '%Y-%m-%d').date()
|
||
except ValueError:
|
||
click.echo("Error: End date must be in YYYY-MM-DD format", err=True)
|
||
sys.exit(1)
|
||
|
||
# Get periods
|
||
period_manager = PeriodManager(db_path)
|
||
periods = period_manager.list_periods(
|
||
status_filter=status,
|
||
start_date=start_date_obj,
|
||
end_date=end_date_obj
|
||
)
|
||
|
||
if not periods:
|
||
click.echo("No periods found matching criteria.")
|
||
return
|
||
|
||
# Display periods
|
||
click.echo(f"📅 Calculation Periods")
|
||
click.echo("=" * 80)
|
||
click.echo(f"{'ID':<4} {'Start Date':<12} {'End Date':<12} {'Type':<10} {'Status':<12} {'Total Cost':<12}")
|
||
click.echo("-" * 80)
|
||
|
||
for period in periods:
|
||
click.echo(
|
||
f"{period['id']:<4} {period['period_start']:<12} {period['period_end']:<12} "
|
||
f"{period['period_type']:<10} {period['status']:<12} €{float(period['total_costs']):<11.2f}"
|
||
)
|
||
|
||
click.echo(f"\nTotal: {len(periods)} periods")
|
||
|
||
except Exception as e:
|
||
click.echo(f"Error listing periods: {e}", err=True)
|
||
sys.exit(1)
|
||
|
||
|
||
@cost_period.command('show')
|
||
@click.argument('period_id', type=int)
|
||
@click.option('--database', 'db_path', help='Database path (defaults to config)')
|
||
def show_period(period_id: int, db_path: Optional[str]):
|
||
"""Show detailed information for a specific period."""
|
||
try:
|
||
# Get database path
|
||
if not db_path:
|
||
config_manager = ConfigurationManager()
|
||
config = config_manager.get_current_config()
|
||
db_path = config.get('database_path')
|
||
|
||
if not db_path:
|
||
click.echo("Error: No database path specified.", err=True)
|
||
sys.exit(1)
|
||
|
||
# Get period details
|
||
period_manager = PeriodManager(db_path)
|
||
period = period_manager.get_period_by_id(period_id)
|
||
|
||
if not period:
|
||
click.echo(f"Period #{period_id} not found.", err=True)
|
||
sys.exit(1)
|
||
|
||
# Display period details
|
||
click.echo(f"📅 Period #{period['id']} Details")
|
||
click.echo("=" * 40)
|
||
click.echo(f"Start Date: {period['period_start']}")
|
||
click.echo(f"End Date: {period['period_end']}")
|
||
click.echo(f"Type: {period['period_type']}")
|
||
click.echo(f"Status: {period['status']}")
|
||
click.echo(f"Total Costs: €{float(period['total_costs']):.4f}")
|
||
click.echo(f"Active Issues: {period['active_issues_count']}")
|
||
|
||
if period['cost_per_issue'] > 0:
|
||
click.echo(f"Cost per Issue: €{float(period['cost_per_issue']):.4f}")
|
||
|
||
if period['loss_carried_forward'] > 0:
|
||
click.echo(f"Loss Carried Forward: €{float(period['loss_carried_forward']):.4f}")
|
||
|
||
click.echo(f"Created: {period['created_at']}")
|
||
|
||
if period['updated_at']:
|
||
click.echo(f"Updated: {period['updated_at']}")
|
||
|
||
except Exception as e:
|
||
click.echo(f"Error showing period: {e}", err=True)
|
||
sys.exit(1)
|
||
|
||
|
||
@cost_period.command('calculate')
|
||
@click.argument('period_id', type=int)
|
||
@click.option('--database', 'db_path', help='Database path (defaults to config)')
|
||
def calculate_period(period_id: int, db_path: Optional[str]):
|
||
"""Calculate costs for a specific period."""
|
||
try:
|
||
# Get database path
|
||
if not db_path:
|
||
config_manager = ConfigurationManager()
|
||
config = config_manager.get_current_config()
|
||
db_path = config.get('database_path')
|
||
|
||
if not db_path:
|
||
click.echo("Error: No database path specified.", err=True)
|
||
sys.exit(1)
|
||
|
||
# Calculate period costs
|
||
period_manager = PeriodManager(db_path)
|
||
calculation = period_manager.calculate_period_costs(period_id)
|
||
|
||
click.echo(f"📊 Period #{calculation['period_id']} Cost Calculation")
|
||
click.echo("=" * 50)
|
||
click.echo(f"Period: {calculation['period_start']} to {calculation['period_end']}")
|
||
click.echo(f"Monthly Recurring: €{calculation['monthly_costs']:.2f}")
|
||
click.echo(f"One-time Expenses: €{calculation['one_time_costs']:.2f}")
|
||
|
||
if calculation['loss_carried_forward'] > 0:
|
||
click.echo(f"Loss Carried Forward: €{calculation['loss_carried_forward']:.2f}")
|
||
|
||
click.echo(f"Total Period Cost: €{calculation['total_costs']:.2f}")
|
||
click.echo(f"Calculated: {calculation['calculation_date'][:19]}")
|
||
|
||
except Exception as e:
|
||
click.echo(f"Error calculating period: {e}", err=True)
|
||
sys.exit(1)
|
||
|
||
|
||
@cost_period.command('status')
|
||
@click.argument('period_id', type=int)
|
||
@click.argument('new_status', type=click.Choice(['open', 'calculating', 'closed']))
|
||
@click.option('--database', 'db_path', help='Database path (defaults to config)')
|
||
def update_status(period_id: int, new_status: str, db_path: Optional[str]):
|
||
"""Update period status."""
|
||
try:
|
||
# Get database path
|
||
if not db_path:
|
||
config_manager = ConfigurationManager()
|
||
config = config_manager.get_current_config()
|
||
db_path = config.get('database_path')
|
||
|
||
if not db_path:
|
||
click.echo("Error: No database path specified.", err=True)
|
||
sys.exit(1)
|
||
|
||
# Update status
|
||
period_manager = PeriodManager(db_path)
|
||
success = period_manager.update_period_status(period_id, new_status)
|
||
|
||
if success:
|
||
click.echo(f"✅ Period #{period_id} status updated to '{new_status}'")
|
||
else:
|
||
click.echo(f"❌ Failed to update period #{period_id} status", err=True)
|
||
sys.exit(1)
|
||
|
||
except ValueError as e:
|
||
click.echo(f"Error: {e}", err=True)
|
||
sys.exit(1)
|
||
except Exception as e:
|
||
click.echo(f"Error updating status: {e}", err=True)
|
||
sys.exit(1)
|
||
|
||
|
||
@cost_period.command('close')
|
||
@click.argument('period_id', type=int)
|
||
@click.option('--skip-validation', is_flag=True, help='Skip transaction validation')
|
||
@click.option('--database', 'db_path', help='Database path (defaults to config)')
|
||
def close_period(period_id: int, skip_validation: bool, db_path: Optional[str]):
|
||
"""Close a period after final calculations."""
|
||
try:
|
||
# Get database path
|
||
if not db_path:
|
||
config_manager = ConfigurationManager()
|
||
config = config_manager.get_current_config()
|
||
db_path = config.get('database_path')
|
||
|
||
if not db_path:
|
||
click.echo("Error: No database path specified.", err=True)
|
||
sys.exit(1)
|
||
|
||
# Close period
|
||
period_manager = PeriodManager(db_path)
|
||
success = period_manager.close_period(
|
||
period_id,
|
||
validate_transactions=not skip_validation
|
||
)
|
||
|
||
if success:
|
||
click.echo(f"✅ Period #{period_id} has been closed")
|
||
|
||
# Show final calculation
|
||
period = period_manager.get_period_by_id(period_id)
|
||
if period:
|
||
click.echo(f"💰 Final total cost: €{float(period['total_costs']):.4f}")
|
||
else:
|
||
click.echo(f"❌ Failed to close period #{period_id}", err=True)
|
||
sys.exit(1)
|
||
|
||
except ValueError as e:
|
||
click.echo(f"Error: {e}", err=True)
|
||
sys.exit(1)
|
||
except Exception as e:
|
||
click.echo(f"Error closing period: {e}", err=True)
|
||
sys.exit(1)
|
||
|
||
|
||
@cost_period.command('current')
|
||
@click.option('--date', 'date_str', help='Date to find period for (YYYY-MM-DD, defaults to today)')
|
||
@click.option('--database', 'db_path', help='Database path (defaults to config)')
|
||
def current_period(date_str: Optional[str], db_path: Optional[str]):
|
||
"""Show current active period for a given date."""
|
||
try:
|
||
# Get database path
|
||
if not db_path:
|
||
config_manager = ConfigurationManager()
|
||
config = config_manager.get_current_config()
|
||
db_path = config.get('database_path')
|
||
|
||
if not db_path:
|
||
click.echo("Error: No database path specified.", err=True)
|
||
sys.exit(1)
|
||
|
||
# Parse date if provided
|
||
target_date = None
|
||
if date_str:
|
||
try:
|
||
target_date = datetime.strptime(date_str, '%Y-%m-%d').date()
|
||
except ValueError:
|
||
click.echo("Error: Date must be in YYYY-MM-DD format", err=True)
|
||
sys.exit(1)
|
||
|
||
# Get current period
|
||
period_manager = PeriodManager(db_path)
|
||
current = period_manager.get_current_period(target_date)
|
||
|
||
if not current:
|
||
date_display = date_str if date_str else "today"
|
||
click.echo(f"No active period found for {date_display}")
|
||
|
||
# Suggest creating a period
|
||
if not date_str:
|
||
today = date.today()
|
||
# Calculate last day of current month
|
||
if today.month == 12:
|
||
next_month = today.replace(year=today.year + 1, month=1, day=1)
|
||
else:
|
||
next_month = today.replace(month=today.month + 1, day=1)
|
||
last_day = (next_month - timedelta(days=1)).day
|
||
click.echo(f"💡 Create one with: markitect cost period create --start-date {today.year}-{today.month:02d}-01 --end-date {today.year}-{today.month:02d}-{last_day:02d}")
|
||
return
|
||
|
||
# Display current period
|
||
click.echo(f"📅 Current Active Period")
|
||
click.echo("=" * 30)
|
||
click.echo(f"Period #{current['id']}")
|
||
click.echo(f"Dates: {current['period_start']} to {current['period_end']}")
|
||
click.echo(f"Status: {current['status']}")
|
||
click.echo(f"Total Costs: €{float(current['total_costs']):.4f}")
|
||
|
||
except Exception as e:
|
||
click.echo(f"Error getting current period: {e}", err=True)
|
||
sys.exit(1) |