from typing import List, Dict, Any import csv from io import StringIO, BytesIO from datetime import datetime from app.models.expense import Expense, ExpenseSplit from app.models.participant import Participant from sqlalchemy.orm import Session class ExportService: """Service for exporting trip data in various formats""" def export_expenses_csv(self, trip_id: int, db: Session) -> str: """Export trip expenses to CSV format - clear and readable format""" from app.models.trip import Trip # Get trip info for currency trip = db.query(Trip).filter(Trip.id == trip_id).first() if not trip: return "Trip not found." trip_currency = trip.currency_code trip_currency_symbol = get_currency_symbol(trip_currency) # Get all expenses for the trip, ordered by date (newest first) expenses = db.query(Expense).filter(Expense.trip_id == trip_id).order_by(Expense.date_incurred.desc()).all() if not expenses: return "No expenses found for this trip." # Create CSV data output = StringIO() writer = csv.writer(output) # Write trip information header writer.writerow(['TRIP INFORMATION']) writer.writerow(['Trip Name', trip.name]) writer.writerow(['Trip Code', trip.share_code]) writer.writerow(['Currency', f"{trip_currency_symbol} {trip_currency}"]) writer.writerow(['Export Date', datetime.utcnow().strftime('%Y-%m-%d %H:%M:%S')]) writer.writerow([]) # Empty row # Write expenses section header writer.writerow(['EXPENSES']) writer.writerow([ 'Date', 'Time', 'Description', 'Category', 'Amount (Original Currency)', 'Original Currency', f'Amount ({trip_currency})', 'Paid By' ]) # Write expense data - one row per expense for expense in expenses: date_obj = expense.date_incurred amount_in_trip_currency = expense.amount * expense.exchange_rate writer.writerow([ date_obj.strftime('%Y-%m-%d'), date_obj.strftime('%H:%M'), expense.description, expense.category or '', f"{expense.amount:.2f}", expense.currency_code, f"{amount_in_trip_currency:.2f}", self._get_paid_by_name(expense.paid_by_id, db) ]) writer.writerow([]) # Empty row # Write splits section - one row per split, showing who paid and who owes writer.writerow(['EXPENSE SPLITS - WHO PAID AND WHO OWES']) writer.writerow([ 'Expense Description', 'Date', 'Paid By', 'Owed By', 'Percentage', f'Amount Owed ({trip_currency})', 'Settled' ]) for expense in expenses: paid_by_name = self._get_paid_by_name(expense.paid_by_id, db) splits = db.query(ExpenseSplit, Participant).join( Participant, ExpenseSplit.participant_id == Participant.id ).filter(ExpenseSplit.expense_id == expense.id).all() for split, participant in splits: # Convert split amount from expense currency to trip currency split_amount_in_trip_currency = split.amount * expense.exchange_rate writer.writerow([ expense.description, expense.date_incurred.strftime('%Y-%m-%d'), paid_by_name, participant.name, f"{split.percentage:.1f}%", f"{split_amount_in_trip_currency:.2f}", 'Yes' if split.is_settled else 'No' ]) writer.writerow([]) # Empty row # Write summary section writer.writerow(['SUMMARY']) writer.writerow([]) # Total expenses total_expenses = sum(exp.amount * exp.exchange_rate for exp in expenses) writer.writerow(['Total Expenses', f"{trip_currency_symbol}{total_expenses:.2f}"]) writer.writerow(['Number of Expenses', len(expenses)]) writer.writerow([]) # Participant balances writer.writerow(['PARTICIPANT BALANCES']) writer.writerow([ 'Participant', f'Total Paid ({trip_currency})', f'Total Owed ({trip_currency})', f'Balance ({trip_currency})', 'Status' ]) participant_totals = self._calculate_participant_totals(trip_id, db) balances = {} for participant_name, data in sorted(participant_totals.items()): balance = data['total_paid'] - data['total_owed'] balances[participant_name] = balance if balance > 0.01: status = f"Gets back {trip_currency_symbol}{abs(balance):.2f}" elif balance < -0.01: status = f"Owes {trip_currency_symbol}{abs(balance):.2f}" else: status = "Settled" writer.writerow([ participant_name, f"{data['total_paid']:.2f}", f"{data['total_owed']:.2f}", f"{balance:.2f}", status ]) writer.writerow([]) # Empty row # Calculate and show who owes what to whom writer.writerow(['WHO OWES WHOM - SETTLEMENTS']) writer.writerow([ 'From (Pays)', 'To (Receives)', f'Amount ({trip_currency})', 'Note' ]) # Calculate settlements: match people who owe with people who are owed settlements = self._calculate_settlements(balances) if settlements: for settlement in settlements: writer.writerow([ settlement['from'], settlement['to'], f"{settlement['amount']:.2f}", settlement.get('note', '') ]) else: writer.writerow(['', '', '', 'All participants are settled up!']) csv_content = output.getvalue() return csv_content def _calculate_settlements(self, balances: Dict[str, float]) -> List[Dict[str, Any]]: """Calculate who should pay whom to settle all balances""" settlements = [] # Separate people who owe (negative balance) from people who are owed (positive balance) debtors = [(name, abs(balance)) for name, balance in balances.items() if balance < -0.01] creditors = [(name, balance) for name, balance in balances.items() if balance > 0.01] # Sort by amount (largest first) debtors.sort(key=lambda x: x[1], reverse=True) creditors.sort(key=lambda x: x[1], reverse=True) # Match up debts with credits debtor_idx = 0 creditor_idx = 0 while debtor_idx < len(debtors) and creditor_idx < len(creditors): debtor_name, debt_amount = debtors[debtor_idx] creditor_name, credit_amount = creditors[creditor_idx] # Calculate how much to transfer transfer_amount = min(debt_amount, credit_amount) if transfer_amount > 0.01: # Only add if amount is significant settlements.append({ 'from': debtor_name, 'to': creditor_name, 'amount': transfer_amount, 'note': f"{debtor_name} should pay {creditor_name}" }) # Update remaining amounts debt_amount -= transfer_amount credit_amount -= transfer_amount if debt_amount < 0.01: debtor_idx += 1 else: debtors[debtor_idx] = (debtor_name, debt_amount) if credit_amount < 0.01: creditor_idx += 1 else: creditors[creditor_idx] = (creditor_name, credit_amount) return settlements def _get_paid_by_name(self, participant_id: int, db: Session) -> str: """Get participant name by ID""" participant = db.query(Participant).filter(Participant.id == participant_id).first() return participant.name if participant else "Unknown" def _calculate_participant_totals(self, trip_id: int, db: Session) -> Dict[str, Dict[str, float]]: """Calculate total paid and owed amounts for each participant""" participant_totals = {} # Get all expenses and their splits expenses = db.query(Expense).filter(Expense.trip_id == trip_id).all() for expense in expenses: splits = db.query(ExpenseSplit, Participant).join( Participant, ExpenseSplit.participant_id == Participant.id ).filter(ExpenseSplit.expense_id == expense.id).all() # Track who paid for this expense paid_by_name = self._get_paid_by_name(expense.paid_by_id, db) if paid_by_name not in participant_totals: participant_totals[paid_by_name] = {'total_paid': 0.0, 'total_owed': 0.0} participant_totals[paid_by_name]['total_paid'] += expense.amount * expense.exchange_rate # Track who owes for this expense # Convert split amount to trip currency (split.amount is in expense currency) for split, participant in splits: participant_name = participant.name if participant_name not in participant_totals: participant_totals[participant_name] = {'total_paid': 0.0, 'total_owed': 0.0} # Split amount is in expense currency, convert to trip currency split_amount_in_trip_currency = split.amount * expense.exchange_rate participant_totals[participant_name]['total_owed'] += split_amount_in_trip_currency return participant_totals async def export_expenses_pdf(self, trip_id: int, db: Session) -> bytes: """Export trip expenses as a beautiful PDF using HTML -> PDF with Playwright""" try: from playwright.async_api import async_playwright import json import tempfile import os except ImportError as e: raise ImportError(f"PDF generation requires playwright. Please install: pip install playwright && playwright install chromium. Error: {e}") from app.models.trip import Trip # Get trip info trip = db.query(Trip).filter(Trip.id == trip_id).first() if not trip: raise ValueError("Trip not found") trip_currency = trip.currency_code trip_currency_symbol = get_currency_symbol(trip_currency) # Get all expenses expenses = db.query(Expense).filter(Expense.trip_id == trip_id).order_by(Expense.date_incurred.desc()).all() # Calculate totals total_expenses = sum(exp.amount * exp.exchange_rate for exp in expenses) participant_totals = self._calculate_participant_totals(trip_id, db) # Calculate category breakdown category_totals = {} for expense in expenses: category = expense.category or 'Other' category_totals[category] = category_totals.get(category, 0) + expense.amount * expense.exchange_rate # Calculate settlements balances = {name: data['total_paid'] - data['total_owed'] for name, data in participant_totals.items()} settlements = self._calculate_settlements_for_pdf(balances) # Calculate additional stats from collections import defaultdict expenses_by_date = defaultdict(lambda: {'count': 0, 'total': 0.0}) top_expense = None max_expense_amount = 0 for expense in expenses: date_key = expense.date_incurred.strftime('%Y-%m-%d') amount_in_trip_currency = expense.amount * expense.exchange_rate expenses_by_date[date_key]['count'] += 1 expenses_by_date[date_key]['total'] += amount_in_trip_currency if amount_in_trip_currency > max_expense_amount: max_expense_amount = amount_in_trip_currency top_expense = { 'description': expense.description, 'amount': amount_in_trip_currency, 'date': expense.date_incurred.strftime('%Y-%m-%d'), 'category': expense.category or 'N/A', 'paid_by': self._get_paid_by_name(expense.paid_by_id, db) } # Find top day (most expenses) top_day = None max_day_count = 0 for date_key, data in expenses_by_date.items(): if data['count'] > max_day_count: max_day_count = data['count'] top_day = { 'date': date_key, 'count': data['count'], 'total': data['total'] } # Calculate average expense avg_expense = total_expenses / len(expenses) if expenses else 0 # Prepare data for HTML template expenses_data = [] for expense in expenses[:50]: expenses_data.append({ 'id': expense.id, 'date': expense.date_incurred.strftime('%Y-%m-%d'), 'description': expense.description, 'category': expense.category or 'N/A', 'amount': expense.amount * expense.exchange_rate, 'paid_by': self._get_paid_by_name(expense.paid_by_id, db) }) participant_data = [] for name, data in sorted(participant_totals.items()): balance = data['total_paid'] - data['total_owed'] participant_data.append({ 'name': name, 'paid': data['total_paid'], 'owed': data['total_owed'], 'balance': balance, 'status': 'positive' if balance > 0.01 else ('negative' if balance < -0.01 else 'settled') }) # Generate HTML html_content = self._generate_pdf_html( trip=trip, trip_currency_symbol=trip_currency_symbol, total_expenses=total_expenses, expenses_count=len(expenses), avg_expense=avg_expense, category_totals=category_totals, participant_data=participant_data, settlements=settlements, expenses_data=expenses_data, top_day=top_day, top_expense=top_expense ) # Render HTML to PDF using Playwright (async) async with async_playwright() as p: browser = await p.chromium.launch(headless=True) page = await browser.new_page() await page.set_content(html_content, wait_until='networkidle') # Wait for charts to render await page.wait_for_timeout(2000) # Generate PDF pdf_bytes = await page.pdf( format='A4', margin={'top': '0.5in', 'right': '0.5in', 'bottom': '0.5in', 'left': '0.5in'}, print_background=True ) await browser.close() return pdf_bytes def _generate_pdf_html(self, trip, trip_currency_symbol, total_expenses, expenses_count, avg_expense, category_totals, participant_data, settlements, expenses_data, top_day, top_expense) -> str: """Generate beautiful HTML template for PDF""" import json from datetime import datetime category_labels = json.dumps(list(category_totals.keys())) category_values = json.dumps(list(category_totals.values())) participant_names = json.dumps([p['name'] for p in participant_data]) participant_balances = json.dumps([p['balance'] for p in participant_data]) html = f""" Trip Expense Report

{trip.name}

Expense Report • {datetime.utcnow().strftime('%B %d, %Y')}

Overview

Total Expenses
{trip_currency_symbol}{total_expenses:,.2f}
Expenses Count
{expenses_count}
Average Expense
{trip_currency_symbol}{avg_expense:,.2f}
Currency
{trip_currency_symbol} {trip.currency_code}

Key Insights

""" if top_day: html += f"""
Top Spending Day
{top_day['date']}
{top_day['count']} expenses • {trip_currency_symbol}{top_day['total']:,.2f}
""" if top_expense: html += f"""
Largest Expense
{trip_currency_symbol}{top_expense['amount']:,.2f}
{top_expense['description'][:40]}{'...' if len(top_expense['description']) > 40 else ''}
""" html += """

Expenses by Category

View all expenses →

Participant Balances

View participant details →

Settlements

View participant balances →
""" if settlements: for s in settlements: html += f""" """ else: html += """ """ html += """
From To Amount
{s['from']} {s['to']} {trip_currency_symbol}{s['amount']:,.2f}
✓ All participants are settled up!

Participant Summary

View all expenses →
""" for p in participant_data: if p['balance'] > 0.01: status_text = f"Gets back {trip_currency_symbol}{abs(p['balance']):,.2f}" status_class = "badge-success" elif p['balance'] < -0.01: status_text = f"Owes {trip_currency_symbol}{abs(p['balance']):,.2f}" status_class = "badge-danger" else: status_text = "Settled" status_class = "badge-neutral" html += f""" """ html += """
Participant Paid Owed Balance Status
{p['name']} {trip_currency_symbol}{p['paid']:,.2f} {trip_currency_symbol}{p['owed']:,.2f} {trip_currency_symbol}{p['balance']:+,.2f} {status_text}

Expense Details

← Back to overview
""" for exp in expenses_data: html += f""" """ html += f"""
Date Description Category Amount Paid By
{exp['date']} {exp['description'][:50]}{'...' if len(exp['description']) > 50 else ''} {exp['category']} {trip_currency_symbol}{exp['amount']:,.2f} {exp['paid_by']}
""" return html def _create_category_chart(self, category_totals: Dict[str, float], currency_symbol: str) -> BytesIO: """Create a modern pie chart for category breakdown""" import matplotlib matplotlib.use('Agg') # Use non-interactive backend import matplotlib.pyplot as plt # Modern color palette colors_list = [ '#3b82f6', # Blue '#10b981', # Green '#f59e0b', # Amber '#ef4444', # Red '#8b5cf6', # Purple '#ec4899', # Pink '#06b6d4', # Cyan '#f97316', # Orange '#84cc16', # Lime '#6366f1', # Indigo ] fig, ax = plt.subplots(figsize=(7, 5)) fig.patch.set_facecolor('white') ax.set_facecolor('white') categories = list(category_totals.keys()) amounts = list(category_totals.values()) # Use modern colors chart_colors = colors_list[:len(categories)] # Create pie chart with modern styling wedges, texts, autotexts = ax.pie( amounts, labels=categories, autopct='%1.1f%%', colors=chart_colors, startangle=90, textprops={'fontsize': 10, 'fontweight': 'bold', 'color': '#0f172a'}, pctdistance=0.85, labeldistance=1.1, wedgeprops={'edgecolor': 'white', 'linewidth': 2, 'linestyle': 'solid'} ) # Style percentage text for autotext in autotexts: autotext.set_color('white') autotext.set_fontweight('bold') autotext.set_fontsize(9) # Style labels for text in texts: text.set_fontsize(10) text.set_fontweight('500') text.set_color('#475569') ax.set_title('Expenses by Category', fontsize=16, fontweight='bold', pad=25, color='#1e40af') plt.tight_layout() buffer = BytesIO() plt.savefig(buffer, format='png', dpi=150, bbox_inches='tight', facecolor='white', edgecolor='none') plt.close() buffer.seek(0) return buffer def _create_balance_chart(self, participant_totals: Dict[str, Dict[str, float]], currency_symbol: str) -> BytesIO: """Create a modern bar chart for participant balances""" import matplotlib matplotlib.use('Agg') # Use non-interactive backend import matplotlib.pyplot as plt fig, ax = plt.subplots(figsize=(7, 5)) fig.patch.set_facecolor('white') ax.set_facecolor('white') names = list(participant_totals.keys()) balances = [data['total_paid'] - data['total_owed'] for data in participant_totals.values()] # Modern color scheme colors_list = ['#10b981' if b > 0.01 else '#ef4444' if b < -0.01 else '#64748b' for b in balances] # Create horizontal bar chart bars = ax.barh(names, balances, color=colors_list, height=0.6, edgecolor='white', linewidth=2) # Add value labels on bars with better positioning max_abs_balance = max(abs(b) for b in balances) if balances else 1 for i, (name, balance) in enumerate(zip(names, balances)): label_x = balance + (max_abs_balance * 0.03 if balance >= 0 else -max_abs_balance * 0.03) ax.text( label_x, i, f"{currency_symbol}{balance:+,.2f}", va='center', fontweight='bold', fontsize=10, color='#0f172a' ) # Zero line ax.axvline(x=0, color='#cbd5e1', linestyle='-', linewidth=1.5, zorder=0) # Styling ax.set_xlabel(f'Balance ({currency_symbol})', fontweight='bold', fontsize=12, color='#475569') ax.set_title('Participant Balances', fontsize=16, fontweight='bold', pad=25, color='#1e40af') ax.grid(axis='x', alpha=0.2, linestyle='--', color='#cbd5e1') ax.spines['top'].set_visible(False) ax.spines['right'].set_visible(False) ax.spines['left'].set_color('#e2e8f0') ax.spines['bottom'].set_color('#e2e8f0') # Y-axis styling ax.tick_params(colors='#64748b', labelsize=10) ax.set_yticks(range(len(names))) ax.set_yticklabels(names, fontweight='500') plt.tight_layout() buffer = BytesIO() plt.savefig(buffer, format='png', dpi=150, bbox_inches='tight', facecolor='white', edgecolor='none') plt.close() buffer.seek(0) return buffer def _calculate_settlements_for_pdf(self, balances: Dict[str, float]) -> List[Dict[str, Any]]: """Calculate who should pay whom to settle all balances""" settlements = [] debtors = [(name, abs(balance)) for name, balance in balances.items() if balance < -0.01] creditors = [(name, balance) for name, balance in balances.items() if balance > 0.01] debtors.sort(key=lambda x: x[1], reverse=True) creditors.sort(key=lambda x: x[1], reverse=True) debtor_idx = 0 creditor_idx = 0 while debtor_idx < len(debtors) and creditor_idx < len(creditors): debtor_name, debt_amount = debtors[debtor_idx] creditor_name, credit_amount = creditors[creditor_idx] transfer_amount = min(debt_amount, credit_amount) if transfer_amount > 0.01: settlements.append({ 'from': debtor_name, 'to': creditor_name, 'amount': transfer_amount }) debt_amount -= transfer_amount credit_amount -= transfer_amount if debt_amount < 0.01: debtor_idx += 1 else: debtors[debtor_idx] = (debtor_name, debt_amount) if credit_amount < 0.01: creditor_idx += 1 else: creditors[creditor_idx] = (creditor_name, credit_amount) return settlements def export_participants_csv(self, trip_id: int, db: Session) -> str: """Export trip participants to CSV format""" participants = db.query(Participant).filter( Participant.trip_id == trip_id, Participant.is_active == True ).all() if not participants: return "No participants found for this trip." output = StringIO() writer = csv.writer(output) # Write header writer.writerow(['Name', 'Is Creator', 'Is Active', 'Created At']) # Write participant data for participant in participants: writer.writerow([ participant.name, 'Yes' if participant.is_creator else 'No', 'Yes' if participant.is_active else 'No', participant.created_at.strftime('%Y-%m-%d %H:%M:%S') ]) return output.getvalue() def export_summary_json(self, trip_id: int, db: Session) -> Dict[str, Any]: """Generate a comprehensive summary in JSON format""" # Get trip info from app.models.trip import Trip trip = db.query(Trip).filter(Trip.id == trip_id).first() if not trip: return {"error": "Trip not found"} # Get expenses expenses = db.query(Expense).filter(Expense.trip_id == trip_id).order_by(Expense.date_incurred.desc()).all() # Calculate analytics total_expenses = sum(exp.amount * exp.exchange_rate for exp in expenses) participant_totals = self._calculate_participant_totals(trip_id, db) # Categorize expenses category_totals = {} for expense in expenses: category = expense.category or 'Other' category_totals[category] = category_totals.get(category, 0) + expense.amount * expense.exchange_rate return { "trip": { "id": trip.id, "name": trip.name, "share_code": trip.share_code, "currency_code": trip.currency_code, "created_at": trip.created_at.isoformat() }, "summary": { "total_expenses": total_expenses, "total_expenses_count": len(expenses), "category_breakdown": category_totals, "participant_balances": participant_totals }, "expenses": [ { "id": exp.id, "description": exp.description, "amount": exp.amount, "currency_code": exp.currency_code, "amount_in_trip_currency": exp.amount * exp.exchange_rate, "category": exp.category, "paid_by": self._get_paid_by_name(exp.paid_by_id, db), "date_incurred": exp.date_incurred.isoformat(), "splits": [ { "participant": split.participant.name, "percentage": split.percentage, "amount": split.amount, "is_settled": split.is_settled } for split, split_participant in db.query(ExpenseSplit, Participant).join( Participant, ExpenseSplit.participant_id == Participant.id ).filter(ExpenseSplit.expense_id == exp.id).all() ] } for exp in expenses ], "generated_at": datetime.utcnow().isoformat() } def get_currency_symbol(currency_code: str) -> str: """Get currency symbol for given currency code""" symbols = { 'USD': '$', 'EUR': '€', 'GBP': '£', 'JPY': '¥', 'CNY': '¥', 'INR': '₹', 'AED': 'د.إ', } return symbols.get(currency_code, currency_code) # Singleton instance export_service = ExportService()