157 lines
4.8 KiB
Python
157 lines
4.8 KiB
Python
from datetime import date
|
|
from decimal import Decimal
|
|
|
|
from pydantic import Field, computed_field, field_validator
|
|
from sqlalchemy import Column, Date, ForeignKey, Integer, String, UniqueConstraint
|
|
from sqlalchemy.orm import relationship
|
|
|
|
from src.constants import IBAN_PATTERN, Currency
|
|
from src.models._common import (
|
|
BaseModelDb,
|
|
ExportBaseModel,
|
|
IngestdbBase,
|
|
Sign,
|
|
_hash,
|
|
date2str,
|
|
)
|
|
|
|
|
|
class B4wExport(ExportBaseModel):
|
|
id: int = Field(alias="Id1", description="internal B4w ID")
|
|
account_iban: str = Field(alias="OwnrAcctIBAN", pattern=IBAN_PATTERN)
|
|
account_currency: Currency = Field(alias="OwnrAcctCcy")
|
|
date_valid: date = Field(alias="ValDt")
|
|
date_book: date = Field(alias="BookgDt")
|
|
amount: Decimal = Field(alias="Amt")
|
|
amount_currency: Currency = Field(alias="AmtCcy")
|
|
sign: Sign = Field(alias="CdtDbtInd")
|
|
reference: str = Field(alias="RmtInf")
|
|
other_name: str = Field(alias="RmtdNm")
|
|
other_iban: str = Field(alias="RmtdAcctIBAN")
|
|
category: Category | None = Field(
|
|
alias="Category",
|
|
examples=[
|
|
"Freizeit",
|
|
"Beruf:Gehalt:super Firma",
|
|
"Lebenshaltung:Lebensmitteleinkäuft",
|
|
"",
|
|
],
|
|
)
|
|
notes: str = Field(alias="Notes")
|
|
|
|
def parameters_to_dict(self) -> dict[str, str]:
|
|
return {
|
|
"Id1": f"{self.id}",
|
|
"OwnrAcctIBAN": f"{self.account_iban!s}",
|
|
"OwnrAcctCcy": f"{self.account_currency!s}",
|
|
"ValDt": date2str(self.date_valid),
|
|
"BookgDt": date2str(self.date_book),
|
|
"Amt": f"{self.amount!s}",
|
|
"AmtCcy": f"{self.amount_currency!s}",
|
|
"CdtDbtInd": f"{self.sign!s}",
|
|
"RmtInf": f"{self.reference}",
|
|
"RmtdNm": f"{self.other_name}",
|
|
"RmtdAcctIBAN": f"{self.other_iban}",
|
|
"Category": f"{self.category.to_str() if self.category else ''}",
|
|
"Notes": f"{self.notes}",
|
|
}
|
|
|
|
def parameters_to_str(self):
|
|
parameters = ", ".join(
|
|
f"{key}='{value}'" for key, value in self.parameters_to_dict().items()
|
|
)
|
|
return parameters
|
|
|
|
def __repr__(self):
|
|
return f"B4wExport({self.parameters_to_str()})"
|
|
|
|
def __str__(self):
|
|
return f"B4wExport({self.parameters_to_str()})"
|
|
|
|
@computed_field
|
|
@property
|
|
def hash_id(self) -> str:
|
|
return _hash(self.id)
|
|
|
|
@property
|
|
def amount_signed(self) -> Decimal:
|
|
return self.amount * self.sign.to_sign()
|
|
|
|
@field_validator("category", mode="before")
|
|
@classmethod
|
|
def category_str_is_parsable(
|
|
cls, category_in: str | Category | None
|
|
) -> Category | None:
|
|
if category_in is None or category_in == "":
|
|
return None
|
|
if isinstance(category_in, Category):
|
|
return category_in
|
|
category = Category.from_str(category_in)
|
|
if category is None or category.to_str() != category_in:
|
|
raise ValueError(f"Category string cannot be parsed: {category_in!r}")
|
|
return category
|
|
|
|
|
|
class Category(ExportBaseModel):
|
|
name: str = Field(pattern=r"[^:\r\n]+")
|
|
parent: Category | None = None
|
|
|
|
@computed_field
|
|
@property
|
|
def hash_id(self) -> str:
|
|
return _hash(self.name, self.parent)
|
|
|
|
@classmethod
|
|
def from_str(cls, s: str) -> Category | None:
|
|
if not s:
|
|
return
|
|
components = s.split(":")
|
|
name = components[-1]
|
|
rest = ":".join(components[:-1])
|
|
return cls(name=name, parent=cls.from_str(rest))
|
|
|
|
def to_str(self) -> str | None:
|
|
if not self.parent:
|
|
return self.name
|
|
else:
|
|
return f"{self.parent.to_str()}:{self.name}"
|
|
|
|
def __str__(self):
|
|
return self.to_str()
|
|
|
|
def __repr__(self):
|
|
return f'Category.from_str("{self.__str__()}")'
|
|
|
|
|
|
class CategoryDb(BaseModelDb):
|
|
__tablename__ = "category"
|
|
__table_args__ = (
|
|
UniqueConstraint("name", "parent_id", name="uq_category_name_parent"),
|
|
)
|
|
|
|
id = Column(Integer, primary_key=True, autoincrement=True)
|
|
name = Column(String, nullable=False)
|
|
parent_id = Column(Integer, ForeignKey("category.id"), nullable=True)
|
|
|
|
# Relationships
|
|
parent = relationship(
|
|
"CategoryDb", remote_side="CategoryDb.id", back_populates="children"
|
|
)
|
|
children = relationship("CategoryDb", back_populates="parent")
|
|
|
|
UniqueConstraint("name", "parent_id", name="uq_category_name_parent")
|
|
|
|
|
|
class B4wDb(IngestdbBase):
|
|
__tablename__ = "import_b4w"
|
|
|
|
id_b4w = Column(Integer, nullable=False, unique=True)
|
|
date_book = Column(Date, nullable=True)
|
|
other_name = Column(String, nullable=True)
|
|
other_iban = Column(String, nullable=True)
|
|
category_id = Column(Integer, ForeignKey("category.id"), nullable=True)
|
|
notes = Column(String, nullable=True)
|
|
|
|
# relationship
|
|
category = relationship("CategoryDb")
|