diff --git a/src/converters/__init__.py b/src/converters/__init__.py index e69de29..d50875a 100644 --- a/src/converters/__init__.py +++ b/src/converters/__init__.py @@ -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 * diff --git a/src/converters/wise_converter.py b/src/converters/wise_converter.py new file mode 100644 index 0000000..a2f59fd --- /dev/null +++ b/src/converters/wise_converter.py @@ -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()) +