forked from ckh08045/AutoPercenty
209 lines
9.5 KiB
Python
209 lines
9.5 KiB
Python
from datetime import datetime
|
|
# 로거 인스턴스 가져오기
|
|
import logging
|
|
from edit.product_info import ProductInfo
|
|
|
|
logger = logging.getLogger('default_logger')
|
|
|
|
class AutoPercentyProductsDB:
|
|
def __init__(self, mongo_config):
|
|
self.mongo_config = mongo_config
|
|
self.db = mongo_config.client['taobao_project']
|
|
self.collection = self.db['AutoPercenty_products']
|
|
|
|
def initialize_process_steps(self, product_id, user):
|
|
steps = {'tag_modification': 'incomplete',
|
|
'option_modification': 'incomplete',
|
|
'detail_page_modification': 'incomplete',
|
|
'thumbnail_modification': 'incomplete'
|
|
# 'thumbnail_modification': 'incomplete',
|
|
# 'price_modification': 'incomplete',
|
|
# 'title_modification': 'incomplete'
|
|
}
|
|
document = {'product_id': product_id, 'process_steps': steps, 'user': user, 'created_at': datetime.now()}
|
|
self.collection.insert_one(document)
|
|
logger.info(f"Initialized process steps for product_id {product_id} by user {user}.")
|
|
|
|
|
|
# def mark_product_processed(self, product_id, current_user):
|
|
# # 상품 처리 정보가 이미 존재하는지 확인
|
|
# if not self.is_product_processed(product_id):
|
|
# document = {
|
|
# 'product_id': product_id,
|
|
# 'user_name': current_user,
|
|
# 'process_time': datetime.now(),
|
|
# # 초기 단계 정보는 여기에 포함하지 않음
|
|
# }
|
|
# self.collection.insert_one(document)
|
|
# print(f"상품 ID {product_id}의 처리가 시작되었습니다.")
|
|
# else:
|
|
# print(f"상품 ID {product_id}에 대한 처리 정보는 이미 존재합니다.")
|
|
|
|
|
|
def finalize_product_processing(self, product_id, current_user, product_info):
|
|
incomplete_steps = self.get_incomplete_steps(product_id)
|
|
if not incomplete_steps:
|
|
self.collection.update_one(
|
|
{'product_id': product_id},
|
|
{'$set': {
|
|
'finalized': True,
|
|
'finalized_time': datetime.now(),
|
|
'finalized_by': current_user,
|
|
'product_info': product_info.to_dict() # 상품 정보 딕셔너리로 저장
|
|
}}
|
|
)
|
|
logger.info(f"Product processing finalized for {product_id}")
|
|
else:
|
|
logger.warning(f"Cannot finalize product {product_id} with incomplete steps: {incomplete_steps}")
|
|
|
|
def is_product_processed(self, product_id):
|
|
# 상품의 현재 처리 단계 정보를 가져옴
|
|
process_steps = self.get_current_step(product_id)
|
|
|
|
# 상품에 대한 기본 정보 조회
|
|
product_info = self.collection.find_one({'product_id': product_id}, {'_id': 0, 'user_name': 1, 'process_time': 1})
|
|
|
|
if product_info:
|
|
# 모든 처리 단계가 완료되었는지 확인
|
|
all_steps_completed = all(status == 'complete' for status in process_steps.values()) if process_steps else False
|
|
|
|
# 조회 결과에 처리 단계의 완료 여부 추가
|
|
product_info['all_steps_completed'] = all_steps_completed
|
|
|
|
return product_info
|
|
else:
|
|
return False
|
|
|
|
def update_process_step(self, product_id, step_name, user):
|
|
update_result = self.collection.update_one(
|
|
{'product_id': product_id, f'process_steps.{step_name}': 'incomplete'},
|
|
{'$set': {f'process_steps.{step_name}': 'complete', 'last_updated_by': user, 'last_updated_at': datetime.now()}}
|
|
)
|
|
if update_result.modified_count == 0:
|
|
logger.info(f"Step {step_name} for product_id {product_id} is already complete or product does not exist.")
|
|
else:
|
|
logger.info(f"Updated step {step_name} to complete for product_id {product_id} by user {user}.")
|
|
|
|
|
|
def get_incomplete_steps(self, product_id):
|
|
product = self.collection.find_one({'product_id': product_id}, {'process_steps': 1, '_id': 0})
|
|
if product:
|
|
logger.debug("완료되지 않은 스텝 찾음")
|
|
return [step for step, status in product['process_steps'].items() if status == 'incomplete']
|
|
else:
|
|
logger.warning(f"해당 ID의 상품을 찾을 수 없습니다. : {product_id}.")
|
|
return []
|
|
|
|
|
|
def is_step_incomplete(self, product_id):
|
|
"""특정 상품의 단계가 완료되지 않았는지 여부를 반환합니다."""
|
|
product = self.collection.find_one({'product_id': product_id}, {'finalized': 1})
|
|
if product:
|
|
incomplete_steps = self.get_incomplete_steps(product_id)
|
|
if product.get('finalized'):
|
|
logger.debug(f"상품 ID {product_id}: 완료")
|
|
return True
|
|
elif len(incomplete_steps) < 6:
|
|
logger.debug(f"상품 ID {product_id}: 미완성 목록을 반환")
|
|
return incomplete_steps
|
|
else:
|
|
logger.debug(f"상품 ID {product_id}: 기록없음")
|
|
return False
|
|
|
|
def get_current_step(self, product_id):
|
|
result = self.collection.find_one({'product_id': product_id}, {'_id': 0, 'process_steps': 1})
|
|
if result and 'process_steps' in result:
|
|
return result['process_steps']
|
|
else:
|
|
return {}
|
|
|
|
def toggle_process_step(self, product_id, step_name):
|
|
# 상품의 현재 처리 단계 상태를 확인
|
|
product = self.collection.find_one({'product_id': product_id}, {'process_steps': 1})
|
|
|
|
if not product or 'process_steps' not in product:
|
|
print(f"상품 ID {product_id}에 대한 정보가 없거나, 처리 단계 정보가 없습니다.")
|
|
return
|
|
|
|
current_status = product['process_steps'].get(step_name, 'incomplete')
|
|
new_status = 'complete' if current_status == 'incomplete' else 'incomplete'
|
|
|
|
# 처리 단계 상태 전환
|
|
self.collection.update_one(
|
|
{'product_id': product_id},
|
|
{'$set': {f'process_steps.{step_name}': new_status}}
|
|
)
|
|
|
|
print(f"상품 ID {product_id}의 {step_name} 단계가 {new_status} 상태로 전환되었습니다.")
|
|
|
|
def reset_product_by_id(self, product_id):
|
|
# 상품 ID에 해당하는 문서를 데이터베이스에서 삭제
|
|
result = self.collection.delete_one({'product_id': product_id})
|
|
|
|
if result.deleted_count > 0:
|
|
print(f"상품 ID {product_id}에 대한 정보가 성공적으로 초기화되었습니다.")
|
|
else:
|
|
print(f"상품 ID {product_id}에 대한 정보를 찾을 수 없습니다.")
|
|
|
|
def complete_all_steps_for_product(self, product_id):
|
|
# 모든 처리 단계를 'complete'로 설정
|
|
steps = {
|
|
'tag_modification': 'complete',
|
|
'option_modification': 'complete',
|
|
'detail_page_modification': 'complete',
|
|
'thumbnail_modification': 'complete'
|
|
# 'thumbnail_modification': 'complete',
|
|
# 'price_modification': 'complete',
|
|
# 'title_modification': 'complete'
|
|
}
|
|
|
|
# 현재 시간
|
|
current_time = datetime.now()
|
|
|
|
# 상품 ID에 해당하는 문서를 업데이트하여 모든 단계를 완료로 설정하고, 최종 완료 표시
|
|
result = self.collection.update_one(
|
|
{'product_id': product_id},
|
|
{'$set': {
|
|
'process_steps': steps,
|
|
'finalized': True,
|
|
'finalized_time': current_time
|
|
}}
|
|
)
|
|
|
|
if result.matched_count > 0:
|
|
print(f"상품 ID {product_id}의 모든 처리 단계가 완료되었으며, 최종 처리가 완료되었습니다.")
|
|
else:
|
|
print(f"상품 ID {product_id}에 대한 정보를 찾을 수 없습니다.")
|
|
|
|
def get_product_info_by_id(self, product_id):
|
|
"""
|
|
데이터베이스에서 특정 상품 ID에 해당하는 상품 정보를 조회합니다.
|
|
조회된 정보를 바탕으로 ProductInfo 인스턴스를 생성하거나 업데이트합니다.
|
|
|
|
:param product_id: 조회할 상품의 ID
|
|
:return: ProductInfo 인스턴스 또는 조회 실패 시 None
|
|
"""
|
|
try:
|
|
# MongoDB에서 상품 ID에 해당하는 레코드 조회
|
|
product_data = self.collection.find_one({"product_id": product_id})
|
|
if product_data and 'product_info' in product_data:
|
|
# 조회된 데이터를 바탕으로 ProductInfo 인스턴스 생성
|
|
product_info_data = product_data['product_info']
|
|
product_info = ProductInfo()
|
|
|
|
# 데이터베이스에서 조회한 정보로 ProductInfo 인스턴스 속성 설정
|
|
for key, value in product_info_data.items():
|
|
if hasattr(product_info, key):
|
|
setattr(product_info, key, value)
|
|
|
|
# 로그 기록: 상품 정보 로드 성공
|
|
logging.debug(f"ProductInfo loaded successfully for product ID {product_id}.")
|
|
return product_info
|
|
else:
|
|
# 로그 기록: 상품 정보 없음
|
|
logging.warning(f"No product info found for product ID {product_id}.")
|
|
return None
|
|
except Exception as e:
|
|
# 예외 발생 시 로그 기록
|
|
logging.error(f"Error loading product info for product ID {product_id}: {e}", exc_info=True)
|
|
return None |