import argparse import csv import sys from collections import defaultdict from datetime import datetime, time from decimal import Decimal from sqlalchemy import select from allmende_payment_system.database import SessionLocal from allmende_payment_system.models import ( Account, Order, OrderItem, Product, ProductDetails, Transaction, User, ) def init_db(): from allmende_payment_system.database import create_tables create_tables() print("Database initialized and tables created successfully.") def import_transactions(filepath: str): """ Import CSV data from Grist into the database. This can be deleted after the initial import, it's just a one-time helper function to get the existing data into the system. """ REQUIRED_COLUMNS = {"Datum", "Typ", "Partei/Konto", "Stück", "Betrag"} with open(filepath, newline="", encoding="utf-8") as f: reader = csv.DictReader(f) if not REQUIRED_COLUMNS.issubset(reader.fieldnames): missing = REQUIRED_COLUMNS - set(reader.fieldnames) raise ValueError(f"CSV is missing required columns: {missing}") rows = [] for i, row in enumerate(reader, start=2): # start=2 to account for header row try: rows.append( { "datum": datetime.strptime( row["Datum"].strip(), "%d.%m.%y" ).date(), "typ": row["Typ"].strip(), "partei_konto": row["Partei/Konto"].strip(), "stueck": int(row["Stück"] or 0), "betrag": float(row["Betrag"].replace(",", ".").strip()), } ) except ValueError as e: raise ValueError(f"Error in row {i}: {e}") print(f"Parsed {len(rows)} rows successfully.") db = SessionLocal() meal_product = db.execute( select(Product).join(ProductDetails).where(ProductDetails.name == "Essen") ).scalar_one_or_none() if not meal_product: sys.exit( "Error: 'Essen' product not found in database. Please create it before importing." ) kids_meal_product = db.execute( select(Product).join(ProductDetails).where(ProductDetails.name == "Essen Kind") ).scalar_one_or_none() if not kids_meal_product: sys.exit( "Error: 'Essen Kind' product not found in database. Please create it before importing." ) orders_by_account_and_date = defaultdict( lambda: defaultdict(lambda: Order(user_id=1)) ) # find users associated with the account all_users = db.execute(select(User)).scalars().all() users_by_firstname = defaultdict(list) for user in all_users: first_name = user.display_name.split()[0] users_by_firstname[first_name].append(user) for i, row in enumerate(rows, start=2): transaction_type = row["typ"].lower() if transaction_type == "einkauf foodcoop": print( "Skipping 'Einkauf Foodcoop' transaction as it's not relevant for the payment system." ) continue if transaction_type not in [ "essen", "essen kind", "einzahlung", "einkauf", "auszahlung", ]: raise ValueError( f"Skipping row {i} with unknown transaction type: '{transaction_type}'" ) # find or create account for "Partei/Konto" account_name = row["partei_konto"] account = db.execute( select(Account).where(Account.name == account_name) ).scalar_one_or_none() if not account: account = Account(name=account_name) db.add(account) db.flush() for person_name in account_name.split(","): person_name = person_name.strip() if person_name in users_by_firstname: users = users_by_firstname[person_name] if len(users) > 1: print( f"Warning: Multiple users found for name '{person_name}' in row {i}. Not adding any user." ) continue user = users[0] if account not in user.accounts: print(f"Adding user {user.username} to account {account.name}") user.accounts.append(account) db.flush() if transaction_type == "essen" or transaction_type == "essen kind": if row["stueck"] == 0: # if no quantity is given, assume it's 1 for non-zero amounts quantity = row["betrag"] / (3 if transaction_type == "essen" else 1.5) else: quantity = row["stueck"] timestamp = datetime.combine(row["datum"], time(18, 30, 0)) order = orders_by_account_and_date[account.id][row["datum"]] order.items.append( OrderItem( order=order, product=( meal_product if transaction_type == "essen" else kids_meal_product ), total_amount=row["betrag"] * -1, quantity=quantity, ) ) if not order.transaction: transaction = Transaction( type="order", timestamp=timestamp, quantity=quantity, account_id=account.id, order=order, ) db.add(transaction) else: order.transaction._total_amount = Decimal(order.total_amount) elif transaction_type == "einzahlung" or transaction_type == "einkauf": transaction = Transaction( type="deposit" if transaction_type == "einzahlung" else "expense", timestamp=datetime.combine(row["datum"], time(12, 0, 0)), _total_amount=row["betrag"], account_id=account.id, ) db.add(transaction) elif transaction_type == "auszahlung": transaction = Transaction( type="withdrawal", timestamp=datetime.combine(row["datum"], time(12, 0, 0)), _total_amount=row["betrag"], account_id=account.id, ) db.add(transaction) db.flush() db.commit() def import_users(filepath: str): """ Import user data from a CSV file into the database. This can be deleted after the initial import, it's just a one-time helper function to get the existing data into the system. """ REQUIRED_COLUMNS = {"username", "firstname", "lastname"} with open(filepath, newline="", encoding="utf-8") as f: reader = csv.DictReader(f, delimiter=";") if not REQUIRED_COLUMNS.issubset(reader.fieldnames): missing = REQUIRED_COLUMNS - set(reader.fieldnames) raise ValueError(f"CSV is missing required columns: {missing}") user_data = [] for i, row in enumerate(reader, start=2): # start=2 to account for header row try: user_data.append( { "username": row["username"].strip(), "display_name": row["firstname"].strip() + " " + row["lastname"].strip(), } ) except ValueError as e: raise ValueError(f"Error in row {i}: {e}") print(f"Parsed {len(user_data)} rows successfully.") db = SessionLocal() for entry in user_data: user = db.execute( select(User).where(User.username == entry["username"]) ).scalar_one_or_none() if not user: user = User( username=entry["username"], display_name=entry["display_name"], ) db.add(user) db.flush() db.commit() def main(): parser = argparse.ArgumentParser(description="Allmende Payment System CLI") subparsers = parser.add_subparsers(dest="command", help="Available subcommands") subparsers.add_parser("init-db", help="Initialize the database and create tables") import_transactions_parser = subparsers.add_parser( "import-transactions", help="Import transaction data from a CSV file" ) import_transactions_parser.add_argument( "filepath", help="Path to the CSV file to import" ) import_users_parser = subparsers.add_parser( "import-users", help="Import user data from a CSV file" ) import_users_parser.add_argument("filepath", help="Path to the CSV file to import") args = parser.parse_args() if args.command == "init-db": init_db() elif args.command == "import-transactions": import_transactions(args.filepath) elif args.command == "import-users": import_users(args.filepath)