Review Dedup & Quality Filter — 多平台评论在线去重与质量排序

1. 解决的问题

从 Amazon、TikTok Shop、独立站同时采集的评论中，30-40% 是重复或低质量内容（同一用户多平台发布、机器生成水评、极短无意义评论）。直接用于 VOC 分析会严重扭曲洞察结论。

2. 核心算法逻辑

从 Amazon、TikTok Shop、独立站同时采集的评论中，3040% 是重复或低质量内容（同一用户多平台发布、机器生成水评、极短无意义评论）。直接用于 VOC 分析会严重扭曲洞察结论。

3. 业务应用场景

业务背景：每日从 Amazon US/UK、TikTok Shop、独立站采集约 500 条新评论，其中约 180 条（36%）是跨平台重复或水评。人工清洗需 30-45 分钟/天。

业务背景：季度复盘需要分析过去 3 个月全渠道 470 万条评论，识别"静音性能下降"等趋势。人工抽样误差大，全量处理成本高。

4. 输入数据要求

请查看原始代码模板获取输入规格。

5. 输出结果

请查看原始代码模板获取输出规格。

6. 业务价值 / ROI

10-50 万元

7. 代码模板

代码块数量：6 · 路径：未检测到

import hashlib
import heapq
import math
import re
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional, Set, Tuple


@dataclass
class Review:
    review_id: str
    text: str
    platform: str
    timestamp: float = 0.0
    rating: float = 3.0
    verified_purchase: bool = False
    helpful_votes: int = 0
    aspects: Dict[str, str] = field(default_factory=dict)
    quality_score: float = 0.0


PLATFORM_PRIORITY = {"amazon": 3, "independent": 2, "tiktok": 1, "other": 0}


class SimHashSignature:
    def __init__(self, n_bits: int = 64, n_shingles: int = 3):
        self.n_bits = n_bits
        self.n_shingles = n_shingles

    def _shingles(self, text: str) -> List[str]:
        tokens = re.sub(r'[^\w\s]', '', text.lower()).split()
        return [" ".join(tokens[i:i+self.n_shingles])
                for i in range(len(tokens) - self.n_shingles + 1)] or [text[:16]]

    def compute(self, text: str) -> int:
        shingles = self._shingles(text)
        v = [0] * self.n_bits
        for s in shingles:
            h = int(hashlib.md5(s.encode()).hexdigest(), 16)
            for i in range(self.n_bits):
                v[i] += 1 if (h >> i) & 1 else -1
        return sum(1 << i for i in range(self.n_bits) if v[i] > 0)

    def hamming_distance(self, sig_a: int, sig_b: int) -> int:
        xor = sig_a ^ sig_b
        return bin(xor).count('1')

    def similarity(self, sig_a: int, sig_b: int) -> float:
        return 1.0 - self.hamming_distance(sig_a, sig_b) / self.n_bits


class HNSWIndex:
    """轻量 HNSW 近似最近邻索引（FOLD 核心数据结构）"""

    def __init__(self, max_neighbors: int = 16, similarity_fn=None):
        self.max_neighbors = max_neighbors
        self.nodes: Dict[str, int] = {}
        self.signatures: List[Tuple[str, int]] = []
        self.graph: Dict[int, List[int]] = {}
        self.sim_fn = similarity_fn or (lambda a, b: 1.0 - bin(a ^ b).count('1') / 64)

8. 论文来源

• 2601.07449
• 2602.21082
• 2606.03001