""" Clean a messy expenses CSV and print a monthly summary. A function is a named recipe. This file uses small functions so each cleaning step (date, category, amount) stays easy to read. Run generate_data.py first so data/expenses.csv exists. """ import csv from collections import defaultdict from datetime import datetime from pathlib import Path TUTORIAL_DIR = Path(__file__).resolve().parent DEFAULT_INPUT_PATH = TUTORIAL_DIR / "data" / "expenses.csv" DEFAULT_OUTPUT_PATH = TUTORIAL_DIR / "output" / "monthly_summary.csv" # datetime.strptime needs a pattern for each date style we expect. DATE_FORMATS = [ "%Y-%m-%d", # 2026-01-03 "%m/%d/%Y", # 1/8/2026 or 01/15/2026 "%m-%d-%Y", # 01-30-2026 "%B %d, %Y", # January 10, 2026 "%b %d, %Y", # Jan 18, 2026 "%b %d %Y", # Mar 14 2026 (no comma) ] # Map many spellings onto a few tidy category names. # .lower() on the raw value makes "FOOD" and "food" the same key. CATEGORY_ALIASES = { "food": "Food", "groceries": "Food", "dining": "Food", "restaurant": "Food", "coffee": "Food", "transport": "Transport", "uber": "Transport", "lyft": "Transport", "gas": "Transport", "parking": "Transport", "utilities": "Utilities", "electric": "Utilities", "internet": "Utilities", "rent": "Housing", "housing": "Housing", "entertainment": "Entertainment", "movies": "Entertainment", "streaming": "Entertainment", "other": "Other", } def parse_date(raw_date): """Turn a messy date string into a date, or None if it cannot be read.""" if raw_date is None: return None text = str(raw_date).strip() if text == "": return None for pattern in DATE_FORMATS: try: return datetime.strptime(text, pattern).date() except ValueError: # This pattern did not match. An if/try here means "keep going." continue return None def parse_amount(raw_amount): """Turn a messy amount into a float (a number with decimals), or None.""" if raw_amount is None: return None text = str(raw_amount).strip() text = text.replace("$", "") text = text.replace(",", "") text = text.strip() if text == "": return None try: return float(text) except ValueError: return None def normalize_category(raw_category): """Return a tidy category name. Unknown spellings become Other.""" if raw_category is None: return "Other" key = str(raw_category).strip().lower() if key == "": return "Other" return CATEGORY_ALIASES.get(key, "Other") def clean_row(row): """Clean one CSV row. Return a dict, or None if the row is unusable. A dict (dictionary) maps names to values, like {"month": "2026-01", "category": "Food", "amount": 64.18}. """ parsed_date = parse_date(row.get("date")) amount = parse_amount(row.get("amount")) if parsed_date is None or amount is None: return None return { "date": parsed_date, "month": parsed_date.strftime("%Y-%m"), "merchant": str(row.get("merchant", "")).strip(), "category": normalize_category(row.get("category")), "amount": amount, "notes": str(row.get("notes", "")).strip(), } def read_and_clean(csv_path): """Read the CSV file and return (clean_rows, skipped_count).""" clean_rows = [] skipped_count = 0 with Path(csv_path).open("r", newline="", encoding="utf-8") as csv_file: reader = csv.DictReader(csv_file) for row in reader: cleaned = clean_row(row) if cleaned is None: skipped_count += 1 else: clean_rows.append(cleaned) return clean_rows, skipped_count def summarize(clean_rows): """Group amounts by month and category. totals[(month, category)] is the money spent. counts[(month, category)] is how many purchases. """ totals = defaultdict(float) counts = defaultdict(int) for row in clean_rows: key = (row["month"], row["category"]) totals[key] += row["amount"] counts[key] += 1 summary_rows = [] for month, category in sorted(totals.keys()): summary_rows.append( { "month": month, "category": category, "total": round(totals[(month, category)], 2), "transaction_count": counts[(month, category)], } ) return summary_rows def write_summary_csv(summary_rows, csv_path): """Write the grouped totals to a new CSV file.""" csv_path = Path(csv_path) csv_path.parent.mkdir(parents=True, exist_ok=True) fieldnames = ["month", "category", "total", "transaction_count"] with csv_path.open("w", newline="", encoding="utf-8") as csv_file: writer = csv.DictWriter(csv_file, fieldnames=fieldnames) writer.writeheader() for row in summary_rows: writer.writerow( { "month": row["month"], "category": row["category"], "total": f"{row['total']:.2f}", "transaction_count": row["transaction_count"], } ) return csv_path def print_summary(summary_rows, skipped_count, output_path): """Print a table a person can read in the terminal.""" print("Monthly expense summary") print("=======================") if not summary_rows: print("No clean rows to summarize.") return current_month = None month_total = 0.0 grand_total = 0.0 def print_month_total(): print(f" {'Month total':<16} ${month_total:>8.2f}") print() for row in summary_rows: if row["month"] != current_month: if current_month is not None: print_month_total() current_month = row["month"] month_total = 0.0 print(current_month) month_total += row["total"] grand_total += row["total"] label = f"{row['category']} ({row['transaction_count']})" print(f" {label:<16} ${row['total']:>8.2f}") print_month_total() print(f"Grand total: ${grand_total:.2f}") print(f"Skipped {skipped_count} messy row(s) that could not be cleaned.") print(f"Wrote {output_path}") def run(input_path, output_path): """Read, clean, print, and write. Used by the script and by tests.""" input_path = Path(input_path) if not input_path.exists(): print(f"Could not find {input_path}") print("Run generate_data.py first to create data/expenses.csv.") return False clean_rows, skipped_count = read_and_clean(input_path) summary_rows = summarize(clean_rows) saved_path = write_summary_csv(summary_rows, output_path) print_summary(summary_rows, skipped_count, saved_path) return True def main(): run(DEFAULT_INPUT_PATH, DEFAULT_OUTPUT_PATH) if __name__ == "__main__": main()