264 lines
10 KiB
Python
264 lines
10 KiB
Python
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()
|