b3nw
650ea2d087
All checks were successful
Build and Push Docker Image / build (push) Successful in 34s
48 lines
1.7 KiB
Python
48 lines
1.7 KiB
Python
from __future__ import annotations
|
|
|
|
import csv
|
|
import io
|
|
from dataclasses import asdict
|
|
from typing import List, Dict, Any
|
|
|
|
from ...core.models import TransactionRecord, TransactionData, AccountInfo
|
|
|
|
|
|
def parse_csv_content(csv_bytes: bytes) -> List[TransactionRecord]:
|
|
"""
|
|
Parse Schwab transaction CSV bytes into a list of TransactionRecord.
|
|
|
|
Expected headers:
|
|
Date,Action,Symbol,Description,Quantity,Price,Fees & Comm,Amount
|
|
"""
|
|
text_stream = io.StringIO(csv_bytes.decode("utf-8"))
|
|
reader = csv.DictReader(text_stream)
|
|
|
|
records: List[TransactionRecord] = []
|
|
for row in reader:
|
|
records.append(
|
|
TransactionRecord(
|
|
date=(row.get("Date") or "").strip(),
|
|
action=(row.get("Action") or "").strip(),
|
|
symbol=(row.get("Symbol") or None) or None,
|
|
description=(row.get("Description") or "").strip(),
|
|
quantity=(row.get("Quantity") or None) or None,
|
|
price=(row.get("Price") or None) or None,
|
|
fees_comm=(row.get("Fees & Comm") or None) or None,
|
|
amount=(row.get("Amount") or None) or None,
|
|
)
|
|
)
|
|
return records
|
|
|
|
|
|
def to_dicts(transaction_data: TransactionData) -> Dict[str, Any]:
|
|
"""Convert TransactionData to plain dicts for JSON output."""
|
|
return {
|
|
"account_info": asdict(transaction_data.account_info),
|
|
"transactions": [asdict(r) for r in transaction_data.transactions],
|
|
"date_range": transaction_data.date_range,
|
|
"export_date": transaction_data.export_date,
|
|
"total_transactions": transaction_data.total_transactions,
|
|
"source": transaction_data.source,
|
|
}
|