Add CLI interface and csv importer

2026-05-17 10:52:44 +02:00
parent ded6f42cba
commit 3f97facd7c
1 changed files with 145 additions and 0 deletions
--- a/src/allmende_payment_system/cli.py
+++ b/src/allmende_payment_system/cli.py
@@ -0,0 +1,145 @@
+import argparse
+import csv
+from datetime import datetime, time
+import sys
+
+from sqlalchemy import select
+
+from allmende_payment_system.database import SessionLocal
+from allmende_payment_system.models import (
+    Account,
+    Order,
+    OrderItem,
+    ProductDetails,
+    Transaction,
+    Order,
+    Product,
+)
+
+
+def init_db():
+    from allmende_payment_system.database import create_tables
+
+    create_tables()
+    print("Database initialized and tables created successfully.")
+
+
+def import_csv(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."
+        )
+
+    for row in rows:
+        if row["typ"].lower() not in ["essen", "essen kind"]:
+            # TODO: Handle other types
+            continue
+
+        # 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()
+
+        if row["stueck"] == 0:
+            # if no quantity is given, assume it's 1 for non-zero amounts
+            quantity = row["betrag"] / (3 if row["typ"].lower() == "essen" else 1.5)
+        else:
+            quantity = row["stueck"]
+
+        timestamp = datetime.combine(row["datum"], time(18, 30, 0))
+
+        order = Order(user_id=1)
+
+        order.items.append(
+            OrderItem(
+                order=order,
+                product=(
+                    meal_product if row["typ"].lower() == "essen" else kids_meal_product
+                ),
+                total_amount=row["betrag"],
+                quantity=quantity,
+            )
+        )
+
+        transaction = Transaction(
+            type="order",
+            timestamp=timestamp,
+            total_amount=row["betrag"],
+            quantity=quantity,
+            account_id=account.id,
+            order=order,
+        )
+        db.add(transaction)
+        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_csv_parser = subparsers.add_parser(
+        "import-csv", help="Import data from a CSV file"
+    )
+    import_csv_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-csv":
+        import_csv(args.filepath)