115 lines
3.6 KiB
Python
115 lines
3.6 KiB
Python
|
from selenium import webdriver
|
||
|
|
from selenium.webdriver.common.by import By
|
||
|
|
from selenium.webdriver.common.action_chains import ActionChains
|
||
|
|
from selenium.webdriver.support.wait import WebDriverWait
|
||
|
|
from selenium.webdriver.support import expected_conditions as EC
|
||
|
|
from selenium.common.exceptions import TimeoutException
|
||
|
|
from selenium.common.exceptions import ElementNotInteractableException
|
||
|
|
|
||
|
|
import json
|
||
|
|
from types import NoneType
|
||
|
|
|
||
|
|
class MailMessage:
|
||
|
|
def __init__(self, subject: str, sender: str, maintext: str):
|
||
|
|
self.subject = subject
|
||
|
|
self.sender = sender
|
||
|
|
self.maintext = maintext
|
||
|
|
|
||
|
|
def __eq__(self, other):
|
||
|
|
return other == self.subject or other == self.sender or other == self.maintext
|
||
|
|
|
||
|
|
def serialize(self):
|
||
|
|
return {
|
||
|
|
"subject": self.subject,
|
||
|
|
"sender": self.sender,
|
||
|
|
"maintext": self.maintext,
|
||
|
|
"date": "", # fill this in later
|
||
|
|
}
|
||
|
|
|
||
|
|
def serialize_to_string(self) -> str:
|
||
|
|
return json.dumps(self.serialize())
|
||
|
|
|
||
|
|
@classmethod
|
||
|
|
def from_json(cls, json_data):
|
||
|
|
if isinstance(json_data, str):
|
||
|
|
parsed = json.loads(json_data)
|
||
|
|
else:
|
||
|
|
parsed = json_data
|
||
|
|
|
||
|
|
return cls(json_data["subject"], json_data["sender"], json_data["maintext"])
|
||
|
|
|
||
|
|
class MailProvider:
|
||
|
|
def __init__(self, username: str, password: str, cachefile: NoneType | str = None):
|
||
|
|
self.username: str = username
|
||
|
|
self.password: str = password
|
||
|
|
self.sessionmessages = []
|
||
|
|
self.webdriver = None # TODO: fill in type information for this instance
|
||
|
|
self.cachefile = cachefile
|
||
|
|
|
||
|
|
def _get_webdriver(self):
|
||
|
|
self.webdriver = webdriver.Chrome()
|
||
|
|
|
||
|
|
def _get_elem_children(self, elem):
|
||
|
|
return elem.find_elements(By.XPATH, "./child::*")
|
||
|
|
|
||
|
|
def _to_elem(self, xpath):
|
||
|
|
return self.webdriver.find_element(By.XPATH, xpath)
|
||
|
|
|
||
|
|
def _wait_for_elem(self, xpath):
|
||
|
|
try: WebDriverWait(self.webdriver, 20).until(EC.presence_of_element_located((By.XPATH, xpath)))
|
||
|
|
except TimeoutException: pass
|
||
|
|
|
||
|
|
def get_seen_messages(self):
|
||
|
|
try:
|
||
|
|
if not self.cachefile:
|
||
|
|
self.cachefile = self.__class__.__qualname__
|
||
|
|
|
||
|
|
with open(self.cachefile, "r") as f:
|
||
|
|
loaded = json.load(f)["payload"]
|
||
|
|
for entry in loaded:
|
||
|
|
self.seen_messages.append(MailMessage.from_json(entry))
|
||
|
|
except (
|
||
|
|
FileNotFoundError,
|
||
|
|
json.decoder.JSONDecodeError
|
||
|
|
) as e:
|
||
|
|
self.seen_messages = []
|
||
|
|
|
||
|
|
def write_seen_messages(self):
|
||
|
|
if not self.cachefile:
|
||
|
|
self.cachefile = self.__name__
|
||
|
|
|
||
|
|
with open(self.cachefile, "w") as f:
|
||
|
|
data = {"payload": [item.serialize() for item in self.seen_messages]}
|
||
|
|
print(data)
|
||
|
|
f.write(
|
||
|
|
json.dumps(
|
||
|
|
data
|
||
|
|
)
|
||
|
|
)
|
||
|
|
|
||
|
|
def add_to_seen(self, msg):
|
||
|
|
self.seen_messages.append(msg)
|
||
|
|
|
||
|
|
def is_seen(self, msg):
|
||
|
|
for i in self.seen_messages:
|
||
|
|
if i == msg:
|
||
|
|
return True
|
||
|
|
return False
|
||
|
|
|
||
|
|
def _type_in_elem(self, xpath, typed):
|
||
|
|
target_element = self._to_elem(xpath)
|
||
|
|
ActionChains(self.webdriver) \
|
||
|
|
.move_to_element(target_element) \
|
||
|
|
.send_keys(typed) \
|
||
|
|
.perform()
|
||
|
|
|
||
|
|
def _click_elem(self, xpath):
|
||
|
|
try:
|
||
|
|
target_element = self._to_elem(xpath)
|
||
|
|
except:
|
||
|
|
target_element = xpath
|
||
|
|
|
||
|
|
ActionChains(self.webdriver) \
|
||
|
|
.move_to_element(target_element) \
|
||
|
|
.click(target_element) \
|
||
|
|
.perform()
|