올리브영 K-Beauty 트렌드 스캐너

# 예시: 올리브영 카테고리 랭킹 수집 import asyncio from playwright.async_api import async_playwright import random, json CATEGORIES = [ "스킨케어", "선케어", "립메이크업", "마스크팩", "클렌징", "바디케어", ] async def scrape_ranking(category): async with async_playwright() as p: browser = await p.chromium.launch() page = await browser.new_page() await page.set_extra_http_headers({ "User-Agent": random.choice(UA_LIST) }) url = build_category_url(category) await page.goto(url, timeout=30000) await asyncio.sleep(random.uniform(3, 7)) items = await page.query_selector_all( ".product-item" ) results = [] for item in items: results.append({ "name": await item.inner_text(".name"), "price": parse_price(await item.inner_text(".price")), "rating": float(await item.inner_text(".rating")), "reviews": parse_int(await item.inner_text(".review-count")), }) await browser.close() return results

# 예시: 리뷰 형태소 분석 및 키워드 추출 from konlpy.tag import Okt from collections import Counter okt = Okt() INGREDIENT_DICT = { "나이아신아마이드", "히알루론산", "레티놀", "세라마이드", "마데카소사이드", "판테놀", } def analyze_reviews(reviews: list[str]): keywords = Counter() sentiments = {"pos": 0, "neg": 0, "neu": 0} for text in reviews: nouns = okt.nouns(text) for n in nouns: if len(n) >= 2: keywords[n] += 1 score = sentiment_score(text) if score > 0.6: sentiments["pos"] += 1 elif score < 0.4: sentiments["neg"] += 1 else: sentiments["neu"] += 1 ingredients = { k: v for k, v in keywords.items() if k in INGREDIENT_DICT } return keywords.most_common(20), ingredients, sentiments

# 예시: YouTube Data API로 언급 탐지 from googleapiclient.discovery import build yt = build("youtube", "v3", developerKey=YT_KEY) CHANNELS = [ "UCxxxxxx", # 이사배 "UCyyyyyy", # 레오제이 "UCzzzzzz", # 디렉터파이 ] def scan_mentions(channels, product_names): mentions = [] for ch_id in channels: resp = yt.search().list( channelId=ch_id, q="올리브영", publishedAfter=yesterday_rfc3339(), type="video", part="snippet", maxResults=10, ).execute() for item in resp["items"]: title = item["snippet"]["title"] matched = match_products( title, product_names ) if matched: mentions.append({ "channel": ch_id, "video": item["id"]["videoId"], "products": matched, }) return mentions

# 예시: 카테고리 기회 점수 산출 import pandas as pd def calc_opportunity(df: pd.DataFrame): """ 기회 점수 = (7일 리뷰 증가율 × 랭킹 변동폭) ÷ 상위 5개 브랜드 점유율 """ scores = [] for cat in df["category"].unique(): sub = df[df["category"] == cat] review_growth = ( sub["reviews_7d"].sum() / sub["reviews_14d"].sum() ) - 1.0 rank_volatility = sub["rank_change"].abs().mean() top5_share = ( sub.nlargest(5, "reviews_7d")["reviews_7d"].sum() / sub["reviews_7d"].sum() ) score = ( review_growth * rank_volatility ) / max(top5_share, 0.01) scores.append({ "category": cat, "opportunity_score": round(score, 3), "review_growth": f"{review_growth:.1%}", "top5_concentration": f"{top5_share:.1%}", }) return sorted( scores, key=lambda x: x["opportunity_score"], reverse=True )