finanz-cockpit/src/models/b4w.py

from datetime import UTC, date, datetime
from decimal import Decimal

from pydantic import Field, computed_field, field_validator
from sqlalchemy import (
    Date,
    DateTime,
    ForeignKey,
    Index,
    Integer,
    String,
    TypeDecorator,
    text,
)
from sqlalchemy.orm import Mapped, mapped_column, 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", gt=0, decimal_places=2)
    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 | None = Field(alias="RmtdAcctIBAN", pattern=IBAN_PATTERN)
    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 or ''}",
            "Category": f"{self.category.to_str() if self.category else ''}",
            "Notes": f"{self.notes}",
        }

    @computed_field  # type: ignore[prop-decorator]
    @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("other_iban", mode="before")
    @classmethod
    def other_iban_blank_is_none(cls, iban_in: str | None) -> str | None:
        if iban_in is None or (isinstance(iban_in, str) and not iban_in.strip()):
            return None
        return iban_in

    @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  # type: ignore[prop-decorator]
    @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 None
        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 UTCDateTime(TypeDecorator[datetime]):
    """Store UTC timestamps and return timezone-aware UTC values on every dialect."""

    impl = DateTime
    cache_ok = True

    def load_dialect_impl(self, dialect):
        return dialect.type_descriptor(DateTime(timezone=True))

    def process_bind_param(self, value: datetime | None, dialect):
        if value is None:
            return None
        if value.tzinfo is None or value.utcoffset() is None:
            return value.replace(tzinfo=UTC)
        return value.astimezone(UTC)

    def process_result_value(self, value: datetime | None, dialect):
        if value is None:
            return None
        if value.tzinfo is None or value.utcoffset() is None:
            return value.replace(tzinfo=UTC)
        return value.astimezone(UTC)


class CategoryDb(BaseModelDb):
    __tablename__ = "category"
    __table_args__ = (
        Index(
            "uq_category_active_name_parent",
            "name",
            "parent_id",
            unique=True,
            sqlite_where=text("valid_to IS NULL AND parent_id IS NOT NULL"),
            postgresql_where=text("valid_to IS NULL AND parent_id IS NOT NULL"),
        ),
        Index(
            "uq_category_active_root_name",
            "name",
            unique=True,
            sqlite_where=text("valid_to IS NULL AND parent_id IS NULL"),
            postgresql_where=text("valid_to IS NULL AND parent_id IS NULL"),
        ),
    )

    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    name: Mapped[str] = mapped_column(String, nullable=False)
    parent_id: Mapped[int | None] = mapped_column(
        ForeignKey("category.id"), nullable=True
    )
    valid_from: Mapped[datetime | None] = mapped_column(
        UTCDateTime(),
        nullable=True,
        info={
            "description": "UTC start of validity; NULL means valid since the indefinite past."
        },
    )
    valid_to: Mapped[datetime | None] = mapped_column(
        UTCDateTime(),
        nullable=True,
        info={
            "description": "UTC end of validity; NULL means the version is currently valid."
        },
    )
    last_action: Mapped[str] = mapped_column(
        String,
        nullable=False,
        default="created",
        server_default="created",
        info={
            "description": "Most recent lifecycle action recorded for this category version."
        },
    )

    # Relationships
    parent = relationship(
        "CategoryDb", remote_side="CategoryDb.id", back_populates="children"
    )
    children = relationship("CategoryDb", back_populates="parent")


class B4wDb(IngestdbBase):
    __tablename__ = "import_b4w"

    id_b4w: Mapped[int] = mapped_column(Integer, nullable=False, unique=True)
    date_book: Mapped[date | None] = mapped_column(Date, nullable=True)
    other_name: Mapped[str | None] = mapped_column(String, nullable=True)
    other_iban: Mapped[str | None] = mapped_column(String, nullable=True)
    category_id: Mapped[int | None] = mapped_column(
        ForeignKey("category.id"), nullable=True
    )
    notes: Mapped[str | None] = mapped_column(String, nullable=True)

    # relationship
    category = relationship("CategoryDb")