314 lines
10 KiB
Python
314 lines
10 KiB
Python
import pandas as pd
|
|
from pathlib import Path
|
|
import sys
|
|
from datetime import datetime
|
|
from typing import Optional
|
|
|
|
class TricountToCospendConverter:
|
|
"""Convert Tricount CSV exports to Cospend import format."""
|
|
|
|
DEFAULT_COLORS = [
|
|
"#d09e6d", "#0082c9", "#ff6b6b", "#4ecdc4", "#45b7d1",
|
|
"#95e1d3", "#f38181", "#aa96da", "#fcbad3", "#a8d8ea"
|
|
]
|
|
|
|
DEFAULT_CATEGORIES = [
|
|
("Grocery", 1, "🛒", "#ffaa00"),
|
|
("Bar/Party", 2, "🎉", "#aa55ff"),
|
|
("Rent", 3, "🏠", "#da8733"),
|
|
("Bill", 4, "🌩", "#4aa6b0"),
|
|
("Excursion/Culture", 5, "🚸", "#0055ff"),
|
|
("Health", 6, "💚", "#bf090c"),
|
|
("Shopping", 7, "🛍", "#e167d1"),
|
|
("Restaurant", 8, "🍴", "#d0d5e1"),
|
|
("Accommodation", 9, "🛌", "#5de1a3"),
|
|
("Transport", 10, "🚌", "#6f2ee1"),
|
|
("Sport", 11, "🎾", "#69e177")
|
|
]
|
|
|
|
DEFAULT_PAYMENT_MODES = [
|
|
("Cash", 2, "💵", "#556B2F"),
|
|
("Transfer", 4, "⇄", "#00CED1"),
|
|
("Online service", 5, "🌎", "#9932CC"),
|
|
("Card", 3, "💳", "#A9A9A9")
|
|
]
|
|
|
|
def __init__(
|
|
self,
|
|
input_file: str,
|
|
output_file: Optional[str] = None,
|
|
system_columns: Optional[list] = None,
|
|
colors: Optional[list] = None,
|
|
verbose: bool = True
|
|
):
|
|
"""
|
|
Initialize converter.
|
|
|
|
Args:
|
|
input_file: Path to Tricount CSV export
|
|
output_file: Path for Cospend output (default: cospend_output/cospend_import.csv)
|
|
system_columns: Column names to exclude from member detection
|
|
colors: Custom color palette for members
|
|
verbose: Print progress information
|
|
"""
|
|
self.input_file = input_file
|
|
self.output_file = output_file or "cospend_output/cospend_import.csv"
|
|
self.system_columns = system_columns or ['Date', 'Description', 'Category', 'Cost', 'Currency']
|
|
self.colors = colors or self.DEFAULT_COLORS
|
|
self.verbose = verbose
|
|
self.df = None
|
|
self.members = []
|
|
self.skipped_rows = 0
|
|
|
|
def log(self, message: str):
|
|
"""Print message if verbose mode enabled."""
|
|
if self.verbose:
|
|
print(message)
|
|
|
|
def read_csv(self) -> pd.DataFrame:
|
|
"""Read and validate Tricount CSV."""
|
|
self.df = pd.read_csv(self.input_file)
|
|
self.log(f"Reading: {self.input_file}")
|
|
self.log(f"Shape: {self.df.shape}")
|
|
self.log(f"Sample:\n{self.df.head()}\n")
|
|
return self.df
|
|
|
|
def detect_members(self) -> list[str]:
|
|
"""Extract member names from columns."""
|
|
self.members = [
|
|
col for col in self.df.columns
|
|
if col not in self.system_columns
|
|
]
|
|
self.log(f"Detected members: {self.members}\n")
|
|
return self.members
|
|
|
|
def create_members_data(self) -> list[dict]:
|
|
"""Generate Cospend members data."""
|
|
members_data = []
|
|
for idx, member in enumerate(self.members):
|
|
members_data.append({
|
|
'name': member.lower(),
|
|
'weight': 1,
|
|
'active': 1,
|
|
'color': f'"{self.colors[idx % len(self.colors)]}"'
|
|
})
|
|
return members_data
|
|
|
|
def find_payer(self, row: pd.Series) -> Optional[str]:
|
|
"""
|
|
Determine who paid for an expense.
|
|
|
|
Logic:
|
|
- Payer has positive balance OR
|
|
- Payer has large negative value close to cost (full amount case)
|
|
"""
|
|
cost = row['Cost']
|
|
|
|
# Check for positive value (typical case)
|
|
for member in self.members:
|
|
if pd.notna(row[member]) and row[member] > 0:
|
|
return member.lower()
|
|
|
|
# Check for negative value matching cost (paid full amount)
|
|
for member in self.members:
|
|
if pd.notna(row[member]) and row[member] < 0:
|
|
if abs(abs(row[member]) - cost) < 0.01:
|
|
return member.lower()
|
|
|
|
return None
|
|
|
|
def create_expenses_data(self) -> list[dict]:
|
|
"""Convert transactions to Cospend expense format."""
|
|
expenses_data = []
|
|
self.skipped_rows = 0
|
|
|
|
for idx, row in self.df.iterrows():
|
|
if pd.isna(row['Date']):
|
|
self.skipped_rows += 1
|
|
continue
|
|
|
|
# Parse date
|
|
try:
|
|
date_obj = pd.to_datetime(row['Date'])
|
|
date_str = date_obj.strftime('%Y-%m-%d')
|
|
timestamp = int(date_obj.timestamp())
|
|
except:
|
|
self.skipped_rows += 1
|
|
continue
|
|
|
|
# Find payer
|
|
payer = self.find_payer(row)
|
|
if not payer:
|
|
self.log(f"⚠️ Row {idx}: No payer for '{row['Description']}' (Cost: {row['Cost']})")
|
|
for member in self.members:
|
|
if pd.notna(row[member]):
|
|
self.log(f" {member}: {row[member]}")
|
|
self.skipped_rows += 1
|
|
continue
|
|
|
|
# All members are owers (split equally)
|
|
owers = ','.join([m.lower() for m in self.members])
|
|
|
|
expense = {
|
|
'what': f'"{row["Description"]}"',
|
|
'amount': abs(row['Cost']),
|
|
'date': date_str,
|
|
'timestamp': timestamp,
|
|
'payer_name': f'"{payer}"',
|
|
'payer_weight': 1,
|
|
'payer_active': 1,
|
|
'owers': f'"{owers}"',
|
|
'repeat': 'n',
|
|
'repeatfreq': 1,
|
|
'repeatallactive': 0,
|
|
'repeatuntil': '',
|
|
'categoryid': 0,
|
|
'paymentmode': 'n',
|
|
'paymentmodeid': 0,
|
|
'comment': '""',
|
|
'deleted': 0
|
|
}
|
|
expenses_data.append(expense)
|
|
|
|
if self.skipped_rows > 0:
|
|
self.log(f"\n⚠️ Skipped {self.skipped_rows} rows\n")
|
|
|
|
return expenses_data
|
|
|
|
def write_cospend_csv(
|
|
self,
|
|
members_data: list,
|
|
expenses_data: list,
|
|
categories: Optional[list] = None,
|
|
payment_modes: Optional[list] = None
|
|
):
|
|
"""Write Cospend-formatted CSV with all sections."""
|
|
categories = categories or self.DEFAULT_CATEGORIES
|
|
payment_modes = payment_modes or self.DEFAULT_PAYMENT_MODES
|
|
|
|
# Ensure output directory exists
|
|
output_path = Path(self.output_file)
|
|
output_path.parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
with open(self.output_file, 'w', encoding='utf-8') as f:
|
|
# Section 1: Members
|
|
f.write('name,weight,active,color\n')
|
|
for member in members_data:
|
|
f.write(f'"{member["name"]}",{member["weight"]},{member["active"]},{member["color"]}\n')
|
|
f.write('\n')
|
|
|
|
# Section 2: Expenses
|
|
expense_headers = [
|
|
'what', 'amount', 'date', 'timestamp', 'payer_name', 'payer_weight',
|
|
'payer_active', 'owers', 'repeat', 'repeatfreq', 'repeatallactive',
|
|
'repeatuntil', 'categoryid', 'paymentmode', 'paymentmodeid', 'comment', 'deleted'
|
|
]
|
|
f.write(','.join(expense_headers) + '\n')
|
|
for expense in expenses_data:
|
|
row_values = [str(expense[header]) for header in expense_headers]
|
|
f.write(','.join(row_values) + '\n')
|
|
f.write('\n')
|
|
|
|
# Section 3: Categories
|
|
f.write('categoryname,categoryid,icon,color\n')
|
|
for cat in categories:
|
|
f.write(f'"{cat[0]}",{cat[1]},"{cat[2]}","{cat[3]}"\n')
|
|
f.write('\n')
|
|
|
|
# Section 4: Payment modes
|
|
f.write('paymentmodename,paymentmodeid,icon,color\n')
|
|
for pm in payment_modes:
|
|
f.write(f'"{pm[0]}",{pm[1]},"{pm[2]}","{pm[3]}"\n')
|
|
|
|
def convert(self) -> dict:
|
|
"""
|
|
Execute full conversion pipeline.
|
|
|
|
Returns:
|
|
Summary statistics
|
|
"""
|
|
# Read and parse
|
|
self.read_csv()
|
|
self.detect_members()
|
|
|
|
# Convert data
|
|
members_data = self.create_members_data()
|
|
expenses_data = self.create_expenses_data()
|
|
|
|
# Write output
|
|
self.write_cospend_csv(members_data, expenses_data)
|
|
|
|
# Calculate statistics
|
|
payer_counts = {}
|
|
for expense in expenses_data:
|
|
payer = expense['payer_name'].strip('"')
|
|
payer_counts[payer] = payer_counts.get(payer, 0) + 1
|
|
|
|
summary = {
|
|
'members': len(members_data),
|
|
'expenses': len(expenses_data),
|
|
'skipped': self.skipped_rows,
|
|
'payer_distribution': payer_counts,
|
|
'output_file': self.output_file
|
|
}
|
|
|
|
# Print summary
|
|
self.log(f"\n✅ Processed {len(expenses_data)} valid records")
|
|
self.log(f"✅ Exported to: {self.output_file}")
|
|
self.log(f"\n📋 Summary:")
|
|
self.log(f"- Members: {summary['members']}")
|
|
self.log(f"- Expenses: {summary['expenses']}")
|
|
self.log(f"- Skipped: {summary['skipped']}")
|
|
self.log(f"\n💰 Expenses by payer:")
|
|
for payer, count in sorted(payer_counts.items()):
|
|
self.log(f"- {payer}: {count} expenses")
|
|
|
|
return summary
|
|
|
|
|
|
def main():
|
|
"""CLI interface."""
|
|
import argparse
|
|
|
|
parser = argparse.ArgumentParser(
|
|
description='Convert Tricount CSV exports to Cospend format'
|
|
)
|
|
parser.add_argument(
|
|
'input',
|
|
help='Input Tricount CSV file'
|
|
)
|
|
parser.add_argument(
|
|
'-o', '--output',
|
|
help='Output file path (default: cospend_output/cospend_import.csv)',
|
|
default=None
|
|
)
|
|
parser.add_argument(
|
|
'-q', '--quiet',
|
|
action='store_true',
|
|
help='Suppress output'
|
|
)
|
|
|
|
args = parser.parse_args()
|
|
|
|
try:
|
|
converter = TricountToCospendConverter(
|
|
input_file=args.input,
|
|
output_file=args.output,
|
|
verbose=not args.quiet
|
|
)
|
|
converter.convert()
|
|
|
|
except FileNotFoundError:
|
|
print(f"❌ Error: Input file '{args.input}' not found")
|
|
sys.exit(1)
|
|
except Exception as e:
|
|
print(f"❌ Error: {str(e)}")
|
|
if not args.quiet:
|
|
import traceback
|
|
traceback.print_exc()
|
|
sys.exit(1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|