Add subcommand for displaying the transaction log

dibbler/lib/render_transaction_log.py

@@ -0,0 +1,115 @@
+from dibbler.lib.render_tree import render_tree
+from dibbler.models import Transaction, TransactionType
+from dibbler.models.Transaction import EXPECTED_FIELDS
+
+
+def render_transaction_log(transaction_log: list[Transaction]) -> str:
+    """
+    Renders a transaction log as a pretty, human-readable string.
+    """
+
+    aggregated_log = _aggregate_joint_transactions(transaction_log)
+
+    lines = []
+    for transaction in aggregated_log:
+        if isinstance(transaction, list):
+            inner_lines = []
+            lines.append(_render_transaction(transaction[0]))
+            for sub_transaction in transaction[1:]:
+                line = _render_transaction(sub_transaction)
+                inner_lines.append(line)
+            lines.append(inner_lines)
+        else:
+            line = _render_transaction(transaction)
+            lines.append(line)
+
+    return render_tree(lines)
+
+
+def _aggregate_joint_transactions(
+    transactions: list[Transaction],
+) -> list[Transaction | list[Transaction]]:
+    aggregated: list[Transaction | list[Transaction]] = []
+
+    i = 0
+    while i < len(transactions):
+        current = transactions[i]
+
+        # The aggregation is running backwards, so it will hit JOINT transactions first
+        if current.type_ == TransactionType.JOINT:
+            joint_transactions = [current]
+            j = i
+            while j < len(transactions):
+                j += 1
+                next_transaction = transactions[j]
+                if next_transaction.type_ == TransactionType.JOINT_BUY_PRODUCT:
+                    joint_transactions.append(next_transaction)
+                else:
+                    break
+            aggregated.append(joint_transactions)
+            i = j  # Skip processed transactions
+        elif current.type_ == TransactionType.JOINT:
+            # Empty joint transaction?
+            i += 1
+            continue
+        else:
+            aggregated.append(current)
+            i += 1
+    return aggregated
+
+
+def _render_transaction(transaction: Transaction) -> str:
+    match transaction.type_:
+        case TransactionType.ADD_PRODUCT:
+            line = f"ADD_PRODUCT({transaction.id}, {transaction.user.name}"
+            for field in EXPECTED_FIELDS[TransactionType.ADD_PRODUCT]:
+                value = getattr(transaction, field)
+                line += f", {field}={value}"
+            line += ")"
+        case TransactionType.BUY_PRODUCT:
+            line = f"BUY_PRODUCT({transaction.id}, {transaction.user.name}"
+            for field in EXPECTED_FIELDS[TransactionType.BUY_PRODUCT]:
+                value = getattr(transaction, field)
+                line += f", {field}={value}"
+            line += ")"
+        case TransactionType.ADJUST_STOCK:
+            line = f"ADJUST_STOCK({transaction.id}, {transaction.user.name}"
+            for field in EXPECTED_FIELDS[TransactionType.ADJUST_STOCK]:
+                value = getattr(transaction, field)
+                line += f", {field}={value}"
+            line += ")"
+        case TransactionType.ADJUST_PENALTY:
+            line = f"ADJUST_PENALTY({transaction.id}, {transaction.user.name}"
+            for field in EXPECTED_FIELDS[TransactionType.ADJUST_PENALTY]:
+                value = getattr(transaction, field)
+                line += f", {field}={value}"
+            line += ")"
+        case TransactionType.ADJUST_INTEREST:
+            line = f"ADJUST_INTEREST({transaction.id}, {transaction.user.name}"
+            for field in EXPECTED_FIELDS[TransactionType.ADJUST_INTEREST]:
+                value = getattr(transaction, field)
+                line += f", {field}={value}"
+            line += ")"
+        case TransactionType.ADJUST_BALANCE:
+            line = f"ADJUST_BALANCE({transaction.id}, {transaction.user.name}"
+            for field in EXPECTED_FIELDS[TransactionType.ADJUST_BALANCE]:
+                value = getattr(transaction, field)
+                line += f", {field}={value}"
+            line += ")"
+        case TransactionType.JOINT:
+            line = f"JOINT({transaction.id}, {transaction.user.name}"
+            for field in EXPECTED_FIELDS[TransactionType.JOINT]:
+                value = getattr(transaction, field)
+                line += f", {field}={value}"
+            line += ")"
+        case TransactionType.JOINT_BUY_PRODUCT:
+            line = f"JOINT_BUY_PRODUCT({transaction.id}, {transaction.user.name}"
+            for field in EXPECTED_FIELDS[TransactionType.JOINT_BUY_PRODUCT]:
+                value = getattr(transaction, field)
+                line += f", {field}={value}"
+            line += ")"
+        case _:
+            line = (
+                f"UNKNOWN[{transaction.type_}](id={transaction.id}, user_id={transaction.user_id})"
+            )
+    return line

dibbler/lib/render_tree.py

@@ -0,0 +1,115 @@
+_TREE_CHARS = {
+    "normal": {
+        "vertical": "│  ",
+        "branch": "├─ ",
+        "last": "└─ ",
+        "empty": "   ",
+    },
+    "ascii": {
+        "vertical": "|   ",
+        "branch": "|-- ",
+        "last": "`-- ",
+        "empty": "    ",
+    },
+}
+
+assert set(_TREE_CHARS["normal"].keys()) == set(_TREE_CHARS["ascii"].keys())
+assert all(len(v) == 3 for v in _TREE_CHARS["normal"].values())
+assert all(len(v) == 4 for v in _TREE_CHARS["ascii"].values())
+
+
+def render_tree(
+    tree: list[str | list],
+    ascii_only: bool = False,
+) -> str:
+    """
+    Render a tree structure as a string.
+
+    Each item in the `tree` list can be either a string (a leaf node)
+    or another list (a subtree).
+
+    When `ascii_only` is `True`, only ASCII characters are used for drawing the tree.
+
+    Example:
+
+    ```python
+        tree = [
+            "root",
+            [
+                "child1",
+                [
+                    "grandchild1",
+                    "grandchild2",
+                ],
+                "child2",
+            ],
+            "root2",
+        ]
+        print(render_tree(tree, ascii_only=False))
+    ```
+
+    Output:
+
+    ```
+    ├─ root
+    │  ├─ child1
+    │  │  ├─ grandchild1
+    │  │  └─ grandchild2
+    │  └─ child2
+    └─ root2
+    ```
+
+    Example with ASCII only:
+
+    ```python
+        print(render_tree(tree, ascii_only=True))
+    ```
+
+    Output:
+
+    ```
+    |-- root
+    |   |-- child1
+    |   |   |-- grandchild1
+    |   |   `-- grandchild2
+    |   `-- child2
+    `-- root2
+    ```
+    """
+
+    result: list[str] = []
+    for index, item in enumerate(tree):
+        is_last = index == len(tree) - 1
+        item_lines = _render_tree_line(item, is_last, ascii_only)
+        result.extend(item_lines)
+    return "\n".join(result)
+
+
+def _render_tree_line(
+    item: str | list,
+    is_last: bool,
+    ascii_only: bool,
+    prefix: str = "",
+) -> list[str]:
+    chars = _TREE_CHARS["ascii"] if ascii_only else _TREE_CHARS["normal"]
+    lines: list[str] = []
+
+    if isinstance(item, str):
+        line_prefix = chars["last"] if is_last else chars["branch"]
+        item_lines = item.splitlines()
+        for line_index, line in enumerate(item_lines):
+            if line_index == 0:
+                lines.append(f"{prefix}{line_prefix}{line}")
+            else:
+                lines.append(f"{prefix}{chars['vertical']}{line}")
+
+    elif isinstance(item, list):
+        new_prefix = prefix + (chars["empty"] if is_last else chars["vertical"])
+        for sub_index, sub_item in enumerate(item):
+            sub_is_last = sub_index == len(item) - 1
+            sub_lines = _render_tree_line(sub_item, sub_is_last, ascii_only, new_prefix)
+            lines.extend(sub_lines)
+    else:
+        raise ValueError("Item must be either a string or a list.")
+
+    return lines

dibbler/main.py

@@ -23,6 +23,7 @@ subparsers.add_parser("loop", help="Run the dibbler loop")
 subparsers.add_parser("create-db", help="Create the database")
 subparsers.add_parser("slabbedasker", help="Find out who is slabbedasker")
 subparsers.add_parser("seed-data", help="Fill with mock data")
+subparsers.add_parser("transaction-log", help="Print transaction log")
 
 
 def main():
@@ -49,6 +50,11 @@ def main():
 
         seed_test_data.main()
 
+    elif args.subcommand == "transaction-log":
+        import dibbler.subcommands.transaction_log as transaction_log
+
+        transaction_log.main()
+
 
 if __name__ == "__main__":
     main()

dibbler/subcommands/transaction_log.py

@@ -0,0 +1,16 @@
+from dibbler.db import Session
+from dibbler.queries import transaction_log
+from dibbler.lib.render_transaction_log import render_transaction_log
+
+
+def main() -> None:
+    sql_session = Session()
+
+    result = transaction_log(sql_session)
+    rendered = render_transaction_log(result)
+
+    print(rendered)
+
+
+if __name__ == "__main__":
+    main()