BotZ/bot_z/bot_z.py

256 lines
8.7 KiB
Python
Raw Normal View History

2019-01-07 11:16:16 +01:00
# -*- coding: utf-8 -*-
2019-01-14 12:11:17 +01:00
"""
Operator is the main object that interacts with the foreign
service. It exposes methods to login, logout and check in and out.
"""
2019-01-14 12:11:17 +01:00
from datetime import datetime, timedelta
import logging
import os
2019-01-17 18:17:56 +01:00
import pkg_resources
2019-01-21 09:33:28 +01:00
import shutil
import sys
import time
import typing as T
2019-01-14 15:41:08 +01:00
from urllib.parse import urlparse
2019-01-21 09:33:28 +01:00
geckoexe = shutil.which('geckodriver')
if geckoexe is None:
local_path = pkg_resources.resource_filename(__name__, 'bin')
try:
os.stat(os.path.join(local_path, 'geckodriver'))
os.environ['PATH'] = os.environ['PATH'] + \
':' + local_path
except FileNotFoundError:
print("Missing geckodriver executable in path", file=sys.stderr)
raise
2019-01-21 09:33:28 +01:00
from selenium import webdriver as wd # noqa
from selenium.common.exceptions import WebDriverException, NoSuchElementException # noqa
logging.basicConfig(
2019-01-14 12:11:17 +01:00
level=os.environ.get('BOTZ_LOGLEVEL', logging.INFO),
format='%(levelname)s: [%(name)s] -> %(message)s'
)
2019-01-14 12:11:17 +01:00
m_logger = logging.getLogger(__name__)
m_logger.debug("Init at debug")
def safely(f: T.Callable) -> T.Callable:
def _protection(self, *args, **kwargs):
try:
f(self, *args, **kwargs)
except WebDriverException as e:
self.logger.error("Something went wrong: %s", e)
2019-01-14 15:41:08 +01:00
finally:
self.switch_to.default_content()
return _protection
2019-01-14 12:11:17 +01:00
def _is_present(driver: wd.Firefox, xpath: str) -> bool:
try:
driver.find_element_by_xpath(xpath)
return True
except NoSuchElementException:
return False
def is_present(driver: wd.Firefox,
xpath: str,
timeout: T.Optional[timedelta]=None
) -> bool:
"""
Helper function. If an element is present in the DOM tree,
returns true. False otherwise.
"""
if timeout is None:
return _is_present(driver, xpath)
_now = datetime.now()
_elapsed = timedelta(seconds=0)
while _elapsed < timeout:
m_logger.debug("Not yet present: %s", xpath)
if _is_present(driver, xpath):
m_logger.debug("Present: %s", xpath)
return True
time.sleep(0.5)
_elapsed = datetime.now() - _now
return False
class Operator(wd.Firefox):
def __init__(
self,
base_uri: str,
name: str = None,
2019-01-14 12:11:17 +01:00
timeout: int=20,
proxy: T.Optional[T.Tuple[str, int]] = None,
headless: bool = True,
debug: bool = False,
*args, **kwargs) -> None:
"""
Adds some configuration to Firefox.
"""
self.profile = wd.FirefoxProfile()
# Do not send telemetry
self.profile.set_preference('datareporting.policy.dataSubmissionEnabled', False)
self.profile.set_preference('datareporting.healthreport.service.enabled', False)
self.profile.set_preference('datareporting.healthreport.uploadEnabled', False)
self.opts = wd.firefox.options.Options()
self.opts.headless = headless
self.debug = debug
self.base_uri = base_uri
2019-01-14 15:41:08 +01:00
self.uri = urlparse(base_uri)
2019-01-14 12:11:17 +01:00
self.timeout = timedelta(seconds=timeout)
if proxy:
self.profile.set_preference('network.proxy.type', 1)
self.profile.set_preference('network.proxy.http', proxy[0])
self.profile.set_preference('network.proxy.http_port', proxy[1])
self.profile.set_preference('network.proxy.ssl', proxy[0])
self.profile.set_preference('network.proxy.ssl_port', proxy[1])
super().__init__(firefox_profile=self.profile, options=self.opts, *args, **kwargs)
2019-01-14 15:41:08 +01:00
self.fullscreen_window()
self.z_name = name if name is not None else __name__
self.logger = logging.getLogger("{}.{}".format(__name__, self.name))
if debug:
self.logger.setLevel(logging.DEBUG)
self.logger.debug("Debug level")
self._logged_in = False
self._checked_in = False
@safely
2019-01-14 12:11:17 +01:00
def login(self, user: str, password: str, force: bool=False) -> None:
"""
Do the login and proceed.
"""
2019-01-14 12:11:17 +01:00
if self._logged_in:
self.logger.warning("Already logged in: %s", user)
if not force:
return
self.logger.info("Forcing login: %s", user)
# Retrieve login page
self.get(self.base_uri)
2019-01-14 12:11:17 +01:00
_correct_url = 'cpccchk' in self.current_url
_now = datetime.now()
_elapsed = timedelta(seconds=0)
while not _correct_url:
self.logger.debug("Not yet on login page: %s", self.current_url)
time.sleep(0.5)
_correct_url = 'cpccchk' in self.current_url
_elapsed = datetime.now() - _now
if _elapsed > self.timeout:
break
self.logger.debug("After login get: %s", self.current_url)
2019-01-14 12:11:17 +01:00
time.sleep(1)
# Username
user_form = self.find_element_by_name('m_cUserName')
# Password
pass_form = self.find_element_by_name('m_cPassword')
# Login button
login_butt = self.find_element_by_xpath('//input[contains(@id, "_Accedi")]')
# Compile and submit
user_form.send_keys(user)
pass_form.send_keys(password)
do_it = True
2019-01-14 12:11:17 +01:00
if self.debug and not force:
_do_it = input("Really do the login? [y/n] ").lower()
do_it = True if _do_it == "y" else False
if do_it:
login_butt.submit()
2019-01-14 12:11:17 +01:00
time.sleep(5)
self.logger.debug("Login result: %s", self.title)
if 'Routine window' in self.title:
2019-01-14 12:11:17 +01:00
self.logger.debug("Reloading...")
self.refresh()
2019-01-14 12:11:17 +01:00
self.switch_to.alert.accept()
if is_present(self, '//a[contains(@class, "imgMenu_ctrl")]', self.timeout):
self._logged_in = True
self.logger.info("Login success for user: %s", user)
2019-01-14 12:11:17 +01:00
else:
self.logger.error("Login failed: %s", user)
@safely
def logout(self, user: str, force: bool=False) -> None:
"""
Do the logout.
"""
if not self._logged_in:
self.logger.warning("Not yet logged in for user: %s", user)
if not force:
return
self.logger.info("Forcing logout: %s", user)
# Find the Profile menu and open it
profile_butt = self.find_element_by_xpath('//span[contains(@id, "imgNoPhotoLabel")]')
profile_butt.click()
time.sleep(1)
# Find the logout button
logout_butt = self.find_element_by_xpath('//input[@value="Logout user"]')
logout_butt.click()
if "jsp/usut_wapplogout_portlet.jsp" in self.current_url:
self.logger.info("User successfully logged out: %s", user)
else:
self.logger.warning("Logout failed: %s", user)
2019-01-14 15:41:08 +01:00
@property
def logged_in(self) -> bool:
2019-01-14 12:11:17 +01:00
"""
Check if already logged in. Checks if page is '/jsp/home.jsp'
and if login cookie is set (and not expired).
"""
2019-01-14 15:41:08 +01:00
_base_domain = '.'.join(self.uri.netloc.split('.')[-2:])
cookies = [c['name'] for c in self.get_cookies() if _base_domain in c['domain']]
_right_url = "/jsp/home.jsp" in self.current_url
_cookies = "dtLatC" in cookies
return _right_url and _cookies
@safely
def check_in(self, force: bool=False) -> None:
"""
Click the check in button.
"""
2019-01-14 15:41:08 +01:00
if not force and not self.logged_in:
self.logger.warning("Not logged in!")
return
if self._checked_in:
2019-01-14 15:41:08 +01:00
self.logger.warn("Already checked in!")
if not force:
return
2019-01-14 15:41:08 +01:00
iframe = self.find_element_by_xpath('//iframe[contains(@id, "gsmd_container.jsp")]')
self.switch_to.frame(iframe)
2019-01-14 12:11:17 +01:00
enter_butt = self.find_element_by_xpath('//input[@value="Entrata"]')
2019-01-14 15:41:08 +01:00
enter_butt.click()
# Click the check in button and change
# self._checked_in state in case of success
pass
@safely
def check_out(self, force: bool=False) -> None:
"""
Click the check out button.
"""
2019-01-14 15:41:08 +01:00
if not force and not self.logged_in:
self.logger.warning("Not logged in!")
return
if not self._checked_in:
2019-01-14 15:41:08 +01:00
self.logger.warn("Not yet checked in!")
if not force:
return
2019-01-14 15:41:08 +01:00
iframe = self.find_element_by_xpath('//iframe[contains(@id, "gsmd_container.jsp")]')
self.switch_to.frame(iframe)
2019-01-14 12:11:17 +01:00
exit_butt = self.find_element_by_xpath('//input[@value="Uscita"]')
2019-01-14 15:41:08 +01:00
exit_butt.click()
# Click the check in button and change
# self._checked_in state in case of success
pass
2019-01-14 12:11:17 +01:00
def __del__(self) -> None:
self.quit()