coulomb/markitect-main
2
0
Fork 0
Code Issues 61 Pull Requests Actions Projects 12 Releases 1 Wiki Activity

feat: implement cost report template generator with Claude session tracking (issue #119)
Some checks failed
Test Suite / unit-tests (3.11) (push) Has been cancelled
Details
Test Suite / unit-tests (3.12) (push) Has been cancelled
Details
Test Suite / integration-tests (push) Has been cancelled
Details
Test Suite / e2e-tests (push) Has been cancelled
Details
Test Suite / performance-tests (push) Has been cancelled
Details
Test Suite / code-quality (push) Has been cancelled
Details
Test Suite / security-scan (push) Has been cancelled
Details
Test Suite / test-summary (push) Has been cancelled
Details

Browse Source 
Comprehensive cost tracking system implementation including:

- Cost report generator with multiple formats (summary, detailed, audit)
- Full CLI integration with cost management commands
- Claude session cost tracking and estimation
- Professional markdown reports with frontmatter/contentmatter
- Automatic cost note generation for issue implementations
- Complete test coverage (33 test cases)
- Database integration with finance schema initialization

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
Bernd Worsch
2025-10-04 01:31:36 +02:00
parent 59814d84d8
commit dab6b9fdef
14 changed files with 4179 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

553
markitect/finance/cost_manager.py Normal file
View File

@@ -0,0 +1,553 @@
"""
Cost Item Management System for MarkiTect.
This module provides comprehensive cost item lifecycle management including:
- Cost item creation, modification, and lifecycle management
- Category management and validation
- Business rule enforcement and validation
- Integration with period management for cost calculations
The system supports both monthly recurring costs and one-time expenses
with proper validation and audit trails.
"""
import sqlite3
from datetime import date, datetime
from decimal import Decimal
from typing import Optional, List, Dict, Any, Union
from dataclasses import dataclass
from .models import FinanceModels
@dataclass
class CostItem:
"""Data class representing a cost item."""
id: Optional[int] = None
category_id: int = None
name: str = ""
description: Optional[str] = None
cost_type: str = "" # 'monthly' or 'one_time'
amount_eur: Decimal = Decimal('0.00')
currency: str = "EUR"
starting_from_date: date = None
ending_date: Optional[date] = None
is_active: bool = True
created_at: Optional[datetime] = None
updated_at: Optional[datetime] = None
@dataclass
class CostCategory:
"""Data class representing a cost category."""
id: Optional[int] = None
name: str = ""
description: Optional[str] = None
created_at: Optional[datetime] = None
updated_at: Optional[datetime] = None
class CostItemManager:
"""Manager for cost item operations and business logic."""
def __init__(self, db_path: str):
"""
Initialize cost item manager.
Args:
db_path: Path to SQLite database file
"""
self.db_path = db_path
self.finance_models = FinanceModels(db_path)
def create_cost_item(self, cost_item: CostItem) -> Optional[int]:
"""
Create a new cost item with validation.
Args:
cost_item: CostItem instance to create
Returns:
ID of created cost item, or None if creation failed
Raises:
ValueError: If validation fails
sqlite3.Error: If database operation fails
"""
# Validate cost item
self._validate_cost_item(cost_item)
conn = self.finance_models.get_connection()
cursor = conn.cursor()
try:
cursor.execute('''
INSERT INTO cost_items
(category_id, name, description, cost_type, amount_eur, currency,
starting_from_date, ending_date, is_active, created_at, updated_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
''', (
cost_item.category_id,
cost_item.name,
cost_item.description,
cost_item.cost_type,
float(cost_item.amount_eur),
cost_item.currency,
cost_item.starting_from_date.isoformat() if cost_item.starting_from_date else None,
cost_item.ending_date.isoformat() if cost_item.ending_date else None,
cost_item.is_active,
datetime.now().isoformat(),
datetime.now().isoformat()
))
cost_item_id = cursor.lastrowid
conn.commit()
return cost_item_id
except sqlite3.Error as e:
conn.rollback()
raise
finally:
conn.close()
def update_cost_item(self, cost_item_id: int, updates: Dict[str, Any]) -> bool:
"""
Update an existing cost item.
Args:
cost_item_id: ID of cost item to update
updates: Dictionary of field updates
Returns:
True if update was successful, False otherwise
Raises:
ValueError: If validation fails
sqlite3.Error: If database operation fails
"""
# Get existing cost item for validation
existing_item = self.get_cost_item(cost_item_id)
if not existing_item:
raise ValueError(f"Cost item with ID {cost_item_id} not found")
# Apply updates to existing item for validation
# Filter out extra fields that aren't part of CostItem dataclass
cost_item_fields = {
'id', 'category_id', 'name', 'description', 'cost_type',
'amount_eur', 'currency', 'starting_from_date', 'ending_date',
'is_active', 'created_at', 'updated_at'
}
filtered_item = {k: v for k, v in existing_item.items() if k in cost_item_fields}
# Convert date strings back to date objects for validation
if filtered_item.get('starting_from_date'):
filtered_item['starting_from_date'] = date.fromisoformat(filtered_item['starting_from_date'])
if filtered_item.get('ending_date'):
filtered_item['ending_date'] = date.fromisoformat(filtered_item['ending_date'])
if filtered_item.get('amount_eur'):
filtered_item['amount_eur'] = Decimal(str(filtered_item['amount_eur']))
updated_item = CostItem(**filtered_item)
for field, value in updates.items():
if hasattr(updated_item, field):
setattr(updated_item, field, value)
else:
raise ValueError(f"Invalid field: {field}")
# Validate updated item
self._validate_cost_item(updated_item)
# Build dynamic update query
set_clauses = []
values = []
for field, value in updates.items():
if field in ['amount_eur'] and isinstance(value, Decimal):
value = float(value)
elif field in ['starting_from_date', 'ending_date'] and isinstance(value, date):
value = value.isoformat() if value else None
set_clauses.append(f"{field} = ?")
values.append(value)
# Add updated_at timestamp
set_clauses.append("updated_at = ?")
values.append(datetime.now().isoformat())
values.append(cost_item_id)
conn = self.finance_models.get_connection()
cursor = conn.cursor()
try:
cursor.execute(f'''
UPDATE cost_items
SET {', '.join(set_clauses)}
WHERE id = ?
''', values)
success = cursor.rowcount > 0
conn.commit()
return success
except sqlite3.Error as e:
conn.rollback()
raise
finally:
conn.close()
def deactivate_cost_item(self, cost_item_id: int, ending_date: Optional[date] = None) -> bool:
"""
Deactivate a cost item (soft delete).
Args:
cost_item_id: ID of cost item to deactivate
ending_date: Optional ending date (defaults to today)
Returns:
True if deactivation was successful, False otherwise
"""
if ending_date is None:
ending_date = date.today()
updates = {
'is_active': False,
'ending_date': ending_date
}
return self.update_cost_item(cost_item_id, updates)
def get_cost_item(self, cost_item_id: int) -> Optional[Dict[str, Any]]:
"""
Retrieve a cost item by ID.
Args:
cost_item_id: ID of cost item to retrieve
Returns:
Dictionary containing cost item data, or None if not found
"""
conn = self.finance_models.get_connection()
cursor = conn.cursor()
try:
cursor.execute('''
SELECT ci.*, cc.name as category_name
FROM cost_items ci
LEFT JOIN cost_categories cc ON ci.category_id = cc.id
WHERE ci.id = ?
''', (cost_item_id,))
row = cursor.fetchone()
if not row:
return None
# Convert row to dictionary
columns = [desc[0] for desc in cursor.description]
return dict(zip(columns, row))
except sqlite3.Error as e:
return None
finally:
conn.close()
def list_cost_items(self,
active_only: bool = True,
category_id: Optional[int] = None,
cost_type: Optional[str] = None) -> List[Dict[str, Any]]:
"""
List cost items with optional filtering.
Args:
active_only: If True, only return active cost items
category_id: Optional category filter
cost_type: Optional cost type filter ('monthly' or 'one_time')
Returns:
List of cost item dictionaries
"""
conn = self.finance_models.get_connection()
cursor = conn.cursor()
# Build WHERE clause
conditions = []
values = []
if active_only:
conditions.append("ci.is_active = ?")
values.append(True)
if category_id is not None:
conditions.append("ci.category_id = ?")
values.append(category_id)
if cost_type is not None:
conditions.append("ci.cost_type = ?")
values.append(cost_type)
where_clause = "WHERE " + " AND ".join(conditions) if conditions else ""
try:
cursor.execute(f'''
SELECT ci.*, cc.name as category_name
FROM cost_items ci
LEFT JOIN cost_categories cc ON ci.category_id = cc.id
{where_clause}
ORDER BY ci.category_id, ci.name
''', values)
rows = cursor.fetchall()
columns = [desc[0] for desc in cursor.description]
return [dict(zip(columns, row)) for row in rows]
except sqlite3.Error as e:
return []
finally:
conn.close()
def get_active_costs_for_period(self, period_start: date, period_end: date) -> List[Dict[str, Any]]:
"""
Get all active cost items for a specific period.
Args:
period_start: Start date of the period
period_end: End date of the period
Returns:
List of active cost items for the period
"""
conn = self.finance_models.get_connection()
cursor = conn.cursor()
try:
cursor.execute('''
SELECT ci.*, cc.name as category_name
FROM cost_items ci
LEFT JOIN cost_categories cc ON ci.category_id = cc.id
WHERE ci.is_active = TRUE
AND ci.starting_from_date <= ?
AND (ci.ending_date IS NULL OR ci.ending_date >= ?)
ORDER BY ci.category_id, ci.name
''', (period_end.isoformat(), period_start.isoformat()))
rows = cursor.fetchall()
columns = [desc[0] for desc in cursor.description]
return [dict(zip(columns, row)) for row in rows]
except sqlite3.Error as e:
return []
finally:
conn.close()
def calculate_period_costs(self, period_start: date, period_end: date) -> Dict[str, Any]:
"""
Calculate total costs for a specific period.
Args:
period_start: Start date of the period
period_end: End date of the period
Returns:
Dictionary with cost calculations
"""
active_costs = self.get_active_costs_for_period(period_start, period_end)
total_monthly = Decimal('0.00')
total_one_time = Decimal('0.00')
category_totals = {}
for cost_item in active_costs:
amount = Decimal(str(cost_item['amount_eur']))
cost_type = cost_item['cost_type']
category = cost_item['category_name']
if cost_type == 'monthly':
total_monthly += amount
elif cost_type == 'one_time':
total_one_time += amount
# Track by category
if category not in category_totals:
category_totals[category] = {'monthly': Decimal('0.00'), 'one_time': Decimal('0.00')}
category_totals[category][cost_type] += amount
return {
'period_start': period_start,
'period_end': period_end,
'total_monthly': float(total_monthly),
'total_one_time': float(total_one_time),
'total_period': float(total_monthly + total_one_time),
'category_breakdown': {
cat: {
'monthly': float(amounts['monthly']),
'one_time': float(amounts['one_time']),
'total': float(amounts['monthly'] + amounts['one_time'])
}
for cat, amounts in category_totals.items()
},
'active_cost_items': len(active_costs)
}
def _validate_cost_item(self, cost_item: CostItem) -> None:
"""
Validate cost item data.
Args:
cost_item: CostItem to validate
Raises:
ValueError: If validation fails
"""
if not cost_item.name or not cost_item.name.strip():
raise ValueError("Cost item name is required")
if cost_item.cost_type not in ['monthly', 'one_time']:
raise ValueError("Cost type must be 'monthly' or 'one_time'")
if cost_item.amount_eur is None or cost_item.amount_eur < 0:
raise ValueError("Amount must be non-negative")
if not cost_item.starting_from_date:
raise ValueError("Starting date is required")
if cost_item.ending_date and cost_item.ending_date < cost_item.starting_from_date:
raise ValueError("Ending date must be after starting date")
if not cost_item.is_active and cost_item.ending_date is None:
raise ValueError("Inactive cost items must have an ending date")
# Validate category exists
if cost_item.category_id:
category = self.get_category(cost_item.category_id)
if not category:
raise ValueError(f"Category with ID {cost_item.category_id} does not exist")
# Category management methods
def create_category(self, name: str, description: Optional[str] = None) -> Optional[int]:
"""
Create a new cost category.
Args:
name: Category name
description: Optional category description
Returns:
ID of created category, or None if creation failed
"""
if not name or not name.strip():
raise ValueError("Category name is required")
conn = self.finance_models.get_connection()
cursor = conn.cursor()
try:
cursor.execute('''
INSERT INTO cost_categories (name, description, created_at, updated_at)
VALUES (?, ?, ?, ?)
''', (name.strip(), description, datetime.now().isoformat(), datetime.now().isoformat()))
category_id = cursor.lastrowid
conn.commit()
return category_id
except sqlite3.IntegrityError:
conn.rollback()
raise ValueError(f"Category '{name}' already exists")
except sqlite3.Error as e:
conn.rollback()
raise
finally:
conn.close()
def get_category(self, category_id: int) -> Optional[Dict[str, Any]]:
"""
Retrieve a category by ID.
Args:
category_id: ID of category to retrieve
Returns:
Dictionary containing category data, or None if not found
"""
conn = self.finance_models.get_connection()
cursor = conn.cursor()
try:
cursor.execute('SELECT * FROM cost_categories WHERE id = ?', (category_id,))
row = cursor.fetchone()
if not row:
return None
columns = [desc[0] for desc in cursor.description]
return dict(zip(columns, row))
except sqlite3.Error as e:
return None
finally:
conn.close()
def list_categories(self) -> List[Dict[str, Any]]:
"""
List all cost categories.
Returns:
List of category dictionaries
"""
conn = self.finance_models.get_connection()
cursor = conn.cursor()
try:
cursor.execute('SELECT * FROM cost_categories ORDER BY name')
rows = cursor.fetchall()
columns = [desc[0] for desc in cursor.description]
return [dict(zip(columns, row)) for row in rows]
except sqlite3.Error as e:
return []
finally:
conn.close()
def get_category_by_name(self, name: str) -> Optional[Dict[str, Any]]:
"""
Retrieve a category by name.
Args:
name: Name of category to retrieve
Returns:
Dictionary containing category data, or None if not found
"""
conn = self.finance_models.get_connection()
cursor = conn.cursor()
try:
cursor.execute('SELECT * FROM cost_categories WHERE name = ?', (name,))
row = cursor.fetchone()
if not row:
return None
columns = [desc[0] for desc in cursor.description]
return dict(zip(columns, row))
except sqlite3.Error as e:
return None
finally:
conn.close()