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