diff --git a/ofx_processor/senders/__init__.py b/ofx_processor/senders/__init__.py new file mode 100644 index 0000000..ca6a4a7 --- /dev/null +++ b/ofx_processor/senders/__init__.py @@ -0,0 +1,8 @@ +from ofx_processor.senders import sms, email, telegram, home_assistant + +SENDERS = { + "sms": sms.send, + "email": email.send, + "telegram": telegram.send, + "home_assistant": home_assistant.send, +} diff --git a/ofx_processor/senders/email.py b/ofx_processor/senders/email.py new file mode 100644 index 0000000..be9f6b6 --- /dev/null +++ b/ofx_processor/senders/email.py @@ -0,0 +1,24 @@ +from decimal import Decimal + +import click +import requests + +from ofx_processor.utils.config import Config + + +def send(config: Config, amount: Decimal) -> None: + if not config.email_setup: + click.secho("Email is not properly setup", fg="yellow") + return + res = requests.post( + f"https://api.mailgun.net/v3/{config.mailgun_domain}/messages", + auth=("api", config.mailgun_api_key), + data={ + "from": config.mailgun_from, + "to": [config.email_recipient], + "subject": f"Reconciled balance: {amount}", + "text": f"Here's your reconciled balance: {amount}", + }, + ) + if res.status_code >= 400: + click.secho("Error while sending email", fg="yellow") diff --git a/ofx_processor/senders/home_assistant.py b/ofx_processor/senders/home_assistant.py new file mode 100644 index 0000000..3d8ec31 --- /dev/null +++ b/ofx_processor/senders/home_assistant.py @@ -0,0 +1,20 @@ +from decimal import Decimal + +import click +import requests + +from ofx_processor.utils.config import Config + + +def send(config: Config, amount: Decimal) -> None: + if not config.home_assistant_setup: + click.secho("Home Assistant is not properly setup", fg="yellow") + return + res = requests.post( + config.home_assistant_webhook_url, + json={ + "reconciled": str(amount), + }, + ) + if res.status_code >= 400: + click.secho("Error while calling Home Assistant", fg="yellow") diff --git a/ofx_processor/senders/sms.py b/ofx_processor/senders/sms.py new file mode 100644 index 0000000..5a530fb --- /dev/null +++ b/ofx_processor/senders/sms.py @@ -0,0 +1,22 @@ +from decimal import Decimal + +import click +import requests + +from ofx_processor.utils.config import Config + + +def send(config: Config, amount: Decimal) -> None: + if not config.sms_setup: + click.secho("SMS is not properly setup", fg="yellow") + return + res = requests.post( + f"https://smsapi.free-mobile.fr/sendmsg", + json={ + "user": config.sms_user, + "pass": config.sms_key, + "msg": f"Reconciled balance: {amount}", + }, + ) + if res.status_code >= 400: + click.secho("Error while sending SMS", fg="yellow") diff --git a/ofx_processor/senders/telegram.py b/ofx_processor/senders/telegram.py new file mode 100644 index 0000000..26f2207 --- /dev/null +++ b/ofx_processor/senders/telegram.py @@ -0,0 +1,24 @@ +import asyncio +from decimal import Decimal + +import click +import telegram + +from ofx_processor.utils.config import Config + + +def send(config: Config, amount: Decimal) -> None: + if not config.telegram_setup: + click.secho("Telegram is not properly setup", fg="yellow") + return + + try: + asyncio.run(_send_telegram_message(config.telegram_bot_token, config.telegram_bot_chat_id, f"Reconciled balance: {amount}")) + except Exception as e: + click.secho(f"Error while sending Telegram message. {type(e).__name__}: {e}", fg="yellow") + + +async def _send_telegram_message(bot_token: str, chat_id: str, message: str) -> None: + bot = telegram.Bot(bot_token) + async with bot: + await bot.send_message(chat_id=chat_id, text=message) diff --git a/ofx_processor/utils/base_ofx.py b/ofx_processor/utils/base_ofx.py index 685e0fe..97d5f85 100644 --- a/ofx_processor/utils/base_ofx.py +++ b/ofx_processor/utils/base_ofx.py @@ -1,14 +1,12 @@ -import asyncio import sys from decimal import Decimal import click -import requests -import telegram from ofxtools import OFXTree from ofxtools.header import OFXHeaderError from ofxtools.models import Aggregate +from ofx_processor.senders import SENDERS from ofx_processor.utils.base_processor import BaseLine, BaseProcessor from ofx_processor.utils.config import get_config @@ -38,12 +36,10 @@ class OfxBaseProcessor(BaseProcessor): def send_reconciled_amount(self, method): amount = self._get_reconciled_amount() click.secho(f"Reconciled balance: {amount}. Sending via {method}...", fg="blue") - if method == "email": - self._send_mail(amount) - elif method == "sms": - self._send_sms(amount) - elif method == "telegram": - self._send_telegram(amount) + config = get_config(self.account_name) + sender = SENDERS.get(method) + if sender: + sender(config, amount) else: click.secho(f"Method not implemented: {method}.", fg="red", bold=True) @@ -61,53 +57,3 @@ class OfxBaseProcessor(BaseProcessor): ofx = parser.convert() return ofx - def _send_mail(self, amount: Decimal): - config = get_config(self.account_name) - if not config.email_setup: - click.secho("Email is not properly setup", fg="yellow") - return - res = requests.post( - f"https://api.mailgun.net/v3/{config.mailgun_domain}/messages", - auth=("api", config.mailgun_api_key), - data={ - "from": config.mailgun_from, - "to": [config.email_recipient], - "subject": f"Reconciled balance: {amount}", - "text": f"Here's your reconciled balance: {amount}", - }, - ) - if res.status_code >= 400: - click.secho("Error while sending email", fg="yellow") - - def _send_sms(self, amount: Decimal): - config = get_config(self.account_name) - if not config.sms_setup: - click.secho("SMS is not properly setup", fg="yellow") - return - res = requests.post( - f"https://smsapi.free-mobile.fr/sendmsg", - json={ - "user": config.sms_user, - "pass": config.sms_key, - "msg": f"Reconciled balance: {amount}", - }, - ) - if res.status_code >= 400: - click.secho("Error while sending SMS", fg="yellow") - - def _send_telegram(self, amount: Decimal): - config = get_config(self.account_name) - if not config.telegram_setup: - click.secho("Telegram is not properly setup", fg="yellow") - return - - try: - asyncio.run(_send_telegram_message(config.telegram_bot_token, config.telegram_bot_chat_id, f"Reconciled balance: {amount}")) - except Exception as e: - click.secho(f"Error while sending Telegram message. {type(e).__name__}: {e}", fg="yellow") - - -async def _send_telegram_message(bot_token: str, chat_id: str, message: str) -> None: - bot = telegram.Bot(bot_token) - async with bot: - await bot.send_message(chat_id=chat_id, text=message) diff --git a/ofx_processor/utils/config.py b/ofx_processor/utils/config.py index 8592d72..9c43cad 100644 --- a/ofx_processor/utils/config.py +++ b/ofx_processor/utils/config.py @@ -86,6 +86,7 @@ class Config: sms_key: Optional[str] = None telegram_bot_token: Optional[str] = None telegram_bot_chat_id: Optional[str] = None + home_assistant_webhook_url: Optional[str] = None @property def email_setup(self) -> bool: @@ -119,6 +120,15 @@ class Config: ] ) + @property + def home_assistant_setup(self): + """Return true if all fields are setup for home assistant.""" + return all( + [ + self.home_assistant_webhook_url, + ] + ) + def get_config(account: str) -> Config: config = configparser.ConfigParser() @@ -156,6 +166,7 @@ def get_config(account: str) -> Config: sms_key = section.get("sms_key") telegram_bot_token = section.get("telegram_bot_token") telegram_bot_chat_id = section.get("telegram_bot_chat_id") + home_assistant_webhook_url = section.get("home_assistant_webhook_url") except KeyError as e: return handle_config_file_error(config_file, e) @@ -173,6 +184,7 @@ def get_config(account: str) -> Config: sms_key, telegram_bot_token, telegram_bot_chat_id, + home_assistant_webhook_url, ) diff --git a/ofx_processor/utils/utils.py b/ofx_processor/utils/utils.py index e5c2e18..bd97b69 100644 --- a/ofx_processor/utils/utils.py +++ b/ofx_processor/utils/utils.py @@ -6,6 +6,7 @@ import pkgutil import click from ofx_processor import processors +from ofx_processor.senders import SENDERS ARG_TO_OPTION = { "keep": click.option( @@ -19,10 +20,9 @@ ARG_TO_OPTION = { "-s", "--send", help=( - "Send the reconciled amount via the chosen method. " - "Accepted methods: sms, email, telegram" + "Send the reconciled amount via the chosen method." ), - default="", + type=click.Choice(list(SENDERS.keys()), case_sensitive=False), show_default=True, ), "download": click.option(