Files
certain/backend/app/models.py
T
ScarriffleandClaude Opus 5 7fc8763d57 feat: 2FA, Passwortrichtlinie und wiederherstellbares JSON-Backup
Zwei-Faktor per TOTP mit zehn Einmal-Wiederherstellungscodes. Der
Zwischenzustand "Passwort stimmt, Code fehlt" haengt an der vorhandenen
Sitzungstabelle (pending_2fa, zehn Minuten Laufzeit) und gilt nirgends als
angemeldet; nach dem Code wird das Cookie getauscht statt weiterverwendet.
Der Wiederverwendungsschutz merkt sich das tatsaechlich getroffene
Zeitfenster, nicht die aktuelle Uhrzeit - sonst liesse sich ein Code aus dem
naechsten Fenster gleich darauf nochmal einloesen.

Passwortrichtlinie (Laenge, Zeichenklassen, MFA-Pflicht) in der neuen
globalen Tabelle app_settings, durchgesetzt beim Anlegen, Zuruecksetzen und
Aendern von Passwoertern. Bestehende Passwoerter lassen sich nicht gegen eine
neue Richtlinie pruefen - der Hash gibt nichts her. Beim Anmelden liegt das
Passwort aber kurz im Klartext vor: dort wird geprueft und bei Verstoss ein
Wechsel verlangt, mit passendem Grund fuer den Hinweistext.

Nebenbei geschlossen: must_change_password war bisher nur im Frontend
erzwungen, wer die API direkt ansprach kam vorbei. enforce_account_state
haengt jetzt am /api-Sammelrouter und laesst nur noch durch, was man zum
Erfuellen der Pflicht braucht.

Das JSON-Backup liess sich bisher grundsaetzlich nicht zurueckladen: keine
Version, Kategorien nur ueber mehrdeutige Namen, sieben Tabellen fehlten.
Daneben steht nun ein Format v2, das Beziehungen ueber lokale IDs abbildet,
je Benutzer oder systemweit. Zurueckgeladen wird nach einer Pruefansicht,
wahlweise ersetzend oder ergaenzend; vor dem Ersetzen legt der Server einen
Schnappschuss der Datenbank an. Der alte Ansichts-Export bleibt fuer
Tabellenkalkulationen bestehen und wird beim Hochladen ausdruecklich
abgewiesen, statt halb zu raten.

Frontend: wiederverwendbares Passwortfeld mit Augensymbol, animierte
Anforderungsliste, 2FA-Seite, Admin-Karte fuer die Richtlinie und Karte zum
Zurueckladen. Neue Bewegung ist in prefers-reduced-motion hinterlegt; Secret
und Codes sind markierbar, sonst verhinderte das globale user-select das
Kopieren.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-08-20 10:11:22 +02:00

566 lines
24 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""SQLAlchemy-Modelle.
Konventionen (wie in der Vorlage): ``Mapped[...] = mapped_column(...)``, Geld als
Ganzzahl in Cent/Rappen, Enums als kurze VARCHAR-Strings (``native_enum=False``,
ohne DB-CHECK) – das hält Alembic auf SQLite einfach.
Phase 0 umfasst nur Benutzer, Sitzungen und Benutzer-Einstellungen. Fachliche
Tabellen (Konten, Kategorien, Transaktionen, Budget …) kommen in späteren Phasen
hinzu und werden dann alle über ``owner_id`` je Benutzer getrennt.
"""
from __future__ import annotations
import enum
# date/time unter Alias, weil es Spalten namens "date"/"time" gibt, die die
# gleichnamigen Typen sonst in den (späten) Annotationen beschatten.
from datetime import date as date_t, datetime, time as time_t, timezone
from sqlalchemy import (
Boolean,
Date,
DateTime,
Enum as SAEnum,
ForeignKey,
Integer,
Numeric,
String,
Text,
Time,
UniqueConstraint,
)
from sqlalchemy.orm import Mapped, mapped_column, relationship
from .database import Base
def _now() -> datetime:
return datetime.now(timezone.utc)
def _enum(enum_cls: type[enum.Enum]):
"""Enum als VARCHAR ohne DB-CHECK – Alembic-freundlich auf SQLite."""
return SAEnum(enum_cls, native_enum=False, create_constraint=False, length=32)
class Role(str, enum.Enum):
admin = "admin"
user = "user"
class AccountType(str, enum.Enum):
girokonto = "girokonto"
sparkonto = "sparkonto"
bargeld = "bargeld"
kreditkarte = "kreditkarte"
depot = "depot"
sonstiges = "sonstiges"
# Kontotypen, deren Ziel-Transfer standardmäßig als budgetierte Ausgabe zählt
# (Sparen/Investment). Andere Transfers sind budgetneutral.
SAVINGS_ACCOUNT_TYPES = {AccountType.sparkonto, AccountType.depot}
class IdentifierKind(str, enum.Enum):
iban = "iban"
account_no = "account_no"
card_no = "card_no"
text = "text" # freier Erkennungs-Text im Buchungstext (Zielkonto bei Transfers)
class CategoryKind(str, enum.Enum):
income = "income"
expense = "expense"
class TxnType(str, enum.Enum):
expense = "expense"
income = "income"
transfer = "transfer"
class BudgetTreatment(str, enum.Enum):
budgeted = "budgeted"
neutral = "neutral"
class AmountMode(str, enum.Enum):
"""Wie der Betrag in der CSV steht."""
signed = "signed" # eine Spalte, mit Vorzeichen (− = Ausgabe)
debit_credit = "debit_credit" # getrennte Soll-/Haben-Spalten
abs_expense = "abs_expense" # Betrag ohne Vorzeichen, immer Ausgabe
class MatchField(str, enum.Enum):
payee = "payee"
purpose = "purpose"
any = "any" # trifft, wenn das Muster im Empfänger ODER im Zweck vorkommt
class RuleSource(str, enum.Enum):
learned = "learned" # automatisch aus bestätigten Zuordnungen gelernt
manual = "manual" # vom Nutzer als Schlagwort angelegt
class ImportStatus(str, enum.Enum):
committed = "committed"
undone = "undone"
class User(Base):
__tablename__ = "users"
id: Mapped[int] = mapped_column(Integer, primary_key=True)
username: Mapped[str] = mapped_column(String(64), unique=True, nullable=False)
password_hash: Mapped[str] = mapped_column(String(255), nullable=False)
role: Mapped[Role] = mapped_column(_enum(Role), nullable=False, default=Role.user)
display_name: Mapped[str | None] = mapped_column(String(120), nullable=True)
is_active: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True)
# Vom Admin mit Temporär-Passwort angelegt: beim ersten Login ändern lassen.
must_change_password: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
# Warum der Wechsel fällig ist: "temp" (Admin-Passwort) oder "policy"
# (Richtlinie wurde verschärft). Steuert nur den Hinweistext.
must_change_password_reason: Mapped[str | None] = mapped_column(String(16), nullable=True)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_now)
# --- Zwei-Faktor (TOTP) ---
# Das Secret liegt im Klartext. Verschlüsseln brächte hier nichts: der
# Schlüssel läge auf derselben Maschine wie die Datenbank. Wer die DB-Datei
# hat, hätte dann auch den Schlüssel. Schutz kommt aus Dateirechten.
totp_secret: Mapped[str | None] = mapped_column(String(64), nullable=True)
totp_enabled: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
totp_confirmed_at: Mapped[datetime | None] = mapped_column(
DateTime(timezone=True), nullable=True
)
# Zuletzt verbrauchtes 30-Sekunden-Zeitfenster – verhindert, dass derselbe
# Code innerhalb seiner Gültigkeit ein zweites Mal benutzt wird.
totp_last_step: Mapped[int | None] = mapped_column(Integer, nullable=True)
sessions: Mapped[list[Session]] = relationship(
back_populates="user", cascade="all, delete-orphan"
)
settings: Mapped[list[UserSetting]] = relationship(
back_populates="user", cascade="all, delete-orphan"
)
recovery_codes: Mapped[list[RecoveryCode]] = relationship(
back_populates="user", cascade="all, delete-orphan"
)
class Session(Base):
"""Server-seitige Sitzung. Im Cookie liegt nur das Klartext-Token, hier der Hash."""
__tablename__ = "sessions"
id: Mapped[int] = mapped_column(Integer, primary_key=True)
user_id: Mapped[int] = mapped_column(
ForeignKey("users.id", ondelete="CASCADE"), nullable=False, index=True
)
token_hash: Mapped[str] = mapped_column(String(64), unique=True, nullable=False, index=True)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_now)
expires_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
last_seen_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
user_agent: Mapped[str | None] = mapped_column(String(255), nullable=True)
# Passwort stimmt, der zweite Faktor fehlt noch. Solche Sitzungen gelten
# nirgends als angemeldet (siehe services/sessions.validate) und laufen
# nach wenigen Minuten ab.
pending_2fa: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
failed_2fa: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
user: Mapped[User] = relationship(back_populates="sessions")
class RecoveryCode(Base):
"""Einmal-Wiederherstellungscode für den Fall eines verlorenen Authenticators.
Gespeichert wird nur der SHA-256-Hash. Bewusst nicht Argon2: die Codes sind
zufällig und hochentrop, und pro Prüfung stünden bis zu zehn Kandidaten an.
"""
__tablename__ = "recovery_codes"
id: Mapped[int] = mapped_column(Integer, primary_key=True)
user_id: Mapped[int] = mapped_column(
ForeignKey("users.id", ondelete="CASCADE"), nullable=False, index=True
)
code_hash: Mapped[str] = mapped_column(String(64), nullable=False, index=True)
used_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_now)
user: Mapped[User] = relationship(back_populates="recovery_codes")
class AppSetting(Base):
"""Globale, vom Admin gepflegte Einstellungen – im Gegensatz zu
``user_settings`` bewusst ohne ``owner_id``. Beherbergt heute die
Passwortrichtlinie (siehe ``services/policy``)."""
__tablename__ = "app_settings"
key: Mapped[str] = mapped_column(String(64), primary_key=True)
value: Mapped[str | None] = mapped_column(String(255), nullable=True)
class UserSetting(Base):
"""Freie Schlüssel-Wert-Einstellungen je Benutzer.
Bekannte Schlüssel: ``base_currency`` (EUR|CHF), ``budget_start_day`` (int),
``locale`` (de|en), ``theme`` (system|light|dark), ``date_format``.
"""
__tablename__ = "user_settings"
user_id: Mapped[int] = mapped_column(
ForeignKey("users.id", ondelete="CASCADE"), primary_key=True
)
key: Mapped[str] = mapped_column(String(64), primary_key=True)
value: Mapped[str | None] = mapped_column(String(255), nullable=True)
user: Mapped[User] = relationship(back_populates="settings")
# ---------------------------------------------------------------------------
# Phase 1: Konten, Kategorien, Transaktionen, Wechselkurse
# Alle fachlichen Tabellen tragen owner_id und werden nur owner-gescoped gelesen.
# ---------------------------------------------------------------------------
class Account(Base):
__tablename__ = "accounts"
id: Mapped[int] = mapped_column(Integer, primary_key=True)
owner_id: Mapped[int] = mapped_column(
ForeignKey("users.id", ondelete="CASCADE"), nullable=False, index=True
)
name: Mapped[str] = mapped_column(String(120), nullable=False)
type: Mapped[AccountType] = mapped_column(_enum(AccountType), nullable=False, default=AccountType.girokonto)
currency: Mapped[str] = mapped_column(String(3), nullable=False, default="EUR")
icon: Mapped[str] = mapped_column(String(40), nullable=False, default="wallet")
color: Mapped[str] = mapped_column(String(9), nullable=False, default="#3f51b5")
opening_balance_cents: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
opening_date: Mapped[date_t | None] = mapped_column(Date, nullable=True)
is_archived: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
position: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_now)
updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_now, onupdate=_now)
identifiers: Mapped[list[AccountIdentifier]] = relationship(
back_populates="account", cascade="all, delete-orphan"
)
class AccountIdentifier(Base):
"""Erkennungsmerkmal eines Kontos (IBAN/Kontonummer/Kartennummer) für die
Transfer-Erkennung beim Import. ``value_norm`` ist die normalisierte Form."""
__tablename__ = "account_identifiers"
id: Mapped[int] = mapped_column(Integer, primary_key=True)
account_id: Mapped[int] = mapped_column(
ForeignKey("accounts.id", ondelete="CASCADE"), nullable=False, index=True
)
kind: Mapped[IdentifierKind] = mapped_column(_enum(IdentifierKind), nullable=False, default=IdentifierKind.iban)
value: Mapped[str] = mapped_column(String(64), nullable=False)
value_norm: Mapped[str] = mapped_column(String(64), nullable=False, index=True)
account: Mapped[Account] = relationship(back_populates="identifiers")
__table_args__ = (
UniqueConstraint("account_id", "kind", "value_norm", name="uq_ident_account_kind_value"),
)
class CategoryGroup(Base):
__tablename__ = "category_groups"
id: Mapped[int] = mapped_column(Integer, primary_key=True)
owner_id: Mapped[int] = mapped_column(
ForeignKey("users.id", ondelete="CASCADE"), nullable=False, index=True
)
name: Mapped[str] = mapped_column(String(120), nullable=False)
kind: Mapped[CategoryKind] = mapped_column(_enum(CategoryKind), nullable=False, default=CategoryKind.expense)
icon: Mapped[str] = mapped_column(String(40), nullable=False, default="list")
color: Mapped[str] = mapped_column(String(9), nullable=False, default="#6b7480")
position: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
is_builtin: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
categories: Mapped[list[Category]] = relationship(
back_populates="group", cascade="all, delete-orphan"
)
class Category(Base):
__tablename__ = "categories"
id: Mapped[int] = mapped_column(Integer, primary_key=True)
owner_id: Mapped[int] = mapped_column(
ForeignKey("users.id", ondelete="CASCADE"), nullable=False, index=True
)
group_id: Mapped[int] = mapped_column(
ForeignKey("category_groups.id", ondelete="CASCADE"), nullable=False, index=True
)
name: Mapped[str] = mapped_column(String(120), nullable=False)
type: Mapped[CategoryKind] = mapped_column(_enum(CategoryKind), nullable=False, default=CategoryKind.expense)
icon: Mapped[str] = mapped_column(String(40), nullable=False, default="more-horizontal")
color: Mapped[str] = mapped_column(String(9), nullable=False, default="#6b7480")
position: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
is_builtin: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
is_archived: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
group: Mapped[CategoryGroup] = relationship(back_populates="categories")
class Transaction(Base):
__tablename__ = "transactions"
id: Mapped[int] = mapped_column(Integer, primary_key=True)
owner_id: Mapped[int] = mapped_column(
ForeignKey("users.id", ondelete="CASCADE"), nullable=False, index=True
)
type: Mapped[TxnType] = mapped_column(_enum(TxnType), nullable=False)
date: Mapped[date_t] = mapped_column(Date, nullable=False, index=True)
time: Mapped[time_t | None] = mapped_column(Time, nullable=True)
account_id: Mapped[int] = mapped_column(
ForeignKey("accounts.id", ondelete="CASCADE"), nullable=False, index=True
)
to_account_id: Mapped[int | None] = mapped_column(
ForeignKey("accounts.id", ondelete="CASCADE"), nullable=True, index=True
)
# Betrag in Kontowährung (bei Transfers: abgebuchter Betrag vom Quellkonto).
amount_cents: Mapped[int] = mapped_column(Integer, nullable=False)
# Nur bei währungsübergreifenden Transfers: gutgeschriebener Betrag am Zielkonto.
to_amount_cents: Mapped[int | None] = mapped_column(Integer, nullable=True)
# Manuell gesetzter Kurs (Kontowährung je 1 EUR); überschreibt den EZB-Kurs.
fx_rate: Mapped[float | None] = mapped_column(Numeric(18, 8), nullable=True)
# Eingefrorener EUR-Wert von amount_cents zum Buchungsdatum (Pivot für Budget/Dashboard).
amount_eur_cents: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
category_id: Mapped[int | None] = mapped_column(
ForeignKey("categories.id", ondelete="SET NULL"), nullable=True, index=True
)
# Übersteuert die Standard-Regel für Sparen/Depot-Transfers (budgeted|neutral).
budget_treatment: Mapped[BudgetTreatment | None] = mapped_column(_enum(BudgetTreatment), nullable=True)
# Kategorie, gegen die ein budgetierter Transfer zählt (Sparen/Investment).
budget_category_id: Mapped[int | None] = mapped_column(
ForeignKey("categories.id", ondelete="SET NULL"), nullable=True
)
payee: Mapped[str | None] = mapped_column(String(255), nullable=True)
description: Mapped[str | None] = mapped_column(String(255), nullable=True)
notes: Mapped[str | None] = mapped_column(Text, nullable=True)
# Import (Phase 4): Herkunft für Undo und Duplikaterkennung.
import_batch_id: Mapped[int | None] = mapped_column(
ForeignKey("import_batches.id", ondelete="SET NULL"), nullable=True, index=True
)
import_hash: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_now)
updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_now, onupdate=_now)
class ExchangeRate(Base):
"""EZB-Referenzkurs: wie viele Einheiten der Währung ein Euro wert war.
EUR ist implizit 1.0 (keine Zeile). In der Praxis nur CHF. Nicht owner-gescoped
(öffentliche Referenzdaten)."""
__tablename__ = "exchange_rates"
date: Mapped[date_t] = mapped_column(Date, primary_key=True)
currency: Mapped[str] = mapped_column(String(3), primary_key=True)
rate_per_eur: Mapped[float] = mapped_column(Numeric(18, 8), nullable=False)
source: Mapped[str] = mapped_column(String(16), nullable=False, default="ecb")
# ---------------------------------------------------------------------------
# Phase 2: Budget (ein Standard-Budget als Vorlage + Monats-Overrides)
# Planned-Beträge sind in der Hauptwährung des Nutzers.
# ---------------------------------------------------------------------------
class BudgetTemplateLine(Base):
"""Standard-Budget je Kategorie (die editierbare Vorlage)."""
__tablename__ = "budget_template_lines"
id: Mapped[int] = mapped_column(Integer, primary_key=True)
owner_id: Mapped[int] = mapped_column(
ForeignKey("users.id", ondelete="CASCADE"), nullable=False, index=True
)
category_id: Mapped[int] = mapped_column(
ForeignKey("categories.id", ondelete="CASCADE"), nullable=False
)
planned_cents: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
__table_args__ = (
UniqueConstraint("owner_id", "category_id", name="uq_tpl_owner_category"),
)
class PlannedIncomeLine(Base):
"""Geplante Einnahme (z. B. Gehalt) mit optionaler Kategorie und Zielkonto."""
__tablename__ = "planned_income_lines"
id: Mapped[int] = mapped_column(Integer, primary_key=True)
owner_id: Mapped[int] = mapped_column(
ForeignKey("users.id", ondelete="CASCADE"), nullable=False, index=True
)
label: Mapped[str] = mapped_column(String(120), nullable=False)
category_id: Mapped[int | None] = mapped_column(
ForeignKey("categories.id", ondelete="SET NULL"), nullable=True
)
target_account_id: Mapped[int | None] = mapped_column(
ForeignKey("accounts.id", ondelete="SET NULL"), nullable=True
)
planned_cents: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
position: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
class BudgetPeriod(Base):
"""Ein Budget-Monat. Existiert erst, sobald der Monat angepasst wurde."""
__tablename__ = "budget_periods"
id: Mapped[int] = mapped_column(Integer, primary_key=True)
owner_id: Mapped[int] = mapped_column(
ForeignKey("users.id", ondelete="CASCADE"), nullable=False, index=True
)
label: Mapped[str] = mapped_column(String(7), nullable=False) # YYYY-MM (End-/Hauptmonat)
customized: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_now)
lines: Mapped[list[BudgetPeriodLine]] = relationship(
back_populates="period", cascade="all, delete-orphan"
)
__table_args__ = (
UniqueConstraint("owner_id", "label", name="uq_period_owner_label"),
)
class BudgetPeriodLine(Base):
"""Monatsspezifischer Budget-Wert (Snapshot-on-Customize)."""
__tablename__ = "budget_period_lines"
id: Mapped[int] = mapped_column(Integer, primary_key=True)
budget_period_id: Mapped[int] = mapped_column(
ForeignKey("budget_periods.id", ondelete="CASCADE"), nullable=False, index=True
)
category_id: Mapped[int] = mapped_column(
ForeignKey("categories.id", ondelete="CASCADE"), nullable=False
)
planned_cents: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
period: Mapped[BudgetPeriod] = relationship(back_populates="lines")
__table_args__ = (
UniqueConstraint("budget_period_id", "category_id", name="uq_periodline_period_category"),
)
# ---------------------------------------------------------------------------
# Phase 3: Dashboard-Layout (frei anordbare Karten, JSON, je Benutzer)
# ---------------------------------------------------------------------------
class DashboardLayout(Base):
__tablename__ = "dashboard_layouts"
id: Mapped[int] = mapped_column(Integer, primary_key=True)
owner_id: Mapped[int] = mapped_column(
ForeignKey("users.id", ondelete="CASCADE"), nullable=False, unique=True, index=True
)
# JSON-Array: [{i, type, x, y, w, h, props}]
layout: Mapped[str] = mapped_column(Text, nullable=False, default="[]")
updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_now, onupdate=_now)
# ---------------------------------------------------------------------------
# Phase 4: CSV-Import (Templates, Batches für Undo, lernende Kategorie-Regeln)
# ---------------------------------------------------------------------------
class ImportTemplate(Base):
__tablename__ = "import_templates"
id: Mapped[int] = mapped_column(Integer, primary_key=True)
owner_id: Mapped[int] = mapped_column(
ForeignKey("users.id", ondelete="CASCADE"), nullable=False, index=True
)
name: Mapped[str] = mapped_column(String(120), nullable=False)
delimiter: Mapped[str] = mapped_column(String(2), nullable=False, default=";")
decimal_sep: Mapped[str] = mapped_column(String(1), nullable=False, default=",")
date_format: Mapped[str] = mapped_column(String(32), nullable=False, default="%d.%m.%Y")
encoding: Mapped[str] = mapped_column(String(16), nullable=False, default="utf-8")
has_header: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True)
skip_rows: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
amount_mode: Mapped[AmountMode] = mapped_column(_enum(AmountMode), nullable=False, default=AmountMode.signed)
# JSON: Feld -> 0-basierter Spaltenindex (oder null). Felder: date, time,
# amount, debit, credit, currency, payee, purpose, counterparty_iban.
column_map: Mapped[str] = mapped_column(Text, nullable=False, default="{}")
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_now)
class ImportBatch(Base):
__tablename__ = "import_batches"
id: Mapped[int] = mapped_column(Integer, primary_key=True)
owner_id: Mapped[int] = mapped_column(
ForeignKey("users.id", ondelete="CASCADE"), nullable=False, index=True
)
account_id: Mapped[int] = mapped_column(
ForeignKey("accounts.id", ondelete="CASCADE"), nullable=False
)
template_id: Mapped[int | None] = mapped_column(
ForeignKey("import_templates.id", ondelete="SET NULL"), nullable=True
)
filename: Mapped[str] = mapped_column(String(255), nullable=False, default="")
row_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
imported_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
status: Mapped[ImportStatus] = mapped_column(_enum(ImportStatus), nullable=False, default=ImportStatus.committed)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_now)
class CategorizationRule(Base):
"""Regel: Empfänger/Zweck enthält Muster -> Kategorie.
Zwei Quellen (``source``): ``learned`` wird beim Import automatisch aus
bestätigten Zuordnungen gelernt; ``manual`` legt der Nutzer als Schlagwort je
Kategorie an. ``applied_count``/``overridden_count`` zählen, wie oft die Regel
zutraf und wie oft ihr Vorschlag wegkorrigiert wurde – daraus wird ein
manuelles Schlagwort ggf. als „stimmt vielleicht nicht" markiert."""
__tablename__ = "categorization_rules"
id: Mapped[int] = mapped_column(Integer, primary_key=True)
owner_id: Mapped[int] = mapped_column(
ForeignKey("users.id", ondelete="CASCADE"), nullable=False, index=True
)
match_field: Mapped[MatchField] = mapped_column(_enum(MatchField), nullable=False, default=MatchField.payee)
pattern: Mapped[str] = mapped_column(String(255), nullable=False) # normalisiert (lowercase)
category_id: Mapped[int | None] = mapped_column(
ForeignKey("categories.id", ondelete="CASCADE"), nullable=True
)
# Optionale Zusatzbedingung: Regel greift nur bei genau diesem (absoluten) Betrag.
amount_cents: Mapped[int | None] = mapped_column(Integer, nullable=True)
# Wenn True: passende Buchung wird zum Transfer (category_id bleibt leer).
is_transfer: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
source: Mapped[RuleSource] = mapped_column(_enum(RuleSource), nullable=False, default=RuleSource.learned)
priority: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
hits: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
applied_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
overridden_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
# „Behalten, nicht mehr fragen" – unterdrückt das Flaggen trotz Korrekturen.
confirmed: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_now)
__table_args__ = (
UniqueConstraint("owner_id", "match_field", "pattern", name="uq_rule_owner_field_pattern"),
)