100 lines
3.0 KiB
Python
Executable File
100 lines
3.0 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
|
|
import csv
|
|
from decimal import Decimal, InvalidOperation
|
|
import sys
|
|
from pathlib import Path
|
|
import hashlib
|
|
from collections import defaultdict
|
|
|
|
|
|
def transaction_hash(date, payee, desc, amount):
|
|
"""Generate a stable hash for each transaction."""
|
|
base = f"{date}|{payee}|{desc}|{amount}"
|
|
return hashlib.md5(base.encode("utf-8")).hexdigest()[:10] # short ID
|
|
|
|
|
|
def convert(in_path, out_path):
|
|
with open(in_path, newline="", encoding="utf-8-sig") as f_in, \
|
|
open(out_path, "w", newline="", encoding="utf-8") as f_out:
|
|
|
|
reader = csv.DictReader(f_in)
|
|
|
|
# Add UniqueID field to prevent Actual from treating repeated merchants as duplicates
|
|
fieldnames = ["Date", "Payee", "Description", "Amount", "UniqueID"]
|
|
writer = csv.DictWriter(f_out, fieldnames=fieldnames)
|
|
writer.writeheader()
|
|
|
|
seen_hashes = defaultdict(int)
|
|
skipped = 0
|
|
|
|
for row in reader:
|
|
if not row:
|
|
continue
|
|
|
|
date = (row.get("Transaction Date") or "").strip()
|
|
if not date:
|
|
continue
|
|
|
|
desc = (row.get("Transaction Description") or "").strip()
|
|
payee = desc or (row.get("Transaction Type") or "").strip()
|
|
|
|
debit_str = (row.get("Debit Amount") or "").replace(",", "").strip()
|
|
credit_str = (row.get("Credit Amount") or "").replace(",", "").strip()
|
|
amount = None
|
|
|
|
if debit_str:
|
|
try:
|
|
amount = -Decimal(debit_str)
|
|
except InvalidOperation:
|
|
continue
|
|
elif credit_str:
|
|
try:
|
|
amount = Decimal(credit_str)
|
|
except InvalidOperation:
|
|
continue
|
|
else:
|
|
continue
|
|
|
|
# Generate transaction fingerprint
|
|
unique_id = transaction_hash(date, payee, desc, amount)
|
|
|
|
# Check if we've seen this transaction hash already
|
|
seen_hashes[unique_id] += 1
|
|
if seen_hashes[unique_id] > 1:
|
|
# Optional: skip true duplicates or just warn
|
|
print(f"⚠️ Duplicate detected: {date} | {payee} | {amount}")
|
|
skipped += 1
|
|
continue
|
|
|
|
writer.writerow({
|
|
"Date": date,
|
|
"Payee": payee,
|
|
"Description": desc,
|
|
"Amount": f"{amount:.2f}",
|
|
"UniqueID": unique_id,
|
|
})
|
|
|
|
print(f"\n✅ Written cleaned file to {out_path}")
|
|
if skipped:
|
|
print(f"⚠️ Skipped {skipped} duplicate entries based on content hash.")
|
|
|
|
|
|
def main():
|
|
if len(sys.argv) < 2:
|
|
print("Usage: python clean_halifax_actual.py input.csv [output.csv]")
|
|
sys.exit(1)
|
|
|
|
in_path = Path(sys.argv[1])
|
|
|
|
if len(sys.argv) >= 3:
|
|
out_path = Path(sys.argv[2])
|
|
else:
|
|
out_path = in_path.with_name(in_path.stem + "_actual.csv")
|
|
|
|
convert(in_path, out_path)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|