feat: gleiches Schlagwort für Einnahme UND Ausgabe erlaubt (Art je Regel)

- CategorizationRule bekommt `kind` (expense|income|transfer); Unique jetzt
  (owner, match_field, pattern, kind) statt ohne kind
- Konfliktprüfung/Anlage (add/update Keyword, Transfer-Keyword, gelernte Regeln)
  arbeiten kind-bewusst: dasselbe Wort je einmal pro Art möglich
- Migration 0009 mit Backfill (kind aus Kategorie-Art bzw. is_transfer)
- Tests angepasst: quer über Arten erlaubt, gleiche Art weiterhin Konflikt (71 grün)

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Guido
2026-08-20 16:37:35 +02:00
co-authored by Claude Opus 4.8
parent 64cbbc8cf6
commit d1cab19264
5 changed files with 84 additions and 13 deletions
+5 -1
View File
@@ -551,6 +551,10 @@ class CategorizationRule(Base):
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)
# Art der Regel: "expense" | "income" | "transfer". Erlaubt, dasselbe Wort je
# einmal für eine Ausgabe- UND eine Einnahme-Kategorie zu verwenden (das
# Matching filtert ohnehin nach Buchungsart).
kind: Mapped[str] = mapped_column(String(16), nullable=False, default="expense")
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)
@@ -561,5 +565,5 @@ class CategorizationRule(Base):
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"),
UniqueConstraint("owner_id", "match_field", "pattern", "kind", name="uq_rule_owner_field_pattern_kind"),
)
+10 -6
View File
@@ -220,11 +220,14 @@ def _conflict(db: Session, clash: CategorizationRule) -> HTTPException:
)
def _find_any_rule(db: Session, user: User, pattern: str, exclude_id: int | None = None):
def _find_any_rule(db: Session, user: User, pattern: str, kind: str, exclude_id: int | None = None):
"""Sucht ein manuelles Schlagwort gleicher Art (kind). Dasselbe Wort darf je
einmal für Ausgabe, Einnahme und Transfer existieren – Konflikt nur je Art."""
q = db.query(CategorizationRule).filter(
CategorizationRule.owner_id == user.id,
CategorizationRule.match_field == MatchField.any,
CategorizationRule.pattern == pattern,
CategorizationRule.kind == kind,
)
if exclude_id is not None:
q = q.filter(CategorizationRule.id != exclude_id)
@@ -279,7 +282,7 @@ def update_keyword(
if not pattern:
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Leeres Schlagwort")
if pattern != rule.pattern:
clash = _find_any_rule(db, user, pattern, exclude_id=rule.id)
clash = _find_any_rule(db, user, pattern, rule.kind, exclude_id=rule.id)
if clash is not None:
raise _conflict(db, clash)
rule.pattern = pattern
@@ -329,14 +332,15 @@ def add_keyword(
pattern = normalize_text(payload.word)[:255]
if not pattern:
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Leeres Schlagwort")
existing = _find_any_rule(db, user, pattern)
kind = cat.type.value # "expense" | "income" – Konflikt nur innerhalb derselben Art
existing = _find_any_rule(db, user, pattern, kind)
if existing is not None:
if not existing.is_transfer and existing.category_id == cat.id and existing.amount_cents == payload.amount_cents:
return _keyword_out(existing) # schon identisch vorhanden – unverändert zurück
raise _conflict(db, existing)
rule = CategorizationRule(
owner_id=user.id, match_field=MatchField.any, pattern=pattern,
category_id=cat.id, amount_cents=payload.amount_cents,
category_id=cat.id, amount_cents=payload.amount_cents, kind=kind,
source=RuleSource.manual, priority=10,
)
db.add(rule)
@@ -374,7 +378,7 @@ def add_transfer_keyword(
pattern = normalize_text(payload.word)[:255]
if not pattern:
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Leeres Schlagwort")
existing = _find_any_rule(db, user, pattern)
existing = _find_any_rule(db, user, pattern, "transfer")
if existing is not None:
if existing.is_transfer and existing.amount_cents == payload.amount_cents:
return _transfer_keyword_out(existing)
@@ -382,7 +386,7 @@ def add_transfer_keyword(
rule = CategorizationRule(
owner_id=user.id, match_field=MatchField.any, pattern=pattern,
category_id=None, amount_cents=payload.amount_cents, is_transfer=True,
source=RuleSource.manual, priority=10,
kind="transfer", source=RuleSource.manual, priority=10,
)
db.add(rule)
db.commit()
+8 -1
View File
@@ -405,19 +405,26 @@ def _learn_rule(db: Session, user: User, row: dict) -> None:
pattern = normalize_text(payee)[:255]
if not pattern:
return
# Art der gelernten Regel = Art der Kategorie (Einnahme/Ausgabe). Gehört zur
# Eindeutigkeit (owner, feld, wort, kind), damit dasselbe Wort je Art zählt.
ckind = db.query(Category.type).filter(
Category.id == cat, Category.owner_id == user.id
).scalar()
kind = ckind.value if ckind is not None else "expense"
existing = (
db.query(CategorizationRule)
.filter(
CategorizationRule.owner_id == user.id,
CategorizationRule.match_field == MatchField.payee,
CategorizationRule.pattern == pattern,
CategorizationRule.kind == kind,
)
.first()
)
if existing is None:
db.add(CategorizationRule(
owner_id=user.id, match_field=MatchField.payee, pattern=pattern,
category_id=cat, source=RuleSource.learned, priority=1, hits=1,
category_id=cat, kind=kind, source=RuleSource.learned, priority=1, hits=1,
))
# Sichtbar machen (Session ist autoflush=False), damit dasselbe Muster ein
# zweites Mal im selben Import die vorhandene Regel findet statt zu doppeln.
@@ -0,0 +1,48 @@
"""Regeln: Art (kind) je Regel + Unique inkl. kind.
Erlaubt, dasselbe Schlagwort einmal für eine Ausgabe- UND einmal für eine
Einnahme-Kategorie (bzw. als Transfer) zu verwenden. ``kind`` wird aus der
zugehörigen Kategorie bzw. ``is_transfer`` zurückgefüllt.
Revision ID: 0009
Revises: 0008
Create Date: 2026-08-20
"""
from alembic import op
import sqlalchemy as sa
revision = "0009"
down_revision = "0008"
branch_labels = None
depends_on = None
def upgrade() -> None:
op.add_column(
"categorization_rules",
sa.Column("kind", sa.String(length=16), nullable=False, server_default="expense"),
)
# Backfill: Transfer-Regeln -> "transfer", sonst die Art der Kategorie.
op.execute("UPDATE categorization_rules SET kind = 'transfer' WHERE is_transfer = 1")
op.execute(
"UPDATE categorization_rules SET kind = ("
" SELECT c.type FROM categories c WHERE c.id = categorization_rules.category_id"
") WHERE is_transfer = 0 AND category_id IS NOT NULL"
)
# Unique-Constraint um kind erweitern (SQLite: über batch neu aufbauen).
with op.batch_alter_table("categorization_rules") as batch:
batch.drop_constraint("uq_rule_owner_field_pattern", type_="unique")
batch.create_unique_constraint(
"uq_rule_owner_field_pattern_kind",
["owner_id", "match_field", "pattern", "kind"],
)
def downgrade() -> None:
with op.batch_alter_table("categorization_rules") as batch:
batch.drop_constraint("uq_rule_owner_field_pattern_kind", type_="unique")
batch.create_unique_constraint(
"uq_rule_owner_field_pattern",
["owner_id", "match_field", "pattern"],
)
batch.drop_column("kind")
+13 -5
View File
@@ -249,11 +249,19 @@ def test_transfer_rule_and_target_guess(admin):
assert rows[1]["to_account_id"] is None
def test_transfer_keyword_conflicts_with_category_keyword(admin):
a, _ = _expense_cats(admin, 2)
admin.post(f"/api/categories/{a['id']}/keywords", json={"word": "doppelwort"})
# gleiches Wort als Transfer-Regel -> Konflikt
assert admin.post("/api/categories/transfer-keywords", json={"word": "doppelwort"}).status_code == 409
def test_same_word_across_kinds_allowed(admin):
a, b = _expense_cats(admin, 2)
inc = _income_cat(admin)
# Dasselbe Wort für eine AUSGABE- und eine EINNAHME-Kategorie -> beides erlaubt.
r1 = admin.post(f"/api/categories/{a['id']}/keywords", json={"word": "querart"})
assert r1.status_code == 201, r1.text
r2 = admin.post(f"/api/categories/{inc['id']}/keywords", json={"word": "querart"})
assert r2.status_code == 201, r2.text
assert r1.json()["id"] != r2.json()["id"]
# ... und auch als Transfer-Regel (dritte Art) erlaubt.
assert admin.post("/api/categories/transfer-keywords", json={"word": "querart"}).status_code == 201
# Aber ein zweites Ausgabe-Wort gleicher Art bleibt ein Konflikt.
assert admin.post(f"/api/categories/{b['id']}/keywords", json={"word": "querart"}).status_code == 409
def test_no_retroactive_apply(admin):