주말동안 각자 주제 아이디어를 생각해오기로 했음.
나는 분석을 위한 프로젝트가 아니라, 기업이 필요로 하는 프로젝트여야 더 경쟁력이 있다고 판단했고, 각 패션 플랫폼들에 대해서 조사를 하기 시작했다.
https://www.wiseapp.co.kr/insight/detail/523?
패션 전문몰 앱 사용자 수 1위 에이블리, 결제추정금액 1위 무신사
패션 전문몰 앱 '에이블리vs무신사vs지그재그'의 2024년 현황이 궁금하다면? 와이즈앱 인사이트에서 확인하세요!
www.wiseapp.co.kr
https://www.businesspost.co.kr/BP?command=article_view&num=435487
에이블리코퍼레이션 작년 매출 3697억 역대 최대, 4910과 아무드 모두 성장률 높아
에이블리코퍼레이션 작년 매출 3697억 역대 최대, 4910과 아무드 모두 성장률 높아
www.businesspost.co.kr
위의 인사이트를 통해서 에이블리 타 플랫폼과 비교했을 때 높은 유저 수 대비 결제 추정 금액이 낮은 것을 알 수 있었고, 아래 기사를 통해 매출이 많이 상승했지만 여전히 적자임을 알게되었다. 이를 해결하기 위해 고마진 상품(뷰티/라이프)을 패션과 함께 추천해주는 크로스셀링 모델을 구축하는 아이디어를 제시했다.
- 해당 아이디어를 들은 팀원들의 의견
- 1. 의류와 뷰티 크로스셀링 전략이 먹힐 것인가?
- 2. 젊은 층이 많이 사용하는 저가 플랫폼인 에이블리가 객단가를 높이는 것이 가능할까?
- 해당 아이디어를 들은 튜터님의 의견
- 1. 아이디어는 매우 좋지만 크롤링이 가능할지 확인해보아야 한다.
- 2. 어떤 인사이트나 목적이 나올지, 분석 모델과 같은 구체성이 잘 그려지지 않는다.
라는 의견을 들었고, 가장 현실성있는 제안을 하신 팀원의 의견을 중심으로 나와 다른 팀원들의 아이디어를 덧붙여가는 식으로 방향을 잡았다.
가장 현실성있는 제안을 하신 팀원의 의견은 리뷰가 구매 전환에 미치는 영향을 분석하는 것이다.
여기서 우리는 가장 정보가 많고 크롤링이 용이한 무신사 플랫폼을 단독으로 분석하기로 결정을 했다.
분석 목적은 리뷰관리를 통해 수익(판매량)을 높이는 것이고 분석의 가닥은 크게 2가지로 잡았다.
1. 단순히 리뷰수가 많은 상품이 잘 팔리는 것이 아니라 어떤 리뷰의 상품이 실제 구매에 영향을 미치는지 분석한다.
2. 부정 리뷰 텍스트를 확인하여 불만 키워드를 관리한다.
(우선은 이렇고 프로젝트를 진행하면서 더 키워갈 예정)
하지만 이 아이디어 역시 무신사의 크롤링이 가능한지가 가장 핵심이었으며 크롤링 가능여부를 확인을 해보았다.
import requests
import pandas as pd
import time
import sys
# ==========================================
# 0. 기본 설정 및 세션 초기화
# ==========================================
HEADERS = {
"User-Agent": "My-Agent",
"Referer": "https://www.musinsa.com/",
"Origin": "https://www.musinsa.com",
}
# Session 객체를 사용하면 연결을 재사용하여 크롤링 속도가 훨씬 빨라집니다.
session = requests.Session()
session.headers.update(HEADERS)
# ==========================================
# 1. 데이터 탐색 유틸리티 함수
# ==========================================
def extract_product_columns(data):
"""JSON 구조 내에서 'PRODUCT_COLUMN' 캡슐만 모두 찾아 리스트로 반환"""
nodes = []
def _traverse(node):
if isinstance(node, dict):
if node.get("type") == "PRODUCT_COLUMN":
nodes.append(node)
for v in node.values():
if isinstance(v, (dict, list)):
_traverse(v)
elif isinstance(node, list):
for item in node:
if isinstance(item, (dict, list)):
_traverse(item)
_traverse(data)
return nodes
def get_value_by_keys(data_dict, keys_to_check, default=None):
"""여러 예상 키값 중 매칭되는 첫 번째 값을 찾아 반환"""
if isinstance(data_dict, dict):
for k in keys_to_check:
if k in data_dict and data_dict[k] is not None:
return data_dict[k]
for v in data_dict.values():
if isinstance(v, (dict, list)):
res = get_value_by_keys(v, keys_to_check)
if res is not None:
return res
elif isinstance(data_dict, list):
for item in data_dict:
if isinstance(item, (dict, list)):
res = get_value_by_keys(item, keys_to_check)
if res is not None:
return res
return default
# ==========================================
# 2. 핵심 크롤링 함수들
# ==========================================
def fetch_ranking_products(category_code, category_name):
"""특정 카테고리의 랭킹 상품 50개를 수집"""
url = "https://api.musinsa.com/api2/hm/web/v5/pans/ranking/sections/199"
params = {
"storeCode": "musinsa",
"categoryCode": category_code,
"contentsId": "",
"subPan": "product",
"gf": "A",
"ageBand": "AGE_BAND_ALL"
}
response = session.get(url, params=params)
if response.status_code != 200:
print(f"[{category_name}] API 호출 실패: {response.status_code}")
return []
raw_data = response.json()
product_nodes = extract_product_columns(raw_data)
# 중복 제거 (상품 id 기준)
unique_nodes = []
seen = set()
for node in product_nodes:
gno = str(node.get("id", ""))
if gno and gno not in seen:
seen.add(gno)
unique_nodes.append(node)
products_data = []
for item in unique_nodes[:50]: # 상위 50개만 추출
gno = str(item.get("id", ""))
brand = get_value_by_keys(item, ["brandName", "brand_name", "brand"], default="")
name = get_value_by_keys(item, ["goodsName", "goods_name", "productName", "item_name", "text"], default="이름없음")
price = get_value_by_keys(item, ["price", "salePrice", "item_price"], default=0)
sale_rate = get_value_by_keys(item, ["saleRate", "discount_rate"], default=0)
review_cnt = get_value_by_keys(item, ["reviewCount", "reviewCnt", "review_count"], default=50)
products_data.append({
"플랫폼": "무신사",
"카테고리": category_name,
"goodsNo": gno,
"브랜드": brand,
"상품명": name,
"정가": price,
"판매가": price,
"할인율(%)": sale_rate,
"리뷰수": review_cnt,
"리뷰점수": 0,
"조회수": 0,
"누적판매수": 0
})
return products_data
def fetch_product_stats(goods_no):
"""상품의 누적 판매량 및 조회수 수집"""
url = f"https://goods-detail.musinsa.com/api2/goods/{goods_no}/stat"
try:
r = session.get(url, timeout=5)
if r.status_code == 200:
data = r.json().get("data", {})
return data.get("purchaseTotal", 0), data.get("pageViewTotal", 0)
except Exception:
pass
return 0, 0
def fetch_product_reviews(goods_no, target_count, max_pages=30):
"""특정 상품의 리뷰 수집"""
reviews = []
for page in range(max_pages):
if len(reviews) >= target_count:
break
url = "https://goods.musinsa.com/api2/review/v1/view/list"
params = {
"page": page,
"pageSize": 10,
"goodsNo": goods_no,
"sort": "up_cnt_desc",
"selectedSimilarNo": goods_no,
"myFilter": "false",
"hasPhoto": "false",
"isExperience": "false",
}
try:
r = session.get(url, params=params, timeout=10)
if r.status_code != 200: break
data = r.json().get("data", {})
review_list = data.get("list", [])
if not review_list: break
for review in review_list:
profile = review.get("userProfileInfo") or {}
reviews.append({
"goodsNo": str(goods_no),
"리뷰내용": review.get("content", ""),
"평점": int(review.get("grade", 0)),
"체험단": review.get("type") == "experience",
"사이즈": review.get("goodsOption", ""),
"키": profile.get("userHeight", ""),
"몸무게": profile.get("userWeight", ""),
"성별": profile.get("reviewSex", ""),
"작성일": review.get("createDate", ""),
})
if len(reviews) >= target_count:
break
if page >= data.get("page", {}).get("totalPages", 0) - 1:
break
time.sleep(0.3)
except Exception:
break
return reviews
# ==========================================
# 3. 메인 실행 블록
# ==========================================
if __name__ == "__main__":
CATEGORIES = {"002000": "아우터"}
all_products = []
# [1] 랭킹 상품 수집
for code, name in CATEGORIES.items():
products = fetch_ranking_products(code, name)
all_products.extend(products)
print(f"[{name}] 랭킹 {len(products)}개 수집 완료")
time.sleep(0.5)
if not all_products:
print("\n:rotating_light: 상품 데이터를 찾지 못했습니다. 스크립트를 종료합니다.")
sys.exit()
df_products = pd.DataFrame(all_products)
print(f"\n:white_check_mark: 상품 총 {len(df_products)}개 수집 성공")
print("-" * 50)
# [2] 상품별 통계 및 리뷰 수집
all_reviews = []
total_items = len(df_products)
for idx, row in df_products.iterrows():
goods_no = str(row["goodsNo"])
target_count = int(row["리뷰수"]) if pd.notna(row["리뷰수"]) else 50
current = idx + 1
# 통계 수집
sales, views = fetch_product_stats(goods_no)
df_products.at[idx, '누적판매수'] = sales
df_products.at[idx, '조회수'] = views
if not goods_no or target_count == 0:
print(f"[{current}/{total_items}] [스킵] 상품번호: {goods_no}")
continue
print(f"[{current}/{total_items}] 수집중: [{row['브랜드']}] {str(row['상품명'])[:20]} (판매: {sales})")
# 리뷰 수집
reviews = fetch_product_reviews(goods_no, target_count)
print(f" → 리뷰 {len(reviews)}개 완료")
all_reviews.extend(reviews)
time.sleep(0.3)
# [3] 데이터 병합 및 저장
df_reviews = pd.DataFrame(all_reviews)
print(f"\n:white_check_mark: 리뷰 총 {len(df_reviews)}개 수집 완료!")
if not df_reviews.empty:
df_merged = df_products.merge(df_reviews, on="goodsNo", how="left")
df_merged.to_csv("musinsa_merged.csv", index=False, encoding="utf-8-sig")
df_reviews.to_csv("musinsa_reviews.csv", index=False, encoding="utf-8-sig")
df_products.to_csv("musinsa_products.csv", index=False, encoding="utf-8-sig")
print("\n:tada: 모든 데이터 병합 및 CSV 파일 저장 완료!")
else:
df_products.to_csv("musinsa_products.csv", index=False, encoding="utf-8-sig")
print("\n:warning: 수집된 리뷰가 없어 상품 데이터만 저장했습니다.")
일단 이렇게 무신사의 아우터 카테고리의 실시간 랭킹 상품 Top50에 대해서 크롤링을 해보니
이런식으로 50개 중 6개의 상품만 누적판매수 정보가 없음을 확인할 수 있었다.
물론 실시간 정보이고 꾸준히 바뀌겠지만, 상위 랭킹에 있는 상품들은 비교적 누적판매수 정보가 많이 포함이 되어있는 것을 확인할 수 있었다.
내일은, 다른 카테고리와 조회기간을 픽스해서 제대로 크롤링을 마쳐볼 계획이다.