Compare commits

..

35 commits

Author SHA1 Message Date
b731c2d085 fix mise config 2024-11-30 10:17:21 +01:00
5bb5b38a2c fix headless 2024-11-29 21:40:30 +01:00
f6c14d98b1 fix lcl downloader 2024-11-29 21:37:03 +01:00
03e1196f88 update dependencies 2024-11-29 20:24:53 +01:00
cdf719f195 fix lcl downloader 2024-11-29 20:20:11 +01:00
8f1662a861 fix ofx lcl 2024-10-14 23:08:44 +02:00
c7acfcd099 fix dockerfile AS 2024-10-14 20:20:29 +02:00
d8c9240808 fix download lcl 2024-10-14 20:19:38 +02:00
43b76772e1 update readme 2024-10-14 19:58:25 +02:00
8ef7829b73 implement download from new lcl interface 2024-10-14 19:55:37 +02:00
a4eb23280f quiet pip install output 2023-06-15 22:49:13 +02:00
85c36bc14f fix: retry dismiss welcome + update to bookworm 2023-06-15 22:45:30 +02:00
7494f4e51b fix: add sleep before dismissing welcome screen 2023-06-15 22:37:31 +02:00
c36030856d fix: get screenshot dir using config file 2023-06-14 08:58:12 +02:00
3df3d1059c lcl: screenshot when error 2023-06-14 08:54:03 +02:00
14499eab65 fix: sleep 2023-06-14 08:40:09 +02:00
3290134835 lcl: sleep between welcome screen and burger menu 2023-06-14 08:35:53 +02:00
fb55fa4d1a Update dockerfile 2023-06-14 08:18:42 +02:00
2e468f69f1 Bump version 2023-06-14 08:11:15 +02:00
0a4e7985cc update deps 2023-06-14 08:10:48 +02:00
01b486031b Dismiss LCL welcome screen 2023-06-14 08:02:30 +02:00
2e0d5dc419 Bump version 2022-11-05 20:35:05 +01:00
30f1e1c990 Improve logging 2022-11-05 20:34:16 +01:00
a0395fc296 Fix lcl return to legacy 2022-11-05 20:34:03 +01:00
57cdb596c5 Cleanup dockerfile 2022-10-10 22:41:47 +02:00
a54906c6d1 Bump version 2022-10-10 22:41:47 +02:00
3185a7d103 Update deps 2022-10-10 22:41:47 +02:00
e0006873f9 Refactor senders and add home_assistant send method 2022-10-10 22:41:41 +02:00
1a94bd6d36 Add error message if send method is unknown 2022-09-22 21:16:02 +01:00
80a15d6825 Bump version 2022-09-22 21:11:31 +01:00
36f74681b2 Add command to get chat id 2022-09-22 21:11:07 +01:00
b190913e6e Implement send via telegram 2022-09-22 20:54:57 +01:00
9eef4a66a6 Fix legacy 2022-09-22 20:24:02 +01:00
6bfb5ca29d Bump version 2022-09-22 20:03:50 +01:00
b831f2be5d LCL downloader go back to legacy version 2022-09-22 20:03:03 +01:00
18 changed files with 1204 additions and 833 deletions

1
.envrc Normal file
View file

@ -0,0 +1 @@
layout python3

2
.gitignore vendored
View file

@ -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

3
.mise.toml Normal file
View file

@ -0,0 +1,3 @@
[tools]
poetry = {version='latest', pyproject='pyproject.toml'}
python = '3.13' # must be after poetry so the poetry bin is first in PATH

View file

@ -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.13-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"]

View file

@ -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
```

View file

@ -1,8 +1,11 @@
import datetime
import time
from pathlib import Path
import click
from selenium import webdriver
from selenium.common.exceptions import NoSuchElementException
from selenium.webdriver import Keys
from selenium.webdriver.common.by import By
from selenium.webdriver.support.select import Select
@ -25,25 +28,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 +64,59 @@ 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=30)
self._type_date(By.ID, "mat-input-0", start)
self._type_date(By.ID, "mat-input-1", end)
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(self, by: By, value: str, value_to_type: str):
self.selenium.find_element(by, value).send_keys(value_to_type)
def _clear(self, by: By, value: str):
self.selenium.find_element(by, value).clear()
def _type_date(self, by: By, value: str, _date: datetime.date):
# The new version of the form doesn't behave nicely when typing the date.
self._type(by, value, "1")
self._type(by, value, Keys.ARROW_LEFT*5)
self._type(by, value, Keys.BACKSPACE*5)
self._type(by, value, _date.strftime("%d/%m"))
self._type(by, value, Keys.ARROW_RIGHT*5)
self._type(by, value, Keys.BACKSPACE*4)
self._type(by, value, _date.strftime("%Y"))
def _get_last_download_file_name(self, wait_seconds: int = 30):
end_time = time.time() + wait_seconds
while time.time() < end_time:

View file

@ -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]:
with open(self.filename, "w") as temp_file:
temp_file.writelines(data[1:])
new_lines = [line for line in data if is_valid_line(line)]
with open(self.filename, "w") as temp_file:
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)
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:

View file

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

View file

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

View file

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

View file

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

View file

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

View file

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

View file

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

View file

@ -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(

1679
poetry.lock generated

File diff suppressed because it is too large Load diff

View file

@ -1,6 +1,6 @@
[tool.poetry]
name = "ofx-processor"
version = "4.1.0"
version = "4.5.5"
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
"""

View file

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