diff --git a/.gitignore b/.gitignore index ff8b1cb..803269d 100644 --- a/.gitignore +++ b/.gitignore @@ -14,7 +14,7 @@ wheels/ # Environment variables .env -settings.py +/settings.py # Database instances *.db diff --git a/src/ingest/b4w.py b/src/ingest/b4w.py index 6391289..0316d74 100644 --- a/src/ingest/b4w.py +++ b/src/ingest/b4w.py @@ -7,6 +7,11 @@ from src.ingest._common import _reader_raw, _upsert from src.models.b4w import B4wDb, B4wExport, Category, CategoryDb +def reader_b4w(path: Path) -> list[B4wExport]: + raw = _reader_raw(path) + return list(map(B4wExport.model_validate, raw)) + + def upsert_category( session: Session, category: Category | None, do_commit=True ) -> CategoryDb | None: @@ -65,8 +70,3 @@ def ingest_b4w(path: Path): 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 index 234f8a7..6490df9 100644 --- a/src/ingest/comdirect.py +++ b/src/ingest/comdirect.py @@ -4,14 +4,17 @@ from pathlib import Path from sqlalchemy.orm import Session +from settings import BANK_ACCOUNTS from src.db import get_session from src.ingest._common import _upsert +from src.models._common import Csv2Dict, CsV2DictMulti from src.models.comdirect import ( ComdirectCreditCard, ComdirectDepot, ComdirectGiro, ComdirectGiroDb, ) +from src.models.settings import Account, Source PATTERN_ACCOUNT_NAME = r'^"Umsätze ([^"]+)";"Zeitraum: .*$' PATTERN_HEADER = r'"([^"]+)"' @@ -25,12 +28,61 @@ class EXPECT_LINE(Enum): 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), -} +type ComdirectExportModel = ComdirectGiro | ComdirectDepot | ComdirectCreditCard + + +SOURCE_COMDIRECT = BANK_ACCOUNTS.get("Comdirect") +assert isinstance(SOURCE_COMDIRECT, Source) + +SOURCE_PATH = SOURCE_COMDIRECT.source_path +ACCOUNTS = SOURCE_COMDIRECT.accounts + + +def reader_comdirect(path: Path) -> dict[str, list[ComdirectExportModel]]: + result = {} + raw = _reader_raw_comdirect(path) + for account in ACCOUNTS: + data_raw: Csv2Dict = raw.pop(account.name) + data: list = list(map(account.model.model_validate, data_raw)) + result[account.name] = data + assert not raw + return result + + +def _reader_raw_comdirect(path: Path) -> CsV2DictMulti: + + 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 def upsert_comdirect_giro( @@ -69,62 +121,24 @@ def upsert_comdirect_giro( 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, + assert SOURCE_COMDIRECT + data_collection = reader_comdirect(SOURCE_PATH) + for account_name, data in data_collection.items(): + account = SOURCE_COMDIRECT.get_account(account_name) + assert isinstance(account, Account) + if account_name == "Comdirect": + db_load_function = lambda session_, row_, account_: upsert_comdirect_giro( + session=session_, + data=row_, + account_iban=account_.iban, + account_currency=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 + continue + with get_session() as session: + for row in data: + data_validated = account.model.model_validate(row) + assert isinstance(data_validated, account.model) + db_load_function(session, row, account) + session.commit() diff --git a/src/models/_common.py b/src/models/_common.py index 75df288..fbf530a 100644 --- a/src/models/_common.py +++ b/src/models/_common.py @@ -3,6 +3,7 @@ from datetime import date from decimal import Decimal from enum import Enum +from pydantic import BaseModel from sqlalchemy import Column, Date, Integer, Numeric, String from sqlalchemy.orm import DeclarativeBase @@ -37,6 +38,15 @@ def _parse_decimal_german(decimal_str: str) -> Decimal: return Decimal(decimal_str.replace(".", "").replace(",", ".")) +type Csv2DictRow = dict[str, str] +type Csv2Dict = list[Csv2DictRow] +type CsV2DictMulti = dict[str, Csv2Dict] + + +class ExportBaseModel(BaseModel): + pass + + class BaseModelDb(DeclarativeBase): pass diff --git a/src/models/b4w.py b/src/models/b4w.py index 37b66e7..4ecfbfc 100644 --- a/src/models/b4w.py +++ b/src/models/b4w.py @@ -1,15 +1,15 @@ from datetime import date from decimal import Decimal -from pydantic import BaseModel, Field, computed_field, field_validator +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, IngestdbBase, Sign, _hash +from src.models._common import BaseModelDb, ExportBaseModel, IngestdbBase, Sign, _hash -class B4wExport(BaseModel): +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") @@ -56,7 +56,7 @@ class B4wExport(BaseModel): return category -class Category(BaseModel): +class Category(ExportBaseModel): name: str = Field(pattern=r"[^:\r\n]+") parent: Category | None = None diff --git a/src/models/comdirect.py b/src/models/comdirect.py index a25c05d..9990484 100644 --- a/src/models/comdirect.py +++ b/src/models/comdirect.py @@ -1,10 +1,11 @@ from datetime import date from decimal import Decimal -from pydantic import BaseModel, Field, computed_field, field_validator +from pydantic import Field, computed_field, field_validator from sqlalchemy import Column, Date, String from src.models._common import ( + ExportBaseModel, IngestdbBase, _hash, _parse_date_german, @@ -12,7 +13,7 @@ from src.models._common import ( ) -class ComdirectBase(BaseModel): +class ComdirectBase(ExportBaseModel): date_val: date | None = Field( alias="Wertstellung (Valuta)", examples=["07.09.2026"] ) diff --git a/src/models/settings.py b/src/models/settings.py new file mode 100644 index 0000000..85b5449 --- /dev/null +++ b/src/models/settings.py @@ -0,0 +1,26 @@ +from pathlib import Path + +from pydantic import BaseModel, Field + +from src.constants import IBAN_PATTERN, Currency +from src.models.b4w import B4wExport +from src.models.comdirect import ComdirectCreditCard, ComdirectDepot, ComdirectGiro + + +class Account(BaseModel): + name: str + model: type[B4wExport | ComdirectGiro | ComdirectCreditCard | ComdirectDepot] + iban: str | None = Field(pattern=IBAN_PATTERN, default=None) + currency: Currency | None = None + + +class Source(BaseModel): + name: str + source_path: Path + accounts: list[Account] + + def get_account_names(self) -> list[str]: + return [a.name for a in self.accounts] + + def get_account(self, name: str) -> Account | None: + return next((a for a in self.accounts if a.name == name), None)