mongoDB_test

This commit is contained in:
9700X_PC 2025-02-08 09:54:27 +09:00
parent c41b8a20c1
commit b9f30cfb21
3 changed files with 517 additions and 18 deletions

View File

@ -1,26 +1,57 @@
from mongoDBManager import MongoDBManager
from forbiddenWD_Manager import ForbiddenWordManager
from datetime import datetime, timezone
def insert_forbidden_words(db_manager, words_str, added_by="me", status="인증"):
"""
콤마로 구분된 단어 문자열을 받아 단어별로 문서를 생성하여 DB에 등록하는 함수입니다.
Parameters:
db_manager: MongoDBManager 인스턴스
words_str (str): 콤마로 구분된 단어 목록 (: "apple, banana, orange")
added_by (str): 단어를 추가한 사용자 (기본값: "me")
status (str): 단어의 상태 (기본값: "인증")
"""
# 입력된 문자열을 콤마로 분리하고, 양쪽 공백을 제거한 후 빈 문자열은 제외
words = [word.strip() for word in words_str.split(",") if word.strip()]
# 각 단어별로 문서를 생성하여 DB에 등록
for word in words:
document = {
"word": word,
"has_trademark": True,
"title": "Dummy Title Max",
"registration_date": "2020-03-01",
"applicant_name": "Dummy Applicant",
"category_code": "003",
"category_description": "Dummy description",
"created_at": datetime.now(timezone.utc),
"added_by": added_by,
"status": status
}
db_manager.insert_one(document)
print(f"'{word}'가 금지어로 등록됨.")
if __name__ == "__main__":
db_manager = MongoDBManager()
# MongoDB 연결 URL 설정
db_url = 'mongodb://root:1234@cckb9998.synology.me:27017/'
# MongoDB Manager 및 금지어 Manager 인스턴스 생성
db_manager = MongoDBManager(db_url=db_url)
forbidden_manager = ForbiddenWordManager(db_manager)
forbidden_manager.register_word(
word="example123",
has_trademark=True,
category_code="35",
category_description="광고업; 사업 관리",
registrant="Example Corp",
registration_date="2024-11-03",
added_by="user123"
)
# 관리자 인증 예시
# 콤마로 구분된 단어 목록 (예시)
forbidden_words_str = "나이키,아디다스,샤오미,아레나,화장품,식품,미즈노,산리오,폴로, polo, 애플, 자라, ZARA, 앨범, 게임타이틀, 레스포삭, 루이비통, 다이슨, 레고, 짱구, 라코스테, 블루레이,비비안, 비너스, 캘빈클라인, adidas, nike, 미키마우스, 르카프, BYC, 지오지아, 다이와, daiwa, 시마노, shimano, 골프, 승마, 이니스프리, 니콘, 후지필름, 앱손, 삼성전자, 중고, 갤럭시,타카, 아이폰, 휠체어,쌤쏘나이트,샘소나이트,스케쳐스,축구화,윌슨,아식스,요넥스,코치,COACH,더로우,the row,아미,ami,타미힐피거,Porsche,BMW,Audi,volvo,뉴발란스,비비드,borsalino,보르살리노,닥터마틴,팀버랜드,ArcTeryx,아크테릭스,허먼밀러,Samsonite,렉시떼,콘트라템포,apple,에어팟,프리휠러스,바버,barbour,예티,yeti,버즈릭슨,슈가케인,오닉스,onix,레노버,키즈,아동,유아,버버리,Jolyn,졸린,에어팟프로,아이팟,DANTON,단톤,어그,온러닝,세이코,포켓몬,건담,룰루레몬,리닝,리렌,파타고니아,꼼데가르송,미니로디니,바챠,bachaa,컘퍼,니들스,컨버스,나인웨스트,프라다,펜디,샤넬,르메르,LEMAIRE,오니츠카,요시다,타정,수비드,칼하트,몽벨,피코탄,보테가,베네타,캉골,몬데인,Mondaine,프리츠한센,카이저,이델,르크루제,발레리,키보드,LEMOUTON,르무통,로퍼,유니클로,조말론,Wilson,마샬,Beats,구글,로보락,Nintendo,미니맥시,minimaxi,슈올즈,옥펌기,직펌기,블랙다이아몬드,반스,Gucci,구찌,리복,티어,TYR,언더아머,힐피거,클라터뮤젠,랄프로렌,FCE,에프씨이,로렌랄프로렌,페더드프렌즈,비트라,Vitra,발망,캐피탈,KAPITAL,살로몬,SALOMON,보쉬,bosch,스탠리,stanley,마끼다,makita,디월트,DeWALT,KUOSE,쿠오세,에일릭로봇,Eilik,공식판매점,암웨이,슬라이락,보양,아쿠아픽,물레방아커터,Apple,네이처하이크,쌤소나이트,38explore,조디백,보테,로로피아나,캐나다구스,MKA,ugg,노스페이스,Mercedes,VOLVO,스타벅스,반다이,베어브릭,내셔널지오그래픽,WEDGWOOD,토리버치,뱅앤올룹슨,브론슨,고어텍스,지프,EQLZ,소니,블랙베리,Daiwa,전동건,usm,USM,스노우피크,Jackson,안타,ANTA,필슨,FILSON,맨체스터,푸마,포르쉐,버팔로,파운터코리아,에르메스,데상트,헬리콘텍스,Theory,띠어리,밍크바라클라바,에센셜,마지스,MAGIS,삼손체어,SamSon,벤츠,몽클레어,테크먼트,파운터,BEAMS,락포트,Rockport,오스프리,오클리,BAUER,카시오,노다,norda,리바이스,Citizen,자크뮈스,토즈,빌라쥬드레브,이탈리아아울렛,스톤아일랜드,미우미우,KEEN,Moncler,NVIDIA,빅토리아시크릿,나나미카,쯔리무사,페라가모,베르사체,TIMEX,디올,겐조,타이맥스,Artek,아르텍,라스포르티바,스카르파,FC바르셀로나,버켄스탁,umily,UMILY,PREMIATA,데스포르치,마르티넬리루체,PXG,제이린드버그,테일러메이드,핑,PING,지포어,타이틀리스트,어뉴,왁,Wac,오딧세이,캘러웨이,Callaway,클레버랜드,풋조이,말본,Malbon,스릭슨,미즈노,브릿지스톤,포틴,아담스골프,니켄트,윌슨,나이키골프,코브라,윌리암스,까스텔바작,와이드앵글,르꼬끄,팬텀,몽클레어,데상트,마스터바니,마크앤로나,사우스케이프,톨비스트,레노마골프,벤제프,JDX,마제스티,어뉴골프,PRGR,브릿지스톤,PEARLY GATES,젝시오"
# 콤마로 구분된 단어 목록을 금지어로 등록
insert_forbidden_words(db_manager, forbidden_words_str, added_by="me", status="인증")
# 관리자 인증 예시 (필요시 주석 해제 후 사용)
# forbidden_manager.authenticate_word("example123", admin_user="admin_user")
# 인증된 단어만 조회
words = forbidden_manager.list_all_words(filter_status="인증")
print("인증된 단어 목록:", words)
# 인증된 단어만 조회 (필요시 주석 해제 후 사용)
# words = forbidden_manager.list_all_words(filter_status="인증")
# print("인증된 단어 목록:", words)
# DB 연결 종료
db_manager.close_connection()

263
titleManager/db_test1.py Normal file
View File

@ -0,0 +1,263 @@
from pymongo import MongoClient, ASCENDING
from pymongo.errors import ConnectionFailure
from typing import Optional, List, Dict
import logging
from datetime import datetime, timezone
# --- MongoDBManager 클래스 ---
class MongoDBManager:
def __init__(self, db_url: str, db_name: str = 'AutoPercenty'):
"""MongoDB와의 연결을 설정하고 데이터베이스를 설정합니다."""
try:
self.client = MongoClient(db_url)
self.db = self.client[db_name]
self.collection = None
print("MongoDB 연결 성공!")
except ConnectionFailure as e:
print(f"MongoDB 연결 실패: {e}")
raise
def set_collection(self, collection_name: str):
"""컬렉션을 설정합니다."""
self.collection = self.db[collection_name]
def insert_one(self, document: Dict) -> bool:
"""단일 문서를 컬렉션에 추가합니다."""
try:
self.collection.insert_one(document)
return True
except Exception as e:
print(f"문서 삽입 실패: {e}")
return False
def find_one(self, query: Dict) -> Optional[Dict]:
"""단일 문서를 조회합니다."""
return self.collection.find_one(query)
def find(self, query: Dict, projection: Dict = None) -> List[Dict]:
"""여러 문서를 조회합니다."""
return list(self.collection.find(query, projection))
def update_one(self, query: Dict, update_data: Dict) -> bool:
"""단일 문서를 업데이트합니다."""
result = self.collection.update_one(query, {"$set": update_data})
return result.modified_count > 0
def delete_one(self, query: Dict) -> bool:
"""단일 문서를 삭제합니다."""
result = self.collection.delete_one(query)
return result.deleted_count > 0
def insert_multiple(self, documents: List[Dict]) -> bool:
"""여러 문서를 컬렉션에 추가합니다."""
try:
if len(documents) > 10:
documents = documents[:10] # 최대 10개까지만 추가
self.collection.insert_many(documents)
return True
except Exception as e:
print(f"여러 문서 삽입 실패: {e}")
return False
def find_all(self, query: Dict = {}) -> List[Dict]:
"""모든 문서를 조회합니다."""
return list(self.collection.find(query))
def close_connection(self):
"""MongoDB 연결을 닫습니다."""
self.client.close()
print("MongoDB 연결이 종료되었습니다.")
# --- ForbiddenWordManager 클래스 ---
logger = logging.getLogger('ForbiddenWordManager')
class ForbiddenWordManager:
def __init__(self, db_manager, collection_name='ForbbidenDB'):
self.db_manager = db_manager
self.db_manager.set_collection(collection_name)
def register_word(self, word: str, kipris_results: List[Dict], added_by: str):
"""
Kipris API 결과를 바탕으로 최대 10개의 금지어를 등록.
"""
count = 0
for result in kipris_results:
if count >= 10:
break
category_code = result.get("category_code")
existing_entry = self.db_manager.find_one({"word": word, "category_code": category_code})
if existing_entry:
logger.debug(f"'{word}' (카테고리 코드: {category_code})는 이미 등록되어 있음.")
continue
document = {
"word": word,
"has_trademark": True,
"title": result.get("title"),
"registration_date": result.get("registration_date", "N/A"),
"applicant_name": result.get("applicant_name", "N/A"),
"category_code": category_code,
"category_description": result.get("category_description", "N/A"),
"created_at": datetime.utcnow(),
"added_by": added_by,
"status": "미인증"
}
self.db_manager.insert_one(document)
logger.debug(f"'{word}' (카테고리 코드: {category_code})가 금지어로 등록됨.")
count += 1
if count == 0:
logger.debug(f"'{word}'와 관련된 새로운 카테고리 코드가 발견되지 않음.")
else:
logger.debug(f"'{word}'와 관련된 금지어가 {count}개 등록됨.")
def authenticate_word(self, word: str, admin_user: str):
"""관리자가 단어를 인증"""
result = self.db_manager.update_one(
{"word": word},
{"$set": {"status": "인증", "authenticated_by": admin_user, "authenticated_at": datetime.utcnow()}}
)
# 참고: update_one에서 bool을 반환하므로 matched_count 접근은 불가합니다.
if result:
logger.debug(f"'{word}'가 인증됨.")
return True
else:
logger.debug(f"'{word}'를 찾을 수 없음.")
return False
def update_word(self, word: str, update_fields: dict):
"""금지어 정보 업데이트 (관리자만 가능)"""
update_fields["updated_at"] = datetime.utcnow()
result = self.db_manager.update_one({"word": word}, {"$set": update_fields})
if result:
logger.debug(f"'{word}'의 정보가 업데이트됨.")
return True
else:
logger.debug(f"'{word}'를 찾을 수 없음.")
return False
def delete_word(self, word: str):
"""금지어 삭제"""
result = self.db_manager.delete_one({"word": word})
if result:
logger.debug(f"'{word}'가 삭제됨.")
return True
else:
logger.debug(f"'{word}'를 찾을 수 없음.")
return False
def list_all_words(self, filter_status=None):
"""모든 금지어 목록 조회 (필터링 가능)"""
query = {}
if filter_status:
query["status"] = filter_status
return list(self.db_manager.find(query, {"_id": 0}))
def is_word_forbidden(self, word: str) -> List[Dict]:
"""단어가 금지어 목록에 있는지 확인하고 모든 일치하는 레코드를 반환"""
results = list(self.db_manager.find({"word": word}, {"_id": 0}))
if results:
logger.debug(f"'{word}'에 대한 금지어가 {len(results)}개 발견됨.")
return results
else:
logger.debug(f"'{word}'에 대한 금지어가 발견되지 않음.")
return []
# --- 테스트 코드 ---
if __name__ == "__main__":
# 로깅 설정 (DEBUG 레벨로 설정하여 logger.debug 출력 확인)
logging.basicConfig(level=logging.DEBUG)
# MongoDB 연결 설정 (실제 환경에 맞게 수정)
db_url = 'mongodb://root:1234@cckb9998.synology.me:27017/'
db_name = 'AutoPercenty'
# MongoDBManager 인스턴스 생성
db_manager = MongoDBManager(db_url=db_url, db_name=db_name)
# ForbiddenWordManager 인스턴스 생성 (자동으로 'ForbbidenDB' 컬렉션 사용)
forbidden_manager = ForbiddenWordManager(db_manager)
# # 테스트를 위해 기존 금지어 컬렉션을 초기화 (컬렉션 삭제)
# db_manager.collection.drop()
# print("기존 금지어 컬렉션 초기화 완료.\n")
# # 샘플 금지어 데이터 등록 (예제에서는 'Apple', 'iPhone', 'Max'를 금지어로 등록)
# sample_forbidden_words = [
# {
# "word": "Apple",
# "has_trademark": True,
# "title": "Dummy Title Apple",
# "registration_date": "2020-01-01",
# "applicant_name": "Dummy Applicant",
# "category_code": "001",
# "category_description": "Dummy description",
# "created_at": datetime.now(timezone.utc),
# "added_by": "test",
# "status": "인증"
# },
# {
# "word": "iPhone",
# "has_trademark": True,
# "title": "Dummy Title iPhone",
# "registration_date": "2020-02-01",
# "applicant_name": "Dummy Applicant",
# "category_code": "002",
# "category_description": "Dummy description",
# "created_at": datetime.now(timezone.utc),
# "added_by": "test",
# "status": "인증"
# },
# {
# "word": "Max",
# "has_trademark": True,
# "title": "Dummy Title Max",
# "registration_date": "2020-03-01",
# "applicant_name": "Dummy Applicant",
# "category_code": "003",
# "category_description": "Dummy description",
# "created_at": datetime.now(timezone.utc),
# "added_by": "test",
# "status": "인증"
# }
# ]
# for doc in sample_forbidden_words:
# if db_manager.insert_one(doc):
# print(f"'{doc['word']}' 금지어 등록 완료.")
# else:
# print(f"'{doc['word']}' 금지어 등록 실패.")
# 테스트용 샘플 상품명 설정
sample_product_name = "Apple iPhone 14 Pro Max"
print(f"\n샘플 상품명: {sample_product_name}")
# 상품명을 단어 단위(공백 기준)로 분리
product_words = sample_product_name.split()
print("상품명 단어 목록:", product_words)
# 각 단어에 대해 금지어 여부 확인
forbidden_words_found = {}
for word in product_words:
entries = forbidden_manager.is_word_forbidden(word)
if entries:
forbidden_words_found[word] = entries
print(f"단어 '{word}'는 금지어 목록에 존재합니다.")
else:
print(f"단어 '{word}'는 금지어 목록에 존재하지 않습니다.")
# 최종 필터링 결과 출력
print("\n금지어 필터링 결과:")
if forbidden_words_found:
for word, entries in forbidden_words_found.items():
print(f"'{word}': {entries}")
else:
print("금지어가 발견되지 않았습니다.")
# MongoDB 연결 종료
db_manager.close_connection()

205
titleManager/db_test2.py Normal file
View File

@ -0,0 +1,205 @@
import re
import random
import logging
from datetime import datetime
# =============================================================================
# TitleGenerator 클래스 (테스트에 필요한 부분만 발췌하여 최소 구현)
# =============================================================================
class TitleGenerator:
def __init__(self, locator_manager, browser_controller, logger, toggle_states, gpt_client):
self.logger = logger
self.browser_controller = browser_controller
self.locator_manager = locator_manager
self.toggle_states = toggle_states
# 실제 코드에서는 MongoDBManager 등을 초기화하지만,
# 테스트에서는 더미 객체를 사용하므로 여기서는 None 처리
self.forbidden_word_manager = None
self.gpt_client = gpt_client
# 실제 번역 라이브러리 대신, translate_product_name을 단순 반환으로 구현
self.translator = None
def translate_product_name(self, original_name: str) -> str:
# 테스트에서는 번역 없이 원본 상품명을 그대로 사용
return original_name
def is_word_forbidden(self, word: str) -> bool:
"""더미 금지어 매니저로 위임 (없으면 False 반환)"""
if self.forbidden_word_manager:
return self.forbidden_word_manager.is_word_forbidden(word)
return False
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:
"""검색 결과 상위 제목들에서 유효하지 않은 단어들을 제거"""
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 generate_product_title(self, original_name: str, keyword_name: str, search_result: list, product_category: str) -> str:
"""
상품명을 생성하는 메서드
1. 원본 상품명을 번역(테스트에서는 그대로 사용)
2. GPT Client를 통해 원본과 키워드의 관련성을 판단
3. 검색 결과에서 상위 제목과 가격을 추출하여 처리
4. 키워드, 상위 제목, 원본의 특수 단어 등을 합쳐 최종 키워드 목록을 생성
5. 금지어 필터링 , GPT Client를 통해 최종 상품명을 생성하여 반환
"""
# 1. 번역 (여기서는 그대로 사용)
translated_name = self.translate_product_name(original_name)
self.logger.debug(f"translated_name: {translated_name}")
# 2. 관련성 판단 (더미 GPT Client 사용)
is_related = self.gpt_client.is_related_product(translated_name, keyword_name)
if not is_related:
self.logger.debug("원본상품명과 키워드 간 관련성이 낮아 추가 검증 필요.")
# 3. 검색 결과 처리: 상위 제목, 가격 추출
top_titles = []
top_prices = []
if 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.debug(f"top_titles: {top_titles}")
self.logger.debug(f"top_prices: {top_prices}")
else:
self.logger.warning("검색 결과가 비어 있습니다.")
# 4. 검색된 제목 필터링
filtered_top_titles = self.process_top_titles(top_titles)
# 5. 키워드 상품명에서 첫 4개 단어 추출
essential_keywords = keyword_name.split()[:4]
self.logger.debug(f"essential_keywords (첫 4개): {essential_keywords}")
# [keyword_name]과 필터링된 제목들을 합쳐 단어 단위 분해 후 집합 처리
keyword_title = list(set(
word for title in [keyword_name] + filtered_top_titles
for word in title.split()
))
self.logger.debug(f"초기 keyword_title: {keyword_title}")
# 6. 유효하지 않은 단어 제거
keyword_title = [word for word in keyword_title if self.is_valid_word(word)]
self.logger.debug(f"keyword_title after filtering invalid words: {keyword_title}")
keyword_title = list(set(keyword_title))
self.logger.debug(f"final keyword_title after deduplication: {keyword_title}")
# 7. 원본 상품명에서 특수 단어 추출 후 포함
special_words = self.extract_special_words(original_name)
self.logger.debug(f"special_words from original_name: {special_words}")
keyword_title.extend(special_words)
keyword_title = list(set(keyword_title))
self.logger.debug(f"keyword_title including special words: {keyword_title}")
# 8. 필수 키워드 중 랜덤 2개 선택 후 추가
required_keywords = random.sample(essential_keywords, min(2, len(essential_keywords)))
self.logger.debug(f"randomly selected required_keywords: {required_keywords}")
keyword_title.extend(required_keywords)
keyword_title = list(set(keyword_title))
# 9. 금지어 필터링 (더미 forbidden_word_manager 사용)
keyword_title = [word for word in keyword_title if not self.is_word_forbidden(word)]
self.logger.debug(f"keyword_title after forbidden filter: {keyword_title}")
# 10. 최종 상품명 생성 (더미 GPT Client의 generate_product_name_next 호출)
product_title = self.gpt_client.generate_product_name_next(words=keyword_title, original_name=original_name, top_titles=top_titles)
self.logger.debug(f"final product_title: {product_title}")
return product_title
# =============================================================================
# 더미(Stub) 의존 객체 클래스들
# =============================================================================
class DummyLocatorManager:
def get_locator(self, category, locator_name):
return f"{category}_{locator_name}"
class DummyBrowserController:
def __init__(self):
self.page = None # 테스트에서는 사용하지 않음
class DummyGPTClient:
def is_related_product(self, translated_name, keyword_name):
# 테스트에서는 항상 관련 있다고 가정
return True
def generate_product_name_next(self, words, original_name, top_titles):
# 테스트용: 원본 상품명과 정렬된 단어들을 조합하여 반환
return f"{original_name} - {' '.join(sorted(words))}"
class DummyForbiddenWordManager:
def is_word_forbidden(self, word):
# 예를 들어, "bad"라는 단어는 금지어로 판단
return word.lower() == "bad"
# =============================================================================
# 테스트 함수: generate_product_title 메서드 검증
# =============================================================================
def test_generate_product_title():
# 로깅 설정
logger = logging.getLogger("TitleGeneratorTest")
logger.setLevel(logging.DEBUG)
handler = logging.StreamHandler()
formatter = logging.Formatter("%(levelname)s - %(message)s")
handler.setFormatter(formatter)
if not logger.handlers:
logger.addHandler(handler)
# 더미 토글 상태 (네이버 API 키 등 실제 사용되지 않음)
toggle_states = {"clientID": "dummy_id", "clientSecret": "dummy_secret"}
# 더미 의존 객체 생성
locator_manager = DummyLocatorManager()
browser_controller = DummyBrowserController()
gpt_client = DummyGPTClient()
# TitleGenerator 인스턴스 생성
title_generator = TitleGenerator(locator_manager, browser_controller, logger, toggle_states, gpt_client)
# 금지어 매니저를 더미 객체로 교체 (예: "bad"는 금지어)
title_generator.forbidden_word_manager = DummyForbiddenWordManager()
# 테스트용 샘플 입력값
original_name = "Super Deluxe Coffee Maker"
keyword_name = "Coffee Maker Deluxe"
search_result = [
{"title": "Best Coffee Maker", "price": 15000},
{"title": "Deluxe Espresso Machine", "price": 25000},
{"related_tags": ["coffee", "maker", "espresso"]}
]
product_category = "Kitchen Appliances"
# generate_product_title 메서드 호출
generated_title = title_generator.generate_product_title(
original_name=original_name,
keyword_name=keyword_name,
search_result=search_result,
product_category=product_category
)
print("\nGenerated Product Title:")
print(generated_title)
# =============================================================================
# 메인 실행: 테스트 함수 호출
# =============================================================================
if __name__ == "__main__":
test_generate_product_title()