123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216 |
- '''Created Sep 28, 2021 Levi'''
- import configparser
- import sys
- import traceback
- import ssl
- from time import sleep
- from imaplib import IMAP4
- import telegram
- import keyring
- from bs4 import BeautifulSoup
- from telegram.error import BadRequest
- from imap_tools import MailBox, AND
- # -> for python >=3.10, the dh key too small problem
- context = ssl.SSLContext(ssl.PROTOCOL_TLSv1_2)
- context.set_ciphers('DEFAULT@SECLEVEL=1')
- config = configparser.ConfigParser()
- config.read('bot')
- bot = telegram.Bot(token=config['AUTH']['token'])
- bot2 = telegram.Bot(token=config['AUTH']['bank'])
- def pretty_exc():
- exc_type, _exc_obj, exc_tb = sys.exc_info()
- # fname = os.path.split(exc_tb.tb_frame.f_code.co_filename)[1]
- for tb in list(traceback.format_exception(exc_type, _exc_obj, exc_tb)):
- print(tb)
- def mask(msg, to_):
- return any([to_ == msg.from_,
- to_ in msg.to,
- to_ in msg.cc,
- to_ in msg.bcc])
- def mask2(msg, to_):
- return any([to_ == msg.from_,
- to_ in msg.to,
- to_ in msg.cc,
- to_ in msg.bcc,
- to_ in msg.subject])
- def clean_html(text) -> str:
- soup = BeautifulSoup(text, 'html.parser')
- stripped_text = soup.get_text()
- lines = (line.strip() for line in stripped_text.splitlines())
- # break multi-headlines into a line each
- chunks = (phrase.strip() for line in lines for phrase in line.split(" "))
- # drop blank lines
- text = '\n'.join(chunk for chunk in chunks if chunk)
- return text
- async def mail_alert(container) -> str | None:
- '''mail alert Inbox'''
- # container: deque = deque(maxlen=500)
- # while True:
- try:
- mailbox = MailBox(host=config['email']['host'], ssl_context=context)
- mailbox.login(config['email']['user'], keyring.get_password('yagmail', 'levente.marton@mzk.ro'), initial_folder='Inbox') # noqa E:mypy
- sleep(15)
- msgs = [msgs for msgs in mailbox.fetch(AND(seen=False), mark_seen=False)]
- for msg in msgs:
- chk = mask(msg, 'levente.marton@mzk.ro')
- if chk:
- msg_text = msg.text
- if 'end of day extras/intraday' in msg.subject:
- msg_text = msg.text.replace('<br>', '\n')
- if msg.uid not in container:
- try:
- if len(msg.text) != 0:
- await bot.send_message(chat_id=config['AUTH']['chatid'], text=f'{msg.from_}\nsubj: {msg.subject}\n\n{msg_text}')
- else:
- await bot.send_message(chat_id=config['AUTH']['chatid'], text=f'{msg.from_}\nsubj: {msg.subject}\n\n{msg.html}')
- # container.appendleft(msg.uid)
- print('message appended', len(container))
- except BadRequest:
- try:
- if len(msg.text) != 0:
- await bot.send_message(chat_id=config['AUTH']['chatid'], text=f'{msg.from_}\nsubj: {msg.subject}\n\n{msg_text[:len(msg_text) // 5]}')
- else:
- await bot.send_message(chat_id=config['AUTH']['chatid'], text=f'{msg.from_}\nsubj: {msg.subject}\n\n{msg.html[:len(msg.html) // 5]}')
- # container.appendleft(msg.uid)
- print('message appended', len(container))
- except BadRequest:
- await bot.send_message(chat_id=config['AUTH']['chatid'], text=f'{msg.from_}\nsubj: {msg.subject}\n\nMessage is too long')
- # container.appendleft(msg.uid)
- print('message appended', len(container), '!! last message was to long !!')
- except Exception:
- try:
- mailbox.logout()
- pretty_exc()
- except IMAP4.abort:
- pass
- # break
- mailbox.logout()
- return msg.uid
- except Exception:
- try:
- mailbox.logout()
- except IMAP4.abort:
- pass
- return 'empty box'
- async def mail_alert2(container) -> str | None:
- # container: deque = deque(maxlen=500)
- # while True:
- try:
- mailbox2 = MailBox(host=config['email']['host'], ssl_context=context)
- mailbox2.login(config['email']['user'], keyring.get_password('yagmail', 'levente.marton@mzk.ro'), initial_folder='bank_events') # noqa E:mypy
- sleep(15)
- msgs = [msgs for msgs in mailbox2.fetch(AND(seen=False), mark_seen=True)]
- for msg in msgs:
- chk = mask2(msg, 'account')
- if chk:
- if msg.uid not in container:
- soup = BeautifulSoup(msg.html, 'html.parser')
- msg_text = soup.prettify().replace('<br/>', '').replace('<html>', '').replace('<head>', '').replace('</head>', '').replace('<body>', '').replace('<div>', '').replace('</div>', '').replace('</body>', '').replace('</html>', '')
- msg_text = msg_text[:msg_text.find('Intraday transactions')]
- msg_text = clean_html(msg_text)
- try:
- if len(msg.text) or len(msg.html) != 0:
- await bot2.send_message(chat_id=config['AUTH']['chatid'], text=msg_text)
- else:
- await bot2.send_message(chat_id=config['AUTH']['chatid'], text=msg_text)
- # container.appendleft(msg.uid)
- print('message appended', len(container))
- except BadRequest:
- try:
- if len(msg.text) or len(msg.html) != 0:
- await bot2.send_message(chat_id=config['AUTH']['chatid'], text=msg_text // 5) # noqa E:mypy
- else:
- await bot2.send_message(chat_id=config['AUTH']['chatid'], text=msg_text // 5) # noqa E:mypy
- # container.appendleft(msg.uid)
- print('message appended', len(container))
- except BadRequest:
- await bot2.send_message(chat_id=config['AUTH']['chatid'], text=f'Message is too long')
- # container.appendleft(msg.uid)
- print('message appended', len(container), '!! last message was to long !!')
- except Exception:
- try:
- mailbox2.logout()
- pretty_exc()
- except IMAP4.abort:
- pass
- # break
- mailbox2.logout()
- return msg.uid
- except Exception:
- # pretty_exc()
- try:
- mailbox2.logout()
- except IMAP4.abort:
- pass
- return 'empty box'
- async def mail_alert3(container) -> str | None:
- '''mail alert Inbox'''
- # container: deque = deque(maxlen=500)
- # while True:
- try:
- mailbox3 = MailBox(host=config['email']['host'], ssl_context=context)
- mailbox3.login(config['email']['user'], keyring.get_password('yagmail', 'levente.marton@mzk.ro'), initial_folder='API') # noqa E:mypy
- sleep(15)
- msgs = [msgs for msgs in mailbox3.fetch(AND(seen=False), mark_seen=True)]
- for msg in msgs:
- chk = mask(msg, 'Ciel_notifier@mzk.ro')
- if chk:
- msg_text = msg.text
- # if 'end of day extras/intraday' in msg.subject:
- # msg_text = msg.text.replace('<br>', '\n')
- if msg.uid not in container:
- try:
- if len(msg.text) != 0:
- await bot.send_message(chat_id=config['AUTH']['chatid'], text=f'{msg.from_}\nsubj: {msg.subject}\n\n{msg_text}')
- else:
- await bot.send_message(chat_id=config['AUTH']['chatid'], text=f'{msg.from_}\nsubj: {msg.subject}\n\n{msg.html}')
- # container.appendleft(msg.uid)
- print('message appended', len(container))
- except BadRequest:
- try:
- if len(msg.text) != 0:
- await bot.send_message(chat_id=config['AUTH']['chatid'], text=f'{msg.from_}\nsubj: {msg.subject}\n\n{msg_text[:len(msg_text) // 5]}')
- else:
- await bot.send_message(chat_id=config['AUTH']['chatid'], text=f'{msg.from_}\nsubj: {msg.subject}\n\n{msg.html[:len(msg.html) // 5]}')
- # container.appendleft(msg.uid)
- print('message appended', len(container))
- except BadRequest:
- await bot.send_message(chat_id=config['AUTH']['chatid'], text=f'{msg.from_}\nsubj: {msg.subject}\n\nMessage is too long')
- # container.appendleft(msg.uid)
- print('message appended', len(container), '!! last message was to long !!')
- except Exception:
- try:
- mailbox3.logout()
- pretty_exc()
- except IMAP4.abort:
- pass
- # break
- mailbox3.logout()
- return msg.uid
- except Exception:
- try:
- mailbox3.logout()
- except IMAP4.abort:
- pass
- return 'empty box'
- # if __name__ == '__main__':
- # try:
- # # event = Event()
- # asyncio.run(mail_alert())
- # except KeyboardInterrupt:
- # loop = asyncio.get_event_loop()
- # loop.stop()
|