feat(sync): Enable expiration notifications.

leggen/commands/sync.py

@@ -4,8 +4,9 @@
 
 from leggen.main import cli
 from leggen.utils.database import persist_balance, save_transactions
+from leggen.utils.gocardless import REQUISITION_STATUS
 from leggen.utils.network import get
-from leggen.utils.notifications import send_notification
+from leggen.utils.notifications import send_expire_notification, send_notification
 from leggen.utils.text import error, info
 
 
@@ -21,6 +22,21 @@ def sync(ctx: click.Context):
     for r in res.get("results", []):
         accounts.update(r.get("accounts", []))
 
+    for r in res["results"]:
+        account_status = REQUISITION_STATUS.get(r["status"], "UNKNOWN")
+        if account_status != "LINKED":
+            created_at = datetime.fromisoformat(r["createdAt"])
+            now = datetime.now()
+            if (created_at - now).days <= 15:
+                n = {
+                    "bank": r["institution_id"],
+                    "status": REQUISITION_STATUS.get(r["status"], "UNKNOWN"),
+                    "created_at": created_at.timestamp(),
+                    "requisition_id": r["id"],
+                    "days_left": (created_at - now).days,
+                }
+                send_expire_notification(ctx, n)
+
     info(f"Syncing balances for {len(accounts)} accounts")
 
     for account in accounts:

leggen/notifications/discord.py

@@ -4,7 +4,31 @@
 from leggen.utils.text import info
 
 
-def send_message(ctx: click.Context, transactions: list):
+def send_expire_notification(ctx: click.Context, notification: dict):
+    info("Sending expiration notification to Discord")
+    webhook = DiscordWebhook(url=ctx.obj["notifications"]["discord"]["webhook"])
+
+    embed = DiscordEmbed(
+        title="",
+        description=f"Your account {notification['bank']} ({notification['requisition_id']}) is in {notification['status']} status. Days left: {notification['days_left']}",
+        color="03b2f8",
+    )
+    embed.set_author(
+        name="Leggen",
+        url="https://github.com/elisiariocouto/leggen",
+    )
+    embed.set_footer(text="Expiration notice")
+    embed.set_timestamp()
+
+    webhook.add_embed(embed)
+    response = webhook.execute()
+    try:
+        response.raise_for_status()
+    except Exception as e:
+        raise Exception(f"Discord notification failed: {e}\n{response.text}") from e
+
+
+def send_transactions_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"])
 

leggen/notifications/telegram.py

@@ -15,7 +15,30 @@ def escape_markdown(text: str) -> str:
     )
 
 
-def send_message(ctx: click.Context, transactions: list):
+def send_expire_notification(ctx: click.Context, notification: dict):
+    token = ctx.obj["notifications"]["telegram"]["api-key"]
+    chat_id = ctx.obj["notifications"]["telegram"]["chat-id"]
+    bot_url = f"https://api.telegram.org/bot{token}/sendMessage"
+    info("Sending expiration notification to Telegram")
+    message = "*💲 [Leggen](https://github.com/elisiariocouto/leggen)*\n"
+    message += f"Your account {notification['bank']} ({notification['requisition_id']}) is in {notification['status']} status. Days left: {notification['days_left']}\n"
+
+    res = requests.post(
+        bot_url,
+        json={
+            "chat_id": chat_id,
+            "text": escape_markdown(message),
+            "parse_mode": "MarkdownV2",
+        },
+    )
+
+    try:
+        res.raise_for_status()
+    except Exception as e:
+        raise Exception(f"Telegram notification failed: {e}\n{res.text}") from e
+
+
+def send_transaction_message(ctx: click.Context, transactions: list):
     token = ctx.obj["notifications"]["telegram"]["api-key"]
     chat_id = ctx.obj["notifications"]["telegram"]["chat-id"]
     bot_url = f"https://api.telegram.org/bot{token}/sendMessage"

leggen/utils/notifications.py

@@ -2,7 +2,26 @@
 
 import leggen.notifications.discord as discord
 import leggen.notifications.telegram as telegram
-from leggen.utils.text import info, warning
+from leggen.utils.text import error, info, warning
+
+
+def send_expire_notification(ctx: click.Context, notification: dict):
+    discord_enabled = ctx.obj.get("notifications", {}).get("discord", False)
+    telegram_enabled = ctx.obj.get("notifications", {}).get("telegram", False)
+
+    if not discord_enabled and not telegram_enabled:
+        warning("No notification engine is enabled, skipping notifications")
+        error(
+            f"Your account {notification['bank']} ({notification['requisition_id']}) is in {notification['status']} status. Days left: {notification['days_left']}"
+        )
+
+    if discord_enabled:
+        info("Sending expiration notification to Discord")
+        discord.send_expire_notification(ctx, notification)
+
+    if telegram_enabled:
+        info("Sending expiration notification to Telegram")
+        telegram.send_expire_notification(ctx, notification)
 
 
 def send_notification(ctx: click.Context, transactions: list):
@@ -39,8 +58,8 @@ def send_notification(ctx: click.Context, transactions: list):
 
     if discord_enabled:
         info(f"Sending {len(notification_transactions)} transactions to Discord")
-        discord.send_message(ctx, notification_transactions)
+        discord.send_transactions_message(ctx, notification_transactions)
 
     if telegram_enabled:
         info(f"Sending {len(notification_transactions)} transactions to Telegram")
-        telegram.send_message(ctx, notification_transactions)
+        telegram.send_transaction_message(ctx, notification_transactions)