from datetime import date from decimal import Decimal from pydantic import ConfigDict, Field, computed_field, field_validator from sqlalchemy import Date, String from sqlalchemy.orm import Mapped, mapped_column from src.models.common import ( ExportBaseModel, IngestdbBase, _parse_date_german, _parse_decimal_german, date2str, decimal2str, hashify, ) class ComdirectBase(ExportBaseModel): model_config = ConfigDict(populate_by_name=True) date_val: date | None = Field( alias="Wertstellung (Valuta)", examples=["07.09.2026"] ) amount_eur: Decimal = Field(alias="Umsatz in EUR", examples=["32,43"]) @field_validator("date_val", mode="before") @classmethod def parse_date_val(cls, date_str: str | None) -> date | None: if date_str is None: return None return _parse_date_german(date_str) @field_validator("amount_eur", mode="before") @classmethod def parse_amount(cls, decimal_str: str) -> Decimal: return _parse_decimal_german(decimal_str) @classmethod def _date2str(cls, date_value: date | None) -> str: return date2str(date_value, default="--", date_format="%d.%m.%Y") def parameters_to_dict(self, by_name=True) -> dict[str, str]: return { "date_val" if by_name else "Wertstellung (Valuta)": self._date2str( self.date_val ), "amount_eur" if by_name else "Umsatz in EUR": decimal2str(self.amount_eur), } def get_orig_headers(self, sep: str = ";", bracket: str = '"'): return sep.join( f"{bracket}{txt}{bracket}" for txt in self.parameters_to_dict(by_name=False) ) def get_orig_row(self, sep: str = ";", bracket: str = '"'): return sep.join( f"{bracket}{txt}{bracket}" for txt in self.parameters_to_dict(by_name=False).values() ) class ComdirectGiro(ComdirectBase): # # Fields from parents: # date_val # amount_eur date_book: date | None = Field(alias="Buchungstag", examples=["07.09.2026"]) type: str = Field( alias="Vorgang", examples=["Lastschrift / Belastung", "Kartenumsatz"] ) text: str = Field(alias="Buchungstext") @computed_field # type: ignore[prop-decorator] @property def hash_id(self) -> str: return hashify(self.date_val, self.date_book, self.text) @field_validator("date_book", mode="before") @classmethod def parse_date_book(cls, date_str: str) -> date | None: return _parse_date_german(date_str) def parameters_to_dict(self, by_name: bool = True) -> dict[str, str]: parameters = dict(**super().parameters_to_dict(by_name=by_name)) parameters.update( { "date_book" if by_name else "Buchungstag": self._date2str( self.date_book ), "type" if by_name else "Vorgang": self.type, "text" if by_name else "Buchungstext": self.text, } ) return parameters class ComdirectCreditCard(ComdirectGiro): # # Fields from parents: date_book: date | None = Field(alias="Buchungstag", examples=["07.09.2026"]) date_val: date | None = Field(alias="Umsatztag", default=None) type: str = Field( alias="Vorgang", examples=["Lastschrift / Belastung", "Kartenumsatz"], default="", ) reference: str = Field(alias="Referenz", default="") text: str = Field(alias="Buchungstext", default="") def parameters_to_dict(self, by_name: bool = True) -> dict[str, str]: parameters = dict(**super().parameters_to_dict(by_name=by_name)) parameters.update( { "Umsatztag": self._date2str(self.date_val), "Referenz": self.reference, } ) return parameters class ComdirectDepot(ComdirectBase): # # Fields from parents: # amount_eur date_val: date | None = Field(alias="Geschäftstag") num: Decimal = Field(alias="Stück / Nom.") name: str = Field(alias="Bezeichnung") wkn: str = Field(alias="WKN") currency: str = Field(alias="Währung") spot_price: Decimal = Field(alias="Ausführungskurs") @field_validator("num", "spot_price", mode="before") @classmethod def parse_decimals(cls, decimal_in: str) -> Decimal: return _parse_decimal_german(decimal_in) def parameters_to_dict(self, by_name: bool = True) -> dict[str, str]: parameters = dict(**super().parameters_to_dict(by_name=by_name)) parameters.update( { "Geschäftstag": self._date2str(self.date_val), "Stück / Nom.": decimal2str(self.num), "Bezeichnung": self.name, "WKN": self.wkn, "Währung": self.currency, "Ausführungskurs": decimal2str(self.spot_price), } ) return parameters class ComdirectBaseDb(IngestdbBase): __abstract__ = True class ComdirectGiroDb(ComdirectBaseDb): __tablename__ = "comdirect_giro" type: Mapped[str | None] = mapped_column(String, nullable=True) date_book: Mapped[date | None] = mapped_column(Date, nullable=True)