diff --git a/leggen/commands/sync.py b/leggen/commands/sync.py index c02b71a..9eaf882 100644 --- a/leggen/commands/sync.py +++ b/leggen/commands/sync.py @@ -4,8 +4,9 @@ import click 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: diff --git a/leggen/notifications/discord.py b/leggen/notifications/discord.py index 3799726..0420357 100644 --- a/leggen/notifications/discord.py +++ b/leggen/notifications/discord.py @@ -4,7 +4,31 @@ from discord_webhook import DiscordEmbed, DiscordWebhook 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"]) diff --git a/leggen/notifications/telegram.py b/leggen/notifications/telegram.py index 11ea3ee..3b03855 100644 --- a/leggen/notifications/telegram.py +++ b/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" diff --git a/leggen/utils/notifications.py b/leggen/utils/notifications.py index cf79a64..42b9a9f 100644 --- a/leggen/utils/notifications.py +++ b/leggen/utils/notifications.py @@ -2,7 +2,26 @@ import click 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)