Spaces:
Running
Running
| # -*- coding: utf-8 -*- | |
| """app.ipynb | |
| Automatically generated by Colab. | |
| Original file is located at | |
| https://colab.research.google.com/drive/1hl-apQjNm2ijxBoDj-PV0ua7-4taVJo2 | |
| """ | |
| import os | |
| import sys | |
| __import__('pysqlite3') | |
| import sys | |
| sys.modules['sqlite3'] = sys.modules.pop('pysqlite3') | |
| import shutil | |
| import torch | |
| import re | |
| import chromadb | |
| from langchain_chroma import Chroma | |
| import pandas as pd | |
| import gradio as gr | |
| import ast | |
| import numpy as np | |
| from PIL import Image | |
| from collections import Counter, defaultdict | |
| from transformers import CLIPProcessor, CLIPModel | |
| from langchain_huggingface import HuggingFaceEmbeddings | |
| from langchain_openai import ChatOpenAI | |
| from langchain_core.prompts import PromptTemplate | |
| from sentence_transformers import util | |
| import sys | |
| import gspread | |
| from oauth2client.service_account import ServiceAccountCredentials | |
| import datetime | |
| import json | |
| key_content = os.environ.get("GSPREAD_KEY") | |
| # 만약 Secret이 있다면, 즉석에서 'fashion_key.json' 파일을 만들어줌 | |
| if key_content: | |
| with open("fashion_key.json", "w") as f: | |
| f.write(key_content) | |
| print("✅ 보안 키 파일 생성 완료!") | |
| else: | |
| print("⚠️ 경고: GSPREAD_KEY 환경변수가 없습니다. (로컬 실행이면 무시)") | |
| # [1] 구글 시트 연결 설정 함수 | |
| def get_google_sheet(): | |
| # 1. 인증 범위 설정 | |
| scope = ['https://spreadsheets.google.com/feeds', 'https://www.googleapis.com/auth/drive'] | |
| # 2. 다운받은 키 파일 이름 (같은 폴더에 있어야 함) | |
| creds = ServiceAccountCredentials.from_json_keyfile_name('fashion_key.json', scope) | |
| client = gspread.authorize(creds) | |
| # 3. 스프레드시트 열기 (공유한 파일 이름 정확히 입력!) | |
| sh = client.open('패션추천시스템_사용자정보') | |
| return sh | |
| # [2] 검색 데이터 저장 함수 (시트1에 저장) | |
| def log_search_to_sheet(bg, cat, min_p, max_p, style, txt, check, persona): | |
| try: | |
| sh = get_google_sheet() | |
| worksheet = sh.get_worksheet(0) # 첫 번째 시트 선택 | |
| # 헤더가 없으면 추가 (옵션) | |
| if not worksheet.row_values(1): | |
| worksheet.append_row(['시간', '브랜드', '카테고리', '최소가격', '최대가격', '스타일', '텍스트', '성별교차', '결과페르소나']) | |
| # 데이터 추가 | |
| timestamp = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") | |
| worksheet.append_row([timestamp, bg, cat, min_p, max_p, style, txt, "O" if check else "X", persona]) | |
| except Exception as e: | |
| print(f"구글 시트 저장 실패: {e}") | |
| # [3] 사이즈 추천 데이터 저장 함수 (시트2에 저장) | |
| def log_size_to_sheet(sel_product, u_info, rec_size, reason): | |
| try: | |
| sh = get_google_sheet() | |
| # 두 번째 시트가 없으면 생성, 있으면 가져오기 | |
| try: | |
| worksheet = sh.get_worksheet(1) | |
| except: | |
| worksheet = sh.add_worksheet(title="사이즈추천", rows="100", cols="20") | |
| if not worksheet.row_values(1): | |
| worksheet.append_row(['시간', '선택상품', '유저정보', '추천사이즈', '추천이유']) | |
| timestamp = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") | |
| user_info_str = json.dumps(u_info, ensure_ascii=False) | |
| worksheet.append_row([timestamp, sel_product, user_info_str, rec_size, reason]) | |
| except Exception as e: | |
| print(f"구글 시트 저장 실패: {e}") | |
| os.environ["OPENAI_API_KEY"] = os.environ.get("OPENAI_API_KEY") | |
| device = "cuda" if torch.cuda.is_available() else "cpu" | |
| base_path = "." | |
| product_db_path = os.path.join(base_path, "product_db") | |
| review_db_path = os.path.join(base_path, "review_db") | |
| style_db_path = os.path.join(base_path, "style_db") | |
| product_csv_path = os.path.join(base_path, "final_prod_data.csv") | |
| size_csv_path = os.path.join(base_path, "size_sorted.csv") | |
| review_csv_path = os.path.join(base_path, "final_review_df.csv") | |
| try: | |
| prod_data = pd.read_csv(product_csv_path) | |
| prod_data['primary_id'] = prod_data['primary_id'].astype(str) | |
| except: | |
| prod_data = pd.DataFrame() | |
| try: | |
| prod_size_df = pd.read_csv(size_csv_path) | |
| prod_size_df = prod_size_df.replace('-', np.nan) | |
| prod_size_df['primary_id'] = prod_size_df['primary_id'].astype(str) | |
| except: | |
| prod_size_df = pd.DataFrame() | |
| try: | |
| review_df = pd.read_csv(review_csv_path) | |
| review_df['primary_id'] = review_df['primary_id'].astype(int) | |
| review_ids = set(review_df['primary_id'].astype(str)) | |
| except: | |
| print(f"리뷰 데이터 로드 실패: {e}") | |
| review_ids = set() | |
| clip_model = CLIPModel.from_pretrained("patrickjohncyh/fashion-clip").to(device) | |
| clip_processor = CLIPProcessor.from_pretrained("patrickjohncyh/fashion-clip") | |
| review_emb_func = HuggingFaceEmbeddings( | |
| model_name='jhgan/ko-sroberta-multitask', | |
| model_kwargs={'device': device}, | |
| encode_kwargs={'normalize_embeddings': True} | |
| ) | |
| llm = ChatOpenAI(model = "gpt-4o-mini", temperature=0.7) # 결과값 확인 후 temperature 값 조정 필요 | |
| try: | |
| product_client = chromadb.PersistentClient(path=product_db_path) | |
| product_collection = product_client.get_collection("product_vectorstore") | |
| except Exception as e: | |
| print(f"{e} DB 연결 실패") | |
| try: | |
| style_client = chromadb.PersistentClient(path=style_db_path) | |
| style_collection = style_client.get_collection("style_vectorstore") | |
| except Exception as e: | |
| print(f"{e} DB 연결 실패") | |
| try: | |
| review_db = Chroma( | |
| persist_directory=review_db_path, | |
| embedding_function=review_emb_func, | |
| collection_name="review_vectorstore" | |
| ) | |
| except Exception as e: | |
| print(f"{e} DB 연결 실패") | |
| def get_style_options(): | |
| try: | |
| data = style_collection.get(include=["metadatas"], limit=9999) | |
| names = [m.get("display_name") for m in data["metadatas"] if m.get("display_name")] | |
| names = sorted(set(names)) | |
| return ["선택 안 함"] + names | |
| except Exception as e: | |
| print("ERROR in get_style_options:", e) | |
| return ["선택 안 함"] | |
| style_key_options = get_style_options() | |
| ''' | |
| 추천 상품의 사이즈와 유사도 점수 가장 높은 상품의 추천 사이즈 간 유클리디언 거리 측정 | |
| (단, 추천 상품의 사이즈 데이터가 하나 밖에 없을 경우 상세 사이즈 수치만 제시) | |
| ''' | |
| def set_fin_size(id, size_label): | |
| ''' | |
| [1단계] 유사도 점수가 가장 높은 상품 사이즈의 row 가져오기 | |
| ''' | |
| if prod_size_df.empty: | |
| return None | |
| try: | |
| r = prod_size_df[ | |
| (prod_size_df['primary_id'] == str(id)) & | |
| (prod_size_df['size_cleaned'].astype(str) == str(size_label)) | |
| ] | |
| if r.empty: | |
| return None | |
| numeric_cols = r.select_dtypes(include=[np.number]).columns.tolist() | |
| numeric_cols = [c for c in numeric_cols if c not in ['primary_id']] | |
| return r.iloc[0][numeric_cols].to_dict() | |
| except: | |
| return None | |
| def find_closest_size(target_id, ideal_spec): | |
| ''' | |
| [2단계] 추천 상품의 사이즈 중 ideal_spec에 가장 가까운 사이즈 찾기 | |
| ''' | |
| if prod_size_df.empty or not ideal_spec: | |
| return None | |
| target_r = prod_size_df[prod_size_df['primary_id'] == str(target_id)] | |
| if target_r.empty: | |
| return None | |
| if len(target_r) == 1: | |
| only_size = target_r.iloc[0]['size_cleaned'] | |
| return only_size | |
| if not ideal_spec: | |
| return None | |
| min_dist = float('inf') | |
| best_size = None | |
| common_keys = [k for k in ideal_spec.keys() if k in target_r.columns] | |
| if not common_keys: | |
| return None | |
| for _, row in target_r.iterrows(): | |
| try: | |
| c_specs = [float(ideal_spec[k]) for k in common_keys if pd.notna(row[k])] | |
| t_specs = [float(row[k]) for k in common_keys if pd.notna(row[k])] | |
| if not c_specs or len(c_specs) != len(t_specs): | |
| continue | |
| dist = np.linalg.norm(np.array(c_specs) - np.array(t_specs)) | |
| if dist < min_dist: | |
| min_dist = dist | |
| best_size = row['size_cleaned'] | |
| except: | |
| continue | |
| return best_size | |
| def search_recom_products(brand_lb, cat, | |
| sel_key, txt, img, user_info, | |
| min_price, max_price, check_option): | |
| #================================================================================= | |
| # 추천 1단계: 프롬프트에 입력된 스타일 값을 바탕으로 벡터 유사도 계산 후 top-5개 추천 | |
| #================================================================================ | |
| query_vec, source = get_search_vector(sel_key, txt, img) | |
| if query_vec is None: | |
| return [], "검색 조건을 입력하세요", {}, "", [] | |
| gender_option = [user_info['gender'], 'all'] # 남성: male, 여성: female, 공용: all nan: all로 바꾸기 | |
| filter = { | |
| "$and": [ | |
| {"brand_label": {"$eq": brand_lb}}, | |
| {"sub_category": {"$eq": cat}} | |
| ] | |
| } | |
| if min_price is not None and max_price is not None: | |
| if min_price > max_price: | |
| min_price, max_price = max_price, min_price | |
| filter["$and"].append({"price": {"$gte": min_price}}) | |
| filter["$and"].append({"price": {"$lte": max_price}}) | |
| if not check_option: | |
| filter["$and"].append({"gender": {"$in": gender_option}}) | |
| candidates = product_collection.query( | |
| query_embeddings=[query_vec], | |
| n_results=20, | |
| where=filter, | |
| include=["metadatas", "embeddings", "distances"] | |
| ) | |
| if not candidates['ids'][0]: | |
| return [], "조건에 맞는 상품이 없습니다.", {}, "", [] | |
| candidate_list_text = "" | |
| candidate_map = {} | |
| for i, meta in enumerate(candidates['metadatas'][0]): | |
| p_id = meta.get('primary_id') | |
| name = meta.get('product_name') | |
| brand = meta.get('brand') | |
| price = meta.get('price') | |
| dist = candidates['distances'][0][i] | |
| sim = 1 / (1 + dist) # 유사도 점수로 변환 | |
| candidate_list_text += f"[{i+1}] ID: {p_id} | 브랜드: {brand} | 제품명: {name} | 스타일유사도: {sim:.4f}\n" | |
| candidate_map[str(i+1)] = { | |
| "id": p_id, "name": name, "brand": brand, "price": price, | |
| "cat": meta.get('sub_category'), "brand_lb": meta.get('brand_label'), | |
| "idx": i | |
| } | |
| print(candidate_map) | |
| style_requests = [] | |
| if txt: | |
| style_requests.append(f"텍스트 요청: '{txt}'") | |
| if sel_key and sel_key != '선택 안 함': | |
| style_key_data = style_collection.get( | |
| where={"display_name": {"$eq": sel_key}}, | |
| include=["metadatas"] | |
| ) | |
| if style_key_data["metadatas"]: | |
| style_desc = style_key_data["metadatas"][0].get("description") | |
| if style_desc: | |
| style_requests.append(f"선택 스타일 키워드 설명: '{style_desc}'") | |
| if img is not None: | |
| style_requests.append("이미지 기반 스타일 검색 포함") | |
| if style_requests: | |
| user_input_str = ", ".join(style_requests) | |
| # 브랜드 라벨도 추가해서 특정 무드의 브랜드들을 선호하는 소비자의 성향 반영 필 | |
| selection_persona_prompt = PromptTemplate.from_template(""" | |
| 당신은 전문 패션 MD입니다. | |
| 사용자 프로필과 스타일 유사도 데이터를 바탕으로 선정된 후보 목록 20개를 분석하여, 가장 적합한 **제품 5개를 선정**하고 **패션 페르소나**를 정의하세요. | |
| [사용자 프로필] | |
| - 성별/체형: {g} / {h}cm / {w}kg | |
| - 선호 스타일 요청: "{s}" | |
| [후보 제품 목록 (유사도 순)] | |
| {c_list} | |
| [사고 과정 (Chain of thought)] | |
| 1. 사용자 분석: 사용자 프로필 정보를 고려하여 사용자의 성별/체형에서 사용자의 선호 스타일을 표현하기 위해서는 어떤 실루엣, 소재, 디자인이 가장 적합한지를 기준을 세우세요. | |
| 2. 추천 제품 선정: 위에서 세운 기준에 근거하여 사용자에게 가장 적합한 제품 5개를 선정하세요. (단순 유사도 순이 아닌, 스타일 적합성 우선) | |
| 3. 페르소나 도출: 선별된 5개 제품과 사용자의 특징을 종합하여, 이 사용자가 추구하는 패션 아이덴티티(페르소나)를 구체적인 코디 예시를 들어 묘사합니다. | |
| [출력 형식 및 제약 사] | |
| 1. 사고 과정(설명)은 출력하지 말고, 오직 **결과 데이터**만 아래 형식으로 출력하세요. | |
| 2. 제품번호는 후보 목록에 표시된 **순번(숫자)**만 적으세요. | |
| SELECTED_IDS = [순번1, 순번2, 순번3, 순번4, 순번5] | |
| ||| | |
| PERSONA = (여기에 3줄 내외의 페르소나 정의 작성) | |
| """) | |
| print("LLM이 제품을 선별 및 페르소나 생성 중...") | |
| selection_response = llm.invoke(selection_persona_prompt.format( | |
| g=user_info['gender'], h=user_info['height'], w=user_info['weight'], | |
| s=user_input_str, c_list=candidate_list_text | |
| )).content | |
| try: | |
| parts = selection_response.split("|||") | |
| ids_text = parts[0] | |
| selected_idx = re.findall(r'\d+', ids_text) | |
| selected_idx = [idx for idx in selected_idx if idx in candidate_map] | |
| persona_text = parts[1].replace("PERSONA =", "").strip() | |
| except Exception as e: | |
| print(f"파싱 오류: {e}") | |
| if not selected_idx: | |
| selected_idx = list(candidate_map.keys())[:5] | |
| print("LLM 선택 응답:", selection_response) | |
| print("candidate_map keys:", list(candidate_map.keys())) | |
| print("selected_idx after filtering:", selected_idx) | |
| user_info['p'] = persona_text | |
| final_recom_item = [] | |
| p_map = {} | |
| for idx_str in selected_idx[:5]: | |
| item_data = candidate_map[idx_str] | |
| img_url = get_img_url(item_data['id']) | |
| label = f"[{item_data['brand']}]_{item_data['name']}] - {item_data['price']}원" | |
| final_recom_item.append((img_url, label)) | |
| final_recom_item_gr = [[url if url else "", str(label)] for url, label in final_recom_item] | |
| # Gradio Gallery, Image에 데이터를 넣을 때는 list 형태로 들어가야 됨 | |
| p_map[label] = item_data | |
| print("Gallery 데이터:", final_recom_item_gr) | |
| return final_recom_item_gr, user_info['p'], p_map, user_info | |
| def get_sim_products(target_id, category, brand_label=None, limit=5): | |
| #======================================================================================== | |
| # 추천 2단계: 사이즈 추천 로직 | |
| # (추천 제품과 유사도 높은 제품 구하는 로직: 추천 제품의 리뷰가 임계치 이하일 경우 로직 작동) | |
| #======================================================================================== | |
| target_data = product_collection.get( | |
| ids=[target_id], include=['embeddings', 'metadatas']) | |
| embs = target_data.get('embeddings') | |
| if embs is None or len(embs) == 0: | |
| return {} | |
| target_img = torch.tensor(embs[0], dtype=torch.float32) | |
| try: | |
| meta_size = target_data['metadatas'][0].get('size_vec', '[]') #오 뭐야 왜 size_vec이 없냐 | |
| target_size = torch.tensor(ast.literal_eval(meta_size), dtype=torch.float32) | |
| except: | |
| return {} | |
| review_ids_list = list(review_ids) | |
| filter = {"$and": [{"sub_category": {"$eq": category}}, | |
| {"primary_id": {"$in": review_ids_list}}, # primary_key save_type 문제 flaot타입으로 저장되어서 맞는게 없음 | |
| {"review_count": { "$gt": 0 }}]} | |
| if brand_label is not None: | |
| filter["$and"].append({"brand_label": {"$eq": brand_label}}) | |
| candidates = product_collection.get( | |
| where=filter, include=['embeddings', 'metadatas']) | |
| cand_ids = candidates.get('ids') | |
| if cand_ids is None or len(cand_ids) == 0: | |
| return {} | |
| valid_idx, candi_img, candi_size = [], [], [] | |
| for i, p_id in enumerate(cand_ids): | |
| try: | |
| v_str = candidates['metadatas'][i].get('size_vec') | |
| if v_str: | |
| v = ast.literal_eval(v_str) | |
| if v: | |
| candi_size.append(v) | |
| candi_img.append(candidates['embeddings'][i]) | |
| valid_idx.append(i) | |
| except: | |
| continue | |
| if not valid_idx: | |
| return {} | |
| cand_img_vec = torch.tensor(np.array(candi_img), dtype=torch.float32) | |
| cand_size_vec = torch.tensor(np.array(candi_size), dtype=torch.float32) | |
| scores = (util.cos_sim(target_img, cand_img_vec)[0] * 0.6) + \ | |
| (util.cos_sim(target_size, cand_size_vec)[0] * 0.4) | |
| top_k = torch.argsort(scores, descending=True)[:limit].tolist() | |
| result_map = {} | |
| for i in top_k: | |
| real_idx = valid_idx[i] | |
| p_id = cand_ids[real_idx] | |
| score = float(scores[i]) | |
| result_map[p_id] = score | |
| print(result_map) | |
| return result_map | |
| def recom_size(label, p_map, user_info): | |
| ''' | |
| 추천 2단계: 사이즈 추천 로직 | |
| ''' | |
| # 유저가 상품 선택 후 선택 상품의 한해 사이즈 추천 진행 | |
| if not label: | |
| return "상품을 먼저 선택해주세요" | |
| if label not in p_map: | |
| return f"오류: 선택한 제품 '{label}'의 데이터를 찾을 수 없습니다.\n(검색을 다시 하거나 다른 제품을 선택해보세요)", "" | |
| if user_info is None: | |
| user_info = {} | |
| else: | |
| user_info = dict(user_info) | |
| target_id, cat, bl = p_map[label]['id'], p_map[label]['cat'], p_map[label]['brand_lb'] | |
| # 유저 페르소나의 리뷰와 유사도 계산 | |
| hyde_prompt = PromptTemplate.from_template(""" | |
| 당신은 지금부터 아래 프로필을 가진 사용자가 되어, 방금 구매한 옷에 대해 **만족한 후기**를 작성해야 합니다. | |
| [사용자 프로필] | |
| - 성별: {g} | |
| - 신체 스펙: {h}cm / {w}kg | |
| - 사용자 페르소나: {p} | |
| [구매한 상품 정보] | |
| - 제품 기본 정보: {p_info} | |
| - 카테고리: {cat} (예: 후드티, 와이드 슬랙스) | |
| [작성 지침] | |
| 1. 당신의 신체 스펙과 선호 스타일을 고려하여 사용자가 제품 핏에 만족한다는 가정 하에 리뷰를 작성하세요. | |
| 2. **주의:** S, M, L 같은 **구체적인 사이즈(옵션명)는 절대 언급하지 마세요.** 오직 '핏감(Fit)'과 '착용감'에 대해서만 이야기하세요. | |
| 3. 리뷰는 구매한 상품 정보에 기반하여 작성하세요. | |
| 4. 말투는 자연스러운 한국어 구어체(리뷰톤)를 사용하며 3줄 이내로 작성하세요. | |
| """) | |
| hyde_txt = llm.invoke(hyde_prompt.format( | |
| h=user_info.get('height',''), w=user_info.get('weight',''), | |
| g=user_info['gender'], p=user_info.get('p',''), | |
| p_info=label, cat=cat)).content | |
| hyde_vec = review_emb_func.embed_query(hyde_txt) | |
| print(hyde_txt) | |
| product_score_map = {str(target_id): 1.0} # 상품 유사도 저장 (해당 추천 상품은 유사도 1) | |
| search_pool_ids = [str(target_id)] | |
| method_log = f"해당 추천 상품" | |
| sim_ids_2 = get_sim_products(target_id, cat, brand_label=bl, limit=5) | |
| if sim_ids_2: | |
| product_score_map.update(sim_ids_2) | |
| search_pool_ids.extend(sim_ids_2.keys()) | |
| method_log += f" + 같은 브랜드 그룹 내 유사도 높은 상품 {len(sim_ids_2)}개" | |
| if len(search_pool_ids) < 3: | |
| sim_ids_3 = get_sim_products(target_id, cat, brand_label=None, limit=5) | |
| # 중복 제거 | |
| for id, score in sim_ids_3.items(): | |
| if id not in product_score_map: | |
| product_score_map[id] = score | |
| search_pool_ids.extend(sim_ids_2.keys()) | |
| method_log = f" + 같은 카테고리 내 유사도 높은 상품 {len(sim_ids_3)}개" | |
| gender = user_info.get('gender', 'male') | |
| height = int(user_info.get('height')) | |
| weight = int(user_info.get('weight')) | |
| min_h, max_h = height - 3, height + 3 | |
| min_w, max_w = weight - 3, weight + 3 | |
| final_docs_raw = review_db.similarity_search_by_vector( | |
| hyde_vec, | |
| k=30, | |
| filter={ | |
| "$and": [ | |
| {"product_id": {"$in": search_pool_ids}}, # 리뷰 데이터 primary_id vectorDB 저장 시 float --> int로 타입 변환 후 str 씌우기!!!!!!!! | |
| {"gender": {"$eq": gender}}, | |
| {"height": {"$gte": min_h}}, | |
| {"height": {"$lte": max_h}}, | |
| {"weight": {"$gte": min_w}}, | |
| {"weight": {"$lte": max_w}} | |
| ] | |
| }) | |
| if not final_docs_raw: | |
| relaxed_filter = { | |
| "$and": [ | |
| {"product_id": {"$in": search_pool_ids}}, | |
| {"gender": {"$eq": gender}} | |
| ] | |
| } | |
| final_docs_raw = review_db.similarity_search_by_vector(hyde_vec, k=30, filter=relaxed_filter) # 리스트 타입 반환 [doc1, doc2, doc3...] | |
| if not final_docs_raw: | |
| return "유사 리뷰 데이터 부족", method_log | |
| print(final_docs_raw) | |
| best_doc, max_score, weight_docs = None, -1e9 , [] | |
| size_score = defaultdict(float) | |
| size_count = defaultdict(int) | |
| valid_doc_cnt = 0 | |
| for doc in final_docs_raw: | |
| meta = getattr(doc, 'metadata', {}) or {} | |
| size = meta.get('size') | |
| if size: | |
| size_count[size] += 1 | |
| valid_doc_cnt += 1 | |
| else: | |
| continue | |
| if valid_doc_cnt == 0: | |
| return "best_doc is not found", method_log | |
| for doc in final_docs_raw: | |
| meta = getattr(doc, 'metadata', {}) or {} | |
| prod_id = meta.get('product_id') | |
| size = meta.get('size') | |
| #print("DOC META:", meta) # 디버깅용 코드 | |
| #print("prod_id:", prod_id, 'size:', size) | |
| if not prod_id or not size: | |
| continue | |
| dis = meta.get("score", 0.0) # 상품 유사도 | |
| review_sim = max(0.0, 1.0 - dis) # 리뷰 유사도 | |
| s_cnt = size_count[size] | |
| size_ratio_score =(1 + s_cnt / valid_doc_cnt) | |
| prod_id = str(prod_id) | |
| prod_sim = float(product_score_map.get(prod_id, 0.5)) # 상품 유사 | |
| w_score = prod_sim * review_sim * size_ratio_score # 가중 점수에 (추천 리뷰 사이즈 개수 / 전체 사이즈 개수)까지 곱해주기 | |
| weight_docs.append({ | |
| "doc": doc, | |
| "id": prod_id, | |
| "weight": w_score, | |
| "size": size | |
| }) | |
| size_score[size] += w_score | |
| if w_score > max_score: | |
| max_score = w_score | |
| best_doc = { | |
| "product_id": prod_id, | |
| "size": size | |
| } | |
| if not best_doc: | |
| return "best_doc is not found", method_log | |
| ref_p, ref_sz = best_doc['product_id'], best_doc['size'] | |
| ideal_spec = set_fin_size(ref_p, ref_sz) # 가장 유사한 제품 사이즈 정보 가져오기 | |
| final_sz, info = ref_sz, "(유사 제품 사이즈 추천)" | |
| # 추천 상품 사이즈 옵션 개수 확인 | |
| target_r = prod_size_df[prod_size_df['primary_id'] == str(target_id)] | |
| size_cnt = len(target_r) | |
| if size_cnt == 1: | |
| final_sz = target_r.iloc[0]['size_cleaned'] | |
| info = "(원사이즈 제품)" | |
| elif ideal_spec: | |
| actual_sz = find_closest_size(target_id, ideal_spec) | |
| if actual_sz: | |
| final_sz = actual_sz | |
| info = f"(유사 제품 {ref_p} [{ref_sz}] 실측 기반 매핑)" | |
| weight_docs.sort(key=lambda x: x['weight'], reverse=True) | |
| evidence_txt = '\n'.join([ | |
| f"-[{doc['size']}] {doc['doc'].page_content[:80]}...({doc['weight']:.2f})" | |
| for doc in weight_docs[:5] | |
| ]) | |
| detail_log = "\n".join([ | |
| f"[{i+1}위] ID: {d['id']} ({d['size']}) --> {d['weight']:.2f}점" | |
| for i, d in enumerate(weight_docs[:10]) | |
| ]) | |
| score_dist = ", ".join([ | |
| f"{k}: {v:.1f}" for k, v in sorted(size_score.items(), key=lambda x:x[1], reverse=True) | |
| ]) | |
| # 추천 근거 생성 LLM (상세 사이즈 정보 추가) | |
| prompt = "패션 상품 구매를 고려하는 사용자{p}에게 {s} 사이즈만 추천한다. 다른 사이즈는 절대 언급하지 않는다.근거 {e}는 리뷰 데이터이며, 해당 리뷰 속에서 확인된 실제 착용감·핏·실루엣 관련 내용을 중심으로 간단하고 핵심적으로 요약하여 추천 근거를 생성한다. 비교 표현, 다른 사이즈 언급, 과도한 확장 설명은 사용하지 않는다. 근거:\n{e}\n페르소나: {p}" | |
| msg = llm.invoke(prompt.format( | |
| s=final_sz, e=evidence_txt, p=user_info.get('p',''))).content | |
| return final_sz, msg | |
| # 이미지 url 가져오기 | |
| def get_img_url(id): | |
| if prod_data.empty: | |
| return None | |
| try: | |
| r = prod_data[prod_data['primary_id'] == str(id).strip()] | |
| if not r.empty: | |
| url = r.iloc[0]['thumbnail_img_url'] | |
| if pd.notna(url): | |
| return str(url) | |
| except: | |
| pass | |
| return None | |
| # 검색 벡터 생성 | |
| def get_search_vector(selected_key, user_txt, img_path): | |
| vector_list = [] # input_prompt 요소들의 embedding_vector 리스트 | |
| message_list = [] | |
| if img_path: | |
| image = Image.open(img_path) | |
| inputs = clip_processor(images=image, return_tensors="pt", padding=True).to(device) | |
| with torch.no_grad(): | |
| img_vec = clip_model.get_image_features(**inputs).cpu().numpy().tolist()[0] | |
| vector_list.append(img_vec) | |
| message_list.append("이미지 기반 검색") | |
| if user_txt: | |
| inputs = clip_processor(text=[user_txt], return_tensors="pt", padding=True).to(device) | |
| with torch.no_grad(): | |
| txt_vec = clip_model.get_text_features(**inputs).cpu().numpy().tolist()[0] | |
| vector_list.append(txt_vec) | |
| message_list.append(f"{user_txt}에 대한 검색 중...") | |
| if selected_key and selected_key != '선택 안 함': | |
| items = style_collection.get(include=["metadatas", "embeddings"]) | |
| query_vec = None | |
| for m, e in zip(items["metadatas"], items["embeddings"]): | |
| if m.get("display_name") == selected_key: | |
| vector_list.append(e) | |
| message_list.append(f"{selected_key}에 대한 검색 중...") | |
| break | |
| if not vector_list: | |
| return None, "검색 조건을 입력하세요" | |
| else: | |
| mean_vector = torch.mean(torch.tensor(vector_list), dim=0).tolist() # 멀티모달 사용 시 벡터 concat 후 512차원으로 projection | |
| return mean_vector, ", ".join(message_list) | |
| def on_gallery_click(p_map, evt: gr.SelectData): | |
| """ | |
| 갤러리 클릭 시 해당 인덱스의 제품 라벨(Key)을 찾아 반환하는 함수 | |
| """ | |
| if not p_map: | |
| return "" | |
| keys_list = list(p_map.keys()) | |
| if evt.index < len(keys_list): | |
| selected_label = keys_list[evt.index] | |
| return selected_label # 텍스트 박스에 들어갈 라벨 반환 | |
| return "" | |
| # 브랜드 label 선택 시 category range 변경 | |
| def update_cat(brand_label): | |
| if not brand_label: | |
| return gr.Dropdown(choices=[], value=None) | |
| try: | |
| filtered_df = prod_data[prod_data['brand_label'] == brand_label] | |
| new_choices = sorted(filtered_df['category_final'].dropna().unique().tolist()) | |
| new_value = new_choices[0] if new_choices else None | |
| return gr.Dropdown(choices=new_choices, value=new_value, label="카테고리 (서브)") | |
| except Exception as e: | |
| print(f" 카테고리 업데이트 실패: {e}") | |
| return gr.Dropdown(choices=[], value=None) | |
| # 카테고리 선택 시 가격 range 변경 함수 | |
| def update_price(brand_label, cat): | |
| min_price = int(prod_data['sale_price'].dropna().min()) | |
| max_price = int(prod_data['sale_price'].dropna().max()) | |
| if not brand_label or not cat: | |
| return (f"가격범위: {min_price:,}원 ~ {max_price:,}원", | |
| min_price, max_price, | |
| gr.update(visible=False) | |
| ) | |
| try: | |
| selected_prod_df = prod_data[(prod_data['brand_label'] == brand_label) & (prod_data['category_final'] == cat)] | |
| update_min_price = int(selected_prod_df['sale_price'].min()) | |
| update_max_price = int(selected_prod_df['sale_price'].max()) | |
| price_text_info = f"가격범위: {update_min_price:,}원 ~ {update_max_price:,}원" | |
| return price_text_info, update_min_price, update_max_price, gr.update(visible=False) | |
| except Exception as e: | |
| print(f"가격 업데이트 실패: {e}") | |
| return min_price, max_price | |
| brand_description = { | |
| "미니멀리즘, 높은 퀄리티, 편안함, 타임리스, 웨어러블": | |
| "미니멀하고 타임리스한 디자인을 통해 높은 품질의 편안한 의류를 제공하며, 실용성과 웨어러블함을 중요시합니다.", | |
| "편안함, 스타일, 개성, 실용성, 자유로움": | |
| "이 클러스터의 브랜드들은 편안한 실루엣과 실용적인 디자인을 통해 개인의 개성을 표현하며, 일상에서 자유로운 스타일을 추구합니다.", | |
| "편안함, 일상, 유연함, 트렌디, 개인성": | |
| "일상 속에서 편안하게 착용할 수 있는 트렌디한 디자인을 추구하며, 개인의 개성과 다양한 라이프스타일을 반영하는 브랜드들이 모여 있습니다.", | |
| "편안함, 유니크함, 지속 가능성, 반항적 디자인, 일상성과 특별함": | |
| "이 클러스터의 브랜드들은 일상 속에서 편안함과 유니크함을 추구하며, 반항적이고 독창적인 디자인으로 새로운 흐름을 제안하는 동시에 지속 가능한 삶을 지향합니다.", | |
| "편안함, 개성, 자연스러움, 균형, 다채로움": | |
| "이 클러스터의 브랜드들은 편안함과 개성을 중시하며, 자연스러운 스타일과 균형 잡힌 라이프스타일을 통해 다양성을 존중하는 독창적인 패션을 제안합니다.", | |
| "아웃도어, 기능성, 일상, 스포츠, 혁신": | |
| "스포츠와 아웃도어 활동을 일상 속에서 즐길 수 있도록 기능성과 스타일을 겸비한 다양한 제품을 제공하는 브랜드들이 모인 클러스터입니다.", | |
| "여성, 클래식, 트렌디, 웨어러블, 심플, 감성": | |
| "이 클러스터의 브랜드들은 클래식함과 트렌디함을 조화롭게 담아내며, 편안한 착용감을 제공하는 웨어러블한 디자인과 감성적인 스타일을 강조합니다." | |
| } | |
| def load_stye_collection(): | |
| data = style_collection.get(include=["metadatas"], limit=9999) | |
| desc_dict = {} | |
| for meta in data['metadatas']: | |
| style_key = meta.get('display_name') | |
| style_desc = meta.get('description') | |
| if style_key and style_desc and style_key not in desc_dict: | |
| desc_dict[style_key] = style_desc | |
| return desc_dict | |
| style_description = load_stye_collection() | |
| brand_label_choice, subcat_choice = [], [] | |
| min_price, max_price = 0, 100 | |
| if not prod_data.empty: | |
| brand_label_choice = sorted(prod_data['brand_label'].dropna().unique().tolist()) | |
| subcat_choice = sorted(prod_data['category_final'].dropna().unique().tolist()) | |
| min_price = int(prod_data['sale_price'].dropna().min()) | |
| max_price = int(prod_data['sale_price'].dropna().max()) | |
| # 데이터에 맞는 변수명으로 변경 작업 | |
| def get_brand_desc(brand_label): | |
| return brand_description.get(brand_label, "") | |
| def get_style_desc(style): | |
| return style_description.get(style, "") | |
| # [1] 텍스트를 예쁜 HTML 박스로 감싸주는 함수 | |
| def format_info_box(text): | |
| if not text: | |
| return "" | |
| # 여기에 style을 직접 적으면 절대 무시당하지 않습니다. | |
| return f""" | |
| <div style=" | |
| background-color: #f0f9ff; | |
| color: #333333; | |
| border: 1px solid #bae6fd; | |
| border-radius: 8px; | |
| padding: 12px; | |
| text-align: center; | |
| font-weight: 600; | |
| font-size: 0.95em; | |
| margin-top: 5px; | |
| "> | |
| {text} | |
| </div> | |
| """ | |
| with gr.Blocks() as app: | |
| state_pmap = gr.State({}) # 제품 메타데이터 저장 | |
| state_results = gr.State({}) | |
| with gr.Group(visible=True) as page_input: | |
| gr.Markdown("### 1. 스타일 조건을 입력해주세요") | |
| with gr.Column(): | |
| in_g = gr.Radio(["male", "female"], value="male", label="성별") | |
| check = gr.Checkbox(label="다른 성별의 제품도 볼래요", value=False) | |
| with gr.Row(): | |
| in_h, in_w = gr.Number(175, label="키"), gr.Number(70, label="몸무게") | |
| with gr.Row(): | |
| with gr.Column(): | |
| in_bg = gr.Dropdown(brand_label_choice, value=brand_label_choice[0] if brand_label_choice else None, | |
| label="브랜드 그룹", info="선택 시 브랜드 설명이 아래에 나타납니다.") | |
| initial_bg_text = get_brand_desc(brand_label_choice[0] if brand_label_choice else "") | |
| bg_desc = gr.HTML(value=format_info_box(initial_bg_text), visible=True) | |
| in_cat = gr.Dropdown(subcat_choice, value=subcat_choice[0] if subcat_choice else None, label="카테고리") | |
| price_info = gr.Markdown(value=f"가격범위: {min_price:,}원 ~ {max_price:,}원") | |
| with gr.Row(): | |
| min_num = gr.Number(value=min_price, label="최소 가격 입력") | |
| max_num = gr.Number(value=max_price, label="최대 가격 입력") | |
| price_error = gr.Markdown("", visible=False) | |
| with gr.Column(): | |
| in_sel = gr.Dropdown(choices = style_key_options, value="선택 안 함", | |
| label="스타일", info="원하는 스타일 무드를 선택해보세요.") | |
| style_desc = gr.HTML(value=format_info_box(""), visible=True) | |
| in_txt = gr.Textbox(label="텍스트 (패션 추구미를 직접 입력해보세요)") | |
| in_img = gr.Image(type="filepath", label="이미지를 넣으면 더 정밀한 추천을 받을 수 있어요 (선택 사항)") | |
| btn_s = gr.Button("🔍 검색", variant='primary', size='lg') | |
| # 검색 결과 페이지 | |
| with gr.Group(visible=True) as page_result: | |
| gr.Markdown("### 2. 추천 결과") | |
| out_p = gr.Textbox(label="나만의 패션 페르소나", lines=3) | |
| with gr.Row(): | |
| with gr.Column(scale=1): | |
| gal = gr.Gallery(label="검색 결과", columns=3) | |
| with gr.Column(scale=2): | |
| sel = gr.Textbox(label="사이즈 추천을 원하는 상품") | |
| btn_r = gr.Button("사이즈 추천") | |
| f_sz = gr.Textbox(label="추천 사이즈", lines=1) | |
| out_m = gr.Textbox(label="추천 이유 설명", lines=5, max_lines=10) | |
| with gr.Row(): | |
| btn_back = gr.Button("검색 조건 다시 입력하기", variant='secondary') | |
| def validate_price(min_num, max_num): | |
| if min_num < min_price: | |
| return gr.update(value=f"최소 가격은 {min_price:,}원 이상이어야 합니다.", visible=True) | |
| if max_num > max_price: | |
| return gr.update(value=f"최대 가격은 {max_price:,}원 이하이어야 합니다.", visible=True) | |
| if min_num > max_num: | |
| return gr.update(value="최소 가격이 최대 가격보다 클 수 없습니다.", visible=True) | |
| return gr.update(value="", visible=False) | |
| min_num.change(fn=validate_price, inputs=[min_num, max_num], outputs=[price_error]) | |
| max_num.change(fn=validate_price, inputs=[min_num, max_num], outputs=[price_error]) | |
| in_bg.change(fn=lambda x: format_info_box(get_brand_desc(x)), | |
| inputs=[in_bg], | |
| outputs=[bg_desc]) | |
| in_sel.change(fn=lambda x: format_info_box(get_style_desc(x)), | |
| inputs=[in_sel], | |
| outputs=[style_desc]) | |
| in_bg.change(fn=update_cat, inputs=[in_bg], outputs=[in_cat]) | |
| in_cat.change(fn=update_price, inputs=[in_bg, in_cat], outputs=[price_info, min_num, max_num, price_error]) | |
| # 페이지 전환 함수 | |
| def move_to_result(): | |
| return gr.update(visible=False), gr.update(visible=True) | |
| def move_to_input(): | |
| return gr.update(visible=True), gr.update(visible=False) | |
| def search_and_log(bg, c, sl, tx, im, g, h, w, min_p, max_p, check): | |
| try: | |
| gr.Info("🚀 사용자 데이터 분석을 시작합니다...") | |
| if not bg or not c: | |
| raise gr.Error("브랜드그룹과 카테고리를 반드시 선택하세요") | |
| # 1. 기존 검색 로직 수행 | |
| user_info = {'gender': g, 'height': h, 'weight': w} | |
| results = search_recom_products(bg, c, sl, tx, im, user_info, min_p, max_p, check) | |
| gr.Info("✅ 분석 완료! 스크롤을 내려 결과를 확인하세요") | |
| persona_result = results[1] | |
| try: | |
| # 2. DB에 입력값과 결과 저장 | |
| log_search_to_sheet(bg, c, min_p, max_p, sl, tx, check, persona_result) | |
| except Exception as log_e: | |
| print(f"로그 저장 실패: {log_e}") | |
| return results | |
| except gr.Error as ge: | |
| raise ge | |
| except Exception as e: | |
| print(f"시스템 에러: {e}") | |
| raise gr.Error(f"시스템 오류가 발생했습니다: {str(e)}") | |
| def size_recom_and_log(sel_product, p_map, u_info): | |
| # 1. 기존 사이즈 추천 로직 수행 | |
| # recom_size 함수는 2개의 값을 반환한다고 가정 (f_sz, out_m) | |
| gr.Info("사용자님만의 스타일을 반영한 최적 사이즈를 분석 중입니다.\n 잠시만 기다려주세요~!") | |
| rec_size, reason = recom_size(sel_product, p_map, u_info) | |
| gr.Info("✅ 분석 완료! 결과를 확인하세요") | |
| # 2. DB에 저장 | |
| log_size_to_sheet(sel_product, u_info, rec_size, reason) | |
| return rec_size, reason | |
| # 1. 검색 버튼 연결 | |
| btn_s.click( | |
| fn=search_and_log, | |
| inputs=[in_bg,in_cat,in_sel,in_txt,in_img,in_g,in_h,in_w, min_num, max_num, check], | |
| outputs=[gal, out_p, state_pmap, state_results] | |
| ).then( | |
| fn=move_to_result, | |
| inputs=None, | |
| outputs=[page_input, page_result] | |
| ) | |
| btn_back.click( # 버튼 클릭 시 페이지 이동 | |
| fn=move_to_input, | |
| inputs=None, | |
| outputs=[page_input, page_result] | |
| ) | |
| gal.select(fn=on_gallery_click, inputs=[state_pmap], outputs=sel) | |
| # 3. 사이즈 추천 버튼 연결 | |
| # inputs에 state_persona를 넣어서 recommend_size로 전달 | |
| btn_r.click( | |
| fn=size_recom_and_log, | |
| inputs=[sel, state_pmap, state_results], # -> state_persona 값을 꺼내서 씀 | |
| outputs=[f_sz, out_m] | |
| ) | |
| try: | |
| app.close() | |
| except: | |
| pass | |
| app.queue().launch(theme=gr.themes.Soft(), share=True) |