Add basic Wise format converter
This commit is contained in:
@@ -0,0 +1,25 @@
|
||||
from typing import TYPE_CHECKING, Type
|
||||
|
||||
from base.converter import Converter
|
||||
|
||||
converters_available: list[Type[Converter]] = []
|
||||
|
||||
def converter_factory(name: str | None = None, qual_name: str | None = None) -> Type[Converter]:
|
||||
if name and not qual_name:
|
||||
converter = next((c for c in converters_available if c.converter_name == name), None)
|
||||
if converter is None:
|
||||
raise ValueError(f"No converter matching {name=}")
|
||||
elif qual_name and not name:
|
||||
converter = next((c for c in converters_available if c.qual_name() == qual_name), None)
|
||||
if converter is None:
|
||||
raise ValueError(f"No converter matching {qual_name=}")
|
||||
else:
|
||||
raise ValueError("Exactly one of name or qual_name must be provided.")
|
||||
|
||||
return converter
|
||||
|
||||
def available_converter(converter_cls):
|
||||
converters_available.append(converter_cls)
|
||||
return converter_cls
|
||||
|
||||
from .wise_converter import *
|
||||
|
||||
307
src/converters/wise_converter.py
Normal file
307
src/converters/wise_converter.py
Normal file
@@ -0,0 +1,307 @@
|
||||
import csv
|
||||
from dataclasses import dataclass
|
||||
from enum import Enum
|
||||
import datetime as dt
|
||||
from datetime import datetime
|
||||
import logging
|
||||
|
||||
from base import Converter, PartialTXN, PartialPosting, Record, Amount
|
||||
from base.converter import ConverterConfig
|
||||
from base.rules import RuleSet
|
||||
from base.transaction import TXNFlag
|
||||
|
||||
from . import available_converter
|
||||
|
||||
__all__ = [
|
||||
'WiseRecordStatus',
|
||||
'WiseRecordDirection',
|
||||
'WiseRecord',
|
||||
'WiseConfig',
|
||||
'WiseConverter',
|
||||
]
|
||||
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class WiseRecordStatus(Enum):
|
||||
COMPLETED = 'COMPLETED'
|
||||
CANCELLED = 'CANCELLED'
|
||||
|
||||
class WiseRecordDirection(Enum):
|
||||
OUT = 'OUT'
|
||||
NEUTRAL = 'NEUTRAL'
|
||||
IN = 'IN'
|
||||
|
||||
|
||||
@dataclass(kw_only=True, frozen=True)
|
||||
class WiseRecord(Record):
|
||||
id: str
|
||||
status: WiseRecordStatus
|
||||
direction: WiseRecordDirection
|
||||
created_on: datetime
|
||||
finished_on: datetime | None
|
||||
source_fee: Amount | None
|
||||
target_fee: Amount | None
|
||||
exchange_rate: float
|
||||
|
||||
_display_fields = [
|
||||
('id', "ID"),
|
||||
('status', "Status"),
|
||||
('direction', "Direction"),
|
||||
('created_on', "Created"),
|
||||
('finished_on', "Finished"),
|
||||
('from_source_net', "Source Net Amount"),
|
||||
('source_fee', "Source Fee"),
|
||||
('exchange_rate', "Exchange Rate"),
|
||||
('exchanged_amount', "Total Exchanged"),
|
||||
('target_fee', "Target Fee"),
|
||||
('source_currency', "Source Currency"),
|
||||
('target_currency', "Target Currency"),
|
||||
('source_account', "Source Account"),
|
||||
('target_account', "Target Account"),
|
||||
]
|
||||
|
||||
_match_fields = [
|
||||
'id', 'status', 'direction',
|
||||
'source_currency', 'target_currency',
|
||||
'source_account', 'target_account',
|
||||
]
|
||||
|
||||
@property
|
||||
def source_currency(self):
|
||||
return self.from_source.currency
|
||||
|
||||
@property
|
||||
def target_currency(self):
|
||||
return self.to_target.currency
|
||||
|
||||
@property
|
||||
def from_source_net(self):
|
||||
"""
|
||||
The amount that was taken from the source account not including fees.
|
||||
Use `from_source` to get the amount including fees.
|
||||
"""
|
||||
amount = self.from_source
|
||||
if self.source_fee is not None:
|
||||
amount = amount - self.source_fee
|
||||
return amount
|
||||
|
||||
@property
|
||||
def exchanged_amount(self):
|
||||
"""
|
||||
The amount that was sent to the target, in the target currency.
|
||||
Includes any target fees.
|
||||
May be used to obtain the amount exchanged to the target currency.
|
||||
"""
|
||||
amount = self.to_target
|
||||
if self.target_fee is not None:
|
||||
amount = amount + self.target_fee
|
||||
return amount
|
||||
|
||||
@classmethod
|
||||
def sample_record(cls):
|
||||
self = cls(
|
||||
date=dt.date.today(),
|
||||
source_account="John Doe",
|
||||
target_account="Jane Austen",
|
||||
from_source=Amount(314, 'GBK'),
|
||||
to_target=Amount(314, 'KBG'),
|
||||
fees=tuple(),
|
||||
raw="Raw Data",
|
||||
id="00000",
|
||||
status=WiseRecordStatus.COMPLETED,
|
||||
direction=WiseRecordDirection.IN,
|
||||
created_on=datetime.now(),
|
||||
finished_on=datetime.now(),
|
||||
source_fee=Amount(1, 'GBK'),
|
||||
target_fee=Amount(1, 'KBG'),
|
||||
exchange_rate=1,
|
||||
)
|
||||
return self
|
||||
|
||||
@classmethod
|
||||
def from_row(cls, row):
|
||||
id = row[0]
|
||||
status = row[1]
|
||||
direction = row[2]
|
||||
created_on = row[3]
|
||||
finished_on = row[3]
|
||||
source_fee_amount = float(row[5] or 0)
|
||||
source_fee_currency = row[6]
|
||||
target_fee_amount = float(row[7] or 0)
|
||||
target_fee_currency = row[8]
|
||||
source_name = row[9]
|
||||
source_amount_final = float(row[10] or 0)
|
||||
source_currency = row[11]
|
||||
target_name = row[12]
|
||||
target_amount_final = float(row[13] or 0)
|
||||
target_currency = row[14]
|
||||
exchange_rate = float(row[15] or 0)
|
||||
|
||||
wise_dt_format = '%Y-%m-%d %H:%M:%S'
|
||||
|
||||
created_on_dt = datetime.strptime(created_on, wise_dt_format)
|
||||
finished_on_dt = datetime.strptime(finished_on, wise_dt_format) if finished_on else None
|
||||
|
||||
fees = []
|
||||
|
||||
if source_fee_amount:
|
||||
source_fee = Amount(source_fee_amount, source_fee_currency)
|
||||
fees.append(source_fee)
|
||||
else:
|
||||
source_fee = None
|
||||
if target_fee_amount:
|
||||
target_fee = Amount(target_fee_amount, target_fee_currency)
|
||||
fees.append(target_fee)
|
||||
else:
|
||||
target_fee = None
|
||||
|
||||
raw = ','.join(row)
|
||||
|
||||
self = cls(
|
||||
date=created_on_dt.date(),
|
||||
source_account=source_name,
|
||||
target_account=target_name,
|
||||
from_source=Amount(source_amount_final, source_currency),
|
||||
to_target=Amount(target_amount_final, target_currency),
|
||||
fees=tuple(fees),
|
||||
raw=raw,
|
||||
id=id,
|
||||
status=WiseRecordStatus(status),
|
||||
direction=WiseRecordDirection(direction),
|
||||
created_on=created_on_dt,
|
||||
finished_on=finished_on_dt,
|
||||
source_fee=source_fee,
|
||||
target_fee=target_fee,
|
||||
exchange_rate=float(exchange_rate),
|
||||
)
|
||||
logger.debug(
|
||||
f"Converted Wise row {raw} to record {self!r}"
|
||||
)
|
||||
return self
|
||||
|
||||
|
||||
@dataclass
|
||||
class WiseConfig(ConverterConfig):
|
||||
"""
|
||||
UI NOTE:
|
||||
For batch mode, error if the fields don't exist.
|
||||
For interactive mode, prompt config creation.
|
||||
|
||||
TODO: Actual field defaults for fill-in
|
||||
"""
|
||||
asset_account: str
|
||||
fee_account: str
|
||||
|
||||
required = {
|
||||
'asset_account',
|
||||
'fee_account',
|
||||
}
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, data: dict) -> 'WiseConfig':
|
||||
if (f := next((f for f in cls.required if f not in data), None)) is not None:
|
||||
raise ValueError(f"Wise Configuration missing required field: {f}")
|
||||
return cls(**data)
|
||||
|
||||
@available_converter
|
||||
class WiseConverter(Converter[WiseRecord, WiseConfig]):
|
||||
record_type = WiseRecord
|
||||
config_type = WiseConfig
|
||||
converter_name = 'wise'
|
||||
version = '0'
|
||||
display_name = "Wise Record Converter v0"
|
||||
config_field = 'WISE'
|
||||
|
||||
def __init__(self, config: WiseConfig, **kwargs):
|
||||
self.config = config
|
||||
|
||||
def annotation(self, record: WiseRecord, partial: PartialTXN):
|
||||
...
|
||||
|
||||
def convert(self, record: WiseRecord, ruleset: RuleSet) -> PartialTXN:
|
||||
fields = {}
|
||||
|
||||
match record.direction:
|
||||
# Handle configured default accounts
|
||||
case WiseRecordDirection.OUT:
|
||||
fields['source_account'] = self.config.asset_account.format(currency=record.source_currency)
|
||||
|
||||
case WiseRecordDirection.NEUTRAL:
|
||||
fields['source_account'] = self.config.asset_account.format(currency=record.source_currency)
|
||||
fields['target_account'] = self.config.asset_account.format(currency=record.target_currency)
|
||||
|
||||
case WiseRecordDirection.IN:
|
||||
fields['target_account'] = self.config.asset_account.format(currency=record.target_currency)
|
||||
|
||||
fields |= ruleset.apply(record.match_fields())
|
||||
|
||||
args = {}
|
||||
args['date'] = record.date
|
||||
|
||||
# Convert string flag if it exists
|
||||
if 'flag' in fields:
|
||||
args['flag'] = TXNFlag(fields['flag'])
|
||||
|
||||
# Copy string fields over directly
|
||||
for name in {'payee', 'narration', 'comment', 'document', 'tags', 'links'}:
|
||||
if name in fields:
|
||||
args[name] = fields[name]
|
||||
|
||||
args['source_posting'] = PartialPosting(
|
||||
account=fields.get('source_account', None),
|
||||
amount=-record.from_source_net,
|
||||
total_cost=record.exchanged_amount if record.source_currency != record.target_currency else None,
|
||||
)
|
||||
args['target_posting'] = PartialPosting(
|
||||
account=fields.get('target_account', None),
|
||||
amount=record.to_target,
|
||||
)
|
||||
|
||||
if record.source_fee:
|
||||
args['source_fee_asset_posting'] = PartialPosting(
|
||||
account=fields.get('source_fee_asset_account', fields.get('source_account', None)),
|
||||
amount=-record.source_fee
|
||||
)
|
||||
args['source_fee_expense_posting'] = PartialPosting(
|
||||
account=fields.get('source_fee_expense_account', self.config.fee_account),
|
||||
amount=record.source_fee
|
||||
)
|
||||
|
||||
if record.target_fee:
|
||||
args['target_fee_expense_posting'] = PartialPosting(
|
||||
account=fields.get('target_fee_expense_account', self.config.fee_account),
|
||||
amount=record.target_fee
|
||||
)
|
||||
|
||||
txn = PartialTXN(**args)
|
||||
|
||||
logger.debug(
|
||||
f"Converted Wise Record {record!r} to Partial Transaction {txn!r}"
|
||||
)
|
||||
return txn
|
||||
|
||||
@classmethod
|
||||
def ingest_string(cls, data: str) -> list[WiseRecord]:
|
||||
"""
|
||||
Parse a string of raw input into a list of records.
|
||||
"""
|
||||
reader = csv.reader(data.splitlines())
|
||||
|
||||
records = []
|
||||
for row in reader:
|
||||
record = WiseRecord.from_row(row)
|
||||
if record.status is not WiseRecordStatus.COMPLETED:
|
||||
logging.info(f"Skipping record with non-complete status: {record}")
|
||||
else:
|
||||
records.append(record)
|
||||
|
||||
return records
|
||||
|
||||
@classmethod
|
||||
def ingest_file(cls, path) -> list[WiseRecord]:
|
||||
with open(path) as f:
|
||||
f.readline()
|
||||
return cls.ingest_string(f.read())
|
||||
|
||||
Reference in New Issue
Block a user