mathias2784/finanz-cockpit
1
0
Fork 0
Code Issues 12 Pull Requests Actions Packages Projects 1 Releases Wiki Activity

comdirect ingest: overhaul

Browse Source
This commit is contained in:
mathias2784
2026-09-12 10:13:59 +02:00
parent 69c13f93ac
commit c354523ae9
7 changed files with 125 additions and 74 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+1 -1
View File
@@ -14,7 +14,7 @@ wheels/
# Environment variables
.env
settings.py
/settings.py
# Database instances
*.db
src/ingest/b4w.py
+5 -5
View File
@@ -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))
src/ingest/comdirect.py
+76 -62
View File
@@ -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:
else:
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
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()
src/models/_common.py
+10
View File
@@ -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
src/models/b4w.py
+4 -4
View File
@@ -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
src/models/comdirect.py
+3 -2
View File
@@ -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"]
)
src/models/settings.py
+26
View File
@@ -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)