ckh08045
/
AutoTao
1
0
Fork 0
Code Issues 1 Pull Requests Packages Projects Releases Wiki Activity
AutoTao/qrtest.py

281 lines
12 KiB
Python
Raw Blame History

import sys
import os
import random
import asyncio
import uuid
from PySide6.QtWidgets import QApplication, QLabel, QMainWindow, QVBoxLayout, QWidget
from PySide6.QtGui import QPixmap
from playwright.async_api import async_playwright
import qasync
# ============================================================
# 1. PySide6 GUI: QR 이미지 표시 창
# ============================================================
class QRCodeWindow(QMainWindow):
def __init__(self, img_bytes):
super().__init__()
self.setWindowTitle("타오바오 QR 코드 로그인")
central_widget = QWidget()
self.setCentralWidget(central_widget)
layout = QVBoxLayout(central_widget)
# QR 이미지를 표시할 QLabel
self.label = QLabel()
layout.addWidget(self.label)
pixmap = QPixmap()
if not pixmap.loadFromData(img_bytes):
self.label.setText("이미지를 로드할 수 없습니다.")
else:
self.label.setPixmap(pixmap)
# 이미지 크기에 따라 창 크기 조절
self.resize(pixmap.width(), pixmap.height())
# ============================================================
# 2. 로그인 페이지 접속 함수 (브라우저 실행 및 로그인 페이지 열기)
# ============================================================
async def open_login_page(p):
"""
제공해주신 브라우저 설정 코드를 사용하여 타오바오 로그인 페이지에 접속합니다.
반환값: browser, page, login_url
"""
# 브라우저 경로 설정
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')
# 사용자 에이전트 설정
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",
])
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()
login_url = "https://login.taobao.com/member/login.jhtml"
await page.goto(login_url)
return browser, page, login_url
# ============================================================
# 3. QR 스크린샷 캡쳐 함수 (QR 이미지를 리턴)
# ============================================================
async def capture_qr_screenshot(page):
"""
페이지에서 'div#qrcode-img canvas' 요소의 스크린샷을 찍어 이미지 바이트 데이터를 반환합니다.
"""
await page.wait_for_selector("div#qrcode-img canvas", timeout=10000)
await asyncio.sleep(2) # QR 코드가 완전히 렌더링될 때까지 딜레이
qr_canvas = await page.query_selector("div#qrcode-img canvas")
if not qr_canvas:
print("QR 코드 캔버스를 찾을 수 없습니다.")
return None
img_bytes = await qr_canvas.screenshot()
return img_bytes
# ============================================================
# 4. 로그인 완료 감시 및 검색 페이지 이동 함수
# ============================================================
async def monitor_login(page, login_url, search_url):
"""
1초마다 현재 페이지 URL을 출력하다가 로그인 완료(로그인 페이지 URL과 달라짐)를 감지하면
지정한 검색 URL로 이동합니다.
"""
print("로그인 완료 감시 시작...")
while True:
current_url = page.url
print("현재 URL:", current_url)
if current_url != login_url:
print("로그인 완료 감지됨. 현재 URL:", current_url)
break
await asyncio.sleep(1)
# 지정한 검색 페이지로 이동
await page.goto(search_url)
print("검색 페이지로 이동했습니다:", search_url)
# 상품 카드가 나타날 때까지 대기
try:
await page.wait_for_selector(".doubleCard--gO3Bz6bu", timeout=15000)
print("상품 카드 요소가 나타났습니다.")
except Exception as e:
print("상품 카드 요소를 찾지 못했습니다.", e)
# ============================================================
# 5. 상품정보 추출 함수 (현재 페이지의 모든 상품 카드 처리)
# ============================================================
async def extract_products_from_current_page(page, card_or_img="card"):
"""
현재 페이지의 모든 상품 카드(.doubleCard--gO3Bz6bu)를 순회하며,
상품 이미지 URL, 상품명, 가격, 구매량 정보를 추출하고,
각 상품 카드의 스크린샷을 찍어 로컬 파일에 저장한 후, 파일 경로를 함께 반환합니다.
"""
await page.wait_for_selector(".doubleCard--gO3Bz6bu", timeout=10000)
cards = await page.query_selector_all(".doubleCard--gO3Bz6bu")
products = []
# 로컬에 상품 이미지 저장 폴더 생성
images_folder = "product_images"
os.makedirs(images_folder, exist_ok=True)
for card in cards:
# 상품 이미지 URL 추출
img_elem = await card.query_selector("img.mainPic--Ds3X7I8z")
img_src = await img_elem.get_attribute("src") if img_elem else None
# 상품명 추출
name_elem = await card.query_selector("div.title--qJ7Xg_90 span")
name_text = (await name_elem.inner_text()).strip() if name_elem else ""
# 가격 추출 (정수 부분)
price_elem = await card.query_selector("span.priceInt--yqqZMJ5a")
price_text = (await price_elem.inner_text()).strip() if price_elem else ""
# 구매량 추출
sales_elem = await card.query_selector("span.realSales--XZJiepmt")
sales_text = (await sales_elem.inner_text()).strip() if sales_elem else ""
# 각 상품 카드의 스크린샷을 찍어서 로컬 파일로 저장
unique_filename = f"product_{uuid.uuid4().hex}.png"
local_path = os.path.join(images_folder, unique_filename)
try:
if card_or_img == "card":
await card.screenshot(path=local_path)
elif card_or_img == "img":
if img_elem:
await img_elem.screenshot(path=local_path)
else:
await card.screenshot(path=local_path)
print(f"상품 스크린샷 저장됨: {local_path}")
except Exception as e:
print("상품 스크린샷 저장 실패:", e)
local_path = None
product = {
"image_url": img_src,
"local_image": local_path,
"name": name_text,
"price": price_text,
"sales": sales_text
}
products.append(product)
return products
# ============================================================
# 6. 여러 페이지에서 상품정보 수집 함수
# ============================================================
async def collect_products(page, pages_to_collect):
"""
사용자가 지정한 페이지 수만큼 각 페이지의 상품 정보를 추출합니다.
각 페이지 당 최대 45개 상품이 있다고 가정하며, 페이지 이동 버튼을 클릭하여 이동합니다.
"""
all_products = []
for i in range(pages_to_collect):
print(f"\n--- Page {i+1} 상품정보 추출 시작 ---")
products = await extract_products_from_current_page(page)
print(f"페이지 {i+1}에서 {len(products)}개의 상품 정보 추출됨.")
all_products.extend(products)
# 마지막 페이지가 아니라면 다음 페이지로 이동
if i < pages_to_collect - 1:
try:
# 페이지 이동 버튼 영역 대기
await page.wait_for_selector("div.next-pagination-list", timeout=10000)
next_page = i + 2 # 현재 페이지가 i+1이면, 다음 페이지는 i+2
# 페이지 이동 버튼 클릭 (버튼 내 텍스트가 페이지 번호와 일치)
btn_selector = f"button.next-pagination-item:has-text('{next_page}')"
await page.click(btn_selector)
print(f"페이지 {next_page}로 이동 버튼 클릭됨.")
# 페이지 이동 후 로드 대기
await page.wait_for_load_state("networkidle")
await asyncio.sleep(2)
except Exception as e:
print(f"페이지 {next_page}로 이동 실패: {e}")
break
return all_products
# ============================================================
# 7. 전체 동작을 통합하는 메인 함수 (qasync 사용)
# ============================================================
async def main(pages_to_collect):
async with async_playwright() as p:
# 1) 로그인 페이지 접속
browser, page, login_url = await open_login_page(p)
# 2) QR 스크린샷 캡쳐 (GUI에 표시할 이미지)
img_bytes = await capture_qr_screenshot(page)
if img_bytes is None:
print("QR 코드 이미지를 가져오지 못했습니다.")
await browser.close()
return
# 3) 검색 페이지 URL (로그인 완료 후 이동)
search_url = ("https://s.taobao.com/search?commend=all&ie=utf8page=1&"
"q=%E5%A5%B3%E5%A3%AB%E8%A5%BF%E6%9C%8D&search_type=item")
# 4) GUI 창 생성 (QR 이미지 표시)
app = QApplication.instance()
if app is None:
app = QApplication(sys.argv)
window = QRCodeWindow(img_bytes)
window.show()
# 5) 로그인 완료 감시 및 검색 페이지 이동
await monitor_login(page, login_url, search_url)
# 6) 사용자가 지정한 페이지 수만큼 상품 정보 수집
products = await collect_products(page, pages_to_collect)
print(f"\n총 {len(products)}개의 상품 정보를 수집했습니다.")
for idx, prod in enumerate(products, start=1):
print(f"[{idx}] {prod}")
# 7) 디버깅 후 브라우저 종료 및 GUI 종료 처리
await browser.close()
await asyncio.sleep(2)
app.quit()
# ============================================================
# 8. qasync를 통한 이벤트 루프 실행 및 사용자 입력 처리
# ============================================================
if __name__ == "__main__":
# 사용자에게 수집할 페이지 수 입력 받기 (기본값 1)
page_count_input = input("수집할 페이지 수를 입력하세요 (기본값 1): ")
try:
pages_to_collect = int(page_count_input) if page_count_input.strip() else 1
except Exception:
pages_to_collect = 1
app = QApplication(sys.argv)
loop = qasync.QEventLoop(app)
asyncio.set_event_loop(loop)
with loop:
loop.run_until_complete(main(pages_to_collect))
Reference in New Issue View Git Blame Copy Permalink