Refactor senders and add home_assistant send method

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,
+}

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")

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")

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")

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)

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)

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,
     )
 
 

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(
@@ -20,9 +21,8 @@ ARG_TO_OPTION = {
         "--send",
         help=(
             "Send the reconciled amount via the chosen method."
-            "Accepted methods: sms, email, telegram"
         ),
-        default="",
+        type=click.Choice(list(SENDERS.keys()), case_sensitive=False),
         show_default=True,
     ),
     "download": click.option(