78 lines
2.6 KiB
Python
78 lines
2.6 KiB
Python
from datetime import datetime
|
|
from typing import Any, TypedDict
|
|
|
|
|
|
class ProcessedTransaction(TypedDict):
|
|
"""Transaction data extracted and transformed to desired format."""
|
|
|
|
id: str
|
|
transaction_amount_cents: int
|
|
transaction_date_timestamp: int
|
|
transaction_date_datetime: datetime
|
|
post_date_timestamp: int
|
|
post_date_datetime: datetime
|
|
transaction_description: str
|
|
|
|
|
|
class TransactionProcessor:
|
|
"""Extract and transform transaction data from WellsFargo payload objects."""
|
|
|
|
@staticmethod
|
|
def _timestamp_to_datetime(timestamp_ms: int) -> datetime:
|
|
"""Convert millisecond timestamp to datetime object."""
|
|
return datetime.fromtimestamp(timestamp_ms / 1000.0)
|
|
|
|
@staticmethod
|
|
def _amount_to_cents(amount: float) -> int:
|
|
"""Convert dollar amount to cents (integer)."""
|
|
return round(amount * 100)
|
|
|
|
def extract_transactions(
|
|
self, extracted_objects: list[dict[str, Any]]
|
|
) -> list[ProcessedTransaction]:
|
|
"""
|
|
Extract and transform transactions from extracted WellsFargo payloads.
|
|
|
|
Path: /transactions/transactionData/transactions
|
|
"""
|
|
transactions: list[ProcessedTransaction] = []
|
|
|
|
for obj in extracted_objects:
|
|
# Navigate to transactions list
|
|
txn_list = (
|
|
obj.get("transactions", {})
|
|
.get("transactionData", {})
|
|
.get("transactions", [])
|
|
)
|
|
|
|
if not isinstance(txn_list, list):
|
|
continue
|
|
|
|
for txn in txn_list:
|
|
if not isinstance(txn, dict):
|
|
continue
|
|
|
|
try:
|
|
processed: ProcessedTransaction = {
|
|
"id": txn["id"],
|
|
"transaction_amount_cents": self._amount_to_cents(
|
|
txn["transactionAmount"]
|
|
),
|
|
"transaction_date_timestamp": txn["transactionDate"],
|
|
"transaction_date_datetime": self._timestamp_to_datetime(
|
|
txn["transactionDate"]
|
|
),
|
|
"post_date_timestamp": txn["postDate"],
|
|
"post_date_datetime": self._timestamp_to_datetime(
|
|
txn["postDate"]
|
|
),
|
|
"transaction_description": " ".join(txn["transactionDescription"].split()),
|
|
}
|
|
transactions.append(processed)
|
|
except (KeyError, TypeError, ValueError):
|
|
# Skip malformed transactions
|
|
continue
|
|
|
|
return transactions
|
|
|