130 lines
6.7 KiB
Python
130 lines
6.7 KiB
Python
import os
|
|
import sys
|
|
import random
|
|
import asyncio
|
|
import logging
|
|
|
|
from playwright.async_api import async_playwright
|
|
|
|
# 로거 설정
|
|
logger = logging.getLogger("playwright_scraper")
|
|
logger.setLevel(logging.DEBUG)
|
|
handler = logging.StreamHandler()
|
|
formatter = logging.Formatter("[%(asctime)s] %(levelname)s - %(message)s")
|
|
handler.setFormatter(formatter)
|
|
logger.addHandler(handler)
|
|
|
|
# 로컬 이미지 파일 경로 (실제 파일 경로로 변경)
|
|
LOCAL_IMAGE_PATH = os.path.join(os.path.dirname(__file__), "o1cn.webp")
|
|
|
|
async def run():
|
|
try:
|
|
async with async_playwright() as p:
|
|
# 브라우저 실행 파일 경로 설정
|
|
if getattr(sys, 'frozen', False):
|
|
browser_path = os.path.join(os.path.dirname(sys.executable), 'src', 'browsers', 'chromium-1112', 'chrome-win', 'chrome.exe')
|
|
else:
|
|
browser_path = os.path.join(os.path.dirname(__file__), 'src', 'browsers', 'chromium-1112', 'chrome-win', 'chrome.exe')
|
|
|
|
logger.debug(f"브라우저 경로: {browser_path}")
|
|
|
|
# 사용자 에이전트 설정
|
|
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",
|
|
])
|
|
logger.debug(f"user_agent: {user_agent}")
|
|
|
|
# 브라우저 시작 (headless 모드)
|
|
browser = await p.chromium.launch(
|
|
headless=False,
|
|
executable_path=browser_path,
|
|
args=[
|
|
'--disable-popup-blocking',
|
|
'--start-maximized',
|
|
'--window-size=1920,1080'
|
|
]
|
|
)
|
|
|
|
# 시크릿 브라우저 컨텍스트 생성
|
|
context = await browser.new_context(
|
|
user_agent=user_agent,
|
|
geolocation={"latitude": 37.5665, "longitude": 126.9780},
|
|
locale="ko-KR",
|
|
permissions=["geolocation", "notifications"]
|
|
)
|
|
|
|
# 페이지 열기
|
|
page = await context.new_page()
|
|
await page.goto("https://itemscout.io/sourcing/1688/search")
|
|
logger.info("검색 페이지 접속 완료.")
|
|
|
|
# 파일 업로드: 파일 입력 요소는 <input id="fileUpload" ...> 이므로 선택자 사용
|
|
file_upload_selector = "input#fileUpload" # 또는 xpath "//*[@id='fileUpload']"
|
|
# 파일 업로드
|
|
await page.set_input_files(file_upload_selector, LOCAL_IMAGE_PATH)
|
|
logger.info(f"로컬 이미지 파일 업로드 완료: {LOCAL_IMAGE_PATH}")
|
|
|
|
# 파일 업로드 후 검색 결과가 로드될 때까지 적절히 대기 (필요에 따라 wait_for_selector 조정)
|
|
results_container_selector = "div.mb-9.flex.w-full.flex-wrap.gap-x-5.gap-y-6.text-base-100"
|
|
await page.wait_for_selector(results_container_selector, timeout=10000)
|
|
logger.info("검색 결과 로드 완료.")
|
|
|
|
# 모든 상품 카드 요소 선택 (각 카드가 <a> 태그로 되어 있음)
|
|
product_cards = await page.query_selector_all(f"{results_container_selector} > a")
|
|
num_products = len(product_cards)
|
|
logger.info(f"검색 결과 {num_products}개의 상품 발견.")
|
|
|
|
# 각 상품 카드에서 정보 추출
|
|
for idx, card in enumerate(product_cards, start=1):
|
|
# 상품명: span 요소 (클래스 'truncate')
|
|
product_name_element = await card.query_selector("span.truncate")
|
|
product_name = (await product_name_element.text_content()).strip() if product_name_element else "N/A"
|
|
|
|
# 상품 이미지: 카드 내부 첫 번째 div의 img 태그
|
|
image_element = await card.query_selector("div:first-child img")
|
|
product_image = await image_element.get_attribute("src") if image_element else "N/A"
|
|
|
|
# 가격 정보: div.leading-6 내에 두 개의 span 요소
|
|
price_container = await card.query_selector("div.leading-6")
|
|
price_spans = await price_container.query_selector_all("span") if price_container else []
|
|
price_yuan = (await price_spans[0].text_content()).strip() if len(price_spans) >= 1 else "N/A"
|
|
price_krw = (await price_spans[1].text_content()).strip() if len(price_spans) >= 2 else "N/A"
|
|
|
|
# 재구매율: p 요소 내에 텍스트 "재구매율" 포함, 그 안의 span.font-bold
|
|
repurchase_element = await card.query_selector("p:has-text('재구매율') span.font-bold")
|
|
repurchase_rate = (await repurchase_element.text_content()).strip() if repurchase_element else "N/A"
|
|
|
|
# 판매량: p 요소 내에 텍스트 "판매량" 포함
|
|
sales_element = await card.query_selector("p:has-text('판매량') span.font-bold")
|
|
sales_volume = (await sales_element.text_content()).strip() if sales_element else "N/A"
|
|
|
|
# 평점: p 요소 내에 텍스트 "평점" 포함
|
|
rating_element = await card.query_selector("p:has-text('평점') span.font-bold")
|
|
rating = (await rating_element.text_content()).strip() if rating_element else "N/A"
|
|
|
|
# 최소 구매수량: p 요소 내에 텍스트 "최소 구매수량" 포함
|
|
min_order_element = await card.query_selector("p:has-text('최소 구매수량') span.font-bold")
|
|
min_order = (await min_order_element.text_content()).strip() if min_order_element else "N/A"
|
|
|
|
logger.info(
|
|
f"[상품 {idx}] 상품명: {product_name}\n"
|
|
f" 상품 이미지: {product_image}\n"
|
|
f" 가격: {price_yuan} / {price_krw}\n"
|
|
f" 재구매율: {repurchase_rate}\n"
|
|
f" 판매량: {sales_volume}\n"
|
|
f" 평점: {rating}\n"
|
|
f" 최소 구매수량: {min_order}\n"
|
|
)
|
|
|
|
await browser.close()
|
|
|
|
except Exception as e:
|
|
logger.exception(f"오류 발생: {e}")
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(run())
|