ofx-processor/ofx_processor/utils/base_ofx.py

111 lines
3.6 KiB
Python

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.utils.base_processor import BaseLine, BaseProcessor
from ofx_processor.utils.config import get_config
class OfxBaseLine(BaseLine):
def get_date(self):
return self.data.dtposted.isoformat().split("T")[0]
def get_amount(self):
return int(self.data.trnamt * 1000)
def get_memo(self):
return self.data.memo
def get_payee(self):
return self.data.name
class OfxBaseProcessor(BaseProcessor):
line_class = OfxBaseLine
account_name = ""
def parse_file(self):
ofx = self._parse_file()
return ofx.statements[0].transactions
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)
def _get_reconciled_amount(self) -> Decimal:
ofx = self._parse_file()
return ofx.statements[0].balance.balamt
def _parse_file(self) -> Aggregate:
parser = OFXTree()
try:
parser.parse(self.filename)
except (FileNotFoundError, OFXHeaderError):
click.secho("Couldn't open or parse ofx file", fg="red")
sys.exit(1)
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)