source/text_cleaner.py

"""
    text_cleaner.py

    모듈화된 텍스트 정제기

    함수:
        - clean_text: 단일 문자열 정제 -> str 또는 None 반환
        - clean_dataframe: DataFrame에 clean_text 적용, 정제된 DataFrame 반환 및 실행 시간/삭제 행 수 출력
"""

import re
import emoji
import time
import pandas as pd
from typing import Optional
from soynlp.normalizer import repeat_normalize

# 허용 문자 패턴
_EMOJI_PATTERN = emoji.get_emoji_regexp()
_CLEAN_PATTERN = re.compile(
    rf"[^ .,?!/@\$%~％·∼()\x00-\x7Fㄱ-ㅣ가-힣{_EMOJI_PATTERN}]+"
)
_URL_PATTERN = re.compile(r"https?://\S+")


def clean_text(
    text: str,
    min_length: int = 3,
    num_repeats: int = 2
) -> Optional[str]:
    """
    텍스트 정제:
    - '레시피', '만들기' 키워드, URL, 한글 없는 댓글 None 반환
    - 허용 문자만 남김
    - 이모지 제거
    - 공백 strip
    - 반복문자 정규화
    - min_length 이하이면 None 반환
    이 모든 필터 통과시 정제된 문자열 반환한다.
    """
    # 삭제 조건
    if any(kw in text for kw in ("레시피", "만들기")):
        return None
    if _URL_PATTERN.search(text):
        return None
    if not re.search(r"[가-힣]", text):
        return None

    # 정제 로직
    cleaned = _CLEAN_PATTERN.sub(" ", text)
    cleaned = _EMOJI_PATTERN.sub("", cleaned)
    cleaned = cleaned.strip()
    cleaned = repeat_normalize(cleaned, num_repeats=num_repeats)
    if len(cleaned) < min_length:
        return None
    return cleaned


def clean_dataframe(
    df: pd.DataFrame,
    text_col: str = "comment",
    min_length: int = 3,
    num_repeats: int = 2
) -> pd.DataFrame:
    """
    DataFrame에 clean_text 적용:
    - original 행 수, 삭제된 행 수, 처리 시간 출력
    - 'cleaned' 컬럼에 정제된 텍스트 저장
    - None인 행은 삭제
    """
    start = time.perf_counter()
    total = len(df)
    df['cleaned'] = df[text_col].apply(
        lambda x: clean_text(x, min_length, num_repeats)
    )
    df_clean = df.dropna(subset=['cleaned']).reset_index(drop=True)
    dropped = total - len(df_clean)
    elapsed = time.perf_counter() - start
    print(f"[CLEAN] 처리 시간: {elapsed:.2f}s | 총 행: {total} | 삭제된 행: {dropped}")
    return df_clean

#CLI 지원
if __name__ == "__main__":
    import argparse

    parser = argparse.ArgumentParser(
        description="텍스트 정제 모듈"
    )
    parser.add_argument(
        "--input", "-i",
        required=True,
        help="입력 CSV 파일 경로 (comment 컬럼 포함)"
    )
    parser.add_argument(
        "--output", "-o",
        required=True,
        help="출력 CSV 파일 경로"
    )
    parser.add_argument(
        "--min_length", type=int, default=3,
        help="최소 길이 (default=3)"
    )
    parser.add_argument(
        "--num_repeats", type=int, default=2,
        help="반복문자 정규화 허용 횟수 (default=2)"
    )
    args = parser.parse_args()

    df = pd.read_csv(args.input, encoding="utf-8-sig")
    df_clean = clean_dataframe(
        df,
        text_col="comment",
        min_length=args.min_length,
        num_repeats=args.num_repeats
    )
    df_clean.to_csv(args.output, index=False, encoding="utf-8-sig")
    print(f"정제된 데이터 저장: {args.output}")