Enhance aggregation script to support annual reports
This commit is contained in:
Kevin Bataille
@@ -1,6 +1,6 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Script to aggregate all account statements by month
|
||||
Script to aggregate all account statements by month or year
|
||||
"""
|
||||
|
||||
import os
|
||||
@@ -106,21 +106,28 @@ def process_csv_file(file_path):
|
||||
return transactions
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description='Aggregate all account statements by month')
|
||||
parser = argparse.ArgumentParser(description='Aggregate all account statements by month or year')
|
||||
parser.add_argument('--input-dir', default='output/csv',
|
||||
help='Directory containing CSV files to aggregate (default: output/csv)')
|
||||
parser.add_argument('--output-dir', default='output/reports',
|
||||
help='Directory to save aggregated reports (default: output/reports)')
|
||||
parser.add_argument('--annual', action='store_true',
|
||||
help='Create annual reports instead of monthly reports')
|
||||
parser.add_argument('--year', type=int,
|
||||
help='Generate reports for a specific year only')
|
||||
|
||||
args = parser.parse_args()
|
||||
|
||||
# Create output directory
|
||||
os.makedirs(args.output_dir, exist_ok=True)
|
||||
|
||||
report_type = "Annual" if args.annual else "Monthly"
|
||||
print(f"\n{'='*60}")
|
||||
print(f"Monthly Aggregation of All Account Statements")
|
||||
print(f"{report_type} Aggregation of All Account Statements")
|
||||
print(f"Input Directory: {os.path.abspath(args.input_dir)}")
|
||||
print(f"Output Directory: {os.path.abspath(args.output_dir)}")
|
||||
if args.year:
|
||||
print(f"Year Filter: {args.year}")
|
||||
print(f"Date: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
|
||||
print(f"{'='*60}")
|
||||
|
||||
@@ -160,7 +167,7 @@ def main():
|
||||
])
|
||||
|
||||
# Process each month
|
||||
for (year, month) in sorted(monthly_transactions.keys()):
|
||||
for year, month in sorted(monthly_transactions.keys()):
|
||||
transactions = monthly_transactions[(year, month)]
|
||||
month_name = calendar.month_name[month]
|
||||
|
||||
@@ -180,34 +187,6 @@ def main():
|
||||
transaction_count, institutions_str
|
||||
])
|
||||
|
||||
# Create detailed monthly transactions file for each month
|
||||
for (year, month) in sorted(monthly_transactions.keys()):
|
||||
month_name = calendar.month_name[month].lower()
|
||||
transactions = monthly_transactions[(year, month)]
|
||||
|
||||
# Create filename
|
||||
detail_file = os.path.join(args.output_dir, f'transactions_{year}_{month_name}.csv')
|
||||
|
||||
with open(detail_file, 'w', newline='', encoding='utf-8') as f:
|
||||
writer = csv.DictWriter(f, fieldnames=[
|
||||
'Date', 'Description', 'Category', 'Amount',
|
||||
'Institution', 'Source'
|
||||
])
|
||||
writer.writeheader()
|
||||
|
||||
# Sort transactions by date
|
||||
sorted_transactions = sorted(transactions, key=lambda x: (x['day'], x['description']))
|
||||
|
||||
for transaction in sorted_transactions:
|
||||
writer.writerow({
|
||||
'Date': transaction['date_str'],
|
||||
'Description': transaction['description'],
|
||||
'Category': transaction['category'],
|
||||
'Amount': transaction['amount'],
|
||||
'Institution': transaction['institution'],
|
||||
'Source': transaction['source']
|
||||
})
|
||||
|
||||
# Create yearly summary
|
||||
yearly_summary = defaultdict(lambda: {'income': 0, 'expenses': 0, 'count': 0})
|
||||
for transaction in all_transactions:
|
||||
@@ -231,22 +210,113 @@ def main():
|
||||
year, data['income'], data['expenses'], net_balance, data['count']
|
||||
])
|
||||
|
||||
# Print summary statistics
|
||||
print(f"\n{'='*60}")
|
||||
print(f"Aggregation Complete")
|
||||
print(f"Total Transactions: {len(all_transactions)}")
|
||||
print(f"Months with Data: {len(monthly_transactions)}")
|
||||
print(f"{'='*60}")
|
||||
|
||||
# List generated files
|
||||
# Create annual reports if requested
|
||||
generated_files = [
|
||||
os.path.basename(summary_file),
|
||||
os.path.basename(yearly_file)
|
||||
]
|
||||
for (year, month) in sorted(monthly_transactions.keys()):
|
||||
month_name = calendar.month_name[month].lower()
|
||||
generated_files.append(f'transactions_{year}_{month_name}.csv')
|
||||
|
||||
if args.annual:
|
||||
# Create annual reports
|
||||
for year in sorted(yearly_summary.keys()):
|
||||
if args.year and year != args.year:
|
||||
continue # Skip years not matching filter
|
||||
|
||||
print(f"\nCreating annual report for {year}...")
|
||||
|
||||
# Get all transactions for the year
|
||||
year_transactions = [t for t in all_transactions if t['year'] == year]
|
||||
|
||||
# Group by category for the annual report
|
||||
categories = defaultdict(lambda: {'count': 0, 'total': 0})
|
||||
for transaction in year_transactions:
|
||||
category = transaction['category']
|
||||
amount = transaction['amount']
|
||||
categories[category]['count'] += 1
|
||||
categories[category]['total'] += amount
|
||||
|
||||
# Create annual detailed report
|
||||
annual_file = os.path.join(args.output_dir, f'annual_report_{year}.csv')
|
||||
with open(annual_file, 'w', newline='', encoding='utf-8') as f:
|
||||
writer = csv.writer(f)
|
||||
writer.writerow(['Category', 'Transaction Count', 'Total Amount', 'Percentage'])
|
||||
|
||||
year_total = sum(c['total'] for c in categories.values())
|
||||
|
||||
# Sort categories by total amount
|
||||
sorted_categories = sorted(categories.items(), key=lambda x: x[1]['total'], reverse=True)
|
||||
|
||||
for category, data in sorted_categories:
|
||||
percentage = (data['total'] / year_total) * 100 if year_total != 0 else 0
|
||||
writer.writerow([category, data['count'], data['total'], f"{percentage:.2f}%"])
|
||||
|
||||
# Create annual transactions file
|
||||
annual_transactions_file = os.path.join(args.output_dir, f'annual_transactions_{year}.csv')
|
||||
with open(annual_transactions_file, 'w', newline='', encoding='utf-8') as f:
|
||||
writer = csv.DictWriter(f, fieldnames=[
|
||||
'Date', 'Description', 'Category', 'Amount',
|
||||
'Institution', 'Source'
|
||||
])
|
||||
writer.writeheader()
|
||||
|
||||
# Sort transactions by date
|
||||
sorted_transactions = sorted(year_transactions, key=lambda x: (x['month'], x['day'], x['description']))
|
||||
|
||||
for transaction in sorted_transactions:
|
||||
writer.writerow({
|
||||
'Date': transaction['date_str'],
|
||||
'Description': transaction['description'],
|
||||
'Category': transaction['category'],
|
||||
'Amount': transaction['amount'],
|
||||
'Institution': transaction['institution'],
|
||||
'Source': transaction['source']
|
||||
})
|
||||
|
||||
generated_files.append(os.path.basename(annual_file))
|
||||
generated_files.append(os.path.basename(annual_transactions_file))
|
||||
|
||||
print(f" Created {os.path.basename(annual_file)} and {os.path.basename(annual_transactions_file)}")
|
||||
else:
|
||||
# Create monthly reports (existing functionality)
|
||||
for year, month in sorted(monthly_transactions.keys()):
|
||||
month_name = calendar.month_name[month].lower()
|
||||
transactions = monthly_transactions[(year, month)]
|
||||
|
||||
# Create filename
|
||||
detail_file = os.path.join(args.output_dir, f'transactions_{year}_{month_name}.csv')
|
||||
|
||||
with open(detail_file, 'w', newline='', encoding='utf-8') as f:
|
||||
writer = csv.DictWriter(f, fieldnames=[
|
||||
'Date', 'Description', 'Category', 'Amount',
|
||||
'Institution', 'Source'
|
||||
])
|
||||
writer.writeheader()
|
||||
|
||||
# Sort transactions by date
|
||||
sorted_transactions = sorted(transactions, key=lambda x: (x['day'], x['description']))
|
||||
|
||||
for transaction in sorted_transactions:
|
||||
writer.writerow({
|
||||
'Date': transaction['date_str'],
|
||||
'Description': transaction['description'],
|
||||
'Category': transaction['category'],
|
||||
'Amount': transaction['amount'],
|
||||
'Institution': transaction['institution'],
|
||||
'Source': transaction['source']
|
||||
})
|
||||
|
||||
generated_files.append(f'transactions_{year}_{month_name}.csv')
|
||||
|
||||
# Print summary statistics
|
||||
print(f"\n{'='*60}")
|
||||
print(f"Aggregation Complete")
|
||||
print(f"Total Transactions: {len(all_transactions)}")
|
||||
print(f"Years with Data: {len(yearly_summary)}")
|
||||
if not args.annual:
|
||||
print(f"Months with Data: {len(monthly_transactions)}")
|
||||
print(f"{'='*60}")
|
||||
|
||||
# List generated files
|
||||
print("\nGenerated Files:")
|
||||
for file in generated_files:
|
||||
file_path = os.path.join(args.output_dir, file)
|
||||
|
||||
Reference in New Issue
Block a user