爬了 10 万条评论却不敢用？这套 Python 数据治理方案让 NLP 情感分析靠谱 10 倍

采集层(httpx + 代理IP池)  →  解析层(parsel)
   →  清洗层(Pandas 六维治理)  →  契约层(Pydantic Schema)
   →  分析层(SnowNLP 情感)  →  呈现层(Plotly 质量看板)

import asyncio
import httpx
from parsel import Selector


# ===== 亿牛云代理配置（白名单用户名密码认证）=====
PROXY_USER = "your_username"   # 亿牛云后台获取的认证用户名
PROXY_PASS = "your_password"   # 亿牛云后台获取的认证密码
PROXY_GATEWAY = "proxy.16yun.cn"  # 亿牛云代理网关地址（示例）
PROXY_PORTS = ["31000", "31001", "31002"]  # 多端口构成出口池


class YiniuProxyPool:
    """亿牛云代理池：认证、轮换、健康度评分。

    规模化采集时单一出口 IP 易被风控。本类将多个亿牛云出口
    组成资源池，按健康度加权选择，并据请求成败动态淘汰劣化节点，
    从而把采集成功率稳定在高位，保障下游数据完整性。
    """

    def __init__(self, user: str, pwd: str, gateway: str, ports: list[str]) -> None:
        self._user, self._pwd, self._gw = user, pwd, gateway
        # 每个端口视为一个独立出口端点
        self._endpoints = [self._build(p) for p in ports]
        self._health = {ep: 1.0 for ep in self._endpoints}  # 健康度 ∈ [0,1]
        self._lock = asyncio.Lock()

    def _build(self, port: str) -> str:
        """构造「user:pass@gateway:port」注入式认证串。"""
        return f"http://{self._user}:{self._pwd}@{self._gw}:{port}"

    async def acquire(self) -> str:
        """按健康度择优选出一个出口（并发安全）。"""
        async with self._lock:
            return max(self._health, key=self._health.get)

    def report(self, endpoint: str, ok: bool) -> None:
        """反馈请求结果，成功小幅加分、失败大幅扣分。

        采用 EWMA 思路的简化版：成功 +0.05，失败 -0.2，
        让劣化节点快速冷却、优质节点长期在线。
        """
        delta = 0.05 if ok else -0.2
        self._health[endpoint] = max(0.0, min(1.0, self._health[endpoint] + delta))


async def fetch_one(
    client: httpx.AsyncClient,
    pool: YiniuProxyPool,
    url: str,
    *,
    max_retries: int = 3,
) -> str | None:
    """单页抓取：代理轮换 + 指数退避 + 健康度反馈。

    Args:
        client: 复用连接的 httpx 异步客户端。
        pool: 亿牛云代理池实例。
        url: 目标页面地址。
        max_retries: 最大重试次数，超过返回 None。

    Returns:
        页面 HTML 文本；全部失败返回 None。
    """
    for attempt in range(1, max_retries + 1):
        endpoint = await pool.acquire()
        try:
            resp = await client.get(
                url,
                timeout=10.0,
                proxies={"http://": endpoint, "https://": endpoint},
            )
            resp.raise_for_status()
            # 提前统一编码，把治理动作前移到采集层，避免下游 GBK/UTF-8 混战
            resp.encoding = resp.apparent_encoding or "utf-8"
            pool.report(endpoint, ok=True)
            return resp.text
        except httpx.HTTPError as exc:
            pool.report(endpoint, ok=False)  # 该出口本轮失效，触发健康度下降
            if attempt == max_retries:
                print(f"[ERROR] 抓取失败 {url}: {exc}")
                return None
            # 指数退避 + 抖动：2s、4s、8s，给目标站与代理池喘息
            await asyncio.sleep(2 ** attempt + asyncio.get_event_loop().time() % 1)
    return None


async def crawl_reviews(urls: list[str], *, concurrency: int = 8) -> list[str]:
    """批量抓取，信号量限流 + 亿牛云代理池。

    Args:
        urls: 评论页地址列表。
        concurrency: 最大并发数，需与代理配额匹配，避免超额被打。

    Returns:
        成功抓取的 HTML 列表（已过滤空结果）。
    """
    pool = YiniuProxyPool(PROXY_USER, PROXY_PASS, PROXY_GATEWAY, PROXY_PORTS)
    sem = asyncio.Semaphore(concurrency)  # 并发闸门，保护上游与代理配额

    async def _bounded(u: str) -> str | None:
        async with sem:
            async with httpx.AsyncClient(
                trust_env=False,  # 禁用系统代理，确保亿牛云是唯一出口
                headers={"User-Agent": "Mozilla/5.0 (compatible; DataGovBot/1.0)"},
                follow_redirects=True,
            ) as client:
                return await fetch_one(client, pool, u)

    results = await asyncio.gather(*(_bounded(u) for u in urls))
    return [h for h in results if h]

import pandas as pd


def parse_html_to_records(html: str) -> list[dict]:
    """用 parsel 把单页 HTML 解析成评论字典列表。"""
    sel = Selector(text=html)
    records = []
    for node in sel.css(".review-item"):
        records.append(
            {
                "user_id": node.css(".uid::attr(data-id)").get(""),
                "content": node.css(".text::text").get(""),
                "rating": node.css(".star::attr(data-score)").get(""),
                "created_at": node.css(".time::text").get(""),
            }
        )
    return records


def clean_reviews(raw: pd.DataFrame) -> pd.DataFrame:
    """评论数据治理：唯一性 → 准确性 → 有效性 → 时效性 → 完整性。

    Args:
        raw: 原始解析后的 DataFrame。

    Returns:
        清洗后的 DataFrame，六维中可程序化部分达标。
    """
    df = raw.copy()

    # 唯一性：同一用户同一内容的重复抓取是常事
    df = df.drop_duplicates(subset=["user_id", "content"])

    # 准确性：去 HTML 残留标签、压缩空白、剥离广告噪声
    df["content"] = (
        df["content"]
        .astype(str)
        .str.replace(r"<[^>]+>", "", regex=True)
        .str.replace(r"[\r\n\t]+", " ", regex=True)
        .str.replace(r"\s+", " ", regex=True)
        .str.strip()
    )

    # 有效性：评分转数值、约束到 [0,5] 语义区间
    df["rating"] = pd.to_numeric(df["rating"], errors="coerce")
    df = df[df["rating"].between(0, 5)]

    # 时效性：时间解析为 datetime，无法解析的归入缺失而非默认 1970
    df["created_at"] = pd.to_datetime(df["created_at"], errors="coerce")

    # 完整性：关键分析列缺失即丢弃，绝不用 0 填充误导模型
    df = df.dropna(subset=["content", "rating"])

    # 准确性（二次）：过滤纯表情/超短噪声，避免污染情感分析
    df = df[df["content"].str.len() >= 4]
    return df.reset_index(drop=True)

from pydantic import BaseModel, Field, field_validator


class ReviewRecord(BaseModel):
    """单条评论领域模型，作为数据质量的硬契约。"""

    user_id: str = Field(min_length=1)
    content: str = Field(min_length=4, max_length=2000)
    rating: float = Field(ge=0, le=5)
    created_at: pd.Timestamp | None = None

    @field_validator("content")
    @classmethod
    def _strip(cls, v: str) -> str:
        return v.strip()  # 入库前再 trim 一次，双保险


def validate_batch(rows: list[dict]) -> tuple[list[ReviewRecord], list[dict]]:
    """批量校验，分离「合格」与「被拒」记录。

    Returns:
        (valid_records, rejected_rows)；rejected_rows 即数据血缘中的
        "质量事件"日志，用于回溯哪些源、哪些字段在漂移。
    """
    valid, rejected = [], []
    for row in rows:
        try:
            valid.append(ReviewRecord(**row))
        except Exception as exc:  # noqa: BLE001 治理层需捕获全部校验异常
            rejected.append({"raw": row, "reason": str(exc)})
    return valid, rejected

from snownlp import SnowNLP


def sentiment_score(text: str) -> float:
    """返回 [0,1] 情感得分，越接近 1 越正面。"""
    return round(SnowNLP(text).sentiment, 4)


def label_sentiment(score: float) -> str:
    """连续分映射离散标签，便于分桶统计。"""
    if score >= 0.6:
        return "正面"
    if score <= 0.4:
        return "负面"
    return "中性"

df = pd.DataFrame([r.model_dump() for r in valid])  # valid 来自第三节
df["sentiment"] = df["content"].apply(sentiment_score)
df["label"] = df["sentiment"].apply(label_sentiment)

import asyncio
import plotly.express as px
from collections import Counter


def quality_scorecard(raw: pd.DataFrame, clean: pd.DataFrame) -> dict:
    """按 DAMA 可程序化维度输出质量看板（原始 vs 清洗后）。

    Returns:
        各维度得分字典，值域 [0,1]，越高越好。
    """
    return {
        "completeness": 1 - clean[["content", "rating"]].isna().any(axis=1).mean(),
        "uniqueness": 1 - clean.duplicated(subset=["user_id", "content"]).mean(),
        "validity": clean["rating"].between(0, 5).mean(),
        "timeliness": clean["created_at"].notna().mean(),
        # 噪声率：清洗前后行数差，反映去噪强度
        "noise_reduction": 1 - len(clean) / max(len(raw), 1),
    }


async def run_pipeline(urls: list[str]) -> pd.DataFrame:
    """端到端：采集 → 解析 → 清洗 → 契约 → 情感 → 质量度量。"""
    htmls = await crawl_reviews(urls)
    raw_rows: list[dict] = [r for h in htmls for r in parse_html_to_records(h)]

    raw_df = pd.DataFrame(raw_rows)
    clean_df = clean_reviews(raw_df)

    valid, rejected = validate_batch(clean_df.to_dict("records"))
    print(f"[INFO] 合格 {len(valid)} 条，契约拦截 {len(rejected)} 条")

    card = quality_scorecard(raw_df, clean_df)
    print("[QUALITY]", {k: round(v, 3) for k, v in card.items()})

    df = pd.DataFrame([r.model_dump() for r in valid])
    df["sentiment"] = df["content"].apply(sentiment_score)
    df["label"] = df["sentiment"].apply(label_sentiment)
    return df


def visualize(df: pd.DataFrame) -> None:
    """情感分布柱状图，附样本量与平均分。"""
    dist = Counter(df["label"])
    fig = px.bar(
        x=list(dist.keys()),
        y=list(dist.values()),
        title=f"评论情感分布（N={len(df)}, 平均情感={df['sentiment'].mean():.2f}）",
        labels={"x": "情感标签", "y": "评论数"},
        color=list(dist.keys()),
    )
    fig.show()


if __name__ == "__main__":
    # 替换为真实评论页 URL 列表；规模化时建议分页批采
    urls = ["https://example-shop.com/product/1/reviews?page=1"]
    df = asyncio.run(run_pipeline(urls))
    visualize(df)
    # utf-8-sig 保证 Excel 打开中文不乱码（时效性/可用性细节）
    df.to_csv("reviews_analyzed.csv", index=False, encoding="utf-8-sig")

uv add httpx parsel pandas pydantic snownlp plotly
# 或
pip install httpx parsel pandas pydantic snownlp plotly