feat(notifications): Add support for transaction filter and notificat…

…ions via Discord.

README.md

@@ -29,24 +29,48 @@ Having your bank data in a database, gives you the power to backup, analyze and
   - Sync all transactions with a SQLite or MongoDB database
   - Visualize and query transactions using NocoDB
   - Schedule regular syncs with the database using Ofelia
+  - Send notifications to Discrod when transactions match certain filters
 
 ## 🚀 Installation and Configuration
 
 In order to use `leggen`, you need to create a GoCardless account. GoCardless is a service that provides access to Open Banking APIs. You can create an account at https://gocardless.com/bank-account-data/.
 
-After creating an account and getting your API keys, the best way is to use the [compose file](docker-compose.yml). Open the file and adapt it to your needs. Then run the following command:
+After creating an account and getting your API keys, the best way is to use the [compose file](docker-compose.yml). Open the file and adapt it to your needs.
 
-```bash
-$ docker compose up -d
+### Example Configuration
+
+Create a configuration file at with the following content:
+
+```toml
+[gocardless]
+key = "your-api-key"
+secret = "your-secret-key"
+url = "https://bankaccountdata.gocardless.com/api/v2"
+
+[database]
+sqlite = true
+
+[notifications.discord]
+webhook = "https://discord.com/api/webhooks/..."
+
+[filters]
+enabled = true
+
+[filters.case-insensitive]
+filter1 = "company-name"
 ```
 
-The leggen container will exit, this is expected. Now you can run the following command to create the configuration file:
+### Running Leggen with Docker
+
+After adapting the compose file, run the following command:
 
 ```bash
-$ docker compose run leggen init
+$ docker compose up -d
 ```
 
-Now you need to connect your bank accounts. Run the following command and follow the instructions:
+The leggen container will exit, this is expected since you didn't connect any bank accounts yet.
+
+Run the following command and follow the instructions:
 
 ```bash
 $ docker compose run leggen bank add
@@ -67,15 +91,17 @@ Usage: leggen [OPTIONS] COMMAND [ARGS]...
   Leggen: An Open Banking CLI
 
 Options:
-  --version   Show the version and exit.
-  -h, --help  Show this message and exit.
+  --version          Show the version and exit.
+  -c, --config FILE  Path to TOML configuration file
+                      [env var: LEGGEN_CONFIG_FILE;
+                       default: ~/.config/leggen/config.toml]
+  -h, --help         Show this message and exit.
 
 Command Groups:
   bank  Manage banks connections
 
 Commands:
   balances      List balances of all connected accounts
-  init          Create configuration file
   status        List all connected banks and their status
   sync          Sync all transactions with database
   transactions  List transactions

docker-compose.yml

@@ -6,13 +6,8 @@ services:
     image: elisiariocouto/leggen:latest
     command: sync
     restart: "no"
-    environment:
-      LEGGEN_GC_API_KEY: "changeme"
-      LEGGEN_GC_API_SECRET: "changeme"
-      # Uncomment the following lines if you use MongoDB
-      # LEGGEN_MONGO_URI: "mongodb://leggen:changeme@mongo:27017/leggen"
     volumes:
-      - "./leggen:/root/.config/leggen"
+      - "./leggen:/root/.config/leggen" # Default configuration file should be in this directory, named `config.toml`
       - "./db:/app"
 
   nocodb:

leggen/commands/sync.py

@@ -92,7 +92,7 @@ def save_transactions(ctx: click.Context, account: str):
         }
         transactions.append(t)
 
-    sqlite = ctx.obj["sqlite"]
+    sqlite = ctx.obj.get("database", {}).get("sqlite", True)
     info(
         f"[{account}] Fetched {len(transactions)} transactions, saving to {'SQLite' if sqlite else 'MongoDB'}"
     )
@@ -119,5 +119,5 @@ def sync(ctx: click.Context):
     for account in accounts:
         try:
             save_transactions(ctx, account)
-        except Exception:
-            error(f"[{account}] Error: Sync failed, skipping account.")
+        except Exception as e:
+            error(f"[{account}] Error: Sync failed, skipping account, exception: {e}")

leggen/main.py

@@ -74,30 +74,33 @@ def get_command(self, ctx, name):
         return getattr(mod, name)
 
 
-@click.group(cls=Group, context_settings={"help_option_names": ["-h", "--help"]})
+@click.option(
+    "-c",
+    "--config",
+    type=click.Path(dir_okay=False),
+    default=click.get_app_dir("leggen") / Path("config.toml"),
+    show_default=True,
+    callback=load_config,
+    is_eager=True,
+    expose_value=False,
+    envvar="LEGGEN_CONFIG_FILE",
+    show_envvar=True,
+    help="Path to TOML configuration file",
+)
+@click.group(
+    cls=Group,
+    context_settings={"help_option_names": ["-h", "--help"]},
+)
 @click.version_option(package_name="leggen")
 @click.pass_context
 def cli(ctx: click.Context):
     """
     Leggen: An Open Banking CLI
     """
-    ctx.ensure_object(dict)
 
     # Do not require authentication when printing help messages
     if "--help" in sys.argv[1:] or "-h" in sys.argv[1:]:
         return
 
-    # or when running the init command
-    if ctx.invoked_subcommand == "init":
-        if (click.get_app_dir("leggen") / Path("config.json")).is_file():
-            click.confirm(
-                "Configuration file already exists. Do you want to overwrite it?",
-                abort=True,
-            )
-        return
-    config = load_config()
-    token = get_token(config)
-    ctx.obj["api_url"] = config["api_url"]
-    ctx.obj["sqlite"] = config["sqlite"]
-    ctx.obj["mongo_uri"] = config["mongo_uri"]
+    token = get_token(ctx)
     ctx.obj["headers"] = {"Authorization": f"Bearer {token}"}

leggen/notifications/discord.py

@@ -0,0 +1,30 @@
+import click
+from discord_webhook import DiscordEmbed, DiscordWebhook
+
+from leggen.utils.text import info
+
+
+def send_message(ctx: click.Context, transactions: list):
+    info(f"Got {len(transactions)} new transactions, sending message to Discord")
+    webhook = DiscordWebhook(url=ctx.obj["notifications"]["discord"]["webhook"])
+
+    embed = DiscordEmbed(
+        title="",
+        description=f"{len(transactions)} new transaction matches",
+        color="03b2f8",
+    )
+    embed.set_author(
+        name="Leggen",
+        url="https://github.com/elisiariocouto/leggen",
+    )
+    embed.set_footer(text="Case-insensitive filters")
+    embed.set_timestamp()
+    for transaction in transactions:
+        embed.add_embed_field(
+            name=transaction["name"],
+            value=f"{transaction['value']}{transaction['currency']} ({transaction['date']})",
+        )
+
+    webhook.add_embed(embed)
+    response = webhook.execute()
+    response.raise_for_status()

leggen/utils/auth.py

@@ -7,21 +7,24 @@
 from leggen.utils.text import warning
 
 
-def create_token(config: dict) -> str:
+def create_token(ctx: click.Context) -> str:
     """
     Create a new token
     """
     res = requests.post(
-        f"{config['api_url']}/token/new/",
-        json={"secret_id": config["api_key"], "secret_key": config["api_secret"]},
+        f"{ctx.obj['gocardless']['url']}/token/new/",
+        json={
+            "secret_id": ctx.obj["gocardless"]["key"],
+            "secret_key": ctx.obj["gocardless"]["secret"],
+        },
     )
     res.raise_for_status()
     auth = res.json()
     save_auth(auth)
     return auth["access"]
 
 
-def get_token(config: dict) -> str:
+def get_token(ctx: click.Context) -> str:
     """
     Get the token from the auth file or request a new one
     """
@@ -30,10 +33,11 @@ def get_token(config: dict) -> str:
         with click.open_file(str(auth_file), "r") as f:
             auth = json.load(f)
         if not auth.get("access"):
-            return create_token(config)
+            return create_token(ctx)
 
         res = requests.post(
-            f"{config['api_url']}/token/refresh/", json={"refresh": auth["refresh"]}
+            f"{ctx.obj['gocardless']['url']}/token/refresh/",
+            json={"refresh": auth["refresh"]},
         )
         try:
             res.raise_for_status()
@@ -44,9 +48,9 @@ def get_token(config: dict) -> str:
             warning(
                 f"Token probably expired, requesting a new one.\nResponse: {res.status_code}\n{res.text}"
             )
-            return create_token(config)
+            return create_token(ctx)
     else:
-        return create_token(config)
+        return create_token(ctx)
 
 
 def save_auth(d: dict):

leggen/utils/config.py

@@ -1,29 +1,18 @@
-import json
 import sys
-from pathlib import Path
 
 import click
+import tomllib
 
-from leggen.utils.text import error, info
+from leggen.utils.text import error
 
 
-def save_config(d: dict):
-    Path.mkdir(Path(click.get_app_dir("leggen")), exist_ok=True)
-    config_file = click.get_app_dir("leggen") / Path("config.json")
-
-    with click.open_file(str(config_file), "w") as f:
-        json.dump(d, f)
-        info(f"Wrote configuration file at '{config_file}'")
-
-
-def load_config() -> dict:
-    config_file = click.get_app_dir("leggen") / Path("config.json")
+def load_config(ctx: click.Context, _, filename):
     try:
-        with click.open_file(str(config_file), "r") as f:
-            config = json.load(f)
-            return config
+        with click.open_file(str(filename), "rb") as f:
+            # TODO: Implement configuration file validation (use pydantic?)
+            ctx.obj = tomllib.load(f)
     except FileNotFoundError:
         error(
-            "Configuration file not found. Run `leggen init` to configure your account."
+            "Configuration file not found. Provide a valid configuration file path with leggen --config <path> or LEGGEN_CONFIG=<path> environment variable."
         )
         sys.exit(1)