164 lines
10 KiB
Python
164 lines
10 KiB
Python
from playwright.async_api import async_playwright
|
|
import logging, random, os, time, re
|
|
|
|
# 로거 인스턴스 가져오기
|
|
logger = logging.getLogger('default_logger')
|
|
|
|
class MarketAutomation:
|
|
async def start_browser(self):
|
|
try:
|
|
|
|
"""브라우저 설정 및 인스턴스 생성"""
|
|
self.playwright = await async_playwright().start()
|
|
# 드라이버의 경로 설정
|
|
driver_path = os.path.join(os.path.dirname(__file__), 'drivers', 'webkit-1983', 'playwright.exe') # 경로는 설치 환경에 맞게 조정 필요
|
|
|
|
# WebKit 브라우저로 변경
|
|
self.browser = await self.playwright.webkit.launch(headless=False, executable_path=driver_path) # headless 모드 설정
|
|
self.context = await self.browser.new_context(
|
|
user_agent=random.choice([
|
|
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/109.0.0.0 Safari/537.36",
|
|
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/109.0.0.0 Safari/537.36 Edg/109.0.0.0",
|
|
"Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:108.0) Gecko/20100101 Firefox/108.0",
|
|
"Mozilla/5.0 (Macintosh; Intel Mac OS X 12_0) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/15.0 Safari/605.1.15",
|
|
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/109.0.0.0 Safari/537.36 OPR/85.0.0.0",
|
|
]))
|
|
self.page = await self.context.new_page()
|
|
|
|
try:
|
|
await self.page.goto(self.url, wait_until='networkidle')
|
|
# self.is_page_loaded = True
|
|
logger.debug("브라우저가 성공적으로 시작되었습니다.")
|
|
except Exception as e:
|
|
# logger.error("브라우저 시작 중 오류 발생", exc_info=True)
|
|
self.is_page_loaded = False # 로드 실패 처리
|
|
|
|
except Exception as e:
|
|
logger.error("브라우저 시작 중 오류 발생", exc_info=True)
|
|
|
|
async def goto_market(self, url):
|
|
try:
|
|
await self.page.goto(url)
|
|
await self.page.wait_for_load_state('networkidle') # 네트워크 활동이 일정 시간 동안 없을 때까지 대기
|
|
logger.debug(f"{url} 주소로 이동 성공")
|
|
except Exception as e:
|
|
logger.error("웹 페이지 로드 실패", exc_info=True)
|
|
|
|
async def search_and_delete_product(self, market, product_name, dryrunFactor):
|
|
try:
|
|
product_name = self.normalize_space(product_name)
|
|
|
|
if market == "ESM":
|
|
await self.page.fill('textarea#txtKeyword', product_name)
|
|
await self.page.click('a[href="javascript:;"] img[src*="btn_search1.gif"]')
|
|
await self.page.wait_for_selector('div#gridcolumn-1014') # 검색 결과 로드 대기
|
|
await self.page.click('div#gridcolumn-1014')
|
|
await self.page.wait_for_selector('span.css_btn1 a[onclick*="SellStateDelete"]', state='attached')
|
|
await self.page.click('span.css_btn1 a[onclick*="SellStateDelete"]')
|
|
await self.page.wait_for_load_state('networkidle') # 다음 동작 전 페이지 로딩 완료 대기
|
|
|
|
|
|
elif market == "쿠팡":
|
|
await self.page.fill('input[data-v-671ac22c][type="text"]', product_name)
|
|
# await self.page.fill('#searchContainer > dd > div > dl.search-condition > dd.searchWordDD > span > table > tr.table-row-focused > td:nth-child(2) > input', product_name)
|
|
|
|
await self.page.click('button.searchBtn')
|
|
print("검색버튼 클릭")
|
|
|
|
# 검색 결과 로드 대기
|
|
await self.page.wait_for_selector('#rootContainer > div:nth-child(6) > div.search-result-section')
|
|
|
|
# "검색 결과가 없습니다." 메시지 확인
|
|
no_results_selector = '#rootContainer > div:nth-child(6) > div.search-result-section > div.table-wrapper-container > div.table-wrapper.border-right.hide-see-auto-option > table > tbody > tr > td > span'
|
|
no_results_text = await self.page.text_content(no_results_selector)
|
|
|
|
if "검색 결과가 없습니다." in no_results_text:
|
|
logger.debug("검색 결과가 없습니다.")
|
|
logger.debug(f"{no_results_text}")
|
|
|
|
# 여기서 추가 작업 처리 가능 (예: 사용자에게 알림, 다른 작업 수행 등)
|
|
return # 더 이상의 작업을 진행하지 않고 함수 종료
|
|
|
|
# 검색된 첫 번째 상품의 상품명 가져오기
|
|
first_product_name = await self.page.text_content('#rootContainer > div:nth-child(6) > div.search-result-section > div.table-wrapper-container > div.table-wrapper.border-right.hide-see-auto-option > table > tbody > tr:nth-child(1) > td.fixed-col.left-align.small-horizontal-padding.middle-vertical-padding.border-right-double.editable-cell.editable-popup > div > span')
|
|
first_product_name = self.normalize_space(first_product_name)
|
|
|
|
# first_product_name = await self.page.text_content('#rootContainer > div:nth-child(6) > div.search-result-section > div.table-wrapper-container > div.table-wrapper.border-right.hide-see-auto-option > table > tbody > tr:nth-child(1) > td.fixed-col.left-align.small-horizontal-padding.middle-vertical-padding.border-right-double.editable-cell.editable-popup > div > span')
|
|
logger.debug(f"product_name : {product_name}")
|
|
logger.debug(f"first_product_name : {first_product_name}")
|
|
# 상품명이 검색 키워드와 일치하는지 확인
|
|
if first_product_name == product_name:
|
|
logger.debug("상품명 일치")
|
|
# 해당 상품의 체크박스 선택
|
|
await self.page.check('#rootContainer > div:nth-child(6) > div.search-result-section > div.table-wrapper-container > div.table-wrapper.border-right.hide-see-auto-option > table > tbody > tr:nth-child(1) > td:nth-child(2) > span > input[type="checkbox"]')
|
|
time.sleep(5)
|
|
logger.debug("체크박스 체크")
|
|
|
|
await self.page.click('.wing-web-component:nth-child(10)') # 삭제 버튼
|
|
|
|
await self.page.wait_for_selector('button.confirm.alert-confirm') # 삭제 확인 버튼 대기
|
|
|
|
if not dryrunFactor:
|
|
await self.page.click('button.confirm.alert-confirm') # 삭제 확인 버튼 클릭
|
|
logger.debug("삭제확인 클릭")
|
|
else:
|
|
logger.debug("DryrunFactor 설정됨")
|
|
await self.page.click('.showSweetAlert .cancel') # 삭제 취소 버튼 클릭
|
|
#container-wing-v2 > div:nth-child(5) > div.sweet-alert.showSweetAlert.visible > div.alert-buttons > button.cancel
|
|
first_product_name = None # 검색된 상품 변수 초기화
|
|
await self.page.wait_for_load_state('networkidle') # 삭제 후 페이지 로딩 대기
|
|
else:
|
|
logger.warning("검색된 상품이 없거나 일치하지 않습니다.")
|
|
|
|
# await self.page.wait_for_load_state('networkidle') # 다음 동작 전 페이지 로딩 완료 대기
|
|
|
|
|
|
elif market == "스스":
|
|
await self.page.fill('input#prd_name', product_name)
|
|
await self.page.click('button[ng-click*="vm.viewData.selectedProductStatusType = undefined"]')
|
|
await self.page.wait_for_selector('input.ag-selection-checkbox')
|
|
await self.page.click('input.ag-selection-checkbox')
|
|
await self.page.wait_for_selector('button[ng-click*="vm.func.changeProductStatus(\'DELETE\')"]', state='attached')
|
|
await self.page.click('button[ng-click*="vm.func.changeProductStatus(\'DELETE\')"]')
|
|
await self.page.wait_for_load_state('networkidle') # 다음 동작 전 페이지 로딩 완료 대기
|
|
|
|
elif market == "롯데온":
|
|
await self.page.fill('input#w2input', product_name)
|
|
await self.page.click('input[value="조회"]')
|
|
await self.page.wait_for_selector('input[type="checkbox"][name="wq_uuid_1420_header__column9_checkboxLabel_"]')
|
|
await self.page.click('input[type="checkbox"][name="wq_uuid_1420_header__column9_checkboxLabel_"]')
|
|
await self.page.wait_for_selector('input[value="선택상품삭제"]', state='attached')
|
|
await self.page.click('input[value="선택상품삭제"]')
|
|
await self.page.wait_for_load_state('networkidle') # 다음 동작 전 페이지 로딩 완료 대기
|
|
|
|
elif market == "11번가":
|
|
await self.page.fill('input#prdNm', product_name)
|
|
await self.page.click('button#btnSearch')
|
|
await self.page.wait_for_selector('div[role="checkbox"]')
|
|
await self.page.click('div[role="checkbox"]')
|
|
await self.page.wait_for_selector('a[href="javascript:fnListDelete(\'U\');"]', state='attached')
|
|
await self.page.click('a[href="javascript:fnListDelete(\'U\');"]')
|
|
await self.page.wait_for_load_state('networkidle') # 다음 동작 전 페이지 로딩 완료 대기
|
|
|
|
if first_product_name:
|
|
logger.debug(f"{product_name} 상품 삭제 완료")
|
|
return True
|
|
else:
|
|
logger.debug(f"{product_name} 상품 삭제 실패")
|
|
return False
|
|
except Exception as e:
|
|
logger.error(f"{market}에서 상품 검색 및 삭제 중 오류 발생", exc_info=True)
|
|
return False
|
|
|
|
async def close_browser(self):
|
|
try:
|
|
await self.browser.close()
|
|
logger.debug("브라우저가 성공적으로 닫혔습니다.")
|
|
except Exception as e:
|
|
logger.error("브라우저 종료 중 오류 발생", exc_info=True)
|
|
|
|
def normalize_space(self, text):
|
|
# 연속된 공백을 하나의 공백으로 치환하고, 양쪽 끝의 공백을 제거합니다.
|
|
text = re.sub(r'\s+', ' ', text).strip()
|
|
return text
|