1042 lines
54 KiB
Python
1042 lines
54 KiB
Python
import re
|
|
from typing import List
|
|
from datetime import datetime
|
|
import random
|
|
import json
|
|
from collections import OrderedDict
|
|
from translatepy import Translator
|
|
from titleManager.gpt_client import GPTClient
|
|
# from titleManager.mongoDBManager import MongoDBManager
|
|
from titleManager.forbiddenWD_Manager import ForbiddenWordManager
|
|
from titleManager.naver_parser import NaverParser
|
|
from titleManager.naverAPI import NaverSearchAPI
|
|
from titleManager.kiprisAPI import Kipris_API
|
|
from titleManager.sp_ForbiddenM import SupabaseForbiddenWordManager
|
|
|
|
# from src.wh_search import WhaleSearchParser
|
|
import logging
|
|
|
|
class TitleGenerator:
|
|
def __init__(self, locator_manager, browser_controller, logger, whale_translator, toggle_states, gpt_client, forbidden_word_manager, user_id, supabase_manager):
|
|
self.logger = logger
|
|
self.whale_translator = whale_translator
|
|
self.browser_controller = browser_controller
|
|
self.page = self.browser_controller.page
|
|
self.parsing_page = self.browser_controller.parsing_page
|
|
self.locator_manager = locator_manager
|
|
self.toggle_states = toggle_states
|
|
self.forbidden_word_manager = forbidden_word_manager
|
|
self.user_id = user_id
|
|
self.supabase_manager = supabase_manager
|
|
|
|
# self.search_browser = WhaleSearchParser(self.logger)
|
|
|
|
self.naver_parser = NaverParser()
|
|
client_id = toggle_states.get("clientID", None)
|
|
client_secret = toggle_states.get("clientSecret", None)
|
|
|
|
self.naverAPI = NaverSearchAPI(client_id=client_id, client_secret=client_secret, logger=self.logger)
|
|
self.kipris_api = Kipris_API(logger, apikey='X9Tz3JqC/JcCwxnNewA6qdloIN6QFIitVBgS1a2KVDYk1AmddaDTvzr6+t3dyLZV3gh2TPXdNhxsRQwaKP673Q==')
|
|
self.gpt_client = gpt_client
|
|
self.translator = Translator() # 번역 라이브러리 초기화
|
|
|
|
# 선택자 로드
|
|
self.product_main_image_locator = self.locator_manager.get_locator('TitleLocators', 'product_main_image_locator')
|
|
self.product_name_input_locator = self.locator_manager.get_locator('TitleLocators', 'product_name_input_locator')
|
|
self.suggestion_input_locator = self.locator_manager.get_locator('TitleLocators', 'suggestion_input_locator')
|
|
self.search_button_locator = self.locator_manager.get_locator('TitleLocators', 'search_button_locator')
|
|
self.original_product_name_locator = self.locator_manager.get_locator('TitleLocators', 'original_product_name_locator')
|
|
self.delete_warning_button_locator = self.locator_manager.get_locator('TitleLocators', 'delete_warning_button_locator')
|
|
self.category_suggestion_button_locator = self.locator_manager.get_locator('TitleLocators', 'category_suggestion_button_locator')
|
|
|
|
self.consumer_product_informartion_disclosure_locator = self.locator_manager.get_locator('TitleLocators', 'consumer_product_informartion_disclosure_locator')
|
|
|
|
self.category_main_selector_with_cp = self.locator_manager.get_locator('TitleLocators', 'category_main_selector_with_cp')
|
|
self.category_main_selector_with_ss = self.locator_manager.get_locator('TitleLocators', 'category_main_selector_with_ss')
|
|
self.category_main_selector_with_esm = self.locator_manager.get_locator('TitleLocators', 'category_main_selector_with_esm')
|
|
|
|
self.category_text_locator = self.locator_manager.get_locator('TitleLocators', 'category_text_locator')
|
|
self.category_text_locator_certified = self.locator_manager.get_locator('TitleLocators', 'category_text_locator_certified')
|
|
|
|
self.title_generator_prompt = self.locator_manager.get_locator('TitleLocators', 'title_generator_prompt')
|
|
|
|
# self.title_infos = {
|
|
# "original_name": None,
|
|
# "keyword_name": None,
|
|
# "generated_name": None,
|
|
# "category": None,
|
|
# }
|
|
|
|
self.initial_title_infos()
|
|
|
|
def initial_title_infos(self):
|
|
self.title_infos = {
|
|
"original_name": None,
|
|
"keyword_name": None,
|
|
"generated_name": None,
|
|
"category": None,
|
|
"category_ss": None,
|
|
"category_esm": None,
|
|
"is_group_ESM": False,
|
|
"is_certified_group_SS": False,
|
|
"is_banned_category": False,
|
|
"banned_category_info": None,
|
|
"search_result": {}, # 추가된 필드
|
|
"top_5_titles": [], # 추가된 필드
|
|
"top_5_prices": [], # 추가된 필드
|
|
"keyword_tags": None, # 추가된 필드
|
|
"top_product_prices": None, # 추가된 필드
|
|
}
|
|
|
|
def update_page(self, page1):
|
|
self.page = page1
|
|
self.logger.log(f"page객체 업데이트 : {page1}", level=logging.DEBUG)
|
|
|
|
def update_parsing_page(self, parsing_page):
|
|
self.parsing_page = parsing_page
|
|
self.logger.log(f"paparsing_pagege객체 업데이트 : {parsing_page}", level=logging.DEBUG)
|
|
|
|
def translate_product_name_ori(self, original_name: str) -> str:
|
|
"""텍스트를 한국어로 번역하는 메서드"""
|
|
try:
|
|
translated_name = self.translator.translate(original_name, 'zh', 'ko')
|
|
|
|
return translated_name.result if translated_name else original_name
|
|
except Exception as e:
|
|
self.logger.log(f"번역 중 오류 발생: {e}", level=logging.ERROR, exc_info=True)
|
|
return original_name
|
|
|
|
def translate_product_name(self, original_name: str) -> str:
|
|
"""중국어 상품명을 한국어로 번역하여 JSON 응답에서 번역 결과 텍스트만 반환하는 메서드.
|
|
|
|
응답이 JSON 형식이 아닐 경우에는 원본 텍스트를 반환합니다.
|
|
"""
|
|
try:
|
|
# GPT에게 JSON 형식으로 응답하도록 요청하는 프롬프트 작성
|
|
prompt = (
|
|
f"중국어 상품명 [{original_name}]를 한국어로 번역해줘. "
|
|
f"응답은 아래 JSON 형식으로 해줘:\n"
|
|
f'{{"translation": "번역결과"}}'
|
|
)
|
|
|
|
# GPT 클라이언트에 프롬프트 전달 및 응답 받기
|
|
response = self.gpt_client.ask(prompt)
|
|
|
|
try:
|
|
# 응답이 JSON 형식이라 가정하고 파싱
|
|
translation_data = json.loads(response)
|
|
translated_text = translation_data.get("translation", original_name)
|
|
except json.JSONDecodeError:
|
|
self.logger.log("응답이 유효한 JSON 형식이 아닙니다. 응답 내용: " + response, level=logging.WARNING)
|
|
return original_name
|
|
|
|
return translated_text
|
|
except Exception as e:
|
|
self.logger.log(f"번역 중 오류 발생: {e}", level=logging.ERROR, exc_info=True)
|
|
return original_name
|
|
|
|
def is_word_forbidden(self, word: str) -> bool:
|
|
"""금지어 매니저를 통해 단어가 금지어인지 확인"""
|
|
return self.forbidden_word_manager.is_word_forbidden(word)
|
|
|
|
def search_trademark(self, word: str) -> dict:
|
|
"""키프리스 API로 단어를 검색하는 메서드"""
|
|
return self.kipris_api.search_trademark(word)
|
|
|
|
def is_valid_word(self, word: str) -> bool:
|
|
"""숫자로만 이루어진 단어 또는 영어와 숫자로만 이루어진 단어를 검증하는 함수."""
|
|
return not (word.isdigit() or re.fullmatch(r'[A-Za-z0-9]+', word))
|
|
|
|
def extract_special_words(self, original_name: str) -> list:
|
|
"""원본 상품명에서 숫자로만 이루어진 단어와 영어와 숫자로만 이루어진 단어를 추출하는 함수."""
|
|
return [word for word in original_name.split() if word.isdigit() or re.fullmatch(r'[A-Za-z0-9]+', word)]
|
|
|
|
def filter_invalid_words(self, words: list) -> list:
|
|
"""영어만 이루어진 단어와 영어와 숫자로 이루어진 단어를 제외하는 함수."""
|
|
return [word for word in words if not re.fullmatch(r'[A-Za-z0-9]+', word)]
|
|
|
|
def process_top_titles(self, top_titles: list) -> list:
|
|
"""top_titles에서 유효하지 않은 단어(영어만 이루어진 단어와 영어와 숫자로 이루어진 단어)를 제외하는 함수."""
|
|
filtered_titles = []
|
|
for title in top_titles:
|
|
filtered_words = self.filter_invalid_words(title.split())
|
|
filtered_titles.append(' '.join(filtered_words))
|
|
return filtered_titles
|
|
|
|
def get_search_result_to_crawling(self, keyword_name):
|
|
try:
|
|
# 키워드 상품명으로 검색
|
|
search_result = self.naver_parser.search_and_parse(" ".join(keyword_name.split()[:4]))
|
|
self.logger.log(f"naver_parser search_result : {search_result}", level=logging.DEBUG)
|
|
|
|
# 데이터 변환
|
|
top_products = search_result.get("top_products", [])
|
|
related_tags = search_result.get("related_tags", [])
|
|
|
|
# 변환된 결과 리스트 생성
|
|
result = []
|
|
for product in top_products:
|
|
result.append({
|
|
"title": product.get("title", ""),
|
|
"price": int(product.get("price", "0").replace(",", "")),
|
|
"delivery_fee": int(product.get("delivery_fee_content", "0").replace(",", "")),
|
|
"purchase_count": int(product.get("purchase_count", "")),
|
|
"rank": int(product.get("rank", "")),
|
|
"category": product.get("category", []),
|
|
"manu_tag": product.get("manu_tag", ""),
|
|
})
|
|
|
|
# 태그 추가
|
|
result.append({"related_tags": related_tags})
|
|
|
|
return result
|
|
except Exception as e:
|
|
self.logger.log(f"상품명 검색 중 오류 발생: {e}", level=logging.ERROR, exc_info=True)
|
|
return []
|
|
|
|
def get_search_result_to_API(self, keyword_name):
|
|
try:
|
|
# 키워드 상품명으로 API 검색
|
|
search_result = self.naverAPI.search(keyword_name)
|
|
self.logger.log(f"naverAPI search_result : {search_result}", level=logging.DEBUG)
|
|
|
|
# 데이터 변환
|
|
parsed_products = self.naverAPI.parse_search_results_list(search_result) # 리스트 반환
|
|
result = []
|
|
for product in parsed_products:
|
|
result.append({
|
|
"title": product.get("title", ""),
|
|
"price": int(product.get("price", "0").replace(",", "")),
|
|
"category": product.get("category", []),
|
|
"manu_tag": product.get("manu_tag", ""),
|
|
})
|
|
|
|
# 태그는 없으므로 빈 리스트 추가
|
|
result.append({"related_tags": []})
|
|
|
|
return result
|
|
except Exception as e:
|
|
self.logger.log(f"API 검색 중 오류 발생: {e}", level=logging.ERROR, exc_info=True)
|
|
return []
|
|
|
|
|
|
def process_tags(self, tags: List[str]) -> List[str]:
|
|
"""태그 리스트를 정리하는 메서드."""
|
|
forbidden_words = {"오늘발송", "오늘출발", "당일배송"} # 제거할 키워드
|
|
unique_tags = list(set(tags)) # 중복 제거
|
|
filtered_tags = [tag for tag in unique_tags if tag not in forbidden_words] # 금지어 필터링
|
|
return filtered_tags
|
|
|
|
def add_forbidden_prefix(self, title: str, forbidden_words: List[str]) -> str:
|
|
"""
|
|
상품 제목 앞에 [금지상품: 단어1/단어2] 형태의 접두어를 추가합니다.
|
|
허용된 특수문자는 !$~()._-=+/ 입니다.
|
|
"""
|
|
prefix = "금지상품//"
|
|
if forbidden_words:
|
|
# 여러 단어가 있을 경우 슬래시(/)로 구분합니다.
|
|
prefix += f": {'/'.join(forbidden_words)}"
|
|
prefix += "/"
|
|
return prefix + title
|
|
|
|
def generate_product_title(self, original_name: str, keyword_name: str, search_result: dict, product_category: str) -> str:
|
|
"""상품명을 생성하는 메서드"""
|
|
# 1. 원본 상품명 번역
|
|
translated_name = self.translate_product_name(original_name)
|
|
self.logger.log(f'translated_name : {translated_name}', level=logging.DEBUG)
|
|
|
|
# 2. 검색 결과에서 제목과 가격 가져오기
|
|
top_titles = []
|
|
top_prices = []
|
|
|
|
if search_result:
|
|
# search_result 리스트 처리
|
|
for product in search_result:
|
|
if "title" in product and "price" in product:
|
|
top_titles.append(product["title"])
|
|
top_prices.append(product["price"])
|
|
|
|
self.logger.log(f'top_titles : {top_titles}', level=logging.DEBUG)
|
|
self.logger.log(f'top_prices : {top_prices}', level=logging.DEBUG)
|
|
else:
|
|
self.logger.log("검색 결과가 비어 있습니다.", level=logging.WARNING)
|
|
|
|
# 3. 검색된 제목 필터링
|
|
filtered_top_titles = self.process_top_titles(top_titles)
|
|
|
|
# 3. 키워드 상품명에서 첫 4개 키워드 추출
|
|
essential_keywords = keyword_name.split()[:4]
|
|
self.logger.log(f'essential_keywords (첫 4개 키워드): {essential_keywords}', level=logging.DEBUG)
|
|
|
|
keyword_title = list(set(
|
|
word for title in [keyword_name] + filtered_top_titles
|
|
for word in title.split()
|
|
))
|
|
self.logger.log(f'keyword_title : {keyword_title}', level=logging.DEBUG)
|
|
|
|
# 5. 원본 상품명에서 숫자 또는 영어와 숫자로만 이루어진 단어 추출 및 포함
|
|
special_words = self.extract_special_words(original_name)
|
|
self.logger.log(f'special_words from original_name: {special_words}', level=logging.DEBUG)
|
|
keyword_title.extend(special_words)
|
|
|
|
# 6. 중복 제거
|
|
keyword_title = list(set(keyword_title))
|
|
self.logger.log(f'final keyword_title after KIPRIS search: {keyword_title}', level=logging.DEBUG)
|
|
|
|
# 8. 필수 키워드 중 랜덤으로 2개 선택 후 병합
|
|
required_keywords = random.sample(essential_keywords, min(2, len(essential_keywords)))
|
|
self.logger.log(f'randomly selected required_keywords: {required_keywords}', level=logging.DEBUG)
|
|
keyword_title.extend(required_keywords)
|
|
keyword_title = list(set(keyword_title))
|
|
|
|
# 11. 금지어 처리
|
|
# 먼저 “금지” 등급이 있는 단어 리스트를 추출
|
|
forbidden_grade_words = self.forbidden_word_manager.get_forbidden_grade_words(keyword_title)
|
|
# “비허용” 등급 단어는 제목에서 제거
|
|
keyword_title = [word for word in keyword_title if not self.forbidden_word_manager.is_word_forbidden(word)]
|
|
self.logger.log(f'keyword_title after forbidden filter : {keyword_title}', level=logging.DEBUG)
|
|
|
|
self.title_generator_prompt
|
|
self.logger.log(f'self.title_generator_prompt : {self.title_generator_prompt}', level=logging.DEBUG)
|
|
|
|
# 12. 최종 상품명 생성 (GPT Client 이용)
|
|
product_title = self.gpt_client.generate_product_name_next(
|
|
words=keyword_title, original_name=original_name, top_titles=top_titles, category=product_category, title_generator_prompt=self.title_generator_prompt
|
|
)
|
|
self.logger.log(f'final product_title: {product_title}', level=logging.DEBUG)
|
|
|
|
if not product_title:
|
|
product_title = keyword_title.append(' '.join(keyword_title))
|
|
self.logger.log(f'because product_title is None, final product_title: {product_title}', level=logging.DEBUG)
|
|
|
|
# 13. 만약 “금지” 등급 단어가 있었다면 접두어 추가 (해당 단어도 함께 표기)
|
|
if forbidden_grade_words:
|
|
product_title = self.add_forbidden_prefix(product_title, forbidden_grade_words)
|
|
return product_title
|
|
|
|
|
|
def generate_product_title_ori(self, original_name: str, keyword_name: str, search_result: dict, product_category: str) -> str:
|
|
"""상품명을 생성하는 메서드"""
|
|
# 1. 원본 상품명 번역 및 관련성 판단
|
|
translated_name = self.translate_product_name(original_name)
|
|
self.logger.log(f'translated_name : {translated_name}', level=logging.DEBUG)
|
|
|
|
# 2. 검색 결과에서 제목과 가격 가져오기
|
|
top_titles = []
|
|
top_prices = []
|
|
|
|
if search_result:
|
|
# search_result 리스트 처리
|
|
for product in search_result:
|
|
if "title" in product and "price" in product:
|
|
top_titles.append(product["title"])
|
|
top_prices.append(product["price"])
|
|
|
|
self.logger.log(f'top_titles : {top_titles}', level=logging.DEBUG)
|
|
self.logger.log(f'top_prices : {top_prices}', level=logging.DEBUG)
|
|
else:
|
|
self.logger.log("검색 결과가 비어 있습니다.", level=logging.WARNING)
|
|
|
|
# 3. 검색된 제목 필터링
|
|
filtered_top_titles = self.process_top_titles(top_titles)
|
|
|
|
|
|
# 3. 키워드 상품명에서 첫 4개 키워드 추출
|
|
essential_keywords = keyword_name.split()[:4]
|
|
self.logger.log(f'essential_keywords (첫 4개 키워드): {essential_keywords}', level=logging.DEBUG)
|
|
|
|
keyword_title = list(set(
|
|
word for title in [keyword_name] + filtered_top_titles
|
|
for word in title.split()
|
|
))
|
|
self.logger.log(f'keyword_title : {keyword_title}', level=logging.DEBUG)
|
|
|
|
# 3. 숫자나 영어와 숫자로만 이루어진 단어 필터링
|
|
keyword_title = [word for word in keyword_title if self.is_valid_word(word)]
|
|
self.logger.log(f'keyword_title after filtering invalid words : {keyword_title}', level=logging.DEBUG)
|
|
|
|
# 4. 중복단어 제거
|
|
keyword_title = list(set(keyword_title)) # 중복 제거
|
|
self.logger.log(f'final keyword_title after KIPRIS search: {keyword_title}', level=logging.DEBUG)
|
|
|
|
# 5. 원본 상품명에서 숫자 또는 영어와 숫자로만 이루어진 단어 추출 및 포함
|
|
special_words = self.extract_special_words(original_name)
|
|
self.logger.log(f'special_words from original_name: {special_words}', level=logging.DEBUG)
|
|
keyword_title.extend(special_words)
|
|
|
|
# 5. 숫자나 영어+숫자로만 이루어진 단어 필터링
|
|
keyword_title = [word for word in keyword_title if self.is_valid_word(word)]
|
|
self.logger.log(f'keyword_title after filtering invalid words : {keyword_title}', level=logging.DEBUG)
|
|
|
|
# 6. 중복 제거
|
|
keyword_title = list(set(keyword_title))
|
|
self.logger.log(f'final keyword_title after KIPRIS search: {keyword_title}', level=logging.DEBUG)
|
|
|
|
# 7. 원본 상품명에서 숫자 및 영어+숫자 단어 추출 후 포함
|
|
special_words = self.extract_special_words(original_name)
|
|
self.logger.log(f'special_words from original_name: {special_words}', level=logging.DEBUG)
|
|
keyword_title.extend(special_words)
|
|
keyword_title = list(set(keyword_title))
|
|
self.logger.log(f'final keyword_title including special words: {keyword_title}', level=logging.DEBUG)
|
|
|
|
# 8. 필수 키워드 중 랜덤으로 2개 선택 후 병합
|
|
required_keywords = random.sample(essential_keywords, min(2, len(essential_keywords)))
|
|
self.logger.log(f'randomly selected required_keywords: {required_keywords}', level=logging.DEBUG)
|
|
keyword_title.extend(required_keywords)
|
|
keyword_title = list(set(keyword_title))
|
|
|
|
# 9. 금지어 처리
|
|
# 먼저 “금지” 등급이 있는지 확인 (최종 제목에 접두어 추가 용도)
|
|
has_forbidden_grade = self.forbidden_word_manager.check_forbidden_grade(keyword_title)
|
|
# “비허용” 등급 단어는 제목에서 제거
|
|
keyword_title = [word for word in keyword_title if not self.forbidden_word_manager.is_word_forbidden(word)]
|
|
self.logger.log(f'keyword_title after forbidden filter : {keyword_title}', level=logging.DEBUG)
|
|
|
|
# 10. 최종 상품명 생성 (GPT Client 이용)
|
|
product_title = self.gpt_client.generate_product_name_next(words=keyword_title, original_name=original_name, top_titles=top_titles)
|
|
self.logger.log(f'final product_title: {product_title}', level=logging.DEBUG)
|
|
|
|
# 11. 만약 “금지” 등급 단어가 있었다면 접두어 추가
|
|
if has_forbidden_grade:
|
|
product_title = self.add_forbidden_prefix(product_title)
|
|
return product_title
|
|
|
|
|
|
async def get_initial_info(self, price_setting_diag, use_lens: bool, use_api: bool = False):
|
|
try:
|
|
# 1. 딕셔너리 초기화
|
|
self.initial_title_infos()
|
|
|
|
# 2. 해당상품 정보 추출
|
|
original_name = await self.get_original_product_name()
|
|
self.logger.log(f'original_name: {original_name}', level=logging.DEBUG)
|
|
self.title_infos["original_name"] = original_name
|
|
|
|
keyword_name = await self.get_product_name()
|
|
self.logger.log(f'keyword_name: {keyword_name}', level=logging.DEBUG)
|
|
self.title_infos["keyword_name"] = keyword_name
|
|
|
|
# product_category = await self.get_category(market='ss') # 카테고리 가져오기
|
|
# self.logger.log(f'product_category: {product_category}', level=logging.DEBUG)
|
|
# self.title_infos["category"] = product_category
|
|
|
|
product_category_data = await self.get_category() # ss와 esm에서 카테고리 정보 가져오기
|
|
# ss 마켓 카테고리 처리
|
|
product_category_ss = product_category_data.get("category_text_ss", "")
|
|
is_certified_ss = product_category_data.get("is_certified_ss", False)
|
|
|
|
# esm 마켓 카테고리 처리
|
|
product_category_esm = product_category_data.get("category_text_esm", "")
|
|
is_group_esm = product_category_data.get("is_group_esm", False)
|
|
|
|
# 로깅
|
|
self.logger.log(f'SS 마켓 카테고리: {product_category_ss}', level=logging.DEBUG)
|
|
self.logger.log(f'SS 마켓 인증 필요 여부: {is_certified_ss}', level=logging.DEBUG)
|
|
self.logger.log(f'ESM 마켓 카테고리: {product_category_esm}', level=logging.DEBUG)
|
|
self.logger.log(f'ESM 마켓 그룹상품 여부: {is_group_esm}', level=logging.DEBUG)
|
|
|
|
# title_infos에 저장
|
|
self.title_infos["category"] = product_category_ss
|
|
self.title_infos["category_ss"] = product_category_ss
|
|
self.title_infos["category_esm"] = product_category_esm
|
|
self.title_infos["is_certified_group_SS"] = is_certified_ss
|
|
self.title_infos["is_group_ESM"] = is_group_esm
|
|
|
|
|
|
# 2-1. 금지카테고리 여부 체크: price_setting_diag.get_crmobi_stage() 호출
|
|
crmobi_info = price_setting_diag.get_crmobi_stage(product_category_ss)
|
|
self.logger.log(f'금지카테고리 여부 체크: {crmobi_info}', level=logging.DEBUG)
|
|
|
|
if crmobi_info is not None:
|
|
threshold, unit, extra_cost, banned_flag = crmobi_info
|
|
if banned_flag == 1:
|
|
# 금지카테고리라면 flag를 추가
|
|
self.title_infos["is_banned_category"] = True
|
|
self.title_infos["banned_category_info"] = product_category_ss
|
|
self.logger.log(f"금지카테고리 감지: {product_category_ss}", level=logging.DEBUG)
|
|
else:
|
|
self.title_infos["is_banned_category"] = False
|
|
else:
|
|
self.logger.log("해당 카테고리에 크무비 단계 또는 금지카테고리가 설정되어 있지 않습니다.", level=logging.DEBUG)
|
|
self.title_infos["is_banned_category"] = False
|
|
|
|
# 3. 검색 결과 가져오기 (crawling 또는 API 선택)
|
|
search_result = None
|
|
|
|
# if use_lens:
|
|
# # 기본적으로 쇼핑렌즈 시도
|
|
# self.logger.log("쇼핑렌즈를 통해 검색 결과 가져오는 중...", level=logging.INFO)
|
|
# try:
|
|
# self.search_browser.start_search_browser()
|
|
# product_main_image_url = await self.get_product_main_image_url()
|
|
# search_result = self.search_browser.search_and_parse(product_main_image_url, min_price=20000, top_n=5)
|
|
# # self.logger.log(f"쇼핑렌즈 검색 결과: {json.dumps(search_result, indent=4, ensure_ascii=False)}", level=logging.INFO)
|
|
# self.search_browser.close_search_browser()
|
|
# except Exception as e:
|
|
# self.logger.log(f"쇼핑렌즈 실패: {e}. API로 전환합니다.", level=logging.ERROR, exc_info=True)
|
|
# use_api = True # 크롤링 실패 시 API로 전환
|
|
|
|
if use_lens:
|
|
# 기본적으로 쇼핑렌즈 시도
|
|
self.logger.log("쇼핑렌즈를 통해 검색 결과 가져오는 중...", level=logging.INFO)
|
|
try:
|
|
# self.search_browser.start_search_browser()
|
|
product_main_image_url = await self.get_product_main_image_url()
|
|
search_result = await self.whale_translator.search_and_parse_to_newContext(self.parsing_page, product_main_image_url, min_price=20000, top_n=5)
|
|
# self.logger.log(f"쇼핑렌즈 검색 결과: {json.dumps(search_result, indent=4, ensure_ascii=False)}", level=logging.INFO)
|
|
# self.search_browser.close_search_browser()
|
|
except Exception as e:
|
|
self.logger.log(f"쇼핑렌즈 실패: {e}. API로 전환합니다.", level=logging.ERROR, exc_info=True)
|
|
use_api = True # 크롤링 실패 시 API로 전환
|
|
|
|
if use_api:
|
|
# API 방식으로 처리
|
|
self.logger.log("API를 통해 검색 결과 가져오는 중...", level=logging.INFO)
|
|
try:
|
|
search_result = self.get_search_result_to_API(self.title_infos["keyword_name"])
|
|
self.logger.log(f'API 검색 결과: {search_result}', level=logging.DEBUG)
|
|
except Exception as e:
|
|
self.logger.log(f"API 검색 실패: {e}", level=logging.ERROR, exc_info=True)
|
|
return {} # API 실패 시 빈 값 반환
|
|
|
|
if not search_result:
|
|
self.logger.log("검색 결과를 찾을 수 없습니다.", level=logging.WARNING)
|
|
return {} # 검색 결과가 없을 경우 빈 값 반환
|
|
|
|
self.title_infos["search_result"] = search_result
|
|
|
|
# 4. 검색 결과 처리
|
|
related_tags = search_result[-1].get("related_tags", []) # 마지막 요소에서 태그 추출
|
|
products = search_result[:-1] # 마지막 요소 제외
|
|
|
|
# 상위 제품 정보 저장
|
|
self.title_infos["top_5_titles"] = [p.get("title", "") for p in products[:5]]
|
|
self.title_infos["top_5_prices"] = [p.get("price", 0) for p in products[:5]]
|
|
|
|
self.logger.log(f'제목 리스트: {self.title_infos["top_5_titles"]}', level=logging.DEBUG)
|
|
self.logger.log(f'가격 리스트: {self.title_infos["top_5_prices"]}', level=logging.DEBUG)
|
|
|
|
# 제조사 태그(manu_tags) 수집
|
|
# 5. 제조사 태그(manu_tags) 수집
|
|
manu_tags = [product["manu_tag"] for product in products[:5] if "manu_tag" in product and product["manu_tag"]]
|
|
self.logger.log(f"추출된 manu_tags: {manu_tags}", level=logging.DEBUG)
|
|
|
|
# 관련 태그와 제조사 태그를 결합하여 중복 제거
|
|
if related_tags or manu_tags:
|
|
combined_tags = list(set(related_tags + manu_tags)) # 중복 제거
|
|
processed_tags = self.process_tags(combined_tags)
|
|
self.logger.log(f"처리된 태그: {processed_tags}", level=logging.DEBUG)
|
|
self.title_infos["keyword_tags"] = processed_tags
|
|
else:
|
|
self.logger.log("관련 태그나 제조사 태그를 찾을 수 없습니다.", level=logging.WARNING)
|
|
self.title_infos["keyword_tags"] = []
|
|
|
|
return self.title_infos
|
|
|
|
except Exception as e:
|
|
self.logger.log(f"초기 상품정보 생성 중 오류 발생: {e}", level=logging.ERROR, exc_info=True)
|
|
return {}
|
|
|
|
def set_banned_category_title(self, banned_category_info: dict) -> str:
|
|
"""
|
|
title_infos의 카테고리 정보를 기반으로,
|
|
“[금지카테고리]카테1-카테2-카테3-카테4” 형식의 제목을 생성하여 반환합니다.
|
|
"""
|
|
banned_product_title = "금지카테고리//미지정"
|
|
|
|
if banned_category_info:
|
|
banned_product_title = "금지카테고리//" + banned_category_info
|
|
|
|
self.logger.log(f"생성된 금지카테고리 상품명: {banned_product_title}", level=logging.DEBUG)
|
|
return banned_product_title
|
|
|
|
|
|
async def process_title_for_banned_category(self):
|
|
try:
|
|
product_category_ss = self.title_infos["banned_category_info"]
|
|
banned_product_title = "[금지카테고리]" + product_category_ss
|
|
self.logger.log(f"생성된 금지카테고리 상품명: {banned_product_title}", level=logging.DEBUG)
|
|
|
|
# 상품명 입력: set_product_name은 async 메서드로 가정
|
|
is_success = await self.set_product_name(banned_product_title)
|
|
if is_success:
|
|
self.logger.log("금지카테고리 상품명 설정 성공", level=logging.INFO)
|
|
else:
|
|
self.logger.log("금지카테고리 상품명 설정 실패", level=logging.WARNING)
|
|
|
|
self.logger.log(f"Title_Infos: {self.title_infos}", level=logging.DEBUG)
|
|
return banned_product_title
|
|
|
|
except Exception as e:
|
|
self.logger.log(f"금지카테고리 상품명 설정 중 오류 발생: {e}", level=logging.ERROR, exc_info=True)
|
|
|
|
async def process_title(self):
|
|
try:
|
|
# # 1. 딕셔너리 초기화
|
|
# self.initial_title_infos()
|
|
|
|
# # 2. 해당상품 정보 추출
|
|
# original_name = await self.get_original_product_name()
|
|
# keyword_name = await self.get_product_name()
|
|
# product_category = await self.get_category(market='ss') # 카테고리 가져오기
|
|
|
|
# # 3. 검색결과 가져오기
|
|
# search_result = self.get_search_result(["keyword_name"])
|
|
|
|
# # 4. 검색결과에서 태그 추출
|
|
# related_tags = search_result.get("related_tags", [])
|
|
# manu_tags = [product.get("manu_tag") for product in search_result.get("top_products", []) if product.get("manu_tag")]
|
|
|
|
# combined_tags = list(set(related_tags + manu_tags))
|
|
# processed_tags = self.process_tags(combined_tags)
|
|
|
|
# # 5. price 추출
|
|
# # top_product_prices = [product.get("price") for product in search_result.get("top_products", []) if product.get("price")]
|
|
# top_product_prices = [
|
|
# int(product.get("price").replace(",", ""))
|
|
# for product in search_result.get("top_products", [])
|
|
# if product.get("price")
|
|
# ]
|
|
|
|
# 6. 상품명 생성
|
|
product_title = self.generate_product_title(original_name=self.title_infos["original_name"], keyword_name=self.title_infos["keyword_name"], search_result=self.title_infos["search_result"], product_category=self.title_infos["category"])
|
|
if product_title == "관련성이 없는 상품 - 체크필요":
|
|
return
|
|
|
|
# 7. 결과 저장
|
|
# self.title_infos["original_name"] = original_name
|
|
# self.title_infos["keyword_name"] = keyword_name
|
|
# self.title_infos["category"] = product_category
|
|
|
|
self.title_infos["generated_name"] = product_title
|
|
# self.title_infos["keyword_tags"] = processed_tags
|
|
# self.title_infos["top_product_prices"] = top_product_prices
|
|
|
|
# 8. 상품명 설정
|
|
is_success = await self.set_product_name(product_title)
|
|
|
|
if is_success:
|
|
self.logger.log("상품명 생성 및 설정 성공", level=logging.INFO)
|
|
else:
|
|
self.logger.log("상품명 설정 실패", level=logging.WARNING)
|
|
|
|
self.logger.log(f"Title_Infos : {self.title_infos}", level=logging.DEBUG)
|
|
|
|
return product_title
|
|
|
|
except Exception as e:
|
|
self.logger.log(f"상품명 생성 중 오류 발생: {e}", level=logging.ERROR, exc_info=True)
|
|
|
|
async def get_product_main_image_url(self) -> str:
|
|
"""
|
|
상품 메인 이미지의 URL을 가져오는 메서드입니다.
|
|
|
|
Returns:
|
|
str: 상품 메인 이미지 URL
|
|
"""
|
|
try:
|
|
self.logger.log("상품 메인 이미지 URL을 가져오는 중입니다.", level=logging.DEBUG)
|
|
product_image_element = await self.page.query_selector(self.product_main_image_locator)
|
|
product_image_url = await product_image_element.get_attribute('src') if product_image_element else ""
|
|
self.logger.log(f"상품 메인 이미지 URL: {product_image_url}", level=logging.DEBUG)
|
|
return product_image_url
|
|
except Exception as e:
|
|
self.logger.log(f"상품 메인 이미지 URL 가져오기 중 오류 발생: {e}", level=logging.ERROR, exc_info=True)
|
|
return ""
|
|
|
|
async def get_product_name(self) -> str:
|
|
"""
|
|
노출상품명 입력칸에서 상품명을 가져오는 메서드입니다.
|
|
|
|
Returns:
|
|
str: 상품명 텍스트
|
|
"""
|
|
try:
|
|
self.logger.log("노출상품명 입력칸에서 상품명을 가져오는 중입니다.", level=logging.DEBUG)
|
|
product_name_element = await self.page.query_selector(self.product_name_input_locator)
|
|
product_name = await product_name_element.get_attribute('value') if product_name_element else ""
|
|
self.logger.log(f"상품명: {product_name}", level=logging.DEBUG)
|
|
return product_name
|
|
except Exception as e:
|
|
self.logger.log(f"상품명 가져오기 중 오류 발생: {e}", level=logging.ERROR, exc_info=True)
|
|
return ""
|
|
|
|
async def set_product_name(self, product_name: str) -> bool:
|
|
"""
|
|
노출상품명 입력칸에 상품명을 설정하는 메서드입니다.
|
|
|
|
Args:
|
|
product_name (str): 설정할 상품명
|
|
|
|
Returns:
|
|
bool: 성공 여부 (True: 성공, False: 실패)
|
|
"""
|
|
try:
|
|
self.logger.log(f"노출상품명 입력칸에 '{product_name}' 설정 중입니다.", level=logging.DEBUG)
|
|
product_name_element = await self.page.query_selector(self.product_name_input_locator)
|
|
|
|
if product_name_element:
|
|
await product_name_element.fill(product_name)
|
|
self.logger.log(f"상품명 '{product_name}'이 성공적으로 입력되었습니다.", level=logging.DEBUG)
|
|
return True
|
|
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
|
|
|
|
async def enter_product_name_suggestion(self, suggestion: str):
|
|
"""
|
|
상품명 추천단어를 입력칸에 입력하는 메서드입니다.
|
|
|
|
Args:
|
|
suggestion (str): 입력할 추천 단어
|
|
"""
|
|
try:
|
|
self.logger.log(f"추천 단어를 상품명 추천 입력칸에 입력 중: {suggestion}", level=logging.DEBUG)
|
|
suggestion_input_element = await self.page.query_selector(self.suggestion_input_locator)
|
|
if suggestion_input_element:
|
|
await suggestion_input_element.fill(suggestion)
|
|
self.logger.log(f"추천 단어 '{suggestion}' 입력 완료.", level=logging.DEBUG)
|
|
else:
|
|
self.logger.log("추천 입력칸 요소를 찾을 수 없습니다.", level=logging.ERROR)
|
|
except Exception as e:
|
|
self.logger.log(f"추천 입력 단어 입력 중 오류 발생: {e}", level=logging.ERROR, exc_info=True)
|
|
|
|
async def click_product_name_search_button(self):
|
|
"""
|
|
상품명 추천단어 입력칸의 검색 버튼을 클릭하는 메서드입니다.
|
|
"""
|
|
try:
|
|
self.logger.log("상품명 추천단어 검색 버튼 클릭 중.", level=logging.DEBUG)
|
|
search_button_element = await self.page.query_selector(self.search_button_locator)
|
|
if search_button_element:
|
|
await search_button_element.click()
|
|
self.logger.log("검색 버튼 클릭 완료.", level=logging.DEBUG)
|
|
else:
|
|
self.logger.log("검색 버튼 요소를 찾을 수 없습니다.", level=logging.ERROR)
|
|
except Exception as e:
|
|
self.logger.log(f"상품명 추천 검색 버튼 클릭 중 오류 발생: {e}", level=logging.ERROR, exc_info=True)
|
|
|
|
async def get_original_product_name(self) -> str:
|
|
"""
|
|
원본 상품명을 가져오는 메서드입니다.
|
|
|
|
Returns:
|
|
str: 원본 상품명 텍스트
|
|
"""
|
|
try:
|
|
self.logger.log("원본 상품명을 가져오는 중입니다.", level=logging.DEBUG)
|
|
original_name_element = await self.page.query_selector(self.original_product_name_locator)
|
|
original_name = await original_name_element.inner_text() if original_name_element else ""
|
|
self.logger.log(f"원본 상품명: {original_name}", level=logging.DEBUG)
|
|
return original_name
|
|
except Exception as e:
|
|
self.logger.log(f"원본 상품명 가져오기 중 오류 발생: {e}", level=logging.ERROR, exc_info=True)
|
|
return ""
|
|
|
|
async def delete_warning_word_in_product_name(self):
|
|
"""
|
|
상품명에서 경고 단어를 삭제하는 버튼을 클릭하는 메서드입니다.
|
|
"""
|
|
try:
|
|
self.logger.log("경고 단어 삭제 버튼 클릭 중입니다.", level=logging.DEBUG)
|
|
delete_button_element = await self.page.query_selector(self.delete_warning_button_locator)
|
|
if delete_button_element:
|
|
await delete_button_element.click()
|
|
self.logger.log("경고 단어 삭제 버튼 클릭 완료.", level=logging.DEBUG)
|
|
else:
|
|
self.logger.log("경고 단어 삭제 버튼 요소를 찾을 수 없습니다.", level=logging.ERROR)
|
|
except Exception as e:
|
|
self.logger.log(f"경고 단어 삭제 버튼 클릭 중 오류 발생: {e}", level=logging.ERROR, exc_info=True)
|
|
|
|
async def click_category_suggestion_button(self):
|
|
"""
|
|
카테고리 추천받기 버튼을 클릭하는 메서드입니다.
|
|
"""
|
|
try:
|
|
self.logger.log("카테고리 추천받기 버튼 클릭 중입니다.", level=logging.DEBUG)
|
|
category_suggestion_button_element = await self.page.query_selector(self.category_suggestion_button_locator)
|
|
if category_suggestion_button_element:
|
|
await category_suggestion_button_element.click()
|
|
self.logger.log("카테고리 추천받기 버튼 클릭 완료.", level=logging.DEBUG)
|
|
else:
|
|
self.logger.log("카테고리 추천받기 버튼 요소를 찾을 수 없습니다.", level=logging.ERROR)
|
|
except Exception as e:
|
|
self.logger.log(f"카테고리 추천받기 버튼 클릭 중 오류 발생: {e}", level=logging.ERROR, exc_info=True)
|
|
|
|
async def get_category_ori(self, market='ss') -> str:
|
|
"""
|
|
카테고리를 가져오는 메서드로 인증 필요 여부에 따라 카테고리 선택자를 다르게 처리합니다.
|
|
|
|
Returns:
|
|
str: 카테고리 텍스트
|
|
"""
|
|
try:
|
|
self.logger.log(f"마켓 : {market} - 카테고리 텍스트를 가져오는 중입니다.", level=logging.DEBUG)
|
|
|
|
if market == 'ss':
|
|
category_locator = self.category_main_selector_with_ss
|
|
elif market == 'cp':
|
|
category_locator = self.category_main_selector_with_cp
|
|
elif market == 'esm':
|
|
category_locator = self.category_main_selector_with_esm
|
|
|
|
self.logger.log(f"category_locator : {category_locator}", level=logging.DEBUG)
|
|
|
|
await self.page.wait_for_selector(category_locator, timeout=5000, state="attached") # 요소가 나타날 때까지 대기
|
|
main_category_element = self.page.locator(category_locator) # 대기 후 동기적으로 요소 가져오기
|
|
self.logger.log(f"main_category_element : {main_category_element}", level=logging.DEBUG)
|
|
|
|
if not await main_category_element.count():
|
|
self.logger.log("카테고리 메인 선택자를 찾을 수 없습니다.", level=logging.ERROR)
|
|
return ""
|
|
|
|
# 인증 텍스트 요소 선택
|
|
category_text_element = main_category_element.locator(self.category_text_locator)
|
|
|
|
self.logger.log(f"category_text_element : {category_text_element}", level=logging.DEBUG)
|
|
|
|
if await category_text_element.count():
|
|
category_text = await category_text_element.inner_text()
|
|
|
|
if "인증" in category_text:
|
|
self.logger.log(f"카테고리 인증 필요 발생 category_text = {category_text}", level=logging.DEBUG)
|
|
category_text_certified_element = main_category_element.locator(self.category_text_locator_certified)
|
|
|
|
if await category_text_certified_element.count():
|
|
category_text = await category_text_certified_element.inner_text()
|
|
self.logger.log(f"인증 필요 카테고리 text = {category_text}", level=logging.DEBUG)
|
|
if "그룹상품" in category_text:
|
|
self.logger.log(f"카테고리 그룹상품 발생 category_text = {category_text}", level=logging.DEBUG)
|
|
category_text_certified_element = main_category_element.locator(self.category_text_locator_certified)
|
|
|
|
if await category_text_certified_element.count():
|
|
category_text = await category_text_certified_element.inner_text()
|
|
self.logger.log(f"그룹상품 카테고리 text = {category_text}", level=logging.DEBUG)
|
|
else:
|
|
self.logger.log(f"카테고리 text = {category_text}", level=logging.DEBUG)
|
|
|
|
return category_text
|
|
|
|
else:
|
|
self.logger.log("카테고리 인증 요소를 찾을 수 없습니다.", level=logging.ERROR)
|
|
return ""
|
|
|
|
except Exception as e:
|
|
self.logger.log(f"카테고리 텍스트 가져오기 중 오류 발생: {e}", level=logging.ERROR, exc_info=True)
|
|
return ""
|
|
|
|
|
|
|
|
|
|
|
|
# async def get_category(self) -> dict:
|
|
# """
|
|
# 스마트스토어의 카테고리 정보를 가져오고,
|
|
# '상품정보제공고시' 요소의 위치에 따라 인증 및 그룹상품 여부를 판단합니다.
|
|
|
|
# Returns:
|
|
# dict: {
|
|
# "category_text_ss": <텍스트>,
|
|
# "category_text_esm": <텍스트>,
|
|
# "is_certified_ss": <True/False>,
|
|
# "is_group_esm": <True/False>
|
|
# }
|
|
# """
|
|
# try:
|
|
# category_data = {
|
|
# "category_text_ss": "",
|
|
# "category_text_esm": "",
|
|
# "is_certified_ss": False,
|
|
# "is_group_esm": False,
|
|
# }
|
|
|
|
# # 1. '상품정보제공고시' 요소의 위치를 확인하여 오프셋 계산
|
|
# found_index = None
|
|
# for idx in [5, 6, 7]:
|
|
# locator = f"div#productMainContentContainerId div:nth-child({idx}) > div > div > div.ant-row.ant-row-middle.css-1li46mu > div:nth-child(1) > span"
|
|
# try:
|
|
# await self.page.wait_for_selector(locator, timeout=2000, state="attached")
|
|
# info_element = self.page.locator(locator)
|
|
# text = await info_element.inner_text()
|
|
# if "상품정보제공고시" in text:
|
|
# found_index = idx
|
|
# self.logger.log(f"'상품정보제공고시' 요소가 nth-child {idx}에서 확인됨.", level=logging.DEBUG)
|
|
# break
|
|
# except Exception:
|
|
# continue
|
|
# if not found_index:
|
|
# found_index = 5
|
|
# self.logger.log("상품정보제공고시 요소를 찾지 못해 기본값(nth-child 5)을 사용합니다.", level=logging.WARNING)
|
|
# offset = found_index - 5 # offset: 0, 1, 또는 2
|
|
|
|
# # 2. 마켓별 카테고리 선택자 동적 조정
|
|
# # 기본: SS는 nth-child 8, ESM은 nth-child 9, 여기에 offset을 더함.
|
|
# base_ss = 8 + offset
|
|
# base_esm = 9 + offset
|
|
|
|
# ss_selector = f"div#productMainContentContainerId div:nth-child({base_ss}) > div > div > div.ant-row.ant-row-no-wrap.ant-row-bottom.css-1li46mu > div:nth-child(1) > div > div > span.ant-select-selection-item > div > div"
|
|
# esm_selector = f"div#productMainContentContainerId div:nth-child({base_esm}) > div > div > div.ant-row.ant-row-no-wrap.ant-row-bottom.css-1li46mu > div:nth-child(1) > div > div > span.ant-select-selection-item > div > div"
|
|
|
|
# self.logger.log(f"SS의 동적 카테고리 선택자: {ss_selector}", level=logging.DEBUG)
|
|
# self.logger.log(f"ESM의 동적 카테고리 선택자: {esm_selector}", level=logging.DEBUG)
|
|
|
|
# # 3. SS 카테고리 텍스트 추출
|
|
# await self.page.wait_for_selector(ss_selector, timeout=5000, state="attached")
|
|
# ss_element = self.page.locator(ss_selector)
|
|
# count_ss = await ss_element.count()
|
|
# if count_ss > 1:
|
|
# # 인증 요소 등이 포함되어 있으므로 두 번째 요소에서 실제 텍스트 추출
|
|
# category_text_ss = await ss_element.nth(1).inner_text()
|
|
# self.logger.log("SS: 인증 요소 포함되어 있어 두 번째 요소에서 카테고리 텍스트 추출.", level=logging.DEBUG)
|
|
# category_data["is_certified_ss"] = True
|
|
# elif count_ss == 1:
|
|
# category_text_ss = await ss_element.inner_text()
|
|
# else:
|
|
# self.logger.log("SS의 카테고리 텍스트 요소를 찾을 수 없습니다.", level=logging.ERROR)
|
|
# category_text_ss = ""
|
|
# category_data["category_text_ss"] = category_text_ss.strip()
|
|
|
|
# # 4. ESM 카테고리 텍스트 추출
|
|
# await self.page.wait_for_selector(esm_selector, timeout=5000, state="attached")
|
|
# esm_element = self.page.locator(esm_selector)
|
|
# count_esm = await esm_element.count()
|
|
# if count_esm > 1:
|
|
# # 그룹상품 요소가 포함되어 있는 경우: 두 번째 요소에서 실제 텍스트 추출
|
|
# category_text_esm = await esm_element.nth(1).inner_text()
|
|
# self.logger.log("ESM: 그룹상품 요소 포함되어 있어 두 번째 요소에서 카테고리 텍스트 추출.", level=logging.DEBUG)
|
|
# category_data["is_group_esm"] = True
|
|
# elif count_esm == 1:
|
|
# category_text_esm = await esm_element.inner_text()
|
|
# else:
|
|
# self.logger.log("ESM의 카테고리 텍스트 요소를 찾을 수 없습니다.", level=logging.ERROR)
|
|
# category_text_esm = ""
|
|
# category_data["category_text_esm"] = category_text_esm.strip()
|
|
|
|
# return category_data
|
|
|
|
# except Exception as e:
|
|
# self.logger.log(f"카테고리 텍스트 가져오기 중 오류 발생: {e}", level=logging.ERROR, exc_info=True)
|
|
# return {
|
|
# "category_text_ss": "",
|
|
# "category_text_esm": "",
|
|
# "is_certified_ss": False,
|
|
# "is_group_esm": False,
|
|
# }
|
|
|
|
|
|
|
|
|
|
|
|
|
|
async def get_category(self) -> dict:
|
|
"""
|
|
스마트스토어의 카테고리 정보를 가져오고,
|
|
'상품정보제공고시' 기준 요소의 위치에 따라 인증 및 그룹상품 여부를 판단합니다.
|
|
|
|
Config.ini에서 아래 선택자들을 사용합니다.
|
|
- consumer_product_informartion_disclosure_locator: "상품정보제공고시" 기준 선택자 (예: nth-child(5))
|
|
- category_main_selector_with_ss: SS 기본 카테고리 선택자 (예: nth-child(8))
|
|
- category_main_selector_with_esm: ESM 기본 카테고리 선택자 (예: nth-child(9))
|
|
|
|
반환 예시:
|
|
{
|
|
"category_text_ss": <텍스트>,
|
|
"category_text_esm": <텍스트>,
|
|
"is_certified_ss": <True/False>,
|
|
"is_group_esm": <True/False>
|
|
}
|
|
"""
|
|
try:
|
|
category_data = {
|
|
"category_text_ss": "",
|
|
"category_text_esm": "",
|
|
"is_certified_ss": False,
|
|
"is_group_esm": False,
|
|
}
|
|
|
|
# 1. '상품정보제공고시' 기준 요소의 선택자를 config.ini에서 가져옴
|
|
|
|
self.logger.log(f"기준 선택자(상품정보제공고시): {self.consumer_product_informartion_disclosure_locator}", level=logging.DEBUG)
|
|
|
|
found_index = None
|
|
# consumer_product_informartion_disclosure_locator 기본적으로 nth-child(5)로 설정되어 있음.
|
|
# 실제 위치가 5, 6, 7 중 어느 곳에 있는지 확인합니다.
|
|
|
|
for idx in [5, 6, 7]:
|
|
# 동적으로 nth-child 인덱스를 변경
|
|
info_locator = re.sub(r"nth-child\(\d+\)", f"nth-child({idx})", self.consumer_product_informartion_disclosure_locator, count=1)
|
|
try:
|
|
await self.page.wait_for_selector(info_locator, timeout=2000, state="attached")
|
|
info_element = self.page.locator(info_locator)
|
|
text = await info_element.inner_text()
|
|
if "상품정보제공고시" in text:
|
|
found_index = idx
|
|
self.logger.log(f"'상품정보제공고시' 요소가 nth-child {idx}에서 확인됨.", level=logging.DEBUG)
|
|
break
|
|
except Exception:
|
|
continue
|
|
if not found_index:
|
|
found_index = 5
|
|
self.logger.log("상품정보제공고시 요소를 찾지 못해 기본값(nth-child 5)을 사용합니다.", level=logging.WARNING)
|
|
offset = found_index - 5 # offset: 0, 1 또는 2
|
|
|
|
self.logger.log(f"기본 SS 선택자: {self.category_main_selector_with_ss}", level=logging.DEBUG)
|
|
self.logger.log(f"기본 ESM 선택자: {self.category_main_selector_with_esm}", level=logging.DEBUG)
|
|
|
|
# 기본값: SS는 nth-child 8, ESM은 nth-child 9 → 오프셋 적용
|
|
base_ss = 8 + offset
|
|
base_esm = 9 + offset
|
|
|
|
# 동적으로 nth-child 부분을 변경 (선택자 문자열의 첫 번째 "div:nth-child(숫자)"를 교체)
|
|
ss_selector = re.sub(r"div:nth-child\(\d+\)", f"div:nth-child({base_ss})", self.category_main_selector_with_ss, count=1)
|
|
esm_selector = re.sub(r"div:nth-child\(\d+\)", f"div:nth-child({base_esm})", self.category_main_selector_with_esm, count=1)
|
|
|
|
self.logger.log(f"SS의 동적 카테고리 선택자: {ss_selector}", level=logging.DEBUG)
|
|
self.logger.log(f"ESM의 동적 카테고리 선택자: {esm_selector}", level=logging.DEBUG)
|
|
|
|
# 3. SS 카테고리 텍스트 추출
|
|
await self.page.wait_for_selector(ss_selector, timeout=5000, state="attached")
|
|
ss_element = self.page.locator(ss_selector)
|
|
count_ss = await ss_element.count()
|
|
if count_ss > 1:
|
|
# 요소 2개 이상이면 첫 번째는 인증 요소, 두 번째가 실제 텍스트
|
|
category_text_ss = await ss_element.nth(1).inner_text()
|
|
self.logger.log("SS: 인증 요소 포함되어 있어 두 번째 요소에서 카테고리 텍스트 추출.", level=logging.DEBUG)
|
|
category_data["is_certified_ss"] = True
|
|
elif count_ss == 1:
|
|
category_text_ss = await ss_element.inner_text()
|
|
else:
|
|
self.logger.log("SS의 카테고리 텍스트 요소를 찾을 수 없습니다.", level=logging.ERROR)
|
|
category_text_ss = ""
|
|
category_data["category_text_ss"] = category_text_ss.strip()
|
|
|
|
# 4. ESM 카테고리 텍스트 추출
|
|
await self.page.wait_for_selector(esm_selector, timeout=5000, state="attached")
|
|
esm_element = self.page.locator(esm_selector)
|
|
count_esm = await esm_element.count()
|
|
if count_esm > 1:
|
|
# 요소 2개 이상이면 첫 번째는 그룹상품 태그, 두 번째가 실제 텍스트
|
|
category_text_esm = await esm_element.nth(1).inner_text()
|
|
self.logger.log("ESM: 그룹상품 요소 포함되어 있어 두 번째 요소에서 카테고리 텍스트 추출.", level=logging.DEBUG)
|
|
category_data["is_group_esm"] = True
|
|
elif count_esm == 1:
|
|
category_text_esm = await esm_element.inner_text()
|
|
else:
|
|
self.logger.log("ESM의 카테고리 텍스트 요소를 찾을 수 없습니다.", level=logging.ERROR)
|
|
category_text_esm = ""
|
|
category_data["category_text_esm"] = category_text_esm.strip()
|
|
|
|
return category_data
|
|
|
|
except Exception as e:
|
|
self.logger.log(f"카테고리 텍스트 가져오기 중 오류 발생: {e}", level=logging.ERROR, exc_info=True)
|
|
return {
|
|
"category_text_ss": "",
|
|
"category_text_esm": "",
|
|
"is_certified_ss": False,
|
|
"is_group_esm": False,
|
|
}
|