feat(b4w): add human-readable B4wExport repr/str output; fix comdirect account lookup and export formatting helpers
This commit is contained in:
+1
-1
@@ -10,7 +10,7 @@ class Currency(str, Enum):
|
|||||||
CHF = "CHF"
|
CHF = "CHF"
|
||||||
|
|
||||||
def __str__(self):
|
def __str__(self):
|
||||||
return self.value
|
return f"{self.value}"
|
||||||
|
|
||||||
def __repr__(self):
|
def __repr__(self):
|
||||||
return f"Currency.{self.name}"
|
return f"Currency.{self.name}"
|
||||||
|
|||||||
@@ -31,12 +31,14 @@ class EXPECT_LINE(Enum):
|
|||||||
type ComdirectExportModel = ComdirectGiro | ComdirectDepot | ComdirectCreditCard
|
type ComdirectExportModel = ComdirectGiro | ComdirectDepot | ComdirectCreditCard
|
||||||
|
|
||||||
|
|
||||||
SOURCE_COMDIRECT = BANK_ACCOUNTS.get("Comdirect")
|
SOURCE_COMDIRECT = BANK_ACCOUNTS.get("comdirect")
|
||||||
assert isinstance(SOURCE_COMDIRECT, Source)
|
assert isinstance(SOURCE_COMDIRECT, Source)
|
||||||
|
|
||||||
SOURCE_PATH = SOURCE_COMDIRECT.source_path
|
SOURCE_PATH = SOURCE_COMDIRECT.source_path
|
||||||
ACCOUNTS = SOURCE_COMDIRECT.accounts
|
ACCOUNTS = SOURCE_COMDIRECT.accounts
|
||||||
|
|
||||||
|
COMDIRECT_ACCOUNT_PARSERS = {a.name: a.model.model_validate for a in ACCOUNTS}
|
||||||
|
|
||||||
|
|
||||||
def reader_comdirect(path: Path) -> dict[str, list[ComdirectExportModel]]:
|
def reader_comdirect(path: Path) -> dict[str, list[ComdirectExportModel]]:
|
||||||
result = {}
|
result = {}
|
||||||
@@ -122,11 +124,11 @@ def upsert_comdirect_giro(
|
|||||||
|
|
||||||
def ingest_comdirect(path: Path, account_iban: str, account_currency: str):
|
def ingest_comdirect(path: Path, account_iban: str, account_currency: str):
|
||||||
assert SOURCE_COMDIRECT
|
assert SOURCE_COMDIRECT
|
||||||
data_collection = reader_comdirect(SOURCE_PATH)
|
data_collection = reader_comdirect(path)
|
||||||
for account_name, data in data_collection.items():
|
for account_name, data in data_collection.items():
|
||||||
account = SOURCE_COMDIRECT.get_account(account_name)
|
account = SOURCE_COMDIRECT.get_account(account_name)
|
||||||
assert isinstance(account, Account)
|
assert isinstance(account, Account)
|
||||||
if account_name == "Comdirect":
|
if account_name == "Girokonto":
|
||||||
db_load_function = lambda session_, row_, account_: upsert_comdirect_giro(
|
db_load_function = lambda session_, row_, account_: upsert_comdirect_giro(
|
||||||
session=session_,
|
session=session_,
|
||||||
data=row_,
|
data=row_,
|
||||||
|
|||||||
@@ -34,10 +34,27 @@ def _parse_date_german(date_str: str) -> date | None:
|
|||||||
return date(day=day, month=month, year=year)
|
return date(day=day, month=month, year=year)
|
||||||
|
|
||||||
|
|
||||||
|
def date2str(value: date | None, default="--", date_format="%Y-%m-%d") -> str:
|
||||||
|
if value:
|
||||||
|
return value.strftime(date_format)
|
||||||
|
return default
|
||||||
|
|
||||||
|
|
||||||
def _parse_decimal_german(decimal_str: str) -> Decimal:
|
def _parse_decimal_german(decimal_str: str) -> Decimal:
|
||||||
return Decimal(decimal_str.replace(".", "").replace(",", "."))
|
return Decimal(decimal_str.replace(".", "").replace(",", "."))
|
||||||
|
|
||||||
|
|
||||||
|
def decimal2str(value: Decimal) -> str:
|
||||||
|
value = Decimal(f"{value:.2f}")
|
||||||
|
int_part, _, frac = str(value).partition(".")
|
||||||
|
groups: list[str] = []
|
||||||
|
while len(int_part) > 3:
|
||||||
|
groups.insert(0, int_part[-3:])
|
||||||
|
int_part = int_part[:-3]
|
||||||
|
groups.insert(0, int_part or "0")
|
||||||
|
return f"{'.'.join(groups)},{frac or '00'}"
|
||||||
|
|
||||||
|
|
||||||
type Csv2DictRow = dict[str, str]
|
type Csv2DictRow = dict[str, str]
|
||||||
type Csv2Dict = list[Csv2DictRow]
|
type Csv2Dict = list[Csv2DictRow]
|
||||||
type CsV2DictMulti = dict[str, Csv2Dict]
|
type CsV2DictMulti = dict[str, Csv2Dict]
|
||||||
|
|||||||
+31
-1
@@ -6,7 +6,14 @@ from sqlalchemy import Column, Date, ForeignKey, Integer, String, UniqueConstrai
|
|||||||
from sqlalchemy.orm import relationship
|
from sqlalchemy.orm import relationship
|
||||||
|
|
||||||
from src.constants import IBAN_PATTERN, Currency
|
from src.constants import IBAN_PATTERN, Currency
|
||||||
from src.models._common import BaseModelDb, ExportBaseModel, IngestdbBase, Sign, _hash
|
from src.models._common import (
|
||||||
|
BaseModelDb,
|
||||||
|
ExportBaseModel,
|
||||||
|
IngestdbBase,
|
||||||
|
Sign,
|
||||||
|
_hash,
|
||||||
|
date2str,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
class B4wExport(ExportBaseModel):
|
class B4wExport(ExportBaseModel):
|
||||||
@@ -32,6 +39,29 @@ class B4wExport(ExportBaseModel):
|
|||||||
)
|
)
|
||||||
notes: str = Field(alias="Notes")
|
notes: str = Field(alias="Notes")
|
||||||
|
|
||||||
|
def parameters_to_str(self):
|
||||||
|
return (
|
||||||
|
f"Id1='{self.id}', "
|
||||||
|
f"OwnrAcctIBAN='{self.account_iban!s}', "
|
||||||
|
f"AmtCcy='{self.amount_currency!s}', "
|
||||||
|
f"ValDt='{date2str(self.date_valid)}', "
|
||||||
|
f"BookgDt='{date2str( self.date_book )}', "
|
||||||
|
f"Amt='{self.amount!s}', "
|
||||||
|
f"OwnrAcctCcy='{self.account_currency!s}', "
|
||||||
|
f"CdtDbtInd='{self.sign!s}', "
|
||||||
|
f"RmtInf='{self.reference}', "
|
||||||
|
f"RmtdNm='{self.other_name}', "
|
||||||
|
f"RmtdAcctIBAN='{self.other_iban}', "
|
||||||
|
f"Category='{self.category.to_str() if self.category else ''}', "
|
||||||
|
f"Notes='{self.notes}'"
|
||||||
|
)
|
||||||
|
|
||||||
|
def __repr__(self):
|
||||||
|
return f"B4wExport({self.parameters_to_str()})"
|
||||||
|
|
||||||
|
def __str__(self):
|
||||||
|
return f"B4wExport({self.parameters_to_str()})"
|
||||||
|
|
||||||
@computed_field
|
@computed_field
|
||||||
@property
|
@property
|
||||||
def hash_id(self) -> str:
|
def hash_id(self) -> str:
|
||||||
|
|||||||
@@ -1,3 +1,5 @@
|
|||||||
|
import re
|
||||||
|
|
||||||
import pytest
|
import pytest
|
||||||
from pydantic import ValidationError
|
from pydantic import ValidationError
|
||||||
|
|
||||||
@@ -29,3 +31,22 @@ def test_b4w_export_account_iban_to_long():
|
|||||||
def test_b4w_export_account_iban_to_short():
|
def test_b4w_export_account_iban_to_short():
|
||||||
with pytest.raises(ValidationError):
|
with pytest.raises(ValidationError):
|
||||||
_ = B4wExportFactory.build(OwnrAcctIBAN="DE123")
|
_ = B4wExportFactory.build(OwnrAcctIBAN="DE123")
|
||||||
|
|
||||||
|
|
||||||
|
def test_b4w_export_parameters_to_str():
|
||||||
|
data = B4wExportFactory.build()
|
||||||
|
params = data.parameters_to_str()
|
||||||
|
components = [s.strip() for s in re.split(r",\s*(?=\w+=)", params)]
|
||||||
|
param_names = [component.split("=")[0].strip() for component in components]
|
||||||
|
param_values = [component.split("=")[1].strip() for component in components]
|
||||||
|
assert "OwnrAcctCcy" in param_names
|
||||||
|
assert all(s.startswith("'") and s.endswith("'") for s in param_values)
|
||||||
|
|
||||||
|
|
||||||
|
def test_b4w_export_repr_round_trip():
|
||||||
|
data_batch = B4wExportFactory.batch(size=10)
|
||||||
|
assert isinstance(data_batch, list)
|
||||||
|
for data in data_batch:
|
||||||
|
data_str = str(data)
|
||||||
|
data2 = eval(data_str)
|
||||||
|
assert data == data2
|
||||||
|
|||||||
Reference in New Issue
Block a user