Compare commits

..

No commits in common. "master" and "master" have entirely different histories.

18 changed files with 888 additions and 1147 deletions

1
.envrc
View file

@ -1 +0,0 @@
layout python3

2
.gitignore vendored
View file

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

View file

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

View file

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

View file

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

View file

@ -1,10 +1,8 @@
import datetime
import time import time
from pathlib import Path from pathlib import Path
import click import click
from selenium import webdriver from selenium import webdriver
from selenium.common.exceptions import NoSuchElementException
from selenium.webdriver.common.by import By from selenium.webdriver.common.by import By
from selenium.webdriver.support.select import Select from selenium.webdriver.support.select import Select
@ -27,34 +25,25 @@ class LclDownloader:
download_folder = Path.home() / "Downloads" download_folder = Path.home() / "Downloads"
self.download_folder = download_folder.resolve() self.download_folder = download_folder.resolve()
options = webdriver.FirefoxOptions() options = webdriver.FirefoxOptions()
options.add_argument("-headless") options.headless = True
options.set_preference("browser.download.dir", str(self.download_folder)) options.set_preference("browser.download.dir", str(self.download_folder))
options.set_preference( options.set_preference(
"browser.helperApps.neverAsk.saveToDisk", "application/x-ofx" "browser.helperApps.neverAsk.saveToDisk", "application/x-ofx"
) )
self.selenium = webdriver.Firefox(options=options) self.selenium = webdriver.Firefox(options=options)
self.selenium.implicitly_wait(30) self.selenium.implicitly_wait(30)
self.selenium.set_window_size(1280, 4000)
def download(self) -> str: 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 selenium = self.selenium
click.secho("Logging in to LCL...", fg="blue") click.secho("Logging in to LCL...", fg="blue")
selenium.get("https://monespace.lcl.fr/connexion") selenium.get("https://monespace.lcl.fr/connexion")
try: try:
self._click(By.ID, "popin_tc_privacy_button_2") self._click(By.ID, "popin_tc_privacy_button_2")
click.secho("Accepting privacy policy...", fg="blue") except:
except NoSuchElementException: # If the user has already accepted the privacy policy,
click.secho("Privacy policy already accepted", fg="blue") # the button is not present
pass
login_input = selenium.find_element(By.ID, "identifier") login_input = selenium.find_element(By.ID, "identifier")
login_input.send_keys(self.config.bank_identifier) login_input.send_keys(self.config.bank_identifier)
self._click(By.CLASS_NAME, "app-cta-button") self._click(By.CLASS_NAME, "app-cta-button")
@ -63,46 +52,23 @@ class LclDownloader:
self._click(By.CLASS_NAME, "app-cta-button") self._click(By.CLASS_NAME, "app-cta-button")
click.secho("Logged in!", fg="green") click.secho("Logged in!", fg="green")
retry = True click.secho("Navigating through archives...", fg="blue")
while retry: self._click(By.ID, "linkSynthese")
try: self._click(By.CLASS_NAME, "picDl")
self._click(By.CSS_SELECTOR, ".app-cta-button--primary") self._select(By.ID, "change", index=1)
click.secho("Dismissing welcome screen...", fg="blue") self._select(By.ID, "DS", index=20)
time.sleep(1) self._click(By.ID, "MON04")
except NoSuchElementException: self._click(By.ID, "Valider")
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") click.secho("Found it!", fg="green")
time.sleep(5)
selenium.get("about:downloads") selenium.get("about:downloads")
return self._get_last_download_file_name() return self._get_last_download_file_name()
def _click(self, by: By, value: str): def _click(self, by: By, value: str):
self.selenium.find_element(by, value).click() 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): def _select(self, by: By, value: str, index: int):
Select(self.selenium.find_element(by, value)).select_by_index(index) 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): def _get_last_download_file_name(self, wait_seconds: int = 30):
end_time = time.time() + wait_seconds end_time = time.time() + wait_seconds
while time.time() < end_time: while time.time() < end_time:

View file

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

View file

@ -1,8 +0,0 @@
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

@ -1,24 +0,0 @@
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

@ -1,20 +0,0 @@
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

@ -1,22 +0,0 @@
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

@ -1,24 +0,0 @@
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 from decimal import Decimal
import click import click
import requests
from ofxtools import OFXTree from ofxtools import OFXTree
from ofxtools.header import OFXHeaderError from ofxtools.header import OFXHeaderError
from ofxtools.models import Aggregate from ofxtools.models import Aggregate
from ofx_processor.senders import SENDERS
from ofx_processor.utils.base_processor import BaseLine, BaseProcessor from ofx_processor.utils.base_processor import BaseLine, BaseProcessor
from ofx_processor.utils.config import get_config from ofx_processor.utils.config import get_config
@ -36,12 +36,10 @@ class OfxBaseProcessor(BaseProcessor):
def send_reconciled_amount(self, method): def send_reconciled_amount(self, method):
amount = self._get_reconciled_amount() amount = self._get_reconciled_amount()
click.secho(f"Reconciled balance: {amount}. Sending via {method}...", fg="blue") click.secho(f"Reconciled balance: {amount}. Sending via {method}...", fg="blue")
config = get_config(self.account_name) if method == "email":
sender = SENDERS.get(method) self._send_mail(amount)
if sender: elif method == "sms":
sender(config, amount) self._send_sms(amount)
else:
click.secho(f"Method not implemented: {method}.", fg="red", bold=True)
def _get_reconciled_amount(self) -> Decimal: def _get_reconciled_amount(self) -> Decimal:
ofx = self._parse_file() ofx = self._parse_file()
@ -57,3 +55,36 @@ class OfxBaseProcessor(BaseProcessor):
ofx = parser.convert() ofx = parser.convert()
return ofx 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,4 +1,3 @@
import asyncio
import configparser import configparser
import os import os
import sys import sys
@ -6,7 +5,6 @@ from dataclasses import dataclass
from typing import Optional from typing import Optional
import click import click
import telegram
DEFAULT_CONFIG_DIR = click.get_app_dir("ofx_processor") DEFAULT_CONFIG_DIR = click.get_app_dir("ofx_processor")
DEFAULT_CONFIG_FILENAME = "config.ini" DEFAULT_CONFIG_FILENAME = "config.ini"
@ -17,7 +15,6 @@ def get_default_config():
default_config["DEFAULT"] = { default_config["DEFAULT"] = {
"token": "<YOUR API TOKEN>", "token": "<YOUR API TOKEN>",
"budget": "<YOUR BUDGET ID>", "budget": "<YOUR BUDGET ID>",
"screenshot_dir": "/tmp",
"mailgun_api_key": "", "mailgun_api_key": "",
"mailgun_domain": "", "mailgun_domain": "",
"mailgun_from": "", "mailgun_from": "",
@ -57,27 +54,11 @@ def show_file_name():
click.echo(config_file) 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) @dataclass(frozen=True)
class Config: class Config:
account: str account: str
budget_id: str budget_id: str
token: str token: str
screenshot_dir: str
bank_identifier: Optional[str] = None bank_identifier: Optional[str] = None
bank_password: Optional[str] = None bank_password: Optional[str] = None
mailgun_api_key: Optional[str] = None mailgun_api_key: Optional[str] = None
@ -86,9 +67,6 @@ class Config:
email_recipient: Optional[str] = None email_recipient: Optional[str] = None
sms_user: Optional[str] = None sms_user: Optional[str] = None
sms_key: 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 @property
def email_setup(self) -> bool: def email_setup(self) -> bool:
@ -112,25 +90,6 @@ 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: def get_config(account: str) -> Config:
config = configparser.ConfigParser() config = configparser.ConfigParser()
@ -154,11 +113,7 @@ def get_config(account: str) -> Config:
section = config[account] section = config[account]
budget_id = section["budget"] budget_id = section["budget"]
token = section["token"] token = section["token"]
screenshot_dir = section.get("screenshot_dir") account = section["account"]
if account == "DEFAULT":
ynab_account_id = ""
else:
ynab_account_id = section["account"]
bank_identifier = section.get("bank_identifier") bank_identifier = section.get("bank_identifier")
bank_password = section.get("bank_password") bank_password = section.get("bank_password")
mailgun_api_key = section.get("mailgun_api_key") mailgun_api_key = section.get("mailgun_api_key")
@ -167,17 +122,13 @@ def get_config(account: str) -> Config:
email_recipient = section.get("email_recipient") email_recipient = section.get("email_recipient")
sms_user = section.get("sms_user") sms_user = section.get("sms_user")
sms_key = section.get("sms_key") 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: except KeyError as e:
return handle_config_file_error(config_file, e) return handle_config_file_error(config_file, e)
return Config( return Config(
ynab_account_id, account,
budget_id, budget_id,
token, token,
screenshot_dir,
bank_identifier, bank_identifier,
bank_password, bank_password,
mailgun_api_key, mailgun_api_key,
@ -186,9 +137,6 @@ def get_config(account: str) -> Config:
email_recipient, email_recipient,
sms_user, sms_user,
sms_key, sms_key,
telegram_bot_token,
telegram_bot_chat_id,
home_assistant_webhook_url,
) )

View file

@ -6,7 +6,6 @@ import pkgutil
import click import click
from ofx_processor import processors from ofx_processor import processors
from ofx_processor.senders import SENDERS
ARG_TO_OPTION = { ARG_TO_OPTION = {
"keep": click.option( "keep": click.option(
@ -20,9 +19,10 @@ ARG_TO_OPTION = {
"-s", "-s",
"--send", "--send",
help=( help=(
"Send the reconciled amount via the chosen method." "Send the reconciled amount via the chosen method. "
"Accepted methods: sms, email"
), ),
type=click.Choice(list(SENDERS.keys()), case_sensitive=False), default="",
show_default=True, show_default=True,
), ),
"download": click.option( "download": click.option(

1692
poetry.lock generated

File diff suppressed because it is too large Load diff

View file

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

View file

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