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', # 'price_modification': 'complete', # 'title_modification': 'complete', # 'trans_detailImage': 'complete', # 'upload_to_market': '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