amz_review_analyse/content清洗.py

"""
清洗合并后的评论 CSV（merged_reviews.csv 等同结构文件）。
保留元数据列，对 content 列做文本清洗与过滤。
"""

from __future__ import annotations

import argparse
import re
from pathlib import Path

import pandas as pd

_PROJECT_ROOT = Path(__file__).resolve().parent
DEFAULT_INPUT_PATH = _PROJECT_ROOT / "merged_reviews.csv"
DEFAULT_OUTPUT_PATH = _PROJECT_ROOT / "merged_reviews_cleaned.csv"

CONTENT_COLUMN = "content"
MIN_WORD_COUNT = 4

# 与 content 重复的正文列，输出时一律删除，避免一行存两遍长文本
REDUNDANT_CONTENT_COLUMNS = (
    "content_raw",
    "content_cleaned",
    "Cleaned_Content",
    "cleaned_content",
)

# 亚马逊「视频/图片无法加载」占位文案，后常跟大量空行
_MEDIA_LOAD_NOISE = re.compile(
    r"The\s+media\s+could\s+not\s+be\s+loaded\.?\s*",
    flags=re.IGNORECASE,
)


def strip_empty_lines(text) -> str:
    """去掉仅含空白/空白的行，并将剩余行合并为单行文本。"""
    if pd.isna(text):
        return ""
    text = str(text).replace("\r\n", "\n").replace("\r", "\n")
    lines = [line.strip() for line in text.split("\n")]
    lines = [line for line in lines if line]
    return " ".join(lines)


def clean_amazon_review(text) -> str:
    if pd.isna(text):
        return ""
    text = str(text).strip()

    if text.startswith('"') and text.endswith('"'):
        text = text[1:-1]

    text = strip_empty_lines(text)
    text = _MEDIA_LOAD_NOISE.sub("", text)

    text = re.sub(r"<br\s*/?>", " ", text, flags=re.IGNORECASE)
    text = text.replace("&#160;", " ").replace("&amp;", "&")

    noise_prefixes = [
        r"Why did you pick this product vs others\?:",
        r"Quality:",
        r"Update:",
    ]
    for prefix in noise_prefixes:
        text = re.sub(prefix, "", text, flags=re.IGNORECASE).strip()

    text = re.sub(r"\s+", " ", text).strip()
    return text


def drop_redundant_content_columns(df: pd.DataFrame) -> pd.DataFrame:
    """删除与 content 重复存储的正文列（如 content_raw），避免 CSV 一行两份长文本。"""
    drop_cols = [
        c for c in REDUNDANT_CONTENT_COLUMNS if c in df.columns and c != CONTENT_COLUMN
    ]
    if drop_cols:
        df = df.drop(columns=drop_cols)
    return df


def process_reviews(
    csv_file_path: str | Path,
    *,
    content_column: str = CONTENT_COLUMN,
) -> pd.DataFrame:
    """
    读取合并后的评论 CSV，清洗 content 列并过滤无效/重复行。
    输出保留原表头及元数据（_id, asin, rating, title 等）。
    """
    csv_file_path = Path(csv_file_path)
    df = pd.read_csv(csv_file_path, dtype=str, keep_default_na=False)

    if content_column not in df.columns:
        raise ValueError(
            f"缺少评论正文列 {content_column!r}，当前列: {list(df.columns)}"
        )

    print(f"1. 原始数据量: {len(df)}")

    df = drop_redundant_content_columns(df)
    df[content_column] = df[content_column].apply(clean_amazon_review)

    df = df[df[content_column] != ""]

    df = df[
        df[content_column].str.contains(
            r"[a-zA-Z0-9áéíóúñÁÉÍÓÚÑ]", regex=True, na=False
        )
    ]

    word_counts = df[content_column].apply(lambda x: len(str(x).split()))
    df = df[word_counts >= MIN_WORD_COUNT]

    df = df.drop_duplicates(subset=[content_column], keep="first")
    df = drop_redundant_content_columns(df)

    print(f"2. 清洗及过滤后数据量: {len(df)}")
    return df


def save_cleaned_reviews(df: pd.DataFrame, output_path: str | Path) -> Path:
    output_path = Path(output_path)
    output_path.parent.mkdir(parents=True, exist_ok=True)
    df.to_csv(output_path, index=False, encoding="utf-8-sig")
    print(f"3. 已写入: {output_path}")
    return output_path


def main() -> None:
    parser = argparse.ArgumentParser(description="清洗合并后的评论 CSV")
    parser.add_argument(
        "-i",
        "--input",
        type=Path,
        default=DEFAULT_INPUT_PATH,
        help=f"输入 CSV（默认: {DEFAULT_INPUT_PATH}）",
    )
    parser.add_argument(
        "-o",
        "--output",
        type=Path,
        default=DEFAULT_OUTPUT_PATH,
        help=f"输出 CSV（默认: {DEFAULT_OUTPUT_PATH}）",
    )
    args = parser.parse_args()

    df = process_reviews(args.input)
    save_cleaned_reviews(df, args.output)


if __name__ == "__main__":
    main()