100 lines
3.9 KiB
Python
100 lines
3.9 KiB
Python
import logging
|
|
from datetime import datetime
|
|
from typing import List, Dict, Optional
|
|
|
|
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()}}
|
|
)
|
|
if result.matched_count > 0:
|
|
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.matched_count > 0:
|
|
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.deleted_count > 0:
|
|
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 []
|