From 69c13f93aca6fc639513f059e6fa4224247e2ea5 Mon Sep 17 00:00:00 2001 From: Mathias Schneider Date: Fri, 11 Sep 2026 14:40:38 +0200 Subject: [PATCH] heavy refactor --- .gitignore | 3 + alembic/env.py | 2 +- settings.template.py | 80 ++++++++---- src/{settings => }/constants.py | 0 src/helper_db.py | 117 ----------------- src/{settings => ingest}/__init__.py | 0 src/ingest/_common.py | 44 +++++++ src/ingest/b4w.py | 72 +++++++++++ src/ingest/comdirect.py | 130 +++++++++++++++++++ src/main.py | 38 +++--- src/models.py | 181 --------------------------- src/models/__init__.py | 7 ++ src/models/_common.py | 54 ++++++++ src/models/b4w.py | 120 ++++++++++++++++++ src/models/comdirect.py | 88 +++++++++++++ src/models_db.py | 89 ------------- src/reader.py | 80 ------------ src/settings/models.py | 11 -- tests/test_helper_db.py | 4 +- tests/test_reader.py | 13 +- 20 files changed, 599 insertions(+), 534 deletions(-) rename src/{settings => }/constants.py (100%) delete mode 100644 src/helper_db.py rename src/{settings => ingest}/__init__.py (100%) create mode 100644 src/ingest/_common.py create mode 100644 src/ingest/b4w.py create mode 100644 src/ingest/comdirect.py delete mode 100644 src/models.py create mode 100644 src/models/__init__.py create mode 100644 src/models/_common.py create mode 100644 src/models/b4w.py create mode 100644 src/models/comdirect.py delete mode 100644 src/models_db.py delete mode 100644 src/reader.py delete mode 100644 src/settings/models.py diff --git a/.gitignore b/.gitignore index 2e81ff7..ff8b1cb 100644 --- a/.gitignore +++ b/.gitignore @@ -6,6 +6,9 @@ dist/ wheels/ *.egg-info +# IDE +.idea + # Virtual environments .venv diff --git a/alembic/env.py b/alembic/env.py index c2e9aac..6dfa89b 100644 --- a/alembic/env.py +++ b/alembic/env.py @@ -6,7 +6,7 @@ from sqlalchemy import engine_from_config, pool from alembic import context from src.db import get_set_engine -from src.models_db import BaseModelDb +from src.models._common import BaseModelDb load_dotenv() # reads .env into os.environ diff --git a/settings.template.py b/settings.template.py index 10677d4..7f0e48c 100644 --- a/settings.template.py +++ b/settings.template.py @@ -1,32 +1,62 @@ +from pathlib import Path + from faker import Faker -from src.models import ComdirectCreditCard, ComdirectDepot, ComdirectGiro -from src.settings.constants import Currency -from src.settings.models import AccountInfo +from src.constants import Currency +from src.ingest.b4w import ingest_b4w, reader_b4w +from src.ingest.comdirect import reader_comdirect +from src.models import ( + ComdirectCreditCard, + ComdirectDepot, + ComdirectGiro, +) +from src.models.settings import Account, Source fake = Faker() BANK_ACCOUNTS = { - "comdirect": { - "Girokonto": AccountInfo( - iban=fake.iban(), - ingestion_function=lambda x: ComdirectGiro.model_validate(x), - currency=Currency.EUR, - ), - "Visa-Karte (Kreditkarte)": AccountInfo( - iban=fake.iban(), - ingestion_function=lambda x: ComdirectCreditCard.model_validate(x), - currency=Currency.EUR, - ), - "Tagesgeld PLUS-Konto": AccountInfo( - iban=fake.iban(), - ingestion_function=lambda x: ComdirectGiro.model_validate(x), - currency=Currency.EUR, - ), - "Depot": AccountInfo( - iban=fake.iban(), - ingestion_function=lambda x: ComdirectDepot.model_validate(x), - currency=Currency.EUR, - ), - } + "b4w": Source( + name="b4w", + source_path=Path(""), + reader_function=reader_b4w, + accounts=[ + Account( + name="b4w", + ingestion_function=ingest_b4w, + iban=fake.iban(), + currency=Currency.EUR, + ) + ], + ), + "comdirect": Source( + name="comdirect", + source_path=Path(""), + reader_function=reader_comdirect, + accounts=[ + Account( + name="Girokonto", + ingestion_function=lambda x: ComdirectGiro.model_validate(x), + iban=fake.iban(), + currency=Currency.EUR, + ), + Account( + name="Visa-Karte (Kreditkarte)", + ingestion_function=lambda x: ComdirectCreditCard.model_validate(x), + iban=fake.iban(), + currency=Currency.EUR, + ), + Account( + name="Tagesgeld PLUS-Konto", + ingestion_function=lambda x: ComdirectGiro.model_validate(x), + iban=fake.iban(), + currency=Currency.EUR, + ), + Account( + name="Depot", + ingestion_function=lambda x: ComdirectDepot.model_validate(x), + iban=fake.iban(), + currency=Currency.EUR, + ), + ], + ), } diff --git a/src/settings/constants.py b/src/constants.py similarity index 100% rename from src/settings/constants.py rename to src/constants.py diff --git a/src/helper_db.py b/src/helper_db.py deleted file mode 100644 index 2afe45a..0000000 --- a/src/helper_db.py +++ /dev/null @@ -1,117 +0,0 @@ -from typing import Any - -from sqlalchemy.orm import Session - -from src.models import B4wExport, Category, ComdirectGiro -from src.models_db import B4wDb, BaseModelDb, CategoryDb, ComdirectGiroDb - - -def _query_existing( - session: Session, DbModel: type[BaseModelDb], query_parameters: dict[str, Any] -) -> BaseModelDb | None: - query = session.query(DbModel).filter_by(**query_parameters) - return query.first() - - -def _upsert[DbModelT: BaseModelDb]( - session: Session, - DbModel: type[DbModelT], - parameters_existing: dict[str, Any], - parameters_new: dict[str, Any], - do_update: bool = False, - do_commit: bool = True, -) -> DbModelT: - existing_db_row = _query_existing(session, DbModel, parameters_existing) - - if existing_db_row: - if do_update: - for key, value in parameters_new.items(): - setattr(existing_db_row, key, value) - assert isinstance(existing_db_row, DbModel) - return existing_db_row - - new_db_row = DbModel(**parameters_new) - session.add(new_db_row) - session.flush() - if do_commit: - session.commit() - return new_db_row - - -def upsert_category( - session: Session, category: Category | None, do_commit=True -) -> CategoryDb | None: - if category is None: - return None - parameters = { - "name": category.name, - "parent": upsert_category(session, category.parent), - } - return _upsert( - session, - CategoryDb, - parameters, - parameters, - do_update=False, - do_commit=do_commit, - ) - - -def upsert_b4w( - session: Session, data: B4wExport, do_update=False, do_commit=True -) -> B4wDb | None: - category_db = upsert_category(session, data.category) - - parameters_existing = {"hash_id": data.hash_id} - - parameters_new = { - "id_b4w": data.id, - "hash_id": data.hash_id, - "account_iban": data.account_iban, - "account_currency": data.account_currency.value, - "date_valid": data.date_valid, - "date_book": data.date_book, - "amount": data.amount_signed, - "amount_currency": data.amount_currency.value, - "text": data.reference, - "other_name": data.other_name, - "other_iban": data.other_iban, - "category": category_db, - "notes": data.notes, - } - - return _upsert( - session, - B4wDb, - parameters_existing, - parameters_new, - do_update=do_update, - do_commit=do_commit, - ) - - -def upsert_comdirect_giro( - session: Session, data: ComdirectGiro, do_update=False, do_commit=True -) -> ComdirectGiroDb | None: - parameters_existing = {"hash_id": data.hash_id} - - if not data.date_val: - return - - parameters_new = { - "hash_id": data.hash_id, - "date_valid": data.date_val, - "amount": data.amount_eur, - "date_book": data.date_book, - "type": data.type, - "text": data.text, - } - - return _upsert( - session, - ComdirectGiroDb, - parameters_existing, - parameters_new, - do_update=do_update, - do_commit=do_commit, - ) diff --git a/src/settings/__init__.py b/src/ingest/__init__.py similarity index 100% rename from src/settings/__init__.py rename to src/ingest/__init__.py diff --git a/src/ingest/_common.py b/src/ingest/_common.py new file mode 100644 index 0000000..906a5b8 --- /dev/null +++ b/src/ingest/_common.py @@ -0,0 +1,44 @@ +import csv +from pathlib import Path +from typing import Any + +from sqlalchemy.orm import Session + +from src.models._common import BaseModelDb + + +def _query_existing( + session: Session, DbModel: type[BaseModelDb], query_parameters: dict[str, Any] +) -> BaseModelDb | None: + query = session.query(DbModel).filter_by(**query_parameters) + return query.first() + + +def _upsert[DbModelT: BaseModelDb]( + session: Session, + DbModel: type[DbModelT], + parameters_existing: dict[str, Any], + parameters_new: dict[str, Any], + do_update: bool = False, + do_commit: bool = True, +) -> DbModelT: + existing_db_row = _query_existing(session, DbModel, parameters_existing) + + if existing_db_row: + if do_update: + for key, value in parameters_new.items(): + setattr(existing_db_row, key, value) + assert isinstance(existing_db_row, DbModel) + return existing_db_row + + new_db_row = DbModel(**parameters_new) + session.add(new_db_row) + session.flush() + if do_commit: + session.commit() + return new_db_row + + +def _reader_raw(path: Path): + with path.open() as fid: + return list(csv.DictReader(fid)) diff --git a/src/ingest/b4w.py b/src/ingest/b4w.py new file mode 100644 index 0000000..6391289 --- /dev/null +++ b/src/ingest/b4w.py @@ -0,0 +1,72 @@ +from pathlib import Path + +from sqlalchemy.orm import Session + +from src.db import get_session +from src.ingest._common import _reader_raw, _upsert +from src.models.b4w import B4wDb, B4wExport, Category, CategoryDb + + +def upsert_category( + session: Session, category: Category | None, do_commit=True +) -> CategoryDb | None: + if category is None: + return None + parameters = { + "name": category.name, + "parent": upsert_category(session, category.parent), + } + return _upsert( + session, + CategoryDb, + parameters, + parameters, + do_update=False, + do_commit=do_commit, + ) + + +def upsert_b4w( + session: Session, data: B4wExport, do_update=False, do_commit=True +) -> B4wDb | None: + category_db = upsert_category(session, data.category) + + parameters_existing = {"hash_id": data.hash_id} + + parameters_new = { + "id_b4w": data.id, + "hash_id": data.hash_id, + "account_iban": data.account_iban, + "account_currency": data.account_currency.value, + "date_valid": data.date_valid, + "date_book": data.date_book, + "amount": data.amount_signed, + "amount_currency": data.amount_currency.value, + "text": data.reference, + "other_name": data.other_name, + "other_iban": data.other_iban, + "category": category_db, + "notes": data.notes, + } + + return _upsert( + session, + B4wDb, + parameters_existing, + parameters_new, + do_update=do_update, + do_commit=do_commit, + ) + + +def ingest_b4w(path: Path): + data = reader_b4w(path) + with get_session() as session: + for row in data: + upsert_b4w(session, row, do_commit=False) + session.commit() + + +def reader_b4w(path: Path) -> list[B4wExport]: + raw = _reader_raw(path) + return list(map(B4wExport.model_validate, raw)) diff --git a/src/ingest/comdirect.py b/src/ingest/comdirect.py new file mode 100644 index 0000000..234f8a7 --- /dev/null +++ b/src/ingest/comdirect.py @@ -0,0 +1,130 @@ +import re +from enum import Enum, auto +from pathlib import Path + +from sqlalchemy.orm import Session + +from src.db import get_session +from src.ingest._common import _upsert +from src.models.comdirect import ( + ComdirectCreditCard, + ComdirectDepot, + ComdirectGiro, + ComdirectGiroDb, +) + +PATTERN_ACCOUNT_NAME = r'^"Umsätze ([^"]+)";"Zeitraum: .*$' +PATTERN_HEADER = r'"([^"]+)"' +PATTERN_DATA = r'"([^"]+)"' + + +class EXPECT_LINE(Enum): + ACCOUNT_NAME = auto() + BLANK = auto() + HEADER = auto() + DATA = auto() + + +COMDIRECT_ACCOUNT_PARSERS = { + "Girokonto": lambda row: ComdirectGiro.model_validate(row), + "Visa-Karte (Kreditkarte)": lambda row: ComdirectCreditCard.model_validate(row), + "Tagesgeld PLUS-Konto": lambda row: ComdirectGiro.model_validate(row), + "Depot": lambda row: ComdirectDepot.model_validate(row), +} + + +def upsert_comdirect_giro( + session: Session, + data: ComdirectGiro, + account_iban: str, + account_currency: str, + do_update=False, + do_commit=True, +) -> ComdirectGiroDb | None: + parameters_existing = {"hash_id": data.hash_id} + + if not data.date_val: + return + + parameters_new = { + "hash_id": data.hash_id, + "account_iban": account_iban, + "account_currency": account_currency, + "amount_currency": account_currency, + "date_valid": data.date_val, + "amount": data.amount_eur, + "date_book": data.date_book, + "type": data.type, + "text": data.text, + } + + return _upsert( + session, + ComdirectGiroDb, + parameters_existing, + parameters_new, + do_update=do_update, + do_commit=do_commit, + ) + + +def ingest_comdirect(path: Path, account_iban: str, account_currency: str): + data = reader_comdirect(path).get("Girokonto") + assert isinstance(data, list) + with get_session() as session: + for row in data: + upsert_comdirect_giro( + session, + row, + account_iban, + account_currency, + do_commit=False, + ) + session.commit() + + +def reader_comdirect(path: Path) -> dict[str, list]: + result = {} + raw = _reader_raw_comdirect(path) + for account_name, account_data in raw.items(): + parser = COMDIRECT_ACCOUNT_PARSERS.get(account_name) + if parser is None: + continue + result[account_name] = list(map(parser, account_data)) + return result + + +def _reader_raw_comdirect(path: Path): + + with path.open(encoding="latin1") as fid: + lines = fid.readlines() + + expect_next = EXPECT_LINE.ACCOUNT_NAME + account_name = None + result = {} + for line in lines: + if expect_next == EXPECT_LINE.ACCOUNT_NAME: + account_name = re.findall(PATTERN_ACCOUNT_NAME, line) + if account_name: + account_name = account_name[0] + result[account_name] = {} + expect_next = EXPECT_LINE.BLANK + elif expect_next == EXPECT_LINE.BLANK: + if line == "\n": + expect_next = EXPECT_LINE.HEADER + elif expect_next == EXPECT_LINE.HEADER: + header = re.findall(PATTERN_HEADER, line) + if header: + result[account_name] = [] + expect_next = EXPECT_LINE.DATA + elif expect_next == EXPECT_LINE.DATA: + data = re.findall(PATTERN_DATA, line) + if data: + result[account_name].append( + {column: value for column, value in zip(header, data)} + ) + else: + expect_next = EXPECT_LINE.ACCOUNT_NAME + else: + expect_next = EXPECT_LINE.ACCOUNT_NAME + return result diff --git a/src/main.py b/src/main.py index e7a7aa1..340b7d6 100644 --- a/src/main.py +++ b/src/main.py @@ -1,26 +1,9 @@ import argparse from pathlib import Path -from src.db import get_session -from src.helper_db import upsert_b4w, upsert_comdirect_giro -from src.reader import reader_b4w, reader_comdirect - - -def ingest_b4w(path: Path): - data = reader_b4w(path) - with get_session() as session: - for row in data: - upsert_b4w(session, row, do_commit=False) - session.commit() - - -def ingest_comdirect(path: Path): - data = reader_comdirect(path).get("Girokonto") - assert isinstance(data, list) - with get_session() as session: - for row in data: - upsert_comdirect_giro(session, row, do_commit=False) - session.commit() +from src.ingest.b4w import ingest_b4w +from src.ingest.comdirect import ingest_comdirect +from src.models.settings import Account def main(): @@ -35,11 +18,18 @@ def main(): ) args = parser.parse_args() - path_b4w = Path(args.input_b4w) - path_comdirect = Path(args.input_comdirect) + if args.input_b4w: + ingest_b4w(Path(args.input_b4w)) + if args.input_comdirect: + import settings - ingest_b4w(path_b4w) - ingest_comdirect(path_comdirect) + giro_account = settings.BANK_ACCOUNTS["comdirect"].get_account("Girokonto") + assert isinstance(giro_account, Account) + ingest_comdirect( + Path(args.input_comdirect), + account_iban=giro_account.iban, + account_currency=giro_account.currency, + ) if __name__ == "__main__": diff --git a/src/models.py b/src/models.py deleted file mode 100644 index 804fc05..0000000 --- a/src/models.py +++ /dev/null @@ -1,181 +0,0 @@ -import hashlib -from datetime import date -from decimal import Decimal -from enum import Enum - -from pydantic import BaseModel, Field, computed_field, field_validator - -from src.settings.constants import IBAN_PATTERN, Currency - - -def _hash(*args): - raw = "".join(str(args)) - return hashlib.sha256(raw.encode()).hexdigest() - - -class Sign(str, Enum): - DEBIT = "DBIT" - CREDIT = "CRDT" - - def __str__(self): - return self.value - - def __repr__(self): - return f"Sign.{self.name}" - - def to_sign(self) -> int: - return -1 if self is Sign.DEBIT else 1 - - -class B4wExport(BaseModel): - 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") - - @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(BaseModel): - 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__()}")' - - -def _parse_date_german(date_str: str) -> date | None: - if date_str in ["offen", "neu", "--"]: - return None - day, month, year = list(map(int, date_str.split("."))) - return date(day=day, month=month, year=year) - - -def _parse_decimal_german(decimal_str: str) -> Decimal: - return Decimal(decimal_str.replace(".", "").replace(",", ".")) - - -class ComdirectBase(BaseModel): - 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) -> date | 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) - - -class ComdirectGiro(ComdirectBase): - 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 - @property - def hash_id(self) -> str: - return _hash(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) - - -class ComdirectCreditCard(ComdirectGiro): - # "Buchungstag";"Umsatztag";"Vorgang";"Referenz";"Buchungstext";"Umsatz in EUR"; - 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="") - amount_eur: Decimal | None = Field( - alias="Umsatz in EUR", examples=["32,43"], default=None - ) - - -class ComdirectDepot(ComdirectBase): - 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) diff --git a/src/models/__init__.py b/src/models/__init__.py new file mode 100644 index 0000000..d602c21 --- /dev/null +++ b/src/models/__init__.py @@ -0,0 +1,7 @@ +from src.models.b4w import B4wExport as B4wExport +from src.models.b4w import Category as Category +from src.models.comdirect import ComdirectCreditCard as ComdirectCreditCard +from src.models.comdirect import ComdirectDepot as ComdirectDepot +from src.models.comdirect import ComdirectGiro as ComdirectGiro +from src.models.settings import Account as Account +from src.models.settings import Source as Source diff --git a/src/models/_common.py b/src/models/_common.py new file mode 100644 index 0000000..75df288 --- /dev/null +++ b/src/models/_common.py @@ -0,0 +1,54 @@ +import hashlib +from datetime import date +from decimal import Decimal +from enum import Enum + +from sqlalchemy import Column, Date, Integer, Numeric, String +from sqlalchemy.orm import DeclarativeBase + + +def _hash(*args): + raw = "".join(str(args)) + return hashlib.sha256(raw.encode()).hexdigest() + + +class Sign(str, Enum): + DEBIT = "DBIT" + CREDIT = "CRDT" + + def __str__(self): + return self.value + + def __repr__(self): + return f"Sign.{self.name}" + + def to_sign(self) -> int: + return -1 if self is Sign.DEBIT else 1 + + +def _parse_date_german(date_str: str) -> date | None: + if date_str in ["offen", "neu", "--"]: + return None + day, month, year = list(map(int, date_str.split("."))) + return date(day=day, month=month, year=year) + + +def _parse_decimal_german(decimal_str: str) -> Decimal: + return Decimal(decimal_str.replace(".", "").replace(",", ".")) + + +class BaseModelDb(DeclarativeBase): + pass + + +class IngestdbBase(BaseModelDb): + __abstract__ = True + + id = Column(Integer, primary_key=True, autoincrement=True) + hash_id = Column(String, nullable=False) + account_iban = Column(String, nullable=False) + account_currency = Column(String(3), nullable=False) + date_valid = Column(Date, nullable=False) + amount = Column(Numeric(12, 2), nullable=False) + amount_currency = Column(String(3), nullable=False) + text = Column(String, nullable=True) diff --git a/src/models/b4w.py b/src/models/b4w.py new file mode 100644 index 0000000..37b66e7 --- /dev/null +++ b/src/models/b4w.py @@ -0,0 +1,120 @@ +from datetime import date +from decimal import Decimal + +from pydantic import BaseModel, 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, IngestdbBase, Sign, _hash + + +class B4wExport(BaseModel): + 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") + + @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(BaseModel): + 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") diff --git a/src/models/comdirect.py b/src/models/comdirect.py new file mode 100644 index 0000000..a25c05d --- /dev/null +++ b/src/models/comdirect.py @@ -0,0 +1,88 @@ +from datetime import date +from decimal import Decimal + +from pydantic import BaseModel, Field, computed_field, field_validator +from sqlalchemy import Column, Date, String + +from src.models._common import ( + IngestdbBase, + _hash, + _parse_date_german, + _parse_decimal_german, +) + + +class ComdirectBase(BaseModel): + 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) -> date | 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) + + +class ComdirectGiro(ComdirectBase): + 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 + @property + def hash_id(self) -> str: + return _hash(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) + + +class ComdirectCreditCard(ComdirectGiro): + # "Buchungstag";"Umsatztag";"Vorgang";"Referenz";"Buchungstext";"Umsatz in EUR"; + 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="") + amount_eur: Decimal | None = Field( + alias="Umsatz in EUR", examples=["32,43"], default=None + ) + + +class ComdirectDepot(ComdirectBase): + 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) + + +class ComdirectBaseDb(IngestdbBase): + __abstract__ = True + + +class ComdirectGiroDb(ComdirectBaseDb): + __tablename__ = "comdirect_giro" + + type = Column(String, nullable=True) + date_book = Column(Date, nullable=True) diff --git a/src/models_db.py b/src/models_db.py deleted file mode 100644 index 185b824..0000000 --- a/src/models_db.py +++ /dev/null @@ -1,89 +0,0 @@ -from sqlalchemy import ( - Column, - Date, - ForeignKey, - Integer, - Numeric, - String, - UniqueConstraint, -) -from sqlalchemy.orm import DeclarativeBase, relationship - -from settings import BANK_ACCOUNTS -from src.settings.models import AccountInfo - - -class BaseModelDb(DeclarativeBase): - pass - - -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 IngestdbBase(BaseModelDb): - __abstract__ = True - - id = Column(Integer, primary_key=True, autoincrement=True) - hash_id = Column(String, nullable=False) - account_iban = Column(String, nullable=False) - account_currency = Column(String(3), nullable=False) - date_valid = Column(Date, nullable=False) - amount = Column(Numeric(12, 2), nullable=False) - amount_currency = Column(String(3), nullable=False) - text = Column(String, nullable=True) - - -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") - - -COMDIRECT_INFO = BANK_ACCOUNTS.get("comdirect") -assert isinstance(COMDIRECT_INFO, dict) - -COMDIRECT_GIRO_INFO = COMDIRECT_INFO.get("Girokonto", {}) -assert isinstance(COMDIRECT_GIRO_INFO, AccountInfo) - - -class ComdirectBaseDb(IngestdbBase): - __abstract__ = True - - account_iban = Column(String, nullable=False, default=COMDIRECT_GIRO_INFO.iban) - account_currency = Column( - String(3), nullable=False, default=COMDIRECT_GIRO_INFO.currency - ) - amount_currency = Column( - String(3), nullable=False, default=COMDIRECT_GIRO_INFO.currency - ) - - -class ComdirectGiroDb(ComdirectBaseDb): - __tablename__ = "comdirect_giro" - - type = Column(String, nullable=True) - date_book = Column(Date, nullable=True) diff --git a/src/reader.py b/src/reader.py deleted file mode 100644 index c35bf33..0000000 --- a/src/reader.py +++ /dev/null @@ -1,80 +0,0 @@ -import csv -import re -from enum import Enum, auto -from pathlib import Path - -from settings import BANK_ACCOUNTS -from src.models import B4wExport, ComdirectGiro -from src.settings.models import AccountInfo - - -def _reader_raw(path: Path): - with path.open() as fid: - return list(csv.DictReader(fid)) - - -def reader_b4w(path: Path) -> list[B4wExport]: - raw = _reader_raw(path) - return list(map(B4wExport.model_validate, raw)) - - -PATTERN_ACCOUNT_NAME = r'^"Umsätze ([^"]+)";"Zeitraum: .*$' -PATTERN_HEADER = r'"([^"]+)"' -PATTERN_DATA = r'"([^"]+)"' - - -class EXPECT_LINE(Enum): - ACCOUNT_NAME = auto() - BLANK = auto() - HEADER = auto() - DATA = auto() - - -COMDIRECT_ACCOUNTS = BANK_ACCOUNTS.get("comdirect", {}) -assert isinstance(COMDIRECT_ACCOUNTS, dict) - - -def reader_comdirect(path: Path) -> dict[str, list[ComdirectGiro]]: - result = {} - raw = _reader_raw_comdirect(path) - for account_name, account_data in raw.items(): - account = COMDIRECT_ACCOUNTS.get(account_name) - assert isinstance(account, AccountInfo) - result[account_name] = list(map(account.ingestion_function, account_data)) - return result - - -def _reader_raw_comdirect(path: Path): - - with path.open(encoding="latin1") as fid: - lines = fid.readlines() - - expect_next = EXPECT_LINE.ACCOUNT_NAME - account_name = None - result = {} - for line in lines: - if expect_next == EXPECT_LINE.ACCOUNT_NAME: - account_name = re.findall(PATTERN_ACCOUNT_NAME, line) - if account_name: - account_name = account_name[0] - result[account_name] = {} - expect_next = EXPECT_LINE.BLANK - elif expect_next == EXPECT_LINE.BLANK: - if line == "\n": - expect_next = EXPECT_LINE.HEADER - elif expect_next == EXPECT_LINE.HEADER: - header = re.findall(PATTERN_HEADER, line) - if header: - result[account_name] = [] - expect_next = EXPECT_LINE.DATA - elif expect_next == EXPECT_LINE.DATA: - data = re.findall(PATTERN_DATA, line) - if data: - result[account_name].append( - {column: value for column, value in zip(header, data)} - ) - else: - expect_next = EXPECT_LINE.ACCOUNT_NAME - else: - expect_next = EXPECT_LINE.ACCOUNT_NAME - return result diff --git a/src/settings/models.py b/src/settings/models.py deleted file mode 100644 index 485b8db..0000000 --- a/src/settings/models.py +++ /dev/null @@ -1,11 +0,0 @@ -from collections.abc import Callable - -from pydantic import BaseModel, Field - -from src.settings.constants import IBAN_PATTERN, Currency - - -class AccountInfo(BaseModel): - iban: str = Field(pattern=IBAN_PATTERN) - ingestion_function: Callable - currency: Currency diff --git a/tests/test_helper_db.py b/tests/test_helper_db.py index a7bb12c..a25f97d 100644 --- a/tests/test_helper_db.py +++ b/tests/test_helper_db.py @@ -1,9 +1,9 @@ from sqlalchemy import select from sqlalchemy.orm import Session -from src.helper_db import upsert_category +from src.ingest.b4w import upsert_category from src.models import Category -from src.models_db import CategoryDb +from src.models.b4w import CategoryDb def test_upsert_category_returns_orm_object(db_session: Session): diff --git a/tests/test_reader.py b/tests/test_reader.py index 927af5c..e013f7d 100644 --- a/tests/test_reader.py +++ b/tests/test_reader.py @@ -1,18 +1,23 @@ from pathlib import Path +import pytest + +from src.ingest.comdirect import COMDIRECT_ACCOUNT_PARSERS, reader_comdirect from src.models import ComdirectCreditCard, ComdirectDepot, ComdirectGiro -from src.reader import COMDIRECT_ACCOUNTS, reader_comdirect TEST_DATA_PATH = ".local/data/comdirect/umsaetze_9773063005_20260906-2043.csv" +@pytest.mark.skipif( + not Path(TEST_DATA_PATH).exists(), + reason=f"Testdaten fehlen: {TEST_DATA_PATH}", +) def test_reader_comdirect(): test_path = Path(TEST_DATA_PATH) - assert test_path.exists() data = reader_comdirect(test_path) assert data - assert len(data) == len(COMDIRECT_ACCOUNTS) - assert set(data) == set(COMDIRECT_ACCOUNTS) + assert len(data) == len(COMDIRECT_ACCOUNT_PARSERS) + assert set(data) == set(COMDIRECT_ACCOUNT_PARSERS) assert all(isinstance(row, ComdirectGiro) for row in data["Girokonto"]) assert all( isinstance(row, ComdirectCreditCard) for row in data["Visa-Karte (Kreditkarte)"]