114 lines
3.7 KiB
Python
114 lines
3.7 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)
|
|
else:
|
|
click.secho(f"Method not implemented: {method}.", fg="red", bold=True)
|
|
|
|
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)
|