Skip to content

Commit

Permalink
change main() structure
Browse files Browse the repository at this point in the history
  • Loading branch information
@zdeneklapes
zdeneklapes committed Dec 29, 2023
1 parent 8006662 commit 2272495
Show file tree
Hide file tree
Showing 2 changed files with 66 additions and 52 deletions.
85 changes: 46 additions & 39 deletions bazos/__init__.py
View file
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
import argparse
from pathlib import Path
import sys
from typing import Dict, Any
from typing import Dict, Any, Callable
from distutils.util import strtobool # noqa

from bazos.scrapper import BazosScrapper, BazosUser, BazosDriver
Expand All @@ -14,31 +14,27 @@

def parse_cli_argument() -> Dict[str, Any]:
parser = argparse.ArgumentParser()
BOOL_AS_STR_ARGUMENTS_FALSE = dict(
type=lambda x: bool(strtobool(x)), default=False, nargs="?", const=True
)
BOOL_AS_STR_ARGUMENTS_TRUE = dict(
type=lambda x: bool(strtobool(x)), default=True, nargs="?", const=True
)
BOOL_AS_STR_ARGUMENTS_FALSE = dict(type=lambda x: bool(strtobool(x)), default=False, nargs="?", const=True)
BOOL_AS_STR_ARGUMENTS_TRUE = dict(type=lambda x: bool(strtobool(x)), default=True, nargs="?", const=True)
# true/false
parser.add_argument('--login',
**BOOL_AS_STR_ARGUMENTS_FALSE,
help='Login to bazos')
parser.add_argument('--bazos',
**BOOL_AS_STR_ARGUMENTS_FALSE,
help='Use bazos')
parser.add_argument('--print-rubrics',
**BOOL_AS_STR_ARGUMENTS_FALSE,
help='Print rubrics')
parser.add_argument("--verbose",
**BOOL_AS_STR_ARGUMENTS_TRUE,
help='Verbose')
parser.add_argument("--delete-all",
**BOOL_AS_STR_ARGUMENTS_FALSE,
help='Verbose')
**BOOL_AS_STR_ARGUMENTS_TRUE,
help='Delete all advertisements')
parser.add_argument("--create-all",
**BOOL_AS_STR_ARGUMENTS_TRUE,
help='Verbose')
help='Create all advertisements')
parser.add_argument("--update-all",
**BOOL_AS_STR_ARGUMENTS_TRUE,
help='Update all advertisements with updated data')
parser.add_argument('--remote',
**BOOL_AS_STR_ARGUMENTS_FALSE,
help='Use remote')
Expand All @@ -62,49 +58,60 @@ def parse_cli_argument() -> Dict[str, Any]:
return args


def loop_country(args, bazos_driver, callback: Callable):
for country in args['country']:
if args['verbose']:
print(f"==> Processing country: {country}")

bazos_user = BazosUser(country=country, args=args, driver=bazos_driver.driver)
bazos_scrapper = BazosScrapper(country=country, args=args, user=bazos_user, driver=bazos_driver.driver)
bazos_scrapper.load_page_with_cookies()

# Restore advertisements
if args['delete_all']:
bazos_scrapper.delete_all_advertisements()
if args['create_all']:
bazos_scrapper.create_all_advertisements()


def main():
args = parse_cli_argument()

# Print arguments
if args['verbose']:
print(' '.join(sys.argv))

# Driver
bazos_driver = BazosDriver(args=args, country='cz')
for country in args['country']:
if args['verbose']:
print(f"==> Processing country: {country}")

if args['login']:
bazos_user = BazosUser(country=country, args=args, driver=bazos_driver.driver)
bazos_user.authenticate()
bazos_user.save_user_credentials()
else:
bazos_user = BazosUser(country=country, args=args, driver=bazos_driver.driver)
bazos_user.exists_user_credentials()

# Login
if args['login']:
bazos_user = BazosUser(country='cz', args=args, driver=bazos_driver.driver)
bazos_user.authenticate()
bazos_user.save_user_credentials()
else:
bazos_user = BazosUser(country='cz', args=args, driver=bazos_driver.driver)
bazos_user.exists_user_credentials()

# Rubrics
if args['print_rubrics']:
for country in args['country']:
if args['print_rubrics']:
bazos_user = BazosUser(country=country, args=args, driver=bazos_driver.driver)
bazos_scrapper = BazosScrapper(country=country, args=args, user=bazos_user, driver=bazos_driver.driver)
bazos_scrapper.load_page_with_cookies()
bazos_scrapper.print_all_rubrics_and_categories()

# Bazos
if args['bazos']:
for country in args['country']:
bazos_user = BazosUser(country=country, args=args, driver=bazos_driver.driver)
if args['delete_all'] is False and args['create_all'] is False:
break

if args['verbose']:
print(f"==> Processing country: {country}")
bazos_user = BazosUser(country=country, args=args, driver=bazos_driver.driver)
bazos_scrapper = BazosScrapper(country=country, args=args, user=bazos_user, driver=bazos_driver.driver)
bazos_scrapper.load_page_with_cookies()

bazos_scrapper = BazosScrapper(country=country, args=args, user=bazos_user, driver=bazos_driver.driver)
bazos_scrapper.load_page_with_cookies()
if args['delete_all']:
bazos_scrapper.delete_all_advertisements()

# Restore advertisements
if args['delete_all']:
bazos_scrapper.delete_advertisements()
if args['create_all']:
bazos_scrapper.create_advertisements()
if args['create_all']:
bazos_scrapper.create_all_advertisements()


if __name__ == '__main__':
Expand Down
33 changes: 20 additions & 13 deletions bazos/scrapper.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@
import pickle # nosec
import sys
from os import path
from typing import Literal

from selenium import webdriver
from selenium.webdriver.chrome.service import Service
Expand Down Expand Up @@ -44,12 +45,20 @@ class XPathsBazos:
product_img_input = "//div[@class='maincontent']/form/div[1]/input[3]"


class BazosUrls:
@staticmethod
def base_url(country: Literal["cz", "sk"]) -> str:
return f"https://bazos.{country}"

@staticmethod
def moje_inzeraty_url(country: Literal["cz", "sk"]) -> str:
return path.join(BazosUrls.base_url(country), 'moje-inzeraty.php')


class BazosDriver:
def __init__(self, country: str, args: dict):
self.args = args
self.country = country
self.bazos_base_url = f"https://bazos.{country}"
self.bazos_moje_inzeraty_url = path.join(self.bazos_base_url, 'moje-inzeraty.php')
self.driver = self.get_driver()

def set_chrome_options(self) -> Options:
Expand Down Expand Up @@ -132,7 +141,7 @@ def is_authentication(self) -> bool:
return False

def authenticate(self) -> None:
self.driver.get(self.bazos_moje_inzeraty_url)
self.driver.get(BazosUrls.moje_inzeraty_url(self.country))

# Prepare authentication
telefon_input = self.driver.find_element(By.XPATH, XPathsBazos.auth_phone_input)
Expand Down Expand Up @@ -162,8 +171,6 @@ def __init__(self, country: str, args: dict, user: BazosUser, driver: webdriver.
self.driver = driver
self.args = args
self.country = country
self.bazos_base_url = f"https://bazos.{country}"
self.bazos_moje_inzeraty_url = path.join(self.bazos_base_url, 'moje-inzeraty.php')
self.user = user
self.advertisements: int

Expand Down Expand Up @@ -195,21 +202,21 @@ def print_all_rubrics_and_categories(self):
print(_dict)

def load_page_with_cookies(self) -> None:
self.driver.get(self.bazos_moje_inzeraty_url)
self.driver.get(BazosUrls.moje_inzeraty_url(self.country))
cookies_file = self.args["credentials_path"] / f"{settings.COOKIES_FILE}_{self.country}.pkl"
for cookie_dict in pickle.load(open(cookies_file, 'rb')): # nosec
self.driver.add_cookie(cookie_dict)
self.driver.get(self.bazos_moje_inzeraty_url)
self.driver.get(BazosUrls.moje_inzeraty_url(self.country))

def remove_advertisment(self):
def delete_advertisement(self):
del_btn = self.driver.find_element(By.CLASS_NAME, 'inzeratydetdel').find_element(By.TAG_NAME, 'a')
del_btn.click()
pwd_input = self.driver.find_element(By.XPATH, XPathsBazos.delete_pwd_input)
pwd_input.clear()
pwd_input.send_keys(getattr(self.user, 'password'))
self.driver.find_element(By.XPATH, XPathsBazos.delete_submit).click() # Submit-Delete

def delete_advertisements(self):
def delete_all_advertisements(self):
self.advertisements = len(self.driver.find_elements(By.CLASS_NAME, 'nadpis'))

if self.args['verbose']:
Expand All @@ -221,9 +228,9 @@ def delete_advertisements(self):

#
element.find_element(By.TAG_NAME, 'a').click()
self.remove_advertisment()
self.delete_advertisement()

def add_advertisement(self, product: Product):
def create_advertisement(self, product: Product):
# Rubrik
select_rubrik = Select(self.driver.find_element(By.XPATH, XPathsBazos.product_rubric))
select_rubrik.select_by_visible_text(get_rubric(self.country, product.rubric))
Expand Down Expand Up @@ -251,7 +258,7 @@ def add_advertisement(self, product: Product):

self.driver.find_element(By.XPATH, XPathsBazos.product_submit).click()

def create_advertisements(self) -> None:
def create_all_advertisements(self) -> None:
products = get_all_products(products_path=self.args["items_path"], country=self.country)
self.advertisements = len(products)

Expand All @@ -271,7 +278,7 @@ def create_advertisements(self) -> None:
self.driver.find_element(By.CLASS_NAME, 'pridati').click() # go to add page

self.driver.find_elements(By.CLASS_NAME, 'iconstblcell')[0].click()
self.add_advertisement(product=product)
self.create_advertisement(product=product)

def product_already_advertised(self, product: Product) -> bool:
self.load_page_with_cookies()
Expand Down

0 comments on commit 2272495

Please sign in to comment.