2441c9119c
包含七步编排入口、结构化/向量化/聚类/词频/报告模块与 prompts 配置；忽略原始 CSV 与本地密钥。 Co-authored-by: Cursor <cursoragent@cursor.com>
153 lines
4.3 KiB
Python
153 lines
4.3 KiB
Python
"""
|
||
清洗合并后的评论 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(" ", " ").replace("&", "&")
|
||
|
||
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()
|