뽐뿌 특가 데이터 전처리 하기

import pandas as pd
import numpy as np
import re

from tqdm import tqdm
from kiwipiepy import Kiwi
from datetime import datetime

df = pd.read_csv('./datas/2023-06-30 22:27:20.666568_117980개.csv')
df.head(2)

df.describe()

df.info()

def extract_sales_channel_and_price(title):
    """ 특가 게시물 제목에서 가격, 판매채널 추출
    Args:
        title - 특가 게시물 제목
    Returns:
        str : 아래의 데이터를 가진 str형식 return
            - sales_channel : 특가 판매 채널 (e.g 지마켓)
            - price : 제품/배송비 가격
    """
    # 문자열에서 [...] 혹은 (...) 형태의 구성을 찾아 추출
    pattern = r"\[([^\]]+)\]|\(([^\)]+)\)"
    
    # 가격, 판매채널 추출
    matches = re.findall(pattern, title)
    
    # 판매채널
    sales_channel = matches[0][0] or matches[0][1] if matches else "unknown"
    sales_channel = sales_channel.strip()
    
    # 가격
    price = matches[-1][0] or matches[-1][1] if matches else "unknown"  # Return the last match
    price = price.strip()
    
    return sales_channel, price

# 제목에서 판매채널과 가격 추출
df['sales_channel'], df['price'] = zip(*df['title'].map(extract_sales_channel_and_price))
df[['title', 'sales_channel', 'price']].head(2)

def split_price(price):
    """ 제품/배송비 가격에서 제품 가격과 배송비 분리
    Args:
        price - 제품/배송비 가격
    Returns:
        str : 아래의 데이터를 가진 str형식 return
            - product_price : 특가 제품 가격
            - shipping_cost : 배송비
    """
    # price에 배송비가 없는 경우도 있으므로, 있으면 제품가격과 배송비, 없으면 제품가격과 unknown으로 리턴
    if "/" in price:
        product_price, shipping_cost = price.split("/", 1)  # Split into at most 2 parts
    else:
        product_price = price
        shipping_cost = "unknown"
    return product_price, shipping_cost

# 가격에서 제품 가격과 배송비 구별
df['product_price'], df['shipping_cost'] = zip(*df['price'].map(split_price))
df[['title', 'sales_channel', 'price', 'product_price', 'shipping_cost']].head(2)

df["sales_channel"].value_counts()

def channel_merge(df, channel, change_channel):
    """ 판매 채널 명
    Args:
        df - 특가 게시물 Dataframe
        channel - 변경 전 채널 이름 
        change_channel - 변경될 채널 이름
        
    Returns:
        Dataframe : 통합 채널명으로 변경된 DataFrame
    """
    
    # 통합될 채널 이름 찾기
    temp = df[df["sales_channel"].str.contains(channel, case=False)]
    # 변경할 index 저장
    change_value_idx = temp.index
    # index를 기준으로 변경 될 채널이름으로 변경
    df.loc[change_value_idx, "sales_channel"] = change_channel
    return df

channel_dict = {"네이버":["네이버", "스마트스토어", "스토어팜", "원쁠딜"],
                "11번가":["11번가", "11st", "11마존", "쇼킹딜"],
                "신세계":["신세계", "SSG"],
                "하이마트":["하이마트"],
                "롯데":["롯데", "칠성몰"],
                "카카오":["카카오", "톡딜", "카톡", "톡스토어"],
                "티몬":["티몬", "tmon", "티켓몬스터"],
                "CJ":["CJ"],
                "그립":["그립", "grip"],
                "우체국":["우체국"],
                "쿠팡":["쿠팡", "ㅋㅍ"],
                "보고":["보고","vogo"],
                "인터파크":["인터파크"],
                "AK몰":["ak"],
                "큐텐":["큐텐", "Qo", "큐10", "Q10"],
                "Quube":["Quube"],
                "GS":["gs", "나만의 냉장고"],
                "지마켓/옥션":["지마켓", "옥션","지/옥", "쥐마켓", "g마켓", "g9", "지9", "지구", "gmarket", "지옥", "옥베이"],
                "SK":["sk"],
                "아이허브":["ih"],
                "KT":["kt"],
                "Hmall":["hm", "H패", "현대몰", "h몰"],
                "홈플러스":["홈플"],
                "NS홈쇼핑":["ns"],
                "이마트":["이마트몰"],
                "메가마트":["메가마트"],
                "오늘의집":["오늘의"],
                "전자랜드":["전자랜드"],
                "나이키":["나이키"],
                "예스24":["yes"],
                "코스트코":["코스트코"],
                "Steam":["Steam", "스팀", "Indiegala"],
                "아디다스":["아디다스"],
                "홈앤쇼핑":["홈&"],
                "삼성":["삼성"],
                "신한":["신한"],
                "크록스":["크록스"],
                "국민":["국민", "국카"],
                "다이슨":["다이슨"],
                "리복":["리복"],
                "LF스퀘어몰":["LF"],
                "K쇼핑":["K쇼핑"],
                "CGV":["CGV"],
                "배달의민족":["배민"],
                "동원몰":["동원"],
                "탑텐":["탑텐"],
                "위메프":["위메프"],
                "unknown":["종료", "끌어올림", "끌올", "무배", "다양", "공홈"]
               }

for change_channel, channels in channel_dict.items():
    for channel in channels:
        df = channel_merge(df, channel, change_channel)

df["sales_channel"].value_counts()

def convert_price_to_int(price):
    """ 제품 가격 자료형 변환
    Args:
        price - : Str 형식의 제품 가격
    Returns:
        Nan : unknown일때
        int : Int형 가격
    """
    
    # unknown은 NaN값
    if price == "unknown":
        return np.NaN
    # 그 외 가격은 "원", "," "."을 삭제한 숫자형
    else:
        cleaned_price = price.replace("원", "").replace(",", "").replace(".", "").strip()
        if cleaned_price.isdigit():
            return int(cleaned_price)
        else:
            return np.NaN

df['product_price'] = df['product_price'].map(convert_price_to_int)
df[['title', 'sales_channel', 'price', 'product_price', 'shipping_cost']].head()

# 상위 0.0014 제외
cut = df["product_price"].quantile(0.9986)
print(f"기준 가격 {cut}")
temp = df[df["product_price"] > cut]
temp.sort_values("product_price", ascending=False)[["product_price", "price"]]

top_index = temp.index
df.loc[top_index, "product_price"] = np.nan

# 배송비 확인
df["shipping_cost"].value_counts()

def convert_shipping_cost(cost):
    """ 배송비 변환
    Args:
        cost - Str 형식의 배송비
    Returns:
        Nan : unknown이거나, 그 외 Str형 일때
        0 : 무료 배송일때
        int : 그 외 숫자형 일때
    """
    cost = cost.strip()
    if cost.find("무료") > -1:
        return "0"
    elif cost.find("무배") > -1:
        return "0"
    elif cost.replace("원", "").replace(",", "").replace("~", "").replace(".", "").isdigit():
        return cost.replace("원", "").replace(",", "").replace("~", "").replace(".", "")
    else:
        return np.NaN

df['shipping_cost'] = df['shipping_cost'].map(convert_shipping_cost)
df[['title', 'sales_channel', 'price', 'product_price', 'shipping_cost']].head(2)

# 배송비 아웃라이어 확인
df["shipping_cost"] = df["shipping_cost"].fillna(-1).astype(int).replace({-1: None})

# (상위 0.002 제외)
cut = df["shipping_cost"].quantile(0.998)
print(f"기준 가격 {cut}")
temp = df[df["shipping_cost"] > cut]
temp.sort_values("shipping_cost", ascending=False)[["shipping_cost", "price"]]
top_index = temp.index
df.loc[top_index, "shipping_cost"] = np.nan

def clean_title(title):
    """ 제목 정체
    Args:
        title - Str 형식의 특가 게시물 제목
    Returns:
        title - Str 형식의 판매 채널, 제품 가격 정보, 특수 문자가 제외 된 제목
    """
    # 제목에서 판매 채널 제외
    title = re.sub(r'^\[([^\]]+)\]|\(([^\)]+)\)s*', '', title)
    # 제목에서 가격 정보 제외
    title = re.sub(r'\s*\[([^\]]+)\]|\(([^\)]+)\)$', '', title)
    
    return title

df['title'] = df['title'].astype(str)
df['real_title'] = df['title'].apply(clean_title)
df[['title', 'real_title']].head(2)

def noun_extractor(title):
    """ 명사 추출 및 불용어 처리 함수
    Args:
        title - Str 형식의 판매 채널, 제품 가격 정보, 특수 문자가 제외 된 제목
    Returns:
        results - List : title에서 지정된 불용어를 제외한 명사만 추출된 List
    """
    results = []
    try:
        result = kiwi.analyze(title)
    except:
        return results
    for token, pos, _, _ in result[0][0]:
        if len(token) != 1 and pos.startswith('N') and token not in stopwords:
                results.append(token)
    return results

# 불용어
stopwords = ["할인", "쿠폰", "상품", "무료", "스마일", "적용", "카드", "삼성", "세트", "클럽",
            "프로", "증정", "블랙", "인치", "스클", "박스", "에어", "세대", "무선", "랜드", "머니",
            "가능", "캡슐", "샤오미", "결제", "포인트", "구매", "추가", "최대", "배송", "프리미엄"]

tqdm.pandas()
kiwi = Kiwi()
df["keywords"] = df["real_title"].progress_apply(noun_extractor)

df.loc[df['pop'] == True, 'post_type'] = 'popular/hot'
df.loc[df['hot'] == True, 'post_type'] = 'popular/hot'
df['post_type'].fillna('general', inplace=True)

# 데이터 csv 저장
now = str(datetime.now())
df.to_csv(f"./datas/{now}_preprocessing.csv", index=False)