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
# Created by https://www.gitignore.io/api/osx,pycharm,python
@ -229,4 +228,3 @@ dmypy.json
.idea
.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
RUN apt-get update && apt-get install -y wget
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
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
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
COPY --from=downloader /app/geckodriver /usr/local/bin/geckodriver
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"]

View file

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

View file

@ -1,10 +1,8 @@
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
@ -27,34 +25,25 @@ class LclDownloader:
download_folder = Path.home() / "Downloads"
self.download_folder = download_folder.resolve()
options = webdriver.FirefoxOptions()
options.add_argument("-headless")
options.headless = True
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")
click.secho("Accepting privacy policy...", fg="blue")
except NoSuchElementException:
click.secho("Privacy policy already accepted", fg="blue")
except:
# If the user has already accepted the privacy policy,
# the button is not present
pass
login_input = selenium.find_element(By.ID, "identifier")
login_input.send_keys(self.config.bank_identifier)
self._click(By.CLASS_NAME, "app-cta-button")
@ -63,46 +52,23 @@ class LclDownloader:
self._click(By.CLASS_NAME, "app-cta-button")
click.secho("Logged in!", fg="green")
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("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")
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:

View file

@ -1,6 +1,5 @@
import sys
from datetime import datetime
from operator import truediv
import click
import dateparser
@ -68,27 +67,19 @@ class LclProcessor(OfxBaseProcessor):
click.secho("Couldn't find ofx file", fg="red")
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:
temp_file.writelines(new_lines)
temp_file.writelines(data[1:])
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:
@ -100,7 +91,6 @@ 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

@ -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
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,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")
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)
if method == "email":
self._send_mail(amount)
elif method == "sms":
self._send_sms(amount)
def _get_reconciled_amount(self) -> Decimal:
ofx = self._parse_file()
@ -57,3 +55,36 @@ 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,4 +1,3 @@
import asyncio
import configparser
import os
import sys
@ -6,7 +5,6 @@ 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"
@ -17,7 +15,6 @@ 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": "",
@ -57,27 +54,11 @@ 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
@ -86,9 +67,6 @@ 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:
@ -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:
config = configparser.ConfigParser()
@ -154,11 +113,7 @@ def get_config(account: str) -> Config:
section = config[account]
budget_id = section["budget"]
token = section["token"]
screenshot_dir = section.get("screenshot_dir")
if account == "DEFAULT":
ynab_account_id = ""
else:
ynab_account_id = section["account"]
account = section["account"]
bank_identifier = section.get("bank_identifier")
bank_password = section.get("bank_password")
mailgun_api_key = section.get("mailgun_api_key")
@ -167,17 +122,13 @@ 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(
ynab_account_id,
account,
budget_id,
token,
screenshot_dir,
bank_identifier,
bank_password,
mailgun_api_key,
@ -186,9 +137,6 @@ 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,7 +6,6 @@ import pkgutil
import click
from ofx_processor import processors
from ofx_processor.senders import SENDERS
ARG_TO_OPTION = {
"keep": click.option(
@ -21,8 +20,9 @@ ARG_TO_OPTION = {
"--send",
help=(
"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,
),
"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]
name = "ofx-processor"
version = "4.5.2"
version = "4.1.0"
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.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}
python = ">=3.8,<4"
ofxtools = "^0.9.4"
click = "^8.0.3"
dateparser = "^1.1.0"
requests = "^2.24.0"
selenium = "^4.0.0"
[tool.poetry.dev-dependencies]
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"
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"
[tool.poetry.scripts]
ynab = 'ofx_processor.main:cli'
@ -56,12 +56,12 @@ profile = "black"
legacy_tox_ini = """
[tox]
isolated_build = true
envlist = py310, py311
envlist = py38, py39, py310
[testenv]
allowlist_externals = poetry
whitelist_externals = poetry
commands =
poetry install --sync
poetry install --remove-untracked
poetry run inv test
"""

View file

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