fix ofx lcl

.envrc

@@ -0,0 +1 @@
+layout python3

.gitignore

@@ -1,3 +1,4 @@
+error_download_lcl.png
 geckodriver.log
 
 # Created by https://www.gitignore.io/api/osx,pycharm,python
@@ -228,3 +229,4 @@ dmypy.json
 
 .idea
 .vscode
+.direnv

.tool-versions

@@ -0,0 +1,2 @@
+python 3.11.2
+poetry latest

Dockerfile

@@ -1,13 +1,15 @@
-FROM debian:bullseye AS downloader
+FROM debian:bookworm AS downloader
 WORKDIR /app
 RUN apt-get update && apt-get install -y wget
-ARG GECKODRIVER_VERSION="v0.30.0"
-ARG GECKODRIVER_FILENAME="geckodriver-$GECKODRIVER_VERSION-linux64"
-RUN wget -q https://github.com/mozilla/geckodriver/releases/download/$GECKODRIVER_VERSION/$GECKODRIVER_FILENAME.tar.gz && tar xvf $GECKODRIVER_FILENAME.tar.gz && rm $GECKODRIVER_FILENAME.tar.gz
+ARG GECKODRIVER_VERSION="v0.35.0"
+ARG GECKODRIVER_FILENAME="geckodriver-$GECKODRIVER_VERSION-linux64.tar.gz"
+RUN wget -q https://github.com/mozilla/geckodriver/releases/download/$GECKODRIVER_VERSION/$GECKODRIVER_FILENAME \
+    && tar xvf $GECKODRIVER_FILENAME  \
+    && rm $GECKODRIVER_FILENAME
 
-FROM python:3.10-slim-bullseye as final
+FROM python:3.11-slim-bookworm AS final
 RUN apt-get update && apt-get install -y firefox-esr
 COPY --from=downloader /app/geckodriver /usr/local/bin/geckodriver
 ARG OFX_VERSION
-RUN pip install ofx-processor==$OFX_VERSION
+RUN pip --disable-pip-version-check install --quiet ofx-processor==$OFX_VERSION
 CMD ["ynab", "lcl", "--download"]

README.md

@@ -50,7 +50,7 @@ inv full-test
 poetry version <major/minor/patch>
 git add .
 git commit
-inv tag <version>
+inv tag $(poetry version -s)
 inv publish publish-docker
 ```
 

ofx_processor/downloaders/lcl.py

@@ -1,8 +1,10 @@
+import datetime
 import time
 from pathlib import Path
 
 import click
 from selenium import webdriver
+from selenium.common.exceptions import NoSuchElementException
 from selenium.webdriver.common.by import By
 from selenium.webdriver.support.select import Select
 
@@ -25,25 +27,34 @@ class LclDownloader:
             download_folder = Path.home() / "Downloads"
         self.download_folder = download_folder.resolve()
         options = webdriver.FirefoxOptions()
-        options.headless = True
+        options.add_argument("-headless")
         options.set_preference("browser.download.dir", str(self.download_folder))
         options.set_preference(
             "browser.helperApps.neverAsk.saveToDisk", "application/x-ofx"
         )
         self.selenium = webdriver.Firefox(options=options)
         self.selenium.implicitly_wait(30)
+        self.selenium.set_window_size(1280, 4000)
 
     def download(self) -> str:
+        try:
+            return self._download()
+        except Exception:
+            screenshot = Path(self.config.screenshot_dir) / "error_download_lcl.png"
+            self.selenium.save_screenshot(screenshot)
+            raise
+
+    def _download(self) -> str:
         selenium = self.selenium
 
         click.secho("Logging in to LCL...", fg="blue")
         selenium.get("https://monespace.lcl.fr/connexion")
         try:
             self._click(By.ID, "popin_tc_privacy_button_2")
-        except:
-            # If the user has already accepted the privacy policy,
-            # the button is not present
-            pass
+            click.secho("Accepting privacy policy...", fg="blue")
+        except NoSuchElementException:
+            click.secho("Privacy policy already accepted", fg="blue")
+
         login_input = selenium.find_element(By.ID, "identifier")
         login_input.send_keys(self.config.bank_identifier)
         self._click(By.CLASS_NAME, "app-cta-button")
@@ -52,23 +63,46 @@ class LclDownloader:
         self._click(By.CLASS_NAME, "app-cta-button")
         click.secho("Logged in!", fg="green")
 
-        click.secho("Navigating through archives...", fg="blue")
-        self._click(By.ID, "linkSynthese")
-        self._click(By.CLASS_NAME, "picDl")
-        self._select(By.ID, "change", index=1)
-        self._select(By.ID, "DS", index=20)
-        self._click(By.ID, "MON04")
-        self._click(By.ID, "Valider")
+        retry = True
+        while retry:
+            try:
+                self._click(By.CSS_SELECTOR, ".app-cta-button--primary")
+                click.secho("Dismissing welcome screen...", fg="blue")
+                time.sleep(1)
+            except NoSuchElementException:
+                click.secho("No welcome screen found.", fg="blue")
+                retry = False
+
+        self._click(By.CLASS_NAME, "extended-zone")
+        self._click(By.ID, "export-button")
+
+        end = datetime.date.today() - datetime.timedelta(days=1)
+        start = end - datetime.timedelta(days=9)
+
+        self._type_nth(By.CSS_SELECTOR, "input.range-picker-input", 0, start.strftime("%d/%m/%Y"))
+        self._type_nth(By.CSS_SELECTOR, "input.range-picker-input", 1, end.strftime("%d/%m/%Y"))
+
+        self._click(By.CSS_SELECTOR, "ui-desktop-select button")
+        self._click_nth(By.CSS_SELECTOR, "ui-select-list ul li", 2)
+
+        self._click(By.CLASS_NAME, "download-button")
         click.secho("Found it!", fg="green")
+        time.sleep(5)
         selenium.get("about:downloads")
         return self._get_last_download_file_name()
 
     def _click(self, by: By, value: str):
         self.selenium.find_element(by, value).click()
 
+    def _click_nth(self, by: By, value: str, idx: int):
+        self.selenium.find_elements(by, value)[idx].click()
+
     def _select(self, by: By, value: str, index: int):
         Select(self.selenium.find_element(by, value)).select_by_index(index)
 
+    def _type_nth(self, by: By, value: str, idx: int, value_to_type: str):
+        self.selenium.find_elements(by, value)[idx].send_keys(value_to_type)
+
     def _get_last_download_file_name(self, wait_seconds: int = 30):
         end_time = time.time() + wait_seconds
         while time.time() < end_time:

ofx_processor/processors/lcl.py

@@ -1,5 +1,6 @@
 import sys
 from datetime import datetime
+from operator import truediv
 
 import click
 import dateparser
@@ -67,19 +68,27 @@ class LclProcessor(OfxBaseProcessor):
             click.secho("Couldn't find ofx file", fg="red")
             sys.exit(1)
 
-        if "Content-Type:" in data[0]:
+        new_lines = [line for line in data if is_valid_line(line)]
+
         with open(self.filename, "w") as temp_file:
-                temp_file.writelines(data[1:])
+            temp_file.writelines(new_lines)
 
         ofx = super()._parse_file()
 
-        if "Content-Type:" in data[0]:
         with open(self.filename, "w") as temp_file:
             temp_file.writelines(data)
 
         return ofx
 
 
+def is_valid_line(line):
+    if "Content-Type:" in line:
+        return False
+    if "MKTGINFO" in line:
+        return False
+    return True
+
+
 def main(filename, keep, download, send_method, push_to_ynab):
     """Import LCL bank statement (OFX file)."""
     if download:
@@ -91,6 +100,7 @@ def main(filename, keep, download, send_method, push_to_ynab):
             )
         filename = LclDownloader().download()
     processor = LclProcessor(filename)
+    processor.parse_file()
     if send_method:
         processor.send_reconciled_amount(send_method)
     if push_to_ynab:

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

@@ -2,11 +2,11 @@ import sys
 from decimal import Decimal
 
 import click
-import requests
 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
 
@@ -36,10 +36,12 @@ 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)
+        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)
 
     def _get_reconciled_amount(self) -> Decimal:
         ofx = self._parse_file()
@@ -55,36 +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")

ofx_processor/utils/config.py

@@ -1,3 +1,4 @@
+import asyncio
 import configparser
 import os
 import sys
@@ -5,6 +6,7 @@ from dataclasses import dataclass
 from typing import Optional
 
 import click
+import telegram
 
 DEFAULT_CONFIG_DIR = click.get_app_dir("ofx_processor")
 DEFAULT_CONFIG_FILENAME = "config.ini"
@@ -15,6 +17,7 @@ def get_default_config():
     default_config["DEFAULT"] = {
         "token": "<YOUR API TOKEN>",
         "budget": "<YOUR BUDGET ID>",
+        "screenshot_dir": "/tmp",
         "mailgun_api_key": "",
         "mailgun_domain": "",
         "mailgun_from": "",
@@ -54,11 +57,27 @@ def show_file_name():
     click.echo(config_file)
 
 
+@config.command("telegram", help="Display the bot's chat ID.")
+def print_telegram_chat_id():
+    config = get_config("DEFAULT")
+    click.pause("Please start a conversation with your bot, then press any key...")
+    asyncio.run(print_chat_id(config))
+
+
+async def print_chat_id(config: "Config"):
+    bot = telegram.Bot(config.telegram_bot_token)
+    async with bot:
+        update = (await bot.get_updates())[-1]
+        chat_id = update.message.chat_id
+        click.secho(f"Your chat ID is: {chat_id}", fg="green", bold=True)
+
+
 @dataclass(frozen=True)
 class Config:
     account: str
     budget_id: str
     token: str
+    screenshot_dir: str
     bank_identifier: Optional[str] = None
     bank_password: Optional[str] = None
     mailgun_api_key: Optional[str] = None
@@ -67,6 +86,9 @@ class Config:
     email_recipient: Optional[str] = None
     sms_user: Optional[str] = None
     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:
@@ -90,6 +112,25 @@ class Config:
             ]
         )
 
+    @property
+    def telegram_setup(self) -> bool:
+        """Return true if all fields are setup for telegram."""
+        return all(
+            [
+                self.telegram_bot_token,
+                self.telegram_bot_chat_id,
+            ]
+        )
+
+    @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()
@@ -113,7 +154,11 @@ def get_config(account: str) -> Config:
         section = config[account]
         budget_id = section["budget"]
         token = section["token"]
-        account = section["account"]
+        screenshot_dir = section.get("screenshot_dir")
+        if account == "DEFAULT":
+            ynab_account_id = ""
+        else:
+            ynab_account_id = section["account"]
         bank_identifier = section.get("bank_identifier")
         bank_password = section.get("bank_password")
         mailgun_api_key = section.get("mailgun_api_key")
@@ -122,13 +167,17 @@ def get_config(account: str) -> Config:
         email_recipient = section.get("email_recipient")
         sms_user = section.get("sms_user")
         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)
 
     return Config(
-        account,
+        ynab_account_id,
         budget_id,
         token,
+        screenshot_dir,
         bank_identifier,
         bank_password,
         mailgun_api_key,
@@ -137,6 +186,9 @@ def get_config(account: str) -> Config:
         email_recipient,
         sms_user,
         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(
@@ -19,10 +20,9 @@ ARG_TO_OPTION = {
         "-s",
         "--send",
         help=(
-            "Send the reconciled amount via the chosen method. "
-            "Accepted methods: sms, email"
+            "Send the reconciled amount via the chosen method."
         ),
-        default="",
+        type=click.Choice(list(SENDERS.keys()), case_sensitive=False),
         show_default=True,
     ),
     "download": click.option(

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "ofx-processor"
-version = "4.1.0"
+version = "4.5.2"
 description = "Personal ofx processor"
 readme = "README.md"
 authors = ["Gabriel Augendre <gabriel@augendre.info>"]
@@ -20,28 +20,28 @@ classifiers = [
     "Operating System :: MacOS :: MacOS X",
     "Operating System :: Microsoft :: Windows",
     "Operating System :: Unix",
-    "Programming Language :: Python :: 3.8",
-    "Programming Language :: Python :: 3.9",
     "Programming Language :: Python :: 3.10",
+    "Programming Language :: Python :: 3.11",
     "Programming Language :: Python :: 3 :: Only",
     "Topic :: Utilities",
 ]
 
 [tool.poetry.dependencies]
-python = ">=3.8,<4"
-ofxtools = "^0.9.4"
-click = "^8.0.3"
-dateparser = "^1.1.0"
-requests = "^2.24.0"
-selenium = "^4.0.0"
+python = ">=3.10,<4"
+ofxtools = ">=0.9.4"
+click = ">=8.0.3"
+dateparser = ">=1.1.0"
+requests = ">=2.24.0"
+selenium = ">=4.0.0"
+python-telegram-bot = {version = ">=20.0a4", allow-prereleases = true}
 
 [tool.poetry.dev-dependencies]
-pytest = "^6.0.1"
-pytest-cov = "^3.0.0"
-invoke = "^1.6.0"
-pre-commit = "^2.15.0"
-tox = "^3.24.4"
-pdbpp = "^0.10.3"
+pytest = ">=6.0.1"
+pytest-cov = ">=3.0.0"
+invoke = ">=2.0.0"
+pre-commit = ">=2.15.0"
+tox = ">=3.24.4"
+pdbpp = ">=0.10.3"
 
 [tool.poetry.scripts]
 ynab = 'ofx_processor.main:cli'
@@ -56,12 +56,12 @@ profile = "black"
 legacy_tox_ini = """
 [tox]
 isolated_build = true
-envlist = py38, py39, py310
+envlist = py310, py311
 
 [testenv]
-whitelist_externals = poetry
+allowlist_externals = poetry
 commands =
-    poetry install --remove-untracked
+    poetry install --sync
     poetry run inv test
 """
 

tests/samples/config.ini

@@ -1,6 +1,7 @@
 [DEFAULT]
 token = <YOUR API TOKEN>
 budget = <YOUR BUDGET ID>
+screenshot_dir = /tmp
 
 [bpvf]
 account = <YOUR BPVF ACCOUNT ID>