219 lines
8.2 KiB
Python
219 lines
8.2 KiB
Python
"""
|
|
Export Service — CSV, Excel, and PDF generation for transactions and reports.
|
|
|
|
Memory-efficient exports:
|
|
- CSV: streaming generator (rows written one at a time, never all in memory)
|
|
- Excel: openpyxl write-only mode + DB yield_per(500) avoids loading the full
|
|
result set into Python at once
|
|
"""
|
|
|
|
import io
|
|
import csv
|
|
from datetime import date
|
|
from flask import current_app
|
|
from app.models.transaction import Transaction
|
|
|
|
|
|
# ── CSV export ────────────────────────────────────────────────────────────────
|
|
|
|
def transactions_csv_stream(query):
|
|
"""
|
|
Generator that yields CSV text one row at a time.
|
|
Pass the SQLAlchemy *query* (not a list) — rows are fetched in 500-row batches.
|
|
Use with Flask's stream_with_context() for a true streaming response.
|
|
"""
|
|
buf = io.StringIO()
|
|
writer = csv.writer(buf)
|
|
|
|
writer.writerow(['Date', 'Type', 'Description', 'Category', 'Account', 'Amount', 'Notes'])
|
|
yield buf.getvalue()
|
|
buf.seek(0); buf.truncate()
|
|
|
|
for txn in query.yield_per(500):
|
|
writer.writerow([
|
|
txn.date.strftime('%Y-%m-%d'),
|
|
txn.transaction_type,
|
|
txn.description,
|
|
txn.category.name if txn.category else '',
|
|
txn.account.name if txn.account else '',
|
|
float(txn.amount),
|
|
txn.notes or '',
|
|
])
|
|
yield buf.getvalue()
|
|
buf.seek(0); buf.truncate()
|
|
|
|
|
|
# ── Excel export ──────────────────────────────────────────────────────────────
|
|
|
|
def transactions_to_excel(query, period_label='Transactions'):
|
|
"""
|
|
Return Excel bytes built from a SQLAlchemy *query* using openpyxl write-only
|
|
mode. Rows are fetched 500 at a time so the full result set is never held in
|
|
Python memory simultaneously.
|
|
"""
|
|
from openpyxl import Workbook
|
|
from openpyxl.styles import Font, PatternFill, Alignment
|
|
from openpyxl.cell import WriteOnlyCell
|
|
from openpyxl.utils import get_column_letter
|
|
|
|
symbol = current_app.config.get('APP_CURRENCY_SYMBOL', '$')
|
|
|
|
wb = Workbook(write_only=True)
|
|
ws = wb.create_sheet(title=period_label[:31])
|
|
|
|
col_widths = [12, 10, 40, 18, 18, 16, 30]
|
|
for i, w in enumerate(col_widths, 1):
|
|
ws.column_dimensions[get_column_letter(i)].width = w
|
|
|
|
header_fill = PatternFill(start_color='0F172A', end_color='0F172A', fill_type='solid')
|
|
header_font = Font(color='F1F5F9', bold=True, size=10)
|
|
headers = ['Date', 'Type', 'Description', 'Category', 'Account',
|
|
f'Amount ({symbol})', 'Notes']
|
|
header_row = []
|
|
for h in headers:
|
|
c = WriteOnlyCell(ws, value=h)
|
|
c.font = header_font
|
|
c.fill = header_fill
|
|
header_row.append(c)
|
|
ws.append(header_row)
|
|
|
|
income_fill = PatternFill(start_color='F0FDF4', end_color='F0FDF4', fill_type='solid')
|
|
expense_fill = PatternFill(start_color='FFF7F7', end_color='FFF7F7', fill_type='solid')
|
|
row_font = Font(size=10)
|
|
amt_fmt = '#,##0.00'
|
|
|
|
running_total = 0.0
|
|
for txn in query.yield_per(500):
|
|
fill = income_fill if txn.transaction_type == 'income' else expense_fill
|
|
amt = float(txn.amount)
|
|
running_total += amt
|
|
row_vals = [
|
|
txn.date.strftime('%Y-%m-%d'),
|
|
txn.transaction_type.title(),
|
|
txn.description,
|
|
txn.category.name if txn.category else '',
|
|
txn.account.name if txn.account else '',
|
|
amt,
|
|
txn.notes or '',
|
|
]
|
|
row = []
|
|
for col_idx, value in enumerate(row_vals, 1):
|
|
c = WriteOnlyCell(ws, value=value)
|
|
c.font = row_font
|
|
c.fill = fill
|
|
if col_idx == 6:
|
|
c.number_format = amt_fmt
|
|
c.alignment = Alignment(horizontal='right')
|
|
row.append(c)
|
|
ws.append(row)
|
|
|
|
# Totals row — plain cells (write-only, no random access)
|
|
total_lbl = WriteOnlyCell(ws, value='TOTAL')
|
|
total_lbl.font = Font(bold=True, size=10)
|
|
total_val = WriteOnlyCell(ws, value=running_total)
|
|
total_val.font = Font(bold=True, size=10)
|
|
total_val.number_format = amt_fmt
|
|
total_val.alignment = Alignment(horizontal='right')
|
|
ws.append(['', '', '', '', total_lbl, total_val, ''])
|
|
|
|
output = io.BytesIO()
|
|
wb.save(output)
|
|
output.seek(0)
|
|
return output.read()
|
|
|
|
|
|
# ── PDF export ────────────────────────────────────────────────────────────────
|
|
|
|
def report_to_pdf(html_content):
|
|
"""Convert HTML string to PDF bytes using WeasyPrint."""
|
|
try:
|
|
from weasyprint import HTML, CSS
|
|
pdf = HTML(string=html_content).write_pdf()
|
|
return pdf
|
|
except Exception as e:
|
|
import logging
|
|
logging.getLogger(__name__).error(f'[export] PDF generation failed: {e}')
|
|
return None
|
|
|
|
|
|
def build_report_html(report_data, report_type, currency_symbol='$'):
|
|
"""Build a clean HTML string for PDF export."""
|
|
today = date.today().strftime('%B %d, %Y')
|
|
period = report_data.get('period', '')
|
|
income = report_data.get('income', 0)
|
|
expense = report_data.get('expense', 0)
|
|
net = report_data.get('net', 0)
|
|
savings_rate = report_data.get('savings_rate', 0)
|
|
|
|
def fmt(v):
|
|
return f"{currency_symbol}{abs(float(v)):,.2f}"
|
|
|
|
# Expense categories table rows
|
|
cat_rows = ''
|
|
total_exp = sum(c['total'] for c in report_data.get('expense_categories', []))
|
|
for cat in report_data.get('expense_categories', []):
|
|
pct = round(cat['total'] / total_exp * 100, 1) if total_exp > 0 else 0
|
|
cat_rows += f'''
|
|
<tr>
|
|
<td>{cat['name']}</td>
|
|
<td style="text-align:right">{fmt(cat['total'])}</td>
|
|
<td style="text-align:right">{pct}%</td>
|
|
</tr>'''
|
|
|
|
html = f'''<!DOCTYPE html>
|
|
<html>
|
|
<head>
|
|
<meta charset="UTF-8">
|
|
<style>
|
|
body {{ font-family: Arial, sans-serif; font-size: 12px; color: #1e293b; margin: 40px; }}
|
|
h1 {{ font-size: 22px; color: #0f172a; border-bottom: 2px solid #3b82f6; padding-bottom: 8px; }}
|
|
h2 {{ font-size: 14px; color: #475569; margin-top: 24px; }}
|
|
.summary {{ display: flex; gap: 20px; margin: 16px 0; }}
|
|
.stat {{ background: #f8fafc; border: 1px solid #e2e8f0; border-radius: 8px; padding: 12px 16px; flex: 1; }}
|
|
.stat-label {{ font-size: 10px; text-transform: uppercase; letter-spacing: .06em; color: #94a3b8; }}
|
|
.stat-value {{ font-size: 18px; font-weight: bold; margin-top: 4px; }}
|
|
.income {{ color: #10b981; }}
|
|
.expense {{ color: #ef4444; }}
|
|
.net-pos {{ color: #3b82f6; }}
|
|
.net-neg {{ color: #ef4444; }}
|
|
table {{ width: 100%; border-collapse: collapse; margin-top: 8px; }}
|
|
th {{ background: #0f172a; color: #f1f5f9; padding: 8px 10px; font-size: 10px; text-align: left; }}
|
|
td {{ padding: 7px 10px; border-bottom: 1px solid #f1f5f9; }}
|
|
tr:nth-child(even) {{ background: #f8fafc; }}
|
|
.footer {{ margin-top: 30px; font-size: 10px; color: #94a3b8; text-align: center; }}
|
|
</style>
|
|
</head>
|
|
<body>
|
|
<h1>Personal Finance Report — {period}</h1>
|
|
<p style="color:#94a3b8;font-size:11px;">Generated {today}</p>
|
|
|
|
<div class="summary">
|
|
<div class="stat">
|
|
<div class="stat-label">Income</div>
|
|
<div class="stat-value income">{fmt(income)}</div>
|
|
</div>
|
|
<div class="stat">
|
|
<div class="stat-label">Expenses</div>
|
|
<div class="stat-value expense">{fmt(expense)}</div>
|
|
</div>
|
|
<div class="stat">
|
|
<div class="stat-label">Net</div>
|
|
<div class="stat-value {'net-pos' if net >= 0 else 'net-neg'}">{fmt(net)}</div>
|
|
</div>
|
|
<div class="stat">
|
|
<div class="stat-label">Savings Rate</div>
|
|
<div class="stat-value {'net-pos' if net >= 0 else 'net-neg'}">{savings_rate}%</div>
|
|
</div>
|
|
</div>
|
|
|
|
<h2>Expense Breakdown</h2>
|
|
<table>
|
|
<thead><tr><th>Category</th><th style="text-align:right">Amount</th><th style="text-align:right">% of Total</th></tr></thead>
|
|
<tbody>{cat_rows or '<tr><td colspan="3" style="color:#94a3b8;">No expenses</td></tr>'}</tbody>
|
|
</table>
|
|
|
|
<div class="footer">PFM Personal Finance · pfm.ngodanguyen.tech</div>
|
|
</body>
|
|
</html>'''
|
|
return html
|