import timeimport hashlibimport jsonimport loggingimport requestsimport reimport jiebaimport jieba.analysefrom typing import Dict, List, Tuple, Optionalfrom datetime import datetimeimport pandas as pdimport numpy as npfrom collections import defaultdict, Counter# 日志配置（便于排查商业分析中的数据断层问题）logging.basicConfig(    level=logging.INFO,    format='%(asctime)s - %(levelname)s - %(message)s')logger = logging.getLogger(__name__)class JDCommentAPIClient:    def __init__(self, app_key: str, app_secret: str, access_token: str):        self.app_key = app_key        self.app_secret = app_secret        self.access_token = access_token        self.api_url = "https://api.jd.com/routerjson"        self.session = self._init_session()  # 初始化会话池，提升并发效率        self.stopwords = self._load_stopwords()  # 加载停用词，用于后续文本分析    def _init_session(self) -> requests.Session:        """初始化会话池，配置连接复用与重试，减少接口请求失败率"""        session = requests.Session()        adapter = requests.adapters.HTTPAdapter(            pool_connections=5,  # 连接池数量，匹配QPS限制            pool_maxsize=10,            max_retries=3  # 重试3次，应对临时网络波动        )        session.mount('https://', adapter)        return session    def _generate_sign(self, params: Dict) -> str:        """生成京东API签名，确保请求合法性（官方标准算法）"""        sorted_params = sorted(params.items(), key=lambda x: x[0])        sign_str = self.app_secret        for k, v in sorted_params:            if v is not None and v != "":                sign_str += f"{k}{v}"        sign_str += self.app_secret        return hashlib.md5(sign_str.encode('utf-8')).hexdigest().upper()    def _load_stopwords(self) -> set:        """加载停用词表，用于后续评论文本清洗（提升分析准确性）"""        try:            with open("stopwords.txt", "r", encoding="utf-8") as f:                return set([line.strip() for line in f.readlines() if line.strip()])        except FileNotFoundError:            # 内置基础停用词，避免因文件缺失导致分析中断            return set(["的", "了", "在", "是", "我", "有", "和", "就", "不", "人", "都"])    def get_batch_comments(self, sku_id: str, max_pages: int = 10, page_size: int = 50, 
                          score: int = 0, sort_type: int = 2) -> Tuple[List[Dict], pd.DataFrame]:        """        批量获取商品评论（核心函数，支撑后续所有分析）        :param sku_id: 商品ID（如100012345678）        :param max_pages: 最大获取页数（避免数据量过大）        :param page_size: 每页数量（1-100，匹配接口限制）        :param score: 评分筛选（1-5分，0=全部）        :param sort_type: 排序（2=按有用数倒序，优先获取高价值评论）        :return: 原始评论列表 + 标准化DataFrame（便于后续分析）        """        all_comments = []        page = 1        # 先获取第一页，确认总评论数，避免无效请求        first_comments, total_count = self._get_single_page_comments(            sku_id, page=page, page_size=page_size, score=score, sort_type=sort_type        )        if not first_comments:            logger.warning(f"商品{sku_id}未获取到评论数据")            return [], pd.DataFrame()        all_comments.extend(first_comments)        # 计算总页数（取max_pages与实际页数的最小值，控制数据量）        total_pages = min(max_pages, (total_count + page_size - 1) // page_size)        logger.info(f"商品{sku_id}共{total_count}条评论，计划获取{total_pages}页")        # 批量获取剩余页数        for page in range(2, total_pages + 1):            page_comments, _ = self._get_single_page_comments(                sku_id, page=page, page_size=page_size, score=score, sort_type=sort_type            )            if page_comments:                all_comments.extend(page_comments)                time.sleep(2)  # 控制QPS，避免超限（高级权限可调整为1秒）            else:                logger.warning(f"商品{sku_id}第{page}页评论获取失败，终止后续请求")                break        # 转换为DataFrame，标准化格式（便于情感分析、关键词提取）        comments_df = self._convert_to_standard_df(all_comments)        return all_comments, comments_df    def _get_single_page_comments(self, sku_id: str, page: int, page_size: int, 
                                 score: int, sort_type: int) -> Tuple[List[Dict], int]:        """获取单页评论，封装请求逻辑（内部调用）"""        comment_params = {            "skuId": sku_id,            "pageIndex": page,            "pageSize": page_size,            "score": score,            "sortType": sort_type        }        try:            params = {                "method": "jd.union.open.comment.query",                "app_key": self.app_key,                "access_token": self.access_token,                "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),                "format": "json",                "v": "1.0",                "sign_method": "md5",                "360buy_param_json": json.dumps(comment_params)            }            params["sign"] = self._generate_sign(params)            response = self.session.get(                self.api_url, params=params, timeout=(10, 30)  # 长超时，应对大评论量            )            response.raise_for_status()            result = response.json()            # 处理接口错误（如权限不足、SKU无效）            if "error_response" in result:                error = result["error_response"]                logger.error(f"接口错误：{error.get('msg')}（错误码：{error.get('code')}）")                return [], 0            # 解析评论数据            data = result.get("jd_union_open_comment_query_response", {})            comment_result = data.get("result", {})            return comment_result.get("comments", []), comment_result.get("totalCount", 0)        except requests.exceptions.RequestException as e:            logger.error(f"请求异常：{str(e)}")            return [], 0    def _convert_to_standard_df(self, comments: List[Dict]) -> pd.DataFrame:        """将评论列表转换为标准化DataFrame，新增清洗后字段（支撑后续分析）"""        data = []        for comment in comments:            # 清洗评论内容（去除HTML标签、特殊字符）            cleaned_content = self._clean_comment_content(comment.get("content", ""))            # 提取购买属性（如“颜色：黑色；尺寸：XL”→ 结构化字典）            purchase_attr = self._parse_purchase_attr(comment.get("purchaseAttr", ""))            data.append({                "comment_id": comment.get("id", ""),                "user_id": comment.get("userId", ""),                "user_level": comment.get("userLevel", 0),                "score": comment.get("score", 0),                "content": comment.get("content", ""),                "cleaned_content": cleaned_content,                "creation_time": comment.get("creationTime", ""),                "useful_count": comment.get("usefulVoteCount", 0),                "has_image": len(comment.get("images", [])) > 0,                "has_video": comment.get("videoInfo", {}) != {},                "purchase_attr": json.dumps(purchase_attr),  # 结构化存储                "after_days": comment.get("afterDays", 0)  # 购买后评论天数（判断长期满意度）            })        return pd.DataFrame(data)    def _clean_comment_content(self, content: str) -> str:        """清洗评论内容，去除噪声（提升情感分析、关键词提取准确性）"""        if not content:            return ""        # 去除HTML标签（如<br>）        content = re.sub(r'<[^>]+>', '', content)        # 去除特殊字符（如emoji、符号）        content = re.sub(r'[^\u4e00-\u9fa5a-zA-Z0-9\s]', ' ', content)        # 去除多余空格        content = re.sub(r'\s+', ' ', content).strip()        return content    def _parse_purchase_attr(self, attr_str: str) -> Dict:        """解析购买属性字符串为字典（如“颜色：黑色；尺寸：XL”→ {"颜色":"黑色","尺寸":"XL"}）"""        attr_dict = {}        if not attr_str:            return attr_dict        for attr in attr_str.split(";"):            if "：" in attr:                key, value = attr.split("：", 1)                attr_dict[key.strip()] = value.strip()        return attr_dict

def analyze_sentiment(self, comments_df: pd.DataFrame) -> Tuple[pd.DataFrame, Dict]:        """        情感分析：给每条评论打情感标签，并按维度统计情感分布        :param comments_df: 标准化评论DataFrame        :return: 带情感标签的DataFrame + 维度情感统计（支撑商业决策）        """        if comments_df.empty:            return comments_df, {}        # 加载情感词表（优化基础算法）        positive_words = self._load_sentiment_words("positive_words.txt")        negative_words = self._load_sentiment_words("negative_words.txt")        # 给每条评论打情感分（0=负面，1=正面，0.5=中性）        comments_df["sentiment_score"] = comments_df["cleaned_content"].apply(            lambda x: self._calculate_sentiment_score(x, positive_words, negative_words)        )        # 打情感标签        comments_df["sentiment_label"] = comments_df["sentiment_score"].apply(            lambda x: "positive" if x >= 0.6 else ("negative" if x <= 0.4 else "neutral")        )        # 按核心维度统计情感分布（商业痛点定位关键）        aspect_list = ["质量", "价格", "物流", "外观", "性能", "服务"]        aspect_sentiment = self._analyze_aspect_sentiment(comments_df, aspect_list)        return comments_df, aspect_sentiment    def _load_sentiment_words(self, file_path: str) -> set:        """加载情感词表（自定义词表，适配电商场景）"""        try:            with open(file_path, "r", encoding="utf-8") as f:                return set([line.strip() for line in f.readlines() if line.strip()])        except FileNotFoundError:            # 内置电商场景情感词（如“耐用”“划算”“慢”“破损”）            if "positive" in file_path:                return set(["耐用", "划算", "好看", "快", "满意", "好用", "清晰", "流畅"])            else:                return set(["慢", "破损", "卡顿", "贵", "不满意", "难用", "模糊", "差"])    def _calculate_sentiment_score(self, content: str, positive_words: set, negative_words: set) -> float:        """计算情感得分（结合SnowNLP与自定义词表，提升准确率）"""        if not content:            return 0.5        # 基础SnowNLP得分        from snownlp import SnowNLP        base_score = SnowNLP(content).sentiments        # 结合情感词表调整得分（适配电商场景）        words = jieba.lcut(content)        pos_count = sum(1 for word in words if word in positive_words)        neg_count = sum(1 for word in words if word in negative_words)        if pos_count > neg_count:            # 正面词更多，提升得分            base_score = min(1.0, base_score + 0.1 * (pos_count - neg_count))        elif neg_count > pos_count:            # 负面词更多，降低得分            base_score = max(0.0, base_score - 0.1 * (neg_count - pos_count))        return round(base_score, 4)    def _analyze_aspect_sentiment(self, comments_df: pd.DataFrame, aspect_list: List[str]) -> Dict:        """按维度统计情感分布（如“物流”维度的正面/负面占比）"""        aspect_result = {}        for aspect in aspect_list:            # 筛选提及该维度的评论            aspect_comments = comments_df[                comments_df["cleaned_content"].str.contains(aspect, na=False)            ]            if len(aspect_comments) == 0:                aspect_result[aspect] = {                    "count": 0, "positive_ratio": 0.0, "negative_ratio": 0.0,                    "positive_examples": [], "negative_examples": []                }                continue            # 计算情感占比            total = len(aspect_comments)            positive_count = len(aspect_comments[aspect_comments["sentiment_label"] == "positive"])            negative_count = len(aspect_comments[aspect_comments["sentiment_label"] == "negative"])            # 提取示例评论（用于汇报/决策参考）            positive_examples = aspect_comments[aspect_comments["sentiment_label"] == "positive"][                "content"            ].head(2).tolist()            negative_examples = aspect_comments[aspect_comments["sentiment_label"] == "negative"][                "content"            ].head(2).tolist()            aspect_result[aspect] = {                "count": total,                "positive_ratio": round(positive_count / total * 100, 1),                "negative_ratio": round(negative_count / total * 100, 1),                "positive_examples": [ex[:50] + "..." for ex in positive_examples],                "negative_examples": [ex[:50] + "..." for ex in negative_examples]            }        return aspect_result

def mine_user_demands(self, comments_df: pd.DataFrame) -> Dict:        """        挖掘用户需求：功能需求、改进建议、使用场景（直接支撑产品优化）        :param comments_df: 带情感标签的评论DataFrame        :return: 结构化需求字典        """        if comments_df.empty:            return {"function_demands": {}, "improvement_suggestions": [], "usage_scenes": {}}        # 1. 功能需求挖掘（按“性能、续航”等类别匹配关键词）        function_keywords = {            "性能": ["快", "慢", "流畅", "卡顿", "稳定", "反应快"],            "续航": ["续航", "电池", "电量", "充电", "用得久"],            "外观": ["外观", "颜色", "设计", "大小", "重量", "材质"],            "易用性": ["简单", "方便", "复杂", "麻烦", "操作"],            "价格": ["贵", "便宜", "性价比", "划算", "不值"]        }        function_demands = defaultdict(list)        for func, keywords in function_keywords.items():            for keyword in keywords:                # 筛选提及该关键词的评论                related_comments = comments_df[                    comments_df["cleaned_content"].str.contains(keyword, na=False)                ]                if len(related_comments) == 0:                    continue                # 统计该关键词的情感倾向（需求是否满意）                main_sentiment = self._get_main_sentiment(related_comments)                function_demands[func].append({                    "keyword": keyword,                    "mention_count": len(related_comments),                    "main_sentiment": main_sentiment,                    "examples": related_comments["content"].head(2).apply(lambda x: x[:50] + "...").tolist()                })        # 2. 改进建议提取（正则匹配“希望、建议”等句式）        suggestion_patterns = [            r"如果能(.*?)就好了", r"希望(.*?)", r"建议(.*?)",            r"要是(.*?)就好了", r"应该(.*?)"        ]        improvement_suggestions = []        all_contents = comments_df["content"].tolist()        for content in all_contents:            for pattern in suggestion_patterns:                match = re.search(pattern, content)                if match:                    suggestion = match.group(1).strip()                    # 去重（避免重复建议）                    if not any(s["suggestion"] == suggestion for s in improvement_suggestions):                        improvement_suggestions.append({                            "suggestion": suggestion,                            "original_comment": content[:60] + "..."                        })                    break  # 每条评论只提取1条核心建议        # 3. 使用场景识别（统计“家用、办公”等场景关键词）        scene_keywords = ["家用", "办公", "户外", "旅行", "孩子", "老人", "送礼", "学生"]        usage_scenes = defaultdict(int)        for content in all_contents:            for scene in scene_keywords:                if scene in content:                    usage_scenes[scene] += 1        return {            "function_demands": dict(function_demands),            "improvement_suggestions": improvement_suggestions[:10],  # 取前10条核心建议            "usage_scenes": dict(sorted(usage_scenes.items(), key=lambda x: x[1], reverse=True))        }    def _get_main_sentiment(self, comments_df: pd.DataFrame) -> str:        """获取评论集的主要情感倾向（用于判断需求是否满意）"""        sentiment_count = comments_df["sentiment_label"].value_counts()        return sentiment_count.index[0] if not sentiment_count.empty else "neutral"

def compare_with_competitor(self, target_df: pd.DataFrame, competitor_df: pd.DataFrame,                                target_name: str = "目标商品", competitor_name: str = "竞品") -> Dict:        """        竞品对比分析：从评分、情感、关键词维度找优劣势        :param target_df: 目标商品评论DataFrame        :param competitor_df: 竞品评论DataFrame        :return: 结构化对比结果（支撑差异化运营）        """        if target_df.empty or competitor_df.empty:            return {"error": "目标商品或竞品评论数据为空，无法对比"}        # 1. 基础评分对比        target_avg_score = target_df["score"].mean()        competitor_avg_score = competitor_df["score"].mean()        # 2. 情感分布对比（正面/负面/中性占比）        target_sentiment_dist = target_df["sentiment_label"].value_counts(normalize=True)        competitor_sentiment_dist = competitor_df["sentiment_label"].value_counts(normalize=True)        # 3. 关键词权重对比（提取核心差异关键词）        target_keywords = self._extract_keywords(target_df, top_n=15)        competitor_keywords = self._extract_keywords(competitor_df, top_n=15)        target_keyword_dict = dict(target_keywords)        competitor_keyword_dict = dict(competitor_keywords)        # 优势关键词：目标商品权重高于竞品        advantage_keywords = [            (word, target_w, competitor_w)            for word, target_w in target_keywords            if word in competitor_keyword_dict and target_w > competitor_keyword_dict[word]        ]        # 劣势关键词：竞品权重高于目标商品        disadvantage_keywords = [            (word, competitor_w, target_w)            for word, competitor_w in competitor_keywords            if word in target_keyword_dict and competitor_w > target_keyword_dict[word]        ]        # 4. 核心维度竞争力对比（质量、价格等）        compare_aspects = ["质量", "价格", "物流", "外观", "性能"]        aspect_compare = {}        for aspect in compare_aspects:            target_aspect_data = self._get_aspect_sentiment(target_df, [aspect])[aspect]            competitor_aspect_data = self._get_aspect_sentiment(competitor_df, [aspect])[aspect]            aspect_compare[aspect] = {                f"{target_name}_positive_ratio": target_aspect_data["positive_ratio"],                f"{competitor_name}_positive_ratio": competitor_aspect_data["positive_ratio"],                f"{target_name}_mention_count": target_aspect_data["count"],                f"{competitor_name}_mention_count": competitor_aspect_data["count"]            }        return {            "score_comparison": {                target_name: round(target_avg_score, 1),                competitor_name: round(competitor_avg_score, 1),                "score_gap": round(target_avg_score - competitor_avg_score, 1)            },            "sentiment_distribution": {                target_name: {k: round(v*100, 1) for k, v in target_sentiment_dist.items()},                competitor_name: {k: round(v*100, 1) for k, v in competitor_sentiment_dist.items()}            },            "advantage_keywords": advantage_keywords[:5],  # 前5个核心优势            "disadvantage_keywords": disadvantage_keywords[:5],  # 前5个核心劣势            "aspect_competition": aspect_compare        }    def _extract_keywords(self, comments_df: pd.DataFrame, top_n: int = 15) -> List[Tuple[str, float]]:        """提取评论关键词（TF-IDF算法，突出核心关注点）"""        if comments_df.empty:            return []        # 合并所有清洗后的评论内容        all_content = " ".join(comments_df["cleaned_content"].dropna())        # 提取关键词（只保留名词、动词、形容词，更贴合需求）        keywords = jieba.analyse.extract_tags(            all_content, topK=top_n, withWeight=True, allowPOS=('n', 'v', 'a')        )        return keywords