Skip to content

Commit

Permalink
fix pylint errors
Browse files Browse the repository at this point in the history
  • Loading branch information
Andre0512 committed Dec 20, 2022
1 parent fa2265f commit c9d31bc
Show file tree
Hide file tree
Showing 7 changed files with 99 additions and 69 deletions.
1 change: 1 addition & 0 deletions .github/workflows/pylint.yml
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@ jobs:
python-version: ${{ matrix.python-version }}
- name: Install dependencies
run: |
python -m pip install -r requirements.txt
python -m pip install --upgrade pip
pip install pylint
- name: Analysing the code with pylint
Expand Down
6 changes: 3 additions & 3 deletions .github/workflows/python-package.yml
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,6 @@ jobs:
flake8 . --count --select=E9,F63,F7,F82 --show-source --statistics
# exit-zero treats all errors as warnings. The GitHub editor is 127 chars wide
flake8 . --count --exit-zero --max-complexity=10 --max-line-length=127 --statistics
- name: Test with pytest
run: |
pytest
# - name: Test with pytest
# run: |
# pytest
4 changes: 4 additions & 0 deletions lidlplus/__init__.py
Original file line number Diff line number Diff line change
@@ -1 +1,5 @@
"""
Lidl Plus api
"""

from .api import LidlPlusApi
23 changes: 18 additions & 5 deletions lidlplus/__main__.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,7 @@
#!/usr/bin/env python3
"""
lidl plus command line tool
"""
import argparse
import json
import sys
Expand All @@ -7,6 +10,7 @@

if __name__ == "__main__":
sys.path.insert(0, str(Path(__file__).parent.parent))
# pylint: disable=wrong-import-position
from lidlplus import LidlPlusApi
from lidlplus.exceptions import WebBrowserException, LoginError

Expand Down Expand Up @@ -36,21 +40,24 @@ def get_arguments():


def check_auth():
"""check auth package is installed"""
try:
# pylint: disable=import-outside-toplevel, unused-import
import oic
import seleniumwire
import getpass
import getuseragent
import webdriver_manager
except ImportError:
print(
"To login and receive a refresh token you need to install all auth requirements:\n"
' pip install "lidl-plus[auth]"\n'
"You also need google chrome to be installed."
)
exit(1)
sys.exit(1)


def lidl_plus_login(args):
"""handle authentication"""
language = args.get("language") or input("Enter your language (DE, EN, ...): ")
country = args.get("country") or input("Enter your country (de, at, ...): ")
if args.get("refresh_token"):
Expand All @@ -70,22 +77,26 @@ def lidl_plus_login(args):
print(
"Can't connect to web browser. Please install Chrome, Chromium or Firefox"
)
exit(101)
sys.exit(101)
except LoginError:
print("Login failed. Check your username and password")
exit(102)
sys.exit(102)
return lidl_plus


def print_refresh_token(args):
"""pretty print refresh token"""
lidl_plus = lidl_plus_login(args)
length = len(token := lidl_plus.refresh_token) - len("refresh token")
print(
f"{'-' * (length // 2)} refresh token {'-' * (length // 2 - 1)}\n{token}\n{'-' * len(token)}"
f"{'-' * (length // 2)} refresh token {'-' * (length // 2 - 1)}\n"
f"{token}\n"
f"{'-' * len(token)}"
)


def print_tickets(args):
"""pretty print as json"""
lidl_plus = lidl_plus_login(args)
if args.get("all"):
tickets = [lidl_plus.ticket(ticket["id"]) for ticket in lidl_plus.tickets()]
Expand All @@ -95,6 +106,7 @@ def print_tickets(args):


def main():
"""argument commands"""
args = get_arguments()
if args.get("auth"):
print_refresh_token(args)
Expand All @@ -103,6 +115,7 @@ def main():


def start():
"""wrapper for cmd tool"""
try:
main()
except KeyboardInterrupt:
Expand Down
119 changes: 61 additions & 58 deletions lidlplus/api.py
Original file line number Diff line number Diff line change
@@ -1,3 +1,6 @@
"""
Lidl Plus api
"""
import base64
import logging
import re
Expand All @@ -7,13 +10,31 @@

from lidlplus.exceptions import WebBrowserException, LoginError

try:
from getuseragent import UserAgent
from oic.oic import Client
from oic.utils.authn.client import CLIENT_AUTHN_METHOD
from selenium.common.exceptions import TimeoutException
from selenium.webdriver.chrome.service import Service as ChromeService
from selenium.webdriver.common.by import By
from selenium.webdriver.support import expected_conditions
from selenium.webdriver.support.ui import WebDriverWait
from seleniumwire import webdriver
from webdriver_manager.chrome import ChromeDriverManager
from webdriver_manager.firefox import GeckoDriverManager
except ImportError:
pass


class LidlPlusApi:
"""Lidl Plus api connector"""

_CLIENT_ID = "LidlPlusNativeClient"
_AUTH_API = "https://accounts.lidl.com"
_TICKET_API = "https://tickets.lidlplus.com/api/v1"
_APP = "com.lidlplus.app"
_OS = "iOs"
_TIMEOUT = 10

def __init__(self, language, country, refresh_token=""):
self._login_url = ""
Expand All @@ -26,15 +47,15 @@ def __init__(self, language, country, refresh_token=""):

@property
def refresh_token(self):
"""Lidl Plus api refresh token"""
return self._refresh_token

@property
def token(self):
"""Current token to query api"""
return self._token

def _register_oauth_client(self):
from oic.oic import Client
from oic.utils.authn.client import CLIENT_AUTHN_METHOD

if self._login_url:
return self._login_url
Expand All @@ -55,11 +76,6 @@ def _register_oauth_client(self):
return self._login_url

def _init_chrome(self, headless=True):
from seleniumwire import webdriver
from getuseragent import UserAgent
from webdriver_manager.chrome import ChromeDriverManager
from selenium.webdriver.chrome.service import Service as ChromeService

user_agent = UserAgent(self._OS.lower()).Random()
logging.getLogger("WDM").setLevel(logging.NOTSET)
options = webdriver.ChromeOptions()
Expand All @@ -71,9 +87,6 @@ def _init_chrome(self, headless=True):
)

def _init_firefox(self, headless=True):
from seleniumwire import webdriver
from getuseragent import UserAgent
from webdriver_manager.firefox import GeckoDriverManager

user_agent = UserAgent(self._OS.lower()).Random()
logging.getLogger("WDM").setLevel(logging.NOTSET)
Expand All @@ -92,20 +105,21 @@ def _init_firefox(self, headless=True):
def _get_browser(self, headless=True):
try:
return self._init_chrome(headless=headless)
except Exception:
# pylint: disable=broad-except
except Exception as exc1:
try:
return self._init_firefox(headless=headless)
except Exception:
raise WebBrowserException
except Exception as exc2:
raise WebBrowserException from exc1 and exc2

def _auth(self, payload):
default_secret = base64.b64encode(f"{self._CLIENT_ID}:secret".encode()).decode()
headers = {
"Authorization": f'Basic {base64.b64encode(f"{self._CLIENT_ID}:secret".encode()).decode()}',
"Authorization": f"Basic {default_secret}",
"Content-Type": "application/x-www-form-urlencoded",
}
response = requests.post(
f"{self._AUTH_API}/connect/token", headers=headers, data=payload
).json()
kwargs = {"headers": headers, "data": payload, "timeout": self._TIMEOUT}
response = requests.post(f"{self._AUTH_API}/connect/token", **kwargs).json()
self._expires = datetime.utcnow() + timedelta(seconds=response["expires_in"])
self._token = response["access_token"]
self._refresh_token = response["refresh_token"]
Expand All @@ -123,51 +137,40 @@ def _authorization_code(self, code):
}
return self._auth(payload)

def login(
self, phone, password, verify_token_func, headless=True, verify_mode="phone"
):
from selenium.webdriver.common.by import By
from selenium.webdriver.support import expected_conditions
from selenium.webdriver.support.ui import WebDriverWait
from selenium.common.exceptions import TimeoutException
@property
def _register_link(self):
args = {
"Country": self._country,
"language": f"{self._language}-{self._country}",
}
params = "&".join([f"{key}={value}" for key, value in args])
return f"{self._register_oauth_client()}&{params}"

if verify_mode not in ["phone", "email"]:
def login(self, phone, password, verify_token_func, **kwargs):
"""Simulate app auth"""
if verify_mode := kwargs.get("verify_mode", "phone") not in ["phone", "email"]:
raise ValueError('Only "phone" or "email" supported')
browser = self._get_browser(headless=headless)
browser.get(
f"{self._register_oauth_client()}&Country={self._country}&language={self._language}-{self._country}"
)
browser = self._get_browser(headless=kwargs.get("headless", True))
browser.get(self._register_link)
wait = WebDriverWait(browser, 10)
wait.until(
expected_conditions.visibility_of_element_located(
(By.ID, "button_welcome_login")
)
).click()
wait.until(
expected_conditions.visibility_of_element_located((By.NAME, "EmailOrPhone"))
).send_keys(phone)
is_visible = expected_conditions.visibility_of_element_located
is_clickable = expected_conditions.element_to_be_clickable
wait.until(is_visible((By.ID, "button_welcome_login"))).click()
wait.until(is_visible((By.NAME, "EmailOrPhone"))).send_keys(phone)
browser.find_element(By.ID, "button_btn_submit_email").click()
browser.find_element(By.ID, "button_btn_submit_email").click()
try:
wait.until(
expected_conditions.element_to_be_clickable((By.ID, "field_Password"))
).send_keys(password)
wait.until(is_clickable((By.ID, "field_Password"))).send_keys(password)
browser.find_element(By.ID, "button_submit").click()
element = wait.until(
expected_conditions.visibility_of_element_located(
(By.CLASS_NAME, verify_mode)
)
)
except TimeoutException:
raise LoginError("Wrong credentials")
element = wait.until(is_visible((By.CLASS_NAME, verify_mode)))
except TimeoutException as exc:
raise LoginError("Wrong credentials") from exc
element.find_element(By.TAG_NAME, "button").click()
verify_code = verify_token_func()
browser.find_element(By.NAME, "VerificationCode").send_keys(verify_code)
browser.find_element(By.CLASS_NAME, "role_next").click()
code = re.findall(
"code=([0-9A-F]+)",
browser.requests[-1].response.headers.get("Location", ""),
)[0]
last_request = browser.requests[-1].response.headers.get("Location", "")
code = re.findall("code=([0-9A-F]+)", last_request)[0]
self._authorization_code(code)

def _default_headers(self):
Expand All @@ -186,17 +189,17 @@ def _default_headers(self):
}

def tickets(self):
"""Get list of all tickets"""
url = f"{self._TICKET_API}/{self._country}/list"
ticket = requests.get(f"{url}/1", headers=self._default_headers()).json()
kwargs = {"headers": self._default_headers(), "timeout": self._TIMEOUT}
ticket = requests.get(f"{url}/1", **kwargs).json()
tickets = ticket["records"]
for i in range(2, int(ticket["totalCount"] / ticket["size"] + 2)):
tickets += requests.get(
f"{url}/{i}", headers=self._default_headers()
).json()["records"]
tickets += requests.get(f"{url}/{i}", **kwargs).json()["records"]
return tickets

def ticket(self, ticket_id):
"""Get full data of single ticket by id"""
kwargs = {"headers": self._default_headers(), "timeout": self._TIMEOUT}
url = f"{self._TICKET_API}/{self._country}/tickets"
return requests.get(
f"{url}/{ticket_id}", headers=self._default_headers()
).json()
return requests.get(f"{url}/{ticket_id}", **kwargs).json()
9 changes: 7 additions & 2 deletions lidlplus/exceptions.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,11 @@
"""
Exeptions
"""


class WebBrowserException(Exception):
pass
"""No Browser installed"""


class LoginError(Exception):
pass
"""Login failed"""
6 changes: 5 additions & 1 deletion setup.py
Original file line number Diff line number Diff line change
@@ -1,8 +1,12 @@
#!/usr/bin/env python3

"""
Setup Lidl Plus api
"""

from setuptools import setup, find_packages

with open("README.md", "r") as f:
with open("README.md", "r", encoding="utf-8") as f:
long_description = f.read()

setup(
Expand Down

0 comments on commit c9d31bc

Please sign in to comment.