tegwick
b23ff30e97
- Add `markitect cost session note` command for general work sessions - Support work that is not tied to specific tracked issues - Generate structured cost notes with comprehensive metadata - Include token usage breakdown and cost allocation guidance - Create cost note for agent ecosystem consolidation work (€0.2760) Enhancement allows tracking of general development work like agent optimization, infrastructure improvements, and other non-issue tasks while maintaining proper cost documentation and allocation. 🤖 Generated with [Claude Code](https://claude.ai/code) Co-Authored-By: Claude <noreply@anthropic.com>
1213 lines
46 KiB
Python
1213 lines
46 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('note')
|
||
@click.argument('session_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('--session-type', default='general_work', help='Type of work session')
|
||
@click.option('--summary', help='Work session summary')
|
||
@click.option('--output-dir', default='cost_notes', help='Directory for cost notes')
|
||
def create_session_note(session_title: str, input_tokens: int, output_tokens: int,
|
||
model: str, session_type: str, summary: Optional[str],
|
||
output_dir: str):
|
||
"""Create a cost note for general work sessions without specific issue tracking."""
|
||
try:
|
||
# Create temporary tracker for cost calculation
|
||
tracker = SessionCostTracker("/tmp/dummy.db") # DB not needed for estimation
|
||
|
||
# Calculate session cost
|
||
session_cost = tracker.estimate_session_cost(input_tokens, output_tokens, model)
|
||
|
||
# Create output directory if it doesn't exist
|
||
output_dir_path = Path(output_dir)
|
||
output_dir_path.mkdir(exist_ok=True)
|
||
|
||
# Generate filename
|
||
today = date.today()
|
||
safe_title = "".join(c for c in session_title.lower().replace(' ', '_') if c.isalnum() or c in '_-')
|
||
filename = f"{safe_title}_cost_{today.strftime('%Y-%m-%d')}.md"
|
||
output_path = output_dir_path / filename
|
||
|
||
# Generate cost note content
|
||
cost_note_content = f"""---
|
||
note_type: "session_cost_tracking"
|
||
session_type: "{session_type}"
|
||
session_title: "{session_title}"
|
||
session_date: "{today.strftime('%Y-%m-%d')}"
|
||
claude_model: "{session_cost['model']}"
|
||
total_cost_eur: {session_cost['total_cost_eur']:.4f}
|
||
total_cost_usd: {session_cost['total_cost_usd']:.4f}
|
||
total_tokens: {session_cost['total_tokens']}
|
||
input_tokens: {session_cost['input_tokens']}
|
||
output_tokens: {session_cost['output_tokens']}
|
||
generated_at: "{datetime.now().strftime('%Y-%m-%dT%H:%M:%S')}"
|
||
---
|
||
|
||
# Session Cost Analysis
|
||
**Session**: {session_title}
|
||
**Date**: {today.strftime('%Y-%m-%d')}
|
||
**Claude Model**: {session_cost['model']}
|
||
|
||
## Cost Summary
|
||
- **Total Cost**: €{session_cost['total_cost_eur']:.4f} (${session_cost['total_cost_usd']:.4f} USD)
|
||
- **Token Usage**: {session_cost['total_tokens']:,} tokens
|
||
- Input: {session_cost['input_tokens']:,} tokens × ${session_cost['pricing_rates']['input_per_million']:.2f}/M = ${session_cost['input_cost_usd']:.4f}
|
||
- Output: {session_cost['output_tokens']:,} tokens × ${session_cost['pricing_rates']['output_per_million']:.2f}/M = ${session_cost['output_cost_usd']:.4f}
|
||
|
||
## Session Details
|
||
- **Session Type**: {session_type}"""
|
||
|
||
if summary:
|
||
cost_note_content += f"""
|
||
- **Work Summary**: {summary}"""
|
||
|
||
cost_note_content += f"""
|
||
|
||
## Cost Allocation
|
||
This cost represents general development work and should be allocated to the appropriate cost category based on the nature of the session.
|
||
|
||
**Note**: This is a standalone cost note for work that is not tied to a specific tracked issue. For issue-specific work, use `markitect cost session track` instead.
|
||
|
||
---
|
||
*Generated automatically by MarkiTect Cost Tracking - {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}*
|
||
"""
|
||
|
||
# Save the cost note
|
||
with open(output_path, 'w', encoding='utf-8') as f:
|
||
f.write(cost_note_content)
|
||
|
||
# Display results
|
||
click.echo(f"✅ Session cost note created")
|
||
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']}")
|
||
click.echo(f"📝 Cost note saved: {output_path}")
|
||
|
||
except Exception as e:
|
||
click.echo(f"Error creating session note: {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)
|
||
|
||
|
||
@cost_commands.group(name='allocate')
|
||
def cost_allocate():
|
||
"""Allocate costs to active issues for specified periods."""
|
||
pass
|
||
|
||
|
||
@cost_allocate.command('period')
|
||
@click.argument('period_id', type=int)
|
||
@click.option('--database', 'db_path', help='Database path (defaults to config)')
|
||
@click.option('--dry-run', is_flag=True, help='Show what would be allocated without making changes')
|
||
def allocate_period(period_id: int, db_path: Optional[str], dry_run: bool):
|
||
"""Allocate costs for a specific period to active issues."""
|
||
try:
|
||
# Import allocation engine
|
||
from .allocation_engine import AllocationEngine, AllocationStatus
|
||
|
||
# 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 allocation engine
|
||
engine = AllocationEngine(db_path)
|
||
|
||
if dry_run:
|
||
# TODO: Implement dry-run functionality
|
||
click.echo(f"🔍 Dry run for period {period_id} allocation")
|
||
click.echo("(Dry-run functionality will be implemented in future version)")
|
||
return
|
||
|
||
# Perform allocation
|
||
result = engine.allocate_period_costs(period_id)
|
||
|
||
# Display results based on status
|
||
if result.status == AllocationStatus.SUCCESS:
|
||
click.echo(f"✅ Cost Allocation Complete - Period {period_id}")
|
||
click.echo("=" * 50)
|
||
click.echo(f"Total Costs Allocated: €{result.total_costs:.2f}")
|
||
click.echo(f"Active Issues: {len(result.active_issues)}")
|
||
click.echo(f"Cost Per Issue: €{result.cost_per_issue:.2f}")
|
||
click.echo(f"Allocations Created: {result.allocations_created}")
|
||
click.echo(f"Transactions Created: {result.transactions_created}")
|
||
|
||
if result.active_issues:
|
||
click.echo(f"\nIssues that received allocations:")
|
||
for issue_id in result.active_issues:
|
||
click.echo(f" Issue #{issue_id}: €{result.cost_per_issue:.2f}")
|
||
|
||
elif result.status == AllocationStatus.NO_ACTIVE_ISSUES:
|
||
click.echo(f"⚠️ No Active Issues Found - Period {period_id}")
|
||
click.echo("=" * 45)
|
||
click.echo(f"Total Costs: €{result.total_costs:.2f}")
|
||
click.echo(f"Loss Carried Forward: €{result.loss_carried_forward:.2f}")
|
||
click.echo("All costs have been carried forward to the next period.")
|
||
|
||
elif result.status == AllocationStatus.NO_COSTS_TO_ALLOCATE:
|
||
click.echo(f"ℹ️ No Costs to Allocate - Period {period_id}")
|
||
click.echo("=" * 40)
|
||
click.echo("Period has no costs to allocate.")
|
||
|
||
elif result.status == AllocationStatus.PERIOD_CLOSED:
|
||
click.echo(f"⚠️ Period Already Closed - Period {period_id}")
|
||
click.echo("=" * 40)
|
||
click.echo("This period has already been processed and closed.")
|
||
|
||
else:
|
||
click.echo(f"❌ Allocation Failed - Period {period_id}")
|
||
click.echo("=" * 35)
|
||
click.echo(f"Error: {result.message}")
|
||
sys.exit(1)
|
||
|
||
except Exception as e:
|
||
click.echo(f"Error performing allocation: {e}", err=True)
|
||
sys.exit(1)
|
||
|
||
|
||
@cost_allocate.command('show')
|
||
@click.argument('target', type=str)
|
||
@click.option('--format', 'output_format',
|
||
type=click.Choice(['table', 'json']),
|
||
default='table', help='Output format')
|
||
@click.option('--database', 'db_path', help='Database path (defaults to config)')
|
||
def show_allocations(target: str, output_format: str, db_path: Optional[str]):
|
||
"""Show allocations for an issue (issue:ID) or period (period:ID)."""
|
||
try:
|
||
# Import allocation engine
|
||
from .allocation_engine import AllocationEngine
|
||
|
||
# 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 target (issue:123 or period:456)
|
||
if ':' not in target:
|
||
click.echo("Error: Target must be in format 'issue:ID' or 'period:ID'", err=True)
|
||
sys.exit(1)
|
||
|
||
target_type, target_id_str = target.split(':', 1)
|
||
try:
|
||
target_id = int(target_id_str)
|
||
except ValueError:
|
||
click.echo("Error: Target ID must be a number", err=True)
|
||
sys.exit(1)
|
||
|
||
# Initialize allocation engine
|
||
engine = AllocationEngine(db_path)
|
||
|
||
# Get allocations based on target type
|
||
if target_type == 'issue':
|
||
allocations = engine.get_issue_allocations(target_id)
|
||
title = f"Cost Allocations for Issue #{target_id}"
|
||
elif target_type == 'period':
|
||
allocations = engine.get_period_allocations(target_id)
|
||
title = f"Cost Allocations for Period {target_id}"
|
||
else:
|
||
click.echo("Error: Target type must be 'issue' or 'period'", err=True)
|
||
sys.exit(1)
|
||
|
||
if not allocations:
|
||
click.echo(f"📝 No allocations found for {target}")
|
||
return
|
||
|
||
# Display results
|
||
if output_format == 'json':
|
||
import json
|
||
click.echo(json.dumps(allocations, indent=2, default=str))
|
||
else:
|
||
# Table format
|
||
from tabulate import tabulate
|
||
click.echo(f"\n💰 {title}\n")
|
||
|
||
if target_type == 'issue':
|
||
headers = ['ID', 'Period', 'Amount', 'Date', 'Period Range', 'Transaction']
|
||
rows = []
|
||
for alloc in allocations:
|
||
rows.append([
|
||
alloc['id'],
|
||
alloc['period_id'],
|
||
f"€{alloc['allocated_amount']:.2f}",
|
||
alloc['allocation_date'],
|
||
f"{alloc['period_start']} to {alloc['period_end']}",
|
||
alloc['transaction_id'] or 'N/A'
|
||
])
|
||
else:
|
||
headers = ['ID', 'Issue', 'Amount', 'Date', 'Transaction']
|
||
rows = []
|
||
for alloc in allocations:
|
||
rows.append([
|
||
alloc['id'],
|
||
f"#{alloc['issue_id']}",
|
||
f"€{alloc['allocated_amount']:.2f}",
|
||
alloc['allocation_date'],
|
||
alloc['transaction_id'] or 'N/A'
|
||
])
|
||
|
||
click.echo(tabulate(rows, headers=headers, tablefmt='grid'))
|
||
click.echo(f"\n📊 Total: {len(allocations)} allocations")
|
||
|
||
except Exception as e:
|
||
click.echo(f"Error showing allocations: {e}", err=True)
|
||
sys.exit(1)
|
||
|
||
|
||
@cost_allocate.command('reverse')
|
||
@click.argument('allocation_id', type=int)
|
||
@click.option('--database', 'db_path', help='Database path (defaults to config)')
|
||
@click.confirmation_option(prompt='Are you sure you want to reverse this allocation?')
|
||
def reverse_allocation(allocation_id: int, db_path: Optional[str]):
|
||
"""Reverse a cost allocation (for corrections)."""
|
||
try:
|
||
# Import allocation engine
|
||
from .allocation_engine import AllocationEngine
|
||
|
||
# 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 allocation engine
|
||
engine = AllocationEngine(db_path)
|
||
|
||
# Perform reversal
|
||
success = engine.reverse_allocation(allocation_id)
|
||
|
||
if success:
|
||
click.echo(f"✅ Successfully reversed allocation #{allocation_id}")
|
||
click.echo("A reversal transaction has been created in the audit trail.")
|
||
else:
|
||
click.echo(f"❌ Failed to reverse allocation #{allocation_id}")
|
||
click.echo("Allocation may not exist or may already be reversed.")
|
||
sys.exit(1)
|
||
|
||
except Exception as e:
|
||
click.echo(f"Error reversing allocation: {e}", err=True)
|
||
sys.exit(1) |