import requests from bs4 import BeautifulSoup import pandas as pd import time import json import random import logging import undetected_chromedriver as uc from urllib.parse import urljoin import mysql.connector # Параметри підключення до MySQL db_config = { "host": "10.8.0.1", # Адреса сервера бази даних "user": "integration", # Ім'я користувача "password": "?Q8/{lVK2N08Yk", # Пароль "database": "Salsify" # Назва бази даних } # Підключення до бази даних try: connection = mysql.connector.connect(**db_config) # print("Підключення до бази даних успішне.") except mysql.connector.Error as e: # print(f"Помилка підключення: {e}") exit() # SQL-запит для отримання даних query = """ SELECT SL.SKU, MLW.`Account SKU`, CONCAT('https://www.lowes.com/pd/-/',MLW.`Account SKU`) AS LINK FROM Salsify.MainData SL LEFT JOIN MySamm.Lowes MLW ON MLW.SKU = SL.SKU LEFT JOIN Salsify.Lowes SLW ON SLW.SKU = SL.SKU WHERE MLW.SKU LIKE '%' AND(SL.Status='Active' OR SL.Status='Liquidation') """ try: # Виконання запиту cursor = connection.cursor(dictionary=True) cursor.execute(query) results = cursor.fetchall() # Завантаження даних у DataFrame df_input = pd.DataFrame(results) df_input['Full_URL'] = df_input['LINK'] # Створення списків urls = df_input['Full_URL'].dropna().tolist() # urls = ['https://www.lowes.com/pd/-/1002750654','https://www.lowes.com/pd/-/1002750832','https://www.lowes.com/pd/-/1002751330','https://www.lowes.com/pd/-/1002156222','https://www.lowes.com/pd/-/1002750662','https://www.lowes.com/pd/-/1000061269','https://www.lowes.com/pd/-/1002156730','https://www.lowes.com/pd/-/1002751342','https://www.lowes.com/pd/-/1002750788','https://www.lowes.com/pd/-/1002750794','https://www.lowes.com/pd/-/1002750804','https://www.lowes.com/pd/-/1002751552','https://www.lowes.com/pd/-/1002751024','https://www.lowes.com/pd/-/1002751042','https://www.lowes.com/pd/-/1002751058','https://www.lowes.com/pd/-/1002157238','https://www.lowes.com/pd/-/1002751430','https://www.lowes.com/pd/-/1002751440','https://www.lowes.com/pd/-/1002751460','https://www.lowes.com/pd/-/1002157746','https://www.lowes.com/pd/-/1001731120','https://www.lowes.com/pd/-/1001731094','https://www.lowes.com/pd/-/1001731146','https://www.lowes.com/pd/-/5000165181','https://www.lowes.com/pd/-/5013494627','https://www.lowes.com/pd/-/5013494955','https://www.lowes.com/pd/-/5013494663','https://www.lowes.com/pd/-/5013494819','https://www.lowes.com/pd/-/5013494637','https://www.lowes.com/pd/-/5013494899'] additional_data = df_input['SKU'].tolist() # print("Дані успішно отримані.") finally: # Закриття з'єднання if connection.is_connected(): cursor.close() connection.close() # print("З'єднання з базою даних закрито.") # Налаштування логування logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') # Список для збору даних all_data = [] # Максимальна кількість спроб підключення max_retries = 3 timeout = 20 headers = { "Accept-Language": "en-US,en;q=0.9", "Accept-Encoding": "gzip, deflate, br", "Referer": "https://www.lowes.com/", "Connection": "keep-alive", "Accept-Language": "en-US,en;q=0.9", "Accept-Encoding": "gzip, deflate, br", "Referer": "https://www.lowes.com/", "Connection": "keep-alive", "Upgrade-Insecure-Requests": "1", "Sec-Fetch-Dest": "document", "Sec-Fetch-Mode": "navigate", "Sec-Fetch-Site": "same-origin", "Sec-Fetch-User": "?1", "Cache-Control": "max-age=0", "TE": "trailers", "DNT": "1", } def getDriver(): options = uc.ChromeOptions() options.add_argument("--no-sandbox") # Required for some environments options.add_argument("--disable-blink-features=AutomationControlled") # Hide automation # options.add_argument("--disable-dev-shm-usage") # Fix /dev/shm issue in Docker options.add_argument("--disable-gpu") # Disable GPU acceleration options.add_argument("--headless=new") # Run in headless mode options.add_argument("--start-maximized") # Open maximized window options.add_argument("--disable-extensions") # Disable extensions options.add_argument("--disable-infobars") options = uc.ChromeOptions() return uc.Chrome(options=options, use_subprocess=True) driver = getDriver() # Функція для обробки одного URL def process_url(url): for attempt in range(max_retries): try: logging.error(f"Підключення до {url}, спроба {attempt + 1}...") driver.execute_cdp_cmd("Network.clearBrowserCache", {}) driver.execute_cdp_cmd("Network.clearBrowserCookies", {}) driver.get(url); time.sleep(4) soup = BeautifulSoup(driver.page_source, 'html.parser') links = soup.find_all('a', class_="LinkBase-sc-2ngoxx-0 eViSzl backyard link size--small color--black") pdf_links = [] for link in links: # Витягуємо href href = link.get('href', '-') if href.endswith('.pdf'): formatted_href = urljoin("https:", href).replace(" ", "") span = link.find('span', class_="label link-label") text = span.get_text(strip=True) if span else "No label" pdf_links.append({'PDF Link': formatted_href, 'PDF Text': text}) if not pdf_links: pdf_links.append({'PDF Link': 'PDF not found', 'PDF Text': 'PDF not found'}) return pdf_links except requests.exceptions.RequestException as e: # logging.warning(f"Помилка для {url}: {e}") time.sleep(2 + random.uniform(1, 3)) return [] # Обробка URL-адрес for i, url in enumerate(urls): if i % 15 == 0 and i != 0: driver.quit() time.sleep(3) driver = getDriver() data = process_url(url) if data: # Якщо для одного URL є кілька результатів, додаємо їх у окремі стовпці # row = {'Base URL': urls[i], 'SKU': additional_data[i]} row = {'Base URL': urls[i]} for j, item in enumerate(data): row[f"PDF Link{j+1}"] = item['PDF Link'] row[f"PDF Text{j+1}"] = item['PDF Text'] all_data.append(row) logging.error(f"Processed {i} of {len(urls)}") time.sleep(random.uniform(3, 6)) # Затримка для уникнення блокування # Створення DataFrame # df_output = pd.DataFrame(all_data) print(json.dumps(all_data)) driver.quit() # Збереження в Excel # output_file = "LowesPDF.json" # with open(output_file, "w") as json_file: # json.dump(results, json_file, indent=4) # logging.info(f"Дані збережені в {output_file}")