First Operator stub, does not yet succeedes.
This commit is contained in:
parent
a68eb9d967
commit
1fed09c655
129
bot_z/bot_z.py
129
bot_z/bot_z.py
|
@ -1,3 +1,132 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
|
||||
"""Main module."""
|
||||
|
||||
import logging
|
||||
import os
|
||||
import typing as T
|
||||
|
||||
os.environ['PATH'] = os.environ['PATH'] + \
|
||||
':' + os.path.join(os.path.abspath(os.path.curdir), 'bin')
|
||||
|
||||
from selenium import webdriver as wd
|
||||
from selenium.common.exceptions import WebDriverException
|
||||
|
||||
|
||||
logging.basicConfig(
|
||||
level=logging.INFO,
|
||||
format='%(levelname)s: [%(name)s] -> %(message)s'
|
||||
)
|
||||
|
||||
|
||||
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)
|
||||
|
||||
return _protection
|
||||
|
||||
|
||||
class Operator(wd.Firefox):
|
||||
def __init__(
|
||||
self,
|
||||
base_uri: str,
|
||||
name: str = None,
|
||||
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
|
||||
|
||||
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)
|
||||
|
||||
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
|
||||
def do_login(self, user: str, password: str) -> None:
|
||||
"""
|
||||
Do the login and proceed.
|
||||
"""
|
||||
# Retrieve login page
|
||||
self.get('{}/jsp/login.jsp'.format(self.base_uri))
|
||||
self.logger.debug("After login get: %s", self.current_url)
|
||||
# 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
|
||||
if self.debug:
|
||||
_do_it = input("Really do the login? [y/n]").lower()
|
||||
do_it = True if _do_it == "y" else False
|
||||
if do_it:
|
||||
self.logger.debug("Before clicking: %s", self.current_url)
|
||||
login_butt.submit()
|
||||
self.logger.debug("After clicking: %s", self.current_url)
|
||||
self.logger.debug("Login result: %s", self.title)
|
||||
if '404' not in self.title:
|
||||
# TODO: reckon a proper timeout mechanism
|
||||
# based on cookie expire time
|
||||
self._logged_in = True
|
||||
|
||||
@safely
|
||||
def check_in(self, force: bool=False) -> None:
|
||||
"""
|
||||
Click the check in button.
|
||||
"""
|
||||
# TODO: Check if the page displays you're already
|
||||
# checked in.
|
||||
if self._checked_in:
|
||||
self.logger.warn("Already logged in!")
|
||||
if not force:
|
||||
return
|
||||
# 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.
|
||||
"""
|
||||
# TODO: Check if the page displays you're already
|
||||
# checked out.
|
||||
if not self._checked_in:
|
||||
self.logger.warn("Not yet logged in!")
|
||||
if not force:
|
||||
return
|
||||
# Click the check in button and change
|
||||
# self._checked_in state in case of success
|
||||
pass
|
||||
|
|
Loading…
Reference in New Issue
Block a user