This commit is contained in:
parent
89fdd21058
commit
cc0bfbfa4b
|
|
@ -0,0 +1,484 @@
|
|||
import os
|
||||
import sys
|
||||
import time
|
||||
import json
|
||||
import re
|
||||
import logging
|
||||
from typing import Dict
|
||||
|
||||
from pywinauto import Application, findwindows, timings, clipboard
|
||||
from pywinauto.controls.hwndwrapper import HwndWrapper
|
||||
|
||||
from selenium import webdriver
|
||||
from selenium.webdriver.chrome.service import Service
|
||||
from selenium.webdriver.chrome.options import Options
|
||||
from selenium.webdriver.common.by import By
|
||||
from selenium.webdriver.support.ui import WebDriverWait
|
||||
from selenium.webdriver.support import expected_conditions as EC
|
||||
|
||||
from bs4 import BeautifulSoup
|
||||
|
||||
from PIL import ImageGrab
|
||||
|
||||
from src.shoppingLens import ShoppingLensScraper
|
||||
|
||||
|
||||
class WhaleController:
|
||||
def __init__(self, logger=None):
|
||||
self.logger = logger
|
||||
self.driver = None
|
||||
self.browser_pid = None
|
||||
self.whale_app = None
|
||||
self.whale_window = None
|
||||
self.lens = ShoppingLensScraper(self.logger)
|
||||
# 번역 성공 여부 플래그 (translate_image 용)
|
||||
self.translation_success_flag = False
|
||||
# 이미지 최소 해상도 기준 (width x height)
|
||||
self.min_image_width = 200
|
||||
self.min_image_height = 150
|
||||
|
||||
def get_base_dir(self):
|
||||
"""
|
||||
실행 환경에 따라 base_dir을 설정하는 메서드.
|
||||
cx_Freeze로 패키징된 경우 실행 파일의 경로, 일반 Python 환경일 경우 __file__을 기준으로 설정.
|
||||
"""
|
||||
if getattr(sys, 'frozen', False): # 패키징된 경우
|
||||
base_dir = os.path.dirname(sys.executable)
|
||||
internal_dir = os.path.join(base_dir, '_internal') # _internal 디렉토리 포함
|
||||
if os.path.exists(internal_dir): # _internal 디렉토리가 존재하면 base_dir로 설정
|
||||
return internal_dir
|
||||
else: # 일반 Python 실행 환경
|
||||
base_dir = os.path.dirname(os.path.abspath(__file__))
|
||||
return base_dir
|
||||
|
||||
|
||||
def start_whale_Browser(self):
|
||||
"""웨일 제어"""
|
||||
try:
|
||||
|
||||
base_path = self.get_base_dir()
|
||||
whale_exe_path = os.path.join(base_path, "browsers", "whale", "whale.exe")
|
||||
user_data_dir = os.path.join(base_path, "browsers", "whale", "user_data")
|
||||
cache_dir = os.path.join(base_path, "browsers", "whale", "cache")
|
||||
extension_path = os.path.join(base_path, "browsers", "whale", "extensions", "gadfmnjdnhkncfcibhfleoojcdimdcbd", "1.1.11_0")
|
||||
|
||||
# Selenium WebDriver 설정
|
||||
chromedriver_path = os.path.join(base_path, "browsers", "chromedriver_128.0.6613.137.exe")
|
||||
|
||||
chrome_service = Service(chromedriver_path)
|
||||
|
||||
chrome_options = Options()
|
||||
# chrome_options.add_argument("--headless")
|
||||
chrome_options.add_argument("--no-sandbox")
|
||||
chrome_options.add_argument("--disable-dev-shm-usage")
|
||||
chrome_options.add_argument("--remote-debugging-port=9222")
|
||||
chrome_options.add_argument(f"--user-data-dir={user_data_dir}") # 사용자 데이터 폴더 설정
|
||||
chrome_options.add_argument(f"--disk-cache-dir={cache_dir}") # 캐시 폴더 설정
|
||||
chrome_options.add_argument(f"--load-extension={extension_path}") # 확장 프로그램 로드
|
||||
|
||||
# binary 위치 설정
|
||||
chrome_options.binary_location = whale_exe_path
|
||||
|
||||
|
||||
# 사용자 에이전트 설정
|
||||
chrome_options.add_argument(
|
||||
"user-agent=Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
|
||||
"(KHTML, like Gecko) Chrome/130.0.0.0 Whale/4.29.282.14 Safari/537.36"
|
||||
)
|
||||
|
||||
# 웨일 브라우저 실행 (Selenium)
|
||||
self.driver = webdriver.Chrome(service=chrome_service, options=chrome_options)
|
||||
time.sleep(0.2) # 페이지 로드 잠시 대기
|
||||
|
||||
# Selenium의 서비스 프로세스 PID를 통해 pywinauto로 제어할 창을 찾기 위한 PID 저장
|
||||
self.browser_pid = self.driver.service.process.pid
|
||||
self.logger.log(f"웨일 브라우저 PID: {self.browser_pid}", level=logging.INFO)
|
||||
|
||||
# pywinauto를 사용하여 웨일 창 찾기 (두 창 이름 중 하나로 검색)
|
||||
self.find_whale()
|
||||
|
||||
except Exception as e:
|
||||
self.logger.log(f"웨일 브라우저 시작 중 오류 발생: {e}", level=logging.ERROR, exc_info=True)
|
||||
|
||||
|
||||
def find_whale(self):
|
||||
"""
|
||||
pywinauto를 사용하여 웨일 브라우저 창을 찾고, 창 위치 및 크기를 조정.
|
||||
'whale://new-tab-page-third-party/ - Whale' 또는 '새 시크릿 탭 - Whale'을 찾음.
|
||||
"""
|
||||
try:
|
||||
# 최대 10초 동안 원하는 창이 나타나기를 대기
|
||||
timings.wait_until(10, 0.5, lambda: any(
|
||||
window.name in ['whale://new-tab-page-third-party/ - Whale', '새 시크릿 탭 - Whale']
|
||||
for window in findwindows.find_elements()
|
||||
))
|
||||
windows = findwindows.find_elements()
|
||||
for window in windows:
|
||||
if window.process_id == self.browser_pid or window.name in ['whale://new-tab-page-third-party/ - Whale', '새 시크릿 탭 - Whale']:
|
||||
self.logger.log(f"찾은 창: {window.name}, PID: {window.process_id}", level=logging.INFO)
|
||||
# pywinauto로 애플리케이션 연결
|
||||
self.whale_app = Application(backend="uia").connect(process=window.process_id)
|
||||
self.whale_window = self.whale_app.top_window()
|
||||
# 창 위치 및 크기 조정
|
||||
hwnd_wrapper = HwndWrapper(self.whale_window.handle)
|
||||
hwnd_wrapper.move_window(x=1, y=1, width=1280, height=720)
|
||||
self.whale_window.set_focus()
|
||||
self.logger.log("웨일 브라우저 창을 제어합니다.", level=logging.INFO)
|
||||
return
|
||||
self.logger.log("웨일 브라우저 창을 찾지 못했습니다.", level=logging.WARNING)
|
||||
except Exception as e:
|
||||
self.logger.log(f"웨일 제어 중 오류 발생: {e}", level=logging.ERROR, exc_info=True)
|
||||
|
||||
def safe_click(self, element, right=False):
|
||||
"""
|
||||
click_input() 호출 시 활성 데스크탑 오류가 발생하면 click() 또는 right_click()으로 대체.
|
||||
"""
|
||||
try:
|
||||
if right:
|
||||
element.right_click_input()
|
||||
else:
|
||||
element.click_input()
|
||||
except RuntimeError as e:
|
||||
if "no active desktop" in str(e):
|
||||
self.logger.log("Active desktop 오류 발생 - 메시지 기반 클릭으로 대체합니다.", level=logging.WARNING)
|
||||
if right:
|
||||
element.right_click()
|
||||
else:
|
||||
element.click()
|
||||
else:
|
||||
raise
|
||||
|
||||
def search_and_parse(self, image_url: str, min_price: int = 30000, top_n: int = 5) -> Dict:
|
||||
"""
|
||||
URL로 검색 후 필터링된 상위 제품 정보 반환.
|
||||
:param image_url: 스크래핑할 이미지 URL
|
||||
:param min_price: 최소 가격 필터
|
||||
:param top_n: 상위 N개 제품 정보
|
||||
:return: 요청된 형식의 제품 정보
|
||||
"""
|
||||
|
||||
try:
|
||||
self.driver.get(image_url)
|
||||
|
||||
result_page_url, type_index = self.lens.lens_Search(whale_window=self.whale_window, image_url=image_url)
|
||||
self.logger.log(f"result_page_url: {result_page_url}", level=logging.INFO)
|
||||
|
||||
self.driver.get(result_page_url)
|
||||
|
||||
time.sleep(1)
|
||||
|
||||
# 페이지 소스 출력
|
||||
page_source = self.driver.page_source
|
||||
self.logger.log("페이지소스 가져오기", level=logging.DEBUG)
|
||||
|
||||
# BeautifulSoup로 HTML 파싱
|
||||
soup = BeautifulSoup(page_source, "html.parser")
|
||||
|
||||
# __NEXT_DATA__ 스크립트 태그 찾기
|
||||
next_data_script = soup.find("script", id="__NEXT_DATA__")
|
||||
|
||||
if not next_data_script or not next_data_script.string:
|
||||
raise ValueError("__NEXT_DATA__를 찾을 수 없습니다.")
|
||||
|
||||
# JSON 디코딩
|
||||
next_data = json.loads(next_data_script.string)
|
||||
|
||||
initialState = next_data["props"]["pageProps"]["initialState"]
|
||||
# self.logger.log(initialState, level=logging.INFO)
|
||||
|
||||
result = json.loads(initialState)
|
||||
|
||||
products_data = self.get_product_list(result, type_index, min_price, top_n)
|
||||
|
||||
return products_data
|
||||
|
||||
except Exception as e:
|
||||
self.logger.log(f"렌즈 검색 중 오류 발생: {e}", level=logging.ERROR, exc_info=True)
|
||||
|
||||
|
||||
def get_product_list(self, data: Dict, type_index, min_price: int = 50000, top_n: int = 5) -> list:
|
||||
"""
|
||||
검색 결과에서 제품 리스트를 추출하고 필터링, 정렬 및 결과 구성.
|
||||
:param data: 검색 결과 JSON 데이터
|
||||
:param min_price: 최소 가격
|
||||
:param top_n: 상위 N개 제품 추출
|
||||
:return: 요청된 형식의 제품 정보 (상품별 딕셔너리 리스트)
|
||||
"""
|
||||
try:
|
||||
self.logger.log(f"data 타입 : {type(data)}", level=logging.INFO)
|
||||
|
||||
products = []
|
||||
related_tags = []
|
||||
|
||||
if type_index == 0:
|
||||
# 1번 타입 데이터 처리
|
||||
self.logger.log(f"1번 타입 데이터 처리", level=logging.INFO)
|
||||
|
||||
products_list = data["compositeProducts"]["list"]
|
||||
products = [product["item"] for product in products_list if "item" in product]
|
||||
|
||||
# 1번 타입에서만 관련 태그 추출
|
||||
related_tags = data.get("props", {}).get("pageProps", {}).get("relatedTags", [])
|
||||
related_tags = [tag.strip() for tag in related_tags if tag]
|
||||
|
||||
elif type_index == 1:
|
||||
# 2번 타입 데이터 처리
|
||||
self.logger.log(f"1번 타입 데이터 처리", level=logging.INFO)
|
||||
|
||||
products_list = data["imageSearch"]["searchResult"]["similarImages"]
|
||||
products = products_list # 유사 상품 리스트
|
||||
|
||||
# # 최소 가격 필터링
|
||||
# self.logger.log(f"최소 가격 필터링 : {min_price}", level=logging.INFO)
|
||||
# filtered_products = [
|
||||
# product for product in products if int(product.get("price", 0)) >= min_price
|
||||
# ]
|
||||
|
||||
# # 상위 N개 정렬 (랭킹 기준)
|
||||
# self.logger.log(f"상위 [{top_n}]개 정렬 (랭킹 기준)", level=logging.INFO)
|
||||
|
||||
# sorted_products = sorted(filtered_products, key=lambda p: int(p.get("rank", 0)))[:top_n]
|
||||
|
||||
self.logger.log(f"상위 [{top_n}]개 ", level=logging.INFO)
|
||||
sorted_products = products[:top_n]
|
||||
|
||||
# 상품별 딕셔너리 구성
|
||||
product_data = []
|
||||
|
||||
self.logger.log(f"상품별 딕셔너리 구성", level=logging.INFO)
|
||||
|
||||
for product in sorted_products:
|
||||
product_info = {
|
||||
"title": product.get("productTitle", "") or product.get("title", ""),
|
||||
"price": product.get("price", ""),
|
||||
"category": [
|
||||
product.get("category1Name", ""),
|
||||
product.get("category2Name", ""),
|
||||
product.get("category3Name", ""),
|
||||
product.get("category4Name", ""),
|
||||
],
|
||||
"review_count": product.get("reviewCount", 0) or product.get("reviewCountSum", 0),
|
||||
"purchaseCnt": product.get("purchaseCnt", 0),
|
||||
"overseaTp": product.get("overseaTp", 0),
|
||||
"rank": product.get("rank", 0),
|
||||
"manuTag": product.get("manuTag", ""),
|
||||
}
|
||||
product_data.append(product_info)
|
||||
|
||||
self.logger.log(f"총 {len(product_data)}개의 제품 정보가 처리됨.", level=logging.DEBUG)
|
||||
return product_data
|
||||
|
||||
except KeyError as e:
|
||||
self.logger.log(f"제품 리스트를 처리하는 중 오류 발생: {e}", level=logging.ERROR, exc_info=True)
|
||||
return []
|
||||
|
||||
|
||||
|
||||
def translate_image(self, url: str, path: str = None) -> bool:
|
||||
"""
|
||||
이미지 번역 작업을 수행.
|
||||
1. Selenium으로 URL 이동하여 페이지 로드
|
||||
2. pywinauto로 창을 다시 찾고, 해상도 체크 후 번역 명령 실행
|
||||
3. 번역 후 클립보드에서 이미지 추출(저장 경로가 주어졌다면)
|
||||
"""
|
||||
try:
|
||||
if url.startswith("https://assets.alicdn.com"):
|
||||
self.logger.log("assets.alicdn.com 확인 : 번역작업을 패스합니다.", level=logging.INFO)
|
||||
return False
|
||||
|
||||
# Selenium을 사용하여 URL로 이동
|
||||
self.driver.get(url)
|
||||
self.logger.log(f"{url} 로 이동 중...", level=logging.DEBUG)
|
||||
|
||||
# 이미지 요소가 로딩될 때까지 동적으로 대기 (최대 10초)
|
||||
wait = WebDriverWait(self.driver, 10)
|
||||
wait.until(EC.presence_of_element_located((By.TAG_NAME, "img")))
|
||||
self.logger.log("페이지의 이미지 요소 로딩 완료.", level=logging.INFO)
|
||||
|
||||
# pywinauto를 통해 최신 창 상태 업데이트
|
||||
self.find_whale()
|
||||
|
||||
if not self.check_image_size():
|
||||
self.logger.log("해상도가 기준보다 낮아 작업을 패스합니다.", level=logging.INFO)
|
||||
self.click_back_button()
|
||||
return False
|
||||
|
||||
# 우클릭 후 '이미지 번역' 명령 실행
|
||||
self.translation_success_flag = self.right_click_on_image_and_inspect()
|
||||
self.click_back_button()
|
||||
|
||||
# 결과 저장 (경로가 주어진 경우)
|
||||
if path:
|
||||
try:
|
||||
# 클립보드에서 이미지 가져오기
|
||||
clipboard_image = ImageGrab.grabclipboard()
|
||||
if clipboard_image:
|
||||
clipboard_image.save(path, format='PNG')
|
||||
self.logger.log(f"번역된 이미지가 {path}에 저장되었습니다.", level=logging.INFO)
|
||||
else:
|
||||
self.logger.log("클립보드에 이미지가 존재하지 않아 파일로 저장할 수 없습니다.", level=logging.ERROR)
|
||||
return False
|
||||
except Exception as e:
|
||||
self.logger.log(f"이미지 저장 중 오류 발생: {e}", level=logging.ERROR, exc_info=True)
|
||||
return False
|
||||
|
||||
return True if self.translation_success_flag else False
|
||||
|
||||
except Exception as e:
|
||||
self.logger.log(f"번역 중 오류 발생: {e}", level=logging.ERROR, exc_info=True)
|
||||
return False
|
||||
|
||||
|
||||
def check_image_size(self) -> bool:
|
||||
"""
|
||||
창 제목에서 이미지의 해상도(너비×높이)를 추출하여 기준 충족 여부를 반환.
|
||||
"""
|
||||
try:
|
||||
window_title = self.whale_window.window_text()
|
||||
self.logger.log(f"Window Title: {window_title}", level=logging.DEBUG)
|
||||
match = re.search(r"\((\d+)[×x](\d+)\)", window_title)
|
||||
if match:
|
||||
width = int(match.group(1))
|
||||
height = int(match.group(2))
|
||||
self.logger.log(f"이미지 해상도: {width}×{height}", level=logging.INFO)
|
||||
if width < self.min_image_width or height < self.min_image_height:
|
||||
self.logger.log(f"이미지 해상도가 기준 이하 [{self.min_image_width} x {self.min_image_height}]입니다.", level=logging.INFO)
|
||||
return False
|
||||
return True
|
||||
elif any(ext in window_title.lower() for ext in [".jpg", ".png", ".jpeg", ".gif"]):
|
||||
self.logger.log("해상도 정보는 없지만, 파일 확장자가 감지되어 번역 진행합니다.", level=logging.INFO)
|
||||
return True
|
||||
self.logger.log("해상도 정보와 파일 확장자가 모두 감지되지 않습니다.", level=logging.WARNING)
|
||||
return False
|
||||
except Exception as e:
|
||||
self.logger.log(f"이미지 사이즈 측정 중 오류 발생: {e}", level=logging.ERROR, exc_info=True)
|
||||
return False
|
||||
|
||||
def right_click_on_image_and_inspect(self) -> bool:
|
||||
"""
|
||||
이미지 요소에서 우클릭 후 '이미지 번역 (R)' 메뉴를 선택하여 번역 명령 실행.
|
||||
"""
|
||||
try:
|
||||
# pywinauto를 사용하여 이미지 요소 찾기
|
||||
image = self.whale_window.child_window(title="누락된 이미지 설명을 확인하려면 컨텍스트 메뉴를 여세요.", control_type="Image")
|
||||
if image.exists():
|
||||
self.safe_click(image, right=True)
|
||||
self.logger.log("이미지 요소에서 우클릭 수행.", level=logging.DEBUG)
|
||||
else:
|
||||
self.logger.log("이미지 요소를 찾을 수 없습니다.", level=logging.ERROR)
|
||||
return False
|
||||
|
||||
# '이미지 번역 (R)' 메뉴 항목 선택
|
||||
translate_menu_item = self.whale_window.child_window(title="이미지 번역 (R)", control_type="MenuItem")
|
||||
self.safe_click(translate_menu_item)
|
||||
self.logger.log("이미지 번역 명령 실행.", level=logging.DEBUG)
|
||||
time.sleep(0.5)
|
||||
status = self.check_translation_status()
|
||||
if status == "success":
|
||||
self.logger.log("번역이 성공적으로 완료되었습니다.", level=logging.INFO)
|
||||
return True
|
||||
elif status == "fail":
|
||||
self.logger.log("번역에 실패했습니다.", level=logging.WARNING)
|
||||
return False
|
||||
else:
|
||||
self.logger.log("번역 상태를 확인할 수 없습니다.", level=logging.WARNING)
|
||||
return False
|
||||
except Exception as e:
|
||||
self.logger.log(f"우클릭 및 번역 명령 실행 중 오류 발생: {e}", level=logging.ERROR, exc_info=True)
|
||||
return False
|
||||
|
||||
def check_translation_status(self, max_wait_time: int = 10, check_interval: float = 0.5) -> str:
|
||||
"""
|
||||
번역 진행 상태를 확인.
|
||||
- "success": 번역 성공 (클립보드에 이미지 데이터 존재)
|
||||
- "fail": 번역 불가 (텍스트로 '번역할 영역을 선택하세요.' 감지)
|
||||
- "timeout": 제한 시간 내에 결과 확인 실패
|
||||
"""
|
||||
start_time = time.time()
|
||||
while time.time() - start_time < max_wait_time:
|
||||
try:
|
||||
fail_indicator = self.whale_window.child_window(title="번역할 영역을 선택하세요.", control_type="Text")
|
||||
if fail_indicator.exists():
|
||||
self.logger.log("번역할 문구가 없는 이미지입니다.", level=logging.INFO)
|
||||
return "fail"
|
||||
image_element = self.whale_window.child_window(title="누락된 이미지 설명을 확인하려면 컨텍스트 메뉴를 여세요.", control_type="Image")
|
||||
if image_element.exists():
|
||||
self.safe_click(image_element, right=True)
|
||||
success_indicator = self.whale_window.child_window(title="이미지 복사(C)", control_type="MenuItem")
|
||||
if success_indicator.wait('visible', timeout=5):
|
||||
self.safe_click(success_indicator)
|
||||
time.sleep(0.5)
|
||||
formats = clipboard.GetClipboardFormats()
|
||||
for format_id in formats:
|
||||
format_name = clipboard.GetFormatName(format_id)
|
||||
if format_name in ("CF_BITMAP", "CF_DIB"):
|
||||
image_data = clipboard.GetData(format_id=format_id)
|
||||
if isinstance(image_data, bytes):
|
||||
self.logger.log("번역 성공: 이미지 데이터 복사 완료.", level=logging.INFO)
|
||||
return "success"
|
||||
self.logger.log("번역이 아직 완료되지 않았습니다. 다시 시도 중...", level=logging.INFO)
|
||||
except Exception as e:
|
||||
self.logger.log("클립보드 접근 중 오류 발생", level=logging.ERROR, exc_info=True)
|
||||
return "error"
|
||||
time.sleep(check_interval)
|
||||
self.logger.log("번역 확인 시간 초과.", level=logging.WARNING)
|
||||
return "timeout"
|
||||
|
||||
def click_back_button(self):
|
||||
"""
|
||||
'뒤로' 버튼 클릭하여 이전 페이지로 복귀.
|
||||
"""
|
||||
try:
|
||||
back_button = self.whale_window.child_window(title="뒤로", control_type="Button")
|
||||
if back_button.exists():
|
||||
self.safe_click(back_button)
|
||||
self.logger.log("'뒤로' 버튼 클릭.", level=logging.DEBUG)
|
||||
else:
|
||||
self.logger.log("'뒤로' 버튼을 찾을 수 없습니다.", level=logging.WARNING)
|
||||
except Exception as e:
|
||||
self.logger.log(f"'뒤로' 버튼 클릭 중 오류 발생: {e}", level=logging.ERROR, exc_info=True)
|
||||
|
||||
|
||||
def close_whale_br(self):
|
||||
"""
|
||||
Selenium 드라이버 및 pywinauto로 제어한 웨일 브라우저 종료.
|
||||
"""
|
||||
try:
|
||||
self.logger.log("브라우저 종료 중...", level=logging.DEBUG)
|
||||
if self.driver:
|
||||
self.driver.quit()
|
||||
else:
|
||||
self.logger.log("브라우저 객체를 찾을 수 없습니다.", level=logging.WARNING)
|
||||
if self.whale_app:
|
||||
self.whale_app.kill()
|
||||
self.logger.log("pywinauto로 제어한 웨일 창 종료.", level=logging.INFO)
|
||||
except Exception as e:
|
||||
self.logger.log(f"브라우저 종료 중 오류 발생: {e}", level=logging.ERROR, exc_info=True)
|
||||
|
||||
|
||||
|
||||
# 예시 실행 코드
|
||||
if __name__ == "__main__":
|
||||
# 로깅 설정
|
||||
logging.basicConfig(level=logging.DEBUG)
|
||||
logger = logging.getLogger("WhaleController")
|
||||
|
||||
# WhaleController 인스턴스 생성
|
||||
whale_controller = WhaleController(logger=logger)
|
||||
|
||||
# 웨일 브라우저 시작 (Selenium + pywinauto)
|
||||
whale_controller.start_whale_browser()
|
||||
|
||||
# 검색 및 파싱 예제 (이미지 URL과 가격/상위 제품 개수 지정)
|
||||
test_image_url = "https://file.percenty.co.kr/public/652bed8e865b1f32ea62bf1f/products/6774440131d56e408c286cd4/04ae3cd9-4fdd-4677-8da8-d37dbf240107.jpg"
|
||||
# test_image_url = 'https://file.percenty.co.kr/public/652bed8e865b1f32ea62bf1f/products/677443d531d56e408c286cca/b136631a-66c7-42c5-88a5-fa720215f259.jpg'
|
||||
|
||||
products = whale_controller.search_and_parse(image_url=test_image_url, min_price=50000, top_n=5)
|
||||
logger.info(f"검색 결과: {json.dumps(products, indent=4, ensure_ascii=False)}")
|
||||
|
||||
# 이미지 번역 예제 (번역 후 파일 저장 경로 지정)
|
||||
translation_result = whale_controller.translate_image(url=test_image_url, path="translated_image.png")
|
||||
logger.info(f"번역 작업 결과: {translation_result}")
|
||||
|
||||
# 브라우저 종료
|
||||
whale_controller.close_whale_br()
|
||||
|
|
@ -0,0 +1,347 @@
|
|||
import time
|
||||
import os, sys
|
||||
import re
|
||||
import logging
|
||||
import pyperclip
|
||||
from pywinauto import Application, findwindows, clipboard, timings
|
||||
from pywinauto.controls.hwndwrapper import HwndWrapper
|
||||
from pywinauto.findwindows import ElementNotFoundError
|
||||
from pywinauto.timings import wait_until
|
||||
from PIL import ImageGrab
|
||||
|
||||
class WhaleTranslator:
|
||||
def __init__(self, logger):
|
||||
self.logger = logger
|
||||
|
||||
self.whale_app = None
|
||||
self.whale_window = None
|
||||
|
||||
self.translation_success_flag = False # 번역 성공 플래그
|
||||
self.failure_count = 0 # 실패 횟수
|
||||
|
||||
self.min_image_width = 200
|
||||
self.min_image_height = 150
|
||||
|
||||
def get_base_dir(self):
|
||||
"""
|
||||
실행 환경에 따라 base_dir을 설정하는 메서드.
|
||||
cx_Freeze로 패키징된 경우 실행 파일의 경로, 일반 Python 환경일 경우 __file__을 기준으로 설정.
|
||||
"""
|
||||
if getattr(sys, 'frozen', False): # 패키징된 경우
|
||||
base_dir = os.path.dirname(sys.executable)
|
||||
internal_dir = os.path.join(base_dir, '_internal') # _internal 디렉토리 포함
|
||||
if os.path.exists(internal_dir): # _internal 디렉토리가 존재하면 base_dir로 설정
|
||||
return internal_dir
|
||||
else: # 일반 Python 실행 환경
|
||||
base_dir = os.path.dirname(os.path.abspath(__file__))
|
||||
return base_dir
|
||||
|
||||
def safe_click(self, element, right=False):
|
||||
"""
|
||||
click_input() 호출 시 활성 데스크탑 오류가 발생하면 click() 또는 right_click()을 사용하여 클릭합니다.
|
||||
"""
|
||||
try:
|
||||
if right:
|
||||
element.right_click_input()
|
||||
else:
|
||||
element.click_input()
|
||||
except RuntimeError as e:
|
||||
if "no active desktop" in str(e):
|
||||
self.logger.log("Active desktop 오류 발생 - 메시지 기반 클릭으로 대체합니다.", level=logging.WARNING)
|
||||
if right:
|
||||
element.right_click()
|
||||
else:
|
||||
element.click()
|
||||
else:
|
||||
raise
|
||||
|
||||
def start_whale_browser(self):
|
||||
base_path = self.get_base_dir()
|
||||
whale_exe_path = os.path.join(base_path, "browsers", "whale", "whale.exe")
|
||||
user_data_dir = os.path.join(base_path, "browsers", "whale", "user_data")
|
||||
cache_dir = os.path.join(base_path, "browsers", "whale", "cache")
|
||||
|
||||
self.whale_app = Application(backend="uia").start(
|
||||
f'"{whale_exe_path}" --incognito --user-data-dir="{user_data_dir}" --disk-cache-dir="{cache_dir}"'
|
||||
)
|
||||
|
||||
# 창이 완전히 생성될 때까지 대기
|
||||
self.whale_window = self.find_whale_window()
|
||||
|
||||
if self.whale_window:
|
||||
self.logger.log("웨일 시크릿 모드로 시작 완료.", level=logging.INFO)
|
||||
else:
|
||||
self.logger.log("웨일 창을 찾을 수 없습니다.", level=logging.WARNING)
|
||||
|
||||
def find_whale_window(self):
|
||||
try:
|
||||
# 최대 10초 동안 '새 시크릿 탭 - Whale' 창이 나타나기를 기다림
|
||||
timings.wait_until(10, 0.5, lambda: any(window.name == '새 시크릿 탭 - Whale' for window in findwindows.find_elements()))
|
||||
|
||||
windows = findwindows.find_elements()
|
||||
for window in windows:
|
||||
if window.name == '새 시크릿 탭 - Whale':
|
||||
whale_pid = window.process_id
|
||||
self.whale_app = Application(backend="uia").connect(process=whale_pid)
|
||||
self.whale_window = self.whale_app.top_window()
|
||||
|
||||
# 위치 및 크기 조절
|
||||
self.hwnd_wrapper = HwndWrapper(self.whale_window.handle)
|
||||
self.hwnd_wrapper.move_window(x=1, y=1, width=1280, height=720)
|
||||
self.whale_window.set_focus()
|
||||
|
||||
self.logger.log("웨일 창을 성공적으로 찾았습니다.", level=logging.INFO)
|
||||
return self.whale_window
|
||||
self.logger.log("'새 시크릿 탭 - Whale' 창을 찾을 수 없습니다.", level=logging.ERROR)
|
||||
except Exception as e:
|
||||
self.logger.log(f"웨일 창 탐색 중 오류 발생: {e}", level=logging.ERROR, exc_info=True)
|
||||
return None
|
||||
|
||||
def translate_image(self, url, path=None):
|
||||
|
||||
if url.startswith("https://assets.alicdn.com"):
|
||||
self.logger.log("assets.alicdn.com 확인 : 번역작업을 패스합니다.", level=logging.INFO)
|
||||
return False
|
||||
|
||||
if self.whale_app:
|
||||
try:
|
||||
self.navigate_to_url(url)
|
||||
|
||||
if not self.check_image_size():
|
||||
self.logger.log("해상도가 기준보다 낮아 작업을 패스합니다.", level=logging.INFO)
|
||||
self.click_back_button()
|
||||
return False # 해상도 기준 미달로 작업 종료
|
||||
|
||||
self.translation_success_flag = self.right_click_on_image_and_inspect()
|
||||
self.click_back_button()
|
||||
|
||||
# 경로를 인자로 받을 경우 해당 경로에 파일 저장
|
||||
if path:
|
||||
try:
|
||||
# 클립보드에서 이미지 가져오기
|
||||
clipboard_image = ImageGrab.grabclipboard()
|
||||
if clipboard_image:
|
||||
clipboard_image.save(path, format='PNG')
|
||||
self.logger.log(f"번역된 이미지가 {path}에 저장되었습니다.", level=logging.INFO)
|
||||
else:
|
||||
self.logger.log("클립보드에 이미지가 존재하지 않아 파일로 저장할 수 없습니다.", level=logging.ERROR)
|
||||
return False
|
||||
except Exception as e:
|
||||
self.logger.log(f"이미지 저장 중 오류 발생: {e}", level=logging.ERROR, exc_info=True)
|
||||
return False
|
||||
|
||||
return True if self.translation_success_flag else False
|
||||
|
||||
except Exception as e:
|
||||
self.logger.log(f"번역 중 오류 발생: {e}", level=logging.ERROR, exc_info=True)
|
||||
return False
|
||||
else:
|
||||
self.logger.log('웨일 창을 찾을 수 없습니다.', level=logging.ERROR)
|
||||
return False
|
||||
|
||||
def navigate_to_url(self, url):
|
||||
"""주소창에 URL을 입력하고 페이지 로딩을 대기"""
|
||||
retry_count = 3 # 주소창 찾기 재시도 횟수
|
||||
for attempt in range(retry_count):
|
||||
try:
|
||||
# URL을 클립보드에 복사
|
||||
pyperclip.copy(url)
|
||||
|
||||
# 주소창을 클릭하여 URL 붙여넣기
|
||||
address_bar = self.whale_window.child_window(title="주소창 및 검색창", control_type="Edit")
|
||||
self.safe_click(address_bar)
|
||||
|
||||
# Ctrl + V로 URL 붙여넣기 후 Enter 키 입력
|
||||
address_bar.type_keys("^v{ENTER}", with_spaces=True)
|
||||
self.logger.log(f"{url}로 이동 중...", level=logging.DEBUG)
|
||||
|
||||
# 5초 동안 0.5초 간격으로 이미지 요소가 나타나는지 검사
|
||||
start_time = time.time()
|
||||
while time.time() - start_time < 5:
|
||||
try:
|
||||
# 이미지 요소를 찾으면 즉시 반환
|
||||
image_element = self.whale_window.child_window(
|
||||
title="누락된 이미지 설명을 확인하려면 컨텍스트 메뉴를 여세요.",
|
||||
control_type="Image"
|
||||
)
|
||||
if image_element.exists(timeout=1):
|
||||
self.logger.log("페이지 로딩 완료: 이미지 요소가 나타났습니다.", level=logging.DEBUG)
|
||||
return
|
||||
except Exception as e:
|
||||
self.logger.log(f"이미지 요소 확인 중 오류 발생: {e}", level=logging.DEBUG)
|
||||
|
||||
# 안전장치 1: 현재 창 제목 확인
|
||||
window_title = self.whale_window.window_text()
|
||||
if re.match(r".+\.(jpg|png|jpeg) \(\d+x\d+\)", window_title, re.IGNORECASE):
|
||||
self.logger.log(f"창 제목에 이미지 파일과 해상도가 감지됨: {window_title}", level=logging.INFO)
|
||||
return
|
||||
|
||||
# 안전장치 2: 현재 창의 컨트롤 핸들 요소 출력
|
||||
self.logger.log("지정된 시간 내에 이미지 요소를 찾지 못했습니다. 현재 창의 컨트롤 요소:", level=logging.ERROR)
|
||||
with open("debug_controls.txt", "w", encoding="utf-8") as f:
|
||||
original_stdout = os.sys.stdout
|
||||
os.sys.stdout = f
|
||||
self.whale_window.print_control_identifiers()
|
||||
os.sys.stdout = original_stdout
|
||||
self.logger.log("컨트롤 식별자가 debug_controls.txt에 저장되었습니다.", level=logging.INFO)
|
||||
|
||||
except ElementNotFoundError as e:
|
||||
self.logger.log("주소창 요소를 찾을 수 없습니다. 창을 다시 검색합니다.", level=logging.ERROR, exc_info=True)
|
||||
self.whale_window = self.find_whale_window() # 창을 다시 찾음
|
||||
if not self.whale_window:
|
||||
self.logger.log("웨일 창을 다시 찾을 수 없습니다. 작업 중단.", level=logging.ERROR)
|
||||
return
|
||||
|
||||
except Exception as e:
|
||||
self.logger.log(f"주소창 접근 중 오류 발생: {e}", level=logging.ERROR, exc_info=True)
|
||||
|
||||
self.logger.log("주소창 찾기 및 URL 이동 시도 횟수를 초과했습니다.", level=logging.ERROR)
|
||||
|
||||
def navigate_to_url_for_typing(self, url):
|
||||
"""주소창에 URL을 입력하고 페이지 로딩을 대기"""
|
||||
try:
|
||||
address_bar = self.whale_window.child_window(title="주소창 및 검색창", control_type="Edit")
|
||||
self.safe_click(address_bar)
|
||||
address_bar.type_keys(f"{url}{{ENTER}}", with_spaces=True)
|
||||
self.logger.log(f"{url}로 이동 중...", level=logging.DEBUG)
|
||||
|
||||
# 5초 동안 0.1초 간격으로 이미지 요소가 나타나는지 검사
|
||||
start_time = time.time()
|
||||
while time.time() - start_time < 5:
|
||||
try:
|
||||
image_element = self.whale_window.child_window(
|
||||
title="누락된 이미지 설명을 확인하려면 컨텍스트 메뉴를 여세요.",
|
||||
control_type="Image"
|
||||
)
|
||||
if image_element.exists(timeout=0.5):
|
||||
self.logger.log("페이지 로딩 완료: 이미지 요소가 나타났습니다.", level=logging.DEBUG)
|
||||
return
|
||||
except Exception:
|
||||
pass
|
||||
self.logger.log("지정된 시간 내에 이미지 요소를 찾지 못했습니다.", level=logging.ERROR)
|
||||
except Exception as e:
|
||||
self.logger.log(f"주소창에 접근할 수 없습니다: {e}", level=logging.ERROR, exc_info=True)
|
||||
|
||||
def check_translation_status(self, max_wait_time=10, check_interval=0.5):
|
||||
start_time = time.time()
|
||||
|
||||
while time.time() - start_time < max_wait_time:
|
||||
try:
|
||||
fail_indicator = self.whale_window.child_window(title="번역할 영역을 선택하세요.", control_type="Text")
|
||||
if fail_indicator.exists():
|
||||
self.logger.log("번역할 문구가 없는 이미지 입니다.", level=logging.INFO)
|
||||
return "fail"
|
||||
|
||||
image_element = self.whale_window.child_window(title="누락된 이미지 설명을 확인하려면 컨텍스트 메뉴를 여세요.", control_type="Image")
|
||||
if image_element.exists():
|
||||
self.safe_click(image_element, right=True)
|
||||
success_indicator = self.whale_window.child_window(title="이미지 복사(C)", control_type="MenuItem")
|
||||
|
||||
if success_indicator.wait('visible', timeout=5):
|
||||
self.safe_click(success_indicator)
|
||||
time.sleep(0.5)
|
||||
|
||||
formats = clipboard.GetClipboardFormats()
|
||||
# self.logger.debug(f"클립보드에 있는 형식 목록: {formats}")
|
||||
|
||||
for format_id in formats:
|
||||
format_name = clipboard.GetFormatName(format_id)
|
||||
# self.logger.debug(f"형식 ID {format_id}: {format_name}")
|
||||
|
||||
if format_name in ("CF_BITMAP", "CF_DIB"):
|
||||
image_data = clipboard.GetData(format_id=format_id)
|
||||
if isinstance(image_data, bytes):
|
||||
self.logger.log("번역 성공: 이미지 데이터 복사 완료.", level=logging.INFO)
|
||||
return "success"
|
||||
self.logger.log("번역이 아직 완료되지 않았습니다. 다시 시도 중...", level=logging.INFO)
|
||||
|
||||
except Exception as e:
|
||||
self.logger.log("클립보드 접근 중 오류 발생", level=logging.ERROR, exc_info=True)
|
||||
return "error"
|
||||
|
||||
time.sleep(check_interval)
|
||||
|
||||
self.logger.log("번역 확인 시간 초과.", level=logging.WARNING)
|
||||
return "timeout"
|
||||
|
||||
def right_click_on_image_and_inspect(self):
|
||||
try:
|
||||
image = self.whale_window.child_window(title="누락된 이미지 설명을 확인하려면 컨텍스트 메뉴를 여세요.", control_type="Image")
|
||||
|
||||
if image.exists():
|
||||
self.safe_click(image, right=True)
|
||||
self.logger.log("이미지 요소에서 우클릭을 수행했습니다.", level=logging.DEBUG)
|
||||
else:
|
||||
self.logger.log("이미지 요소를 찾을 수 없습니다.", level=logging.ERROR)
|
||||
return False
|
||||
|
||||
translate_menu_item = self.whale_window.child_window(title="이미지 번역 (R)", control_type="MenuItem")
|
||||
self.safe_click(translate_menu_item)
|
||||
self.logger.log("이미지 번역 명령이 실행되었습니다.", level=logging.DEBUG)
|
||||
time.sleep(0.5)
|
||||
|
||||
status = self.check_translation_status()
|
||||
if status == "success":
|
||||
self.logger.log("번역이 성공적으로 완료되었습니다.", level=logging.INFO)
|
||||
return True
|
||||
elif status == "fail":
|
||||
self.logger.log("번역에 실패했습니다.", level=logging.WARNING)
|
||||
return False
|
||||
else:
|
||||
self.logger.log("번역 상태를 확인할 수 없습니다.", level=logging.WARNING)
|
||||
return False
|
||||
|
||||
except Exception as e:
|
||||
self.logger.log(f"이미지 요소에서 우클릭 중 오류 발생: {e}", level=logging.ERROR, exc_info=True)
|
||||
return False
|
||||
|
||||
def check_image_size(self):
|
||||
"""창 제목에서 이미지의 너비와 높이를 추출하여 반환합니다."""
|
||||
try:
|
||||
window_title = self.whale_window.window_text()
|
||||
self.logger.log(f"Window Title: {window_title}", level=logging.DEBUG)
|
||||
|
||||
# 창 제목에서 해상도 추출 시도
|
||||
match = re.search(r"\((\d+)×(\d+)\)", window_title)
|
||||
if match:
|
||||
width = int(match.group(1))
|
||||
height = int(match.group(2))
|
||||
self.logger.log(f"이미지 해상도: {width}×{height}", level=logging.INFO)
|
||||
|
||||
if width < self.min_image_width or height < self.min_image_height:
|
||||
self.logger.log(f"이미지 해상도가 기준 이하 [{self.min_image_width} x {self.min_image_height}]입니다. 작업을 패스합니다.", level=logging.INFO)
|
||||
return False
|
||||
return True
|
||||
|
||||
elif ".jpg" in window_title or ".png" in window_title or ".jpeg" in window_title:
|
||||
self.logger.log("이미지 해상도가 없지만, 파일 확장자가 .jpg 또는 .jpeg 또는 .png입니다. 번역 작업을 진행합니다.", level=logging.INFO)
|
||||
return True
|
||||
|
||||
self.logger.log("이미지 해상도가 존재하지 않고 파일 확장자도 일치하지 않습니다. 작업을 패스합니다.", level=logging.WARNING)
|
||||
return False
|
||||
except Exception as e:
|
||||
self.logger.log(f"이미지 사이즈 측정 중 오류 발생: {e}", level=logging.ERROR, exc_info=True)
|
||||
return False
|
||||
|
||||
def click_back_button(self):
|
||||
"""'뒤로' 버튼을 클릭합니다."""
|
||||
try:
|
||||
back_button = self.whale_window.child_window(title="뒤로", control_type="Button")
|
||||
if back_button.exists():
|
||||
self.safe_click(back_button)
|
||||
self.logger.log("'뒤로' 버튼을 클릭했습니다.", level=logging.DEBUG)
|
||||
else:
|
||||
self.logger.log("'뒤로' 버튼을 찾을 수 없습니다.", level=logging.WARNING)
|
||||
except Exception as e:
|
||||
self.logger.log(f"'뒤로' 버튼 클릭 중 오류 발생: {e}", level=logging.ERROR, exc_info=True)
|
||||
|
||||
def close_whale_window(self):
|
||||
"""웨일 창을 종료하는 메서드."""
|
||||
try:
|
||||
if self.whale_app:
|
||||
self.whale_app.kill()
|
||||
self.logger.log("웨일 창을 성공적으로 종료했습니다.", level=logging.INFO)
|
||||
else:
|
||||
self.logger.log("웨일 애플리케이션이 시작되지 않았습니다.", level=logging.WARNING)
|
||||
except Exception as e:
|
||||
self.logger.log("웨일 창을 종료하는 중 오류 발생", level=logging.ERROR, exc_info=True)
|
||||
Loading…
Reference in New Issue