k1nq
105 lines
3.6 KiB
Python
105 lines
3.6 KiB
Python
from sqlalchemy.orm import Session
|
|
|
|
from app.categorization.models import CategorizationRule, IncomeSource, Payee, PayeeAlias
|
|
from app.errors import conflict, not_found
|
|
from app.ledger.models import Category
|
|
|
|
|
|
def resolve_category_name(category: Category) -> str | None:
|
|
"""Literal display name, or None when the name localizes via the system key.
|
|
|
|
System categories serialize as a pure ``nameKey`` (for example ``savings``),
|
|
never as a ``category.*`` prefixed string.
|
|
"""
|
|
if category.custom_name is not None:
|
|
return category.custom_name
|
|
if category.name_override is not None:
|
|
return category.name_override
|
|
return None
|
|
|
|
|
|
def resolve_category_name_key(category: Category) -> str | None:
|
|
"""Pure localization key for system categories; None for user categories."""
|
|
return category.system_key
|
|
|
|
|
|
def get_income_source(db: Session, *, space_id, income_source_id) -> IncomeSource:
|
|
obj = db.query(IncomeSource).filter_by(id=income_source_id, space_id=space_id).one_or_none()
|
|
if obj is None:
|
|
raise not_found("INCOME_SOURCE_NOT_FOUND", "Income source not found")
|
|
return obj
|
|
|
|
|
|
def ensure_income_source(db: Session, *, space_id, name: str) -> IncomeSource:
|
|
obj = db.query(IncomeSource).filter_by(space_id=space_id, name=name).one_or_none()
|
|
if obj is None:
|
|
obj = IncomeSource(space_id=space_id, name=name)
|
|
db.add(obj)
|
|
db.flush()
|
|
return obj
|
|
|
|
|
|
def get_or_create_payee(db: Session, *, space_id, name: str, alias: str | None = None) -> Payee:
|
|
payee = db.query(Payee).filter_by(space_id=space_id, name=name).one_or_none()
|
|
if payee is None:
|
|
payee = Payee(space_id=space_id, name=name)
|
|
db.add(payee)
|
|
db.flush()
|
|
if alias:
|
|
add_alias(db, space_id=space_id, payee_id=payee.id, alias=alias)
|
|
return payee
|
|
|
|
|
|
def add_alias(db: Session, *, space_id, payee_id, alias: str) -> PayeeAlias:
|
|
existing = db.query(PayeeAlias).filter_by(space_id=space_id, alias=alias).one_or_none()
|
|
if existing is not None:
|
|
if existing.payee_id != payee_id:
|
|
raise conflict(
|
|
"PAYEE_ALIAS_TAKEN",
|
|
"Alias already mapped to another payee",
|
|
)
|
|
return existing
|
|
obj = PayeeAlias(space_id=space_id, payee_id=payee_id, alias=alias)
|
|
db.add(obj)
|
|
db.flush()
|
|
return obj
|
|
|
|
|
|
def find_payee_by_description(db: Session, *, space_id, description: str) -> Payee | None:
|
|
if not description:
|
|
return None
|
|
alias = (
|
|
db.query(PayeeAlias)
|
|
.filter(PayeeAlias.space_id == space_id, PayeeAlias.alias == description)
|
|
.one_or_none()
|
|
)
|
|
if alias is not None:
|
|
return db.get(Payee, alias.payee_id)
|
|
payee = db.query(Payee).filter_by(space_id=space_id, name=description).one_or_none()
|
|
return payee
|
|
|
|
|
|
def rule_category_for_payee(db: Session, *, space_id, payee_id) -> CategorizationRule | None:
|
|
return (
|
|
db.query(CategorizationRule).filter_by(space_id=space_id, payee_id=payee_id).one_or_none()
|
|
)
|
|
|
|
|
|
def set_payee_rule(db: Session, *, space_id, payee_id, category_id, user_id) -> CategorizationRule:
|
|
payee = db.query(Payee).filter_by(id=payee_id, space_id=space_id).one_or_none()
|
|
if payee is None:
|
|
raise not_found("PAYEE_NOT_FOUND", "Payee not found")
|
|
rule = rule_category_for_payee(db, space_id=space_id, payee_id=payee_id)
|
|
if rule is None:
|
|
rule = CategorizationRule(
|
|
space_id=space_id,
|
|
payee_id=payee_id,
|
|
category_id=category_id,
|
|
created_by_user_id=user_id,
|
|
)
|
|
db.add(rule)
|
|
else:
|
|
rule.category_id = category_id
|
|
db.flush()
|
|
return rule
|