from __future__ import annotations from decimal import Decimal from app.extensions import db from app.models import ( Account, Category, CommunityAccount, CostParticipant, Entry, EntryShareRule, MonthlyIncome, NotificationPreference, User, to_decimal, ) ACCOUNT_TREE = { "gemeinschaftskonto": { "name": "Budgets", "categories": { "wohnen": ["Miete", "Zusatzmiete"], "fixkosten": [ "Internet", "Mobilfunk 1", "Mobilfunk 2", "Rechtsschutz", "Haftpflicht", "Zusatzversicherung", "Altersvorsorge", "Rundfunkbeitrag", ], "mitgliedschaften": [ "Vereinsbeitrag", "Magazin", "Lernplattform", "Streaming 1", "Streaming 2", "Streaming 3", "Cloudspeicher", "Tabler", ], "technik": ["Server", "Hosting", "Domains", "Software"], "energie": ["Strom", "Gas"], "haushalt": ["Lebensmittel", "Drogerie", "Haushaltsbedarf"], "mobilitaet": ["Auto", "Fahrrad"], "finanzen": ["Bankgebuehren", "Kreditrate 1", "Kreditrate 2"], }, }, "sparen": {"name": "Sparen", "categories": {"sparen": ["Sparziel"]}}, "urlaub": {"name": "Urlaub", "categories": {"urlaub": ["Reisebudget"]}}, "freizeit": {"name": "Freizeit", "categories": {"freizeit": ["Freizeitbudget"]}}, "persoenlich-flo": {"name": "Persönlich 1", "categories": {"persoenliche-auszahlung": ["Person 1"]}}, "persoenlich-desi": {"name": "Persönlich 2", "categories": {"persoenliche-auszahlung": ["Person 2"]}}, } COMMUNITY_ACCOUNT_DEFAULTS = [ { "name": "Hauptkonto", "slug": "hauptkonto", "account_type": "shared", "linked_account_slug": None, }, { "name": "Privatkonto 1", "slug": "privatkonto-1", "account_type": "personal", "linked_account_slug": "persoenlich-flo", }, { "name": "Privatkonto 2", "slug": "privatkonto-2", "account_type": "personal", "linked_account_slug": "persoenlich-desi", }, ] LEGACY_COMMUNITY_ACCOUNT_SLUGS = { "hauptkonto": "gemeinschaftskonto", "privatkonto-1": "flo-privat", "privatkonto-2": "desi-privat", } LEGACY_CATEGORY_SLUGS = { ("gemeinschaftskonto", "wohnen"): "miete", ("gemeinschaftskonto", "fixkosten"): "kommunikation", ("gemeinschaftskonto", "mitgliedschaften"): "abos", ("gemeinschaftskonto", "technik"): "server", ("gemeinschaftskonto", "finanzen"): "schulden", ("persoenlich-flo", "persoenliche-auszahlung"): "flo", ("persoenlich-desi", "persoenliche-auszahlung"): "desi", } LEGACY_ENTRY_NAMES = { ("persoenlich-flo", "persoenliche-auszahlung", "Person 1"): "persönliche Auszahlung", ("persoenlich-desi", "persoenliche-auszahlung", "Person 2"): "persönliche Auszahlung", } ENTRY_TARGET_CATEGORY = { "Miete": "wohnen", "Zusatzmiete": "wohnen", "Internet": "fixkosten", "Mobilfunk 1": "fixkosten", "Mobilfunk 2": "fixkosten", "Rechtsschutz": "fixkosten", "Haftpflicht": "fixkosten", "Zusatzversicherung": "fixkosten", "Altersvorsorge": "fixkosten", "Rundfunkbeitrag": "fixkosten", "Lebensmittel": "haushalt", "Drogerie": "haushalt", "Haushaltsbedarf": "haushalt", "Auto": "mobilitaet", "Fahrrad": "mobilitaet", "Vereinsbeitrag": "mitgliedschaften", "Magazin": "mitgliedschaften", "Lernplattform": "mitgliedschaften", "Streaming 1": "mitgliedschaften", "Streaming 2": "mitgliedschaften", "Streaming 3": "mitgliedschaften", "Cloudspeicher": "mitgliedschaften", "Tabler": "mitgliedschaften", "Server": "technik", "Hosting": "technik", "Domains": "technik", "Software": "technik", "Bankgebuehren": "finanzen", "Kreditrate 1": "finanzen", "Kreditrate 2": "finanzen", } EXAMPLE_ENTRY_NAMES = { entry_name for account_data in ACCOUNT_TREE.values() for entries in account_data["categories"].values() for entry_name in entries } EXAMPLE_CATEGORY_KEYS = { (account_slug, category_slug) for account_slug, account_data in ACCOUNT_TREE.items() for category_slug in account_data["categories"].keys() } def slugify(value: str) -> str: return ( value.lower() .replace(" ", "-") .replace("ö", "oe") .replace("ü", "ue") .replace("ä", "ae") .replace("/", "-") .replace("+", "plus") ) def _entry_has_user_data(entry: Entry) -> bool: for value in entry.monthly_values: if ( to_decimal(value.planned_amount) != Decimal("0.00") or value.note or value.created_by is not None or value.updated_by is not None ): return True if entry.share_rules: return True return False def _deactivate_placeholder_entries() -> None: example_entries = Entry.query.filter( Entry.name.in_(sorted(EXAMPLE_ENTRY_NAMES)), Entry.is_active.is_(True), ).all() for entry in example_entries: if _entry_has_user_data(entry): continue entry.is_active = False def _deactivate_placeholder_categories() -> None: categories = ( Category.query.join(Account) .filter(Category.is_active.is_(True), Account.is_active.is_(True)) .all() ) for category in categories: account = category.account if account is None or (account.slug, category.slug) not in EXAMPLE_CATEGORY_KEYS: continue active_entries = [entry for entry in category.entries if entry.is_active] if not active_entries: category.is_active = False category.community_account_id = None continue if all(entry.name in EXAMPLE_ENTRY_NAMES and not _entry_has_user_data(entry) for entry in active_entries): for entry in active_entries: entry.is_active = False category.is_active = False category.community_account_id = None def seed_data(include_example_entries: bool = False) -> None: # Basisdaten nur für die fachliche Grundstruktur, ohne Demo-Benutzer, # Beispiel-Personen oder vorausgefüllte Monatsdaten. community_accounts = {} for sort_order, data in enumerate(COMMUNITY_ACCOUNT_DEFAULTS, start=1): community_account = CommunityAccount.query.filter_by(slug=data["slug"]).first() if community_account is None and LEGACY_COMMUNITY_ACCOUNT_SLUGS.get(data["slug"]): community_account = CommunityAccount.query.filter_by( slug=LEGACY_COMMUNITY_ACCOUNT_SLUGS[data["slug"]] ).first() if community_account is None: community_account = CommunityAccount( name=data["name"], slug=data["slug"], account_type=data["account_type"], linked_account_slug=data["linked_account_slug"], sort_order=sort_order, is_active=True, ) db.session.add(community_account) else: community_account.name = data["name"] community_account.slug = data["slug"] community_account.account_type = data["account_type"] community_account.linked_account_slug = data["linked_account_slug"] community_account.sort_order = sort_order community_account.is_active = True community_accounts[data["slug"]] = community_account db.session.flush() sort_order = 1 account_categories = {} for account_slug, account_data in ACCOUNT_TREE.items(): account = Account.query.filter_by(slug=account_slug).first() if account is None: account = Account( name=account_data["name"], slug=account_slug, sort_order=sort_order, is_active=True, ) db.session.add(account) db.session.flush() else: account.name = account_data["name"] account.sort_order = sort_order account.is_active = True if account_slug == "freizeit": account.sort_order = 2 sort_order += 1 category_sort = 1 account_categories[account_slug] = {} if not include_example_entries: continue for category_slug, entries in account_data["categories"].items(): category = Category.query.filter_by(account_id=account.id, slug=category_slug).first() legacy_slug = LEGACY_CATEGORY_SLUGS.get((account_slug, category_slug)) if category is None and legacy_slug: category = Category.query.filter_by(account_id=account.id, slug=legacy_slug).first() if category is None: category = Category( account_id=account.id, community_account_id=( community_accounts["hauptkonto"].id if account_slug == "gemeinschaftskonto" else None ), name=category_slug.replace("-", " ").title(), slug=category_slug, sort_order=category_sort, is_active=True, ) db.session.add(category) db.session.flush() else: category.name = category_slug.replace("-", " ").title() category.slug = category_slug category.sort_order = category_sort category.is_active = True if account_slug == "gemeinschaftskonto" and category.community_account_id is None: category.community_account_id = community_accounts["hauptkonto"].id account_categories[account_slug][category_slug] = category category_sort += 1 for index, entry_name in enumerate(entries, start=1): default_amount = Decimal("0.00") if entry_name == "Miete": default_amount = Decimal("920.00") elif entry_name == "Lebensmittel": default_amount = Decimal("520.00") elif entry_name == "Streaming 1": default_amount = Decimal("17.99") elif entry_name in {"Sparziel", "Reisebudget", "Freizeitbudget", "persönliche Auszahlung"}: default_amount = Decimal("0.00") entry_slug = slugify(entry_name) entry = Entry.query.filter_by(category_id=category.id, slug=entry_slug).first() if entry is None: entry = ( Entry.query.join(Category) .filter( Category.account_id == account.id, Entry.slug == entry_slug, ) .first() ) if entry is not None: entry.category_id = category.id legacy_entry_name = LEGACY_ENTRY_NAMES.get((account_slug, category_slug, entry_name)) if entry is None and legacy_entry_name: entry = Entry.query.filter_by( category_id=category.id, slug=slugify(legacy_entry_name), ).first() if entry is None: entry = Entry( category_id=category.id, name=entry_name, slug=entry_slug, default_amount=default_amount, amount_type="fixed", benefit_scope=( "all-users" ), is_allocation_target=entry_name in { "Sparziel", "Reisebudget", "Freizeitbudget", "Person 1", "Person 2", }, sort_order=index, is_active=True, ) db.session.add(entry) db.session.flush() else: entry.name = entry_name entry.default_amount = default_amount entry.amount_type = "fixed" entry.benefit_scope = "all-users" entry.is_allocation_target = entry_name in { "Sparziel", "Reisebudget", "Freizeitbudget", "Person 1", "Person 2", } entry.sort_order = index entry.is_active = True gemeinschaft = Account.query.filter_by(slug="gemeinschaftskonto").first() if gemeinschaft: if include_example_entries: target_categories = account_categories["gemeinschaftskonto"] with db.session.no_autoflush: for category in gemeinschaft.categories: for entry in list(category.entries): target_slug = ENTRY_TARGET_CATEGORY.get(entry.name) if not target_slug: continue target_category = target_categories.get(target_slug) if not target_category: continue existing_target_entry = Entry.query.filter_by( category_id=target_category.id, slug=entry.slug, ).first() if existing_target_entry and existing_target_entry.id != entry.id: for monthly_value in entry.monthly_values: monthly_value.entry_id = existing_target_entry.id existing_rule_participants = { rule.participant_id for rule in existing_target_entry.share_rules } for rule in list(entry.share_rules): if rule.participant_id in existing_rule_participants: db.session.delete(rule) continue rule.entry_id = existing_target_entry.id db.session.delete(entry) continue entry.category_id = target_category.id entry.benefit_scope = "all-users" entry.is_allocation_target = entry.name in { "Sparziel", "Reisebudget", "Freizeitbudget", "Person 1", "Person 2", } else: _deactivate_placeholder_categories() _deactivate_placeholder_entries() for category in gemeinschaft.categories: if category.slug not in ACCOUNT_TREE["gemeinschaftskonto"]["categories"]: category.is_active = False elif category.community_account_id is None: category.community_account_id = community_accounts["hauptkonto"].id if not include_example_entries: _deactivate_placeholder_categories() _deactivate_placeholder_entries() db.session.commit() def seed_demo_data() -> None: from datetime import date from flask import current_app seed_data(include_example_entries=True) admin = User.query.filter_by(username="admin").first() if admin is None: admin = User(username="admin", display_name="Admin", email="admin@example.invalid", role="admin") admin.set_password("testpass") db.session.add(admin) editor_a = User.query.filter_by(username="mitglied1").first() if editor_a is None: editor_a = User( username="mitglied1", display_name="Person A", email="person-a@example.invalid", role="editor", ) editor_a.set_password("testpass") db.session.add(editor_a) editor_b = User.query.filter_by(username="mitglied2").first() if editor_b is None: editor_b = User( username="mitglied2", display_name="Person B", email="person-b@example.invalid", role="editor", ) editor_b.set_password("testpass") db.session.add(editor_b) db.session.flush() for user in [admin, editor_a, editor_b]: if user.notification_preference is None: db.session.add(NotificationPreference(user_id=user.id)) participants = { "Person A": CostParticipant.query.filter_by(name="Person A").first(), "Person B": CostParticipant.query.filter_by(name="Person B").first(), "Gast": CostParticipant.query.filter_by(name="Gast").first(), } if participants["Person A"] is None: participants["Person A"] = CostParticipant( name="Person A", is_app_user=True, linked_user_id=editor_a.id, is_external=False ) db.session.add(participants["Person A"]) participants["Person A"].is_app_user = True participants["Person A"].linked_user_id = editor_a.id participants["Person A"].is_external = False participants["Person A"].avatar_url = editor_a.avatar_url if participants["Person B"] is None: participants["Person B"] = CostParticipant( name="Person B", is_app_user=True, linked_user_id=editor_b.id, is_external=False ) db.session.add(participants["Person B"]) participants["Person B"].is_app_user = True participants["Person B"].linked_user_id = editor_b.id participants["Person B"].is_external = False if participants["Gast"] is None: participants["Gast"] = CostParticipant(name="Gast", is_app_user=False, is_external=True) db.session.add(participants["Gast"]) participants["Gast"].is_app_user = False participants["Gast"].is_external = True db.session.flush() shared_entry = Entry.query.join(Category).filter(Entry.name == "Streaming 1").first() if shared_entry is not None: for person in participants.values(): rule = EntryShareRule.query.filter_by( entry_id=shared_entry.id, participant_id=person.id ).first() if rule is None: db.session.add( EntryShareRule( entry_id=shared_entry.id, participant_id=person.id, share_type="equal", ) ) month_service = current_app.extensions["saldo.month_service"] month = month_service.ensure_month(date(2026, 4, 1)) while len(month.incomes) < 2: db.session.add( MonthlyIncome( month_id=month.id, label=f"Einkommen {len(month.incomes) + 1}", amount=Decimal("0.00"), sort_order=len(month.incomes) + 1, ) ) db.session.flush() if not month.incomes: db.session.flush() for income in month.incomes: if income.sort_order == 1: income.label = "Einkommen 1" income.amount = Decimal("3100.00") else: income.label = "Einkommen 2" income.amount = Decimal("2450.00") for allocation in month.allocations: if allocation.target_account.slug == "sparen": allocation.amount = Decimal("350.00") elif allocation.target_account.slug == "urlaub": allocation.amount = Decimal("180.00") elif allocation.target_account.slug == "freizeit": allocation.amount = Decimal("120.00") else: allocation.amount = Decimal("220.00") month_service.refresh_suggestions(month, "Demo-Daten initialisiert") db.session.commit()