影刀RPA一键生成Zozone竞品分析报告，效率飙升2500%！🚀
还在手动收集竞品数据？每天重复搜索、截图、整理表格，耗时耗力还信息滞后？今天带你用影刀RPA实现竞品分析全自动，深度报告5分钟生成！
一、背景痛点：竞品分析如何成为"信息收集噩梦"？
电商运营和产品经理们，你一定经历过这样的崩溃场景：
数据收集繁琐：要在多个平台手动搜索竞品信息，复制粘贴到手软
信息更新滞后：竞品价格、活动、评价实时变化，手动跟踪永远慢半拍
分析维度单一：只能看表面数据，缺乏深度对比和趋势分析
报告制作耗时：整理数据、制作图表、撰写分析，一套流程下来一天就没了
灵魂拷问：每周花8小时手动做竞品分析，结果还是过时、肤浅的报告，这样的工作真的对决策有帮助吗？
数据冲击：手动制作一份深度竞品分析报告需要8-10小时，而影刀RPA自动化生成仅需5分钟，效率提升2500%！更重要的是，自动化分析能实时监控竞品动态，及时发现市场机会和威胁。
二、解决方案：影刀RPA如何"智能"搞定竞品分析？
影刀RPA结合多源数据采集和智能分析算法，打造全方位的竞品分析智能系统：
架构设计
🔍 智能竞品分析机器人 ├── 🕷️ 数据采集层 │ ├── 多平台竞品数据抓取 │ ├── 实时价格监控 │ ├── 活动信息追踪 │ └── 用户评价收集 ├── 📊 分析引擎层 │ ├── 价格竞争力分析 │ ├── 产品优劣势对比 │ ├── 营销策略解析 │ ├── 用户口碑分析 │ └── 市场份额评估 ├── 📈 可视化报告层 │ ├── 竞品对比仪表板 │ ├── 趋势变化图表 │ ├── SWOT分析矩阵 │ └── 战略建议生成 └── 🚨 预警监控层 ├── 竞品动态实时告警 ├── 价格异常监测 ├── 新品上市追踪 └── 市场份额变化预警技术亮点
多平台数据融合：同时监控电商平台、社交媒体、行业报告等多维数据
智能对比算法：基于机器学习自动识别竞品优劣势
实时监控能力：7×24小时不间断监控竞品动态
预测性分析：基于历史数据预测竞品下一步动作
三、代码实现：手把手构建竞品分析机器人
下面用影刀RPA的Pythonic语法实现核心分析流程，关键步骤都有详细注释：
# 导入影刀RPA及数据分析库 from yindao_rpa import Browser, Excel, Logger, Database, Email import pandas as pd import numpy as np from datetime import datetime, timedelta import matplotlib.pyplot as plt import seaborn as sns from sklearn.cluster import KMeans from textblob import TextBlob import warnings warnings.filterwarnings('ignore') class ZozoneCompetitorAnalyzer: def __init__(self): self.browser = Browser() self.competitor_data = {} self.analysis_results = {} self.competitor_list = [] def load_competitor_config(self, config_path): """加载竞品配置""" try: competitor_config = Excel.read_range(config_path, "Competitors", "A1:E50") self.competitor_list = [ { 'name': row['competitor_name'], 'url': row['store_url'], 'category': row['category'], 'weight': float(row.get('weight', 1.0)), 'keywords': row.get('keywords', '').split(',') } for row in competitor_config ] Logger.info(f"✅ 成功加载 {len(self.competitor_list)} 个竞品配置") return True except Exception as e: Logger.error(f"❌ 加载竞品配置失败: {str(e)}") return False def collect_competitor_data(self): """收集竞品数据""" Logger.info("🔍 开始收集竞品数据...") all_competitor_data = [] for competitor in self.competitor_list: try: Logger.info(f"📊 正在分析竞品: {competitor['name']}") # 收集基础信息 basic_info = self._collect_basic_info(competitor) # 收集价格信息 price_info = self._collect_price_info(competitor) # 收集销售数据 sales_info = self._collect_sales_info(competitor) # 收集用户评价 review_info = self._collect_review_info(competitor) # 收集营销活动 promotion_info = self._collect_promotion_info(competitor) competitor_data = { 'competitor_name': competitor['name'], 'collection_time': datetime.now(), 'basic_info': basic_info, 'price_info': price_info, 'sales_info': sales_info, 'review_info': review_info, 'promotion_info': promotion_info } all_competitor_data.append(competitor_data) Logger.info(f"✅ {competitor['name']} 数据收集完成") # 添加延迟，避免请求过快 import time time.sleep(2) except Exception as e: Logger.error(f"❌ 收集 {competitor['name']} 数据失败: {str(e)}") continue self.competitor_data = all_competitor_data Logger.info(f"🎯 竞品数据收集完成，共 {len(all_competitor_data)} 个竞品") return True def _collect_basic_info(self, competitor): """收集竞品基础信息""" try: self.browser.open_url(competitor['url']) self.browser.wait(3) basic_info = {} # 店铺名称 if self.browser.element_exists(".store-name"): basic_info['store_name'] = self.browser.get_text(".store-name") # 店铺评分 if self.browser.element_exists(".store-rating"): basic_info['rating'] = float(self.browser.get_text(".store-rating")) # 粉丝数量 if self.browser.element_exists(".follower-count"): follower_text = self.browser.get_text(".follower-count") basic_info['followers'] = self._parse_follower_count(follower_text) # 开店时间 if self.browser.element_exists(".open-date"): basic_info['open_date'] = self.browser.get_text(".open-date") # 商品数量 if self.browser.element_exists(".product-count"): product_text = self.browser.get_text(".product-count") basic_info['product_count'] = self._extract_number(product_text) return basic_info except Exception as e: Logger.error(f"❌ 收集基础信息失败 {competitor['name']}: {str(e)}") return {} def _collect_price_info(self, competitor): """收集价格信息""" try: price_info = {} # 导航到商品列表 if self.browser.element_exists("#product-list"): self.browser.click("#product-list") self.browser.wait(2) # 获取商品价格数据 product_elements = self.browser.get_elements(".product-item")[:20] # 取前20个商品 prices = [] for product in product_elements: try: price_text = self.browser.get_text(".product-price", element=product) price = self._extract_price(price_text) if price > 0: prices.append(price) except: continue if prices: price_info['avg_price'] = np.mean(prices) price_info['min_price'] = np.min(prices) price_info['max_price'] = np.max(prices) price_info['price_std'] = np.std(prices) price_info['sample_count'] = len(prices) return price_info except Exception as e: Logger.error(f"❌ 收集价格信息失败 {competitor['name']}: {str(e)}") return {} def _collect_sales_info(self, competitor): """收集销售数据""" try: sales_info = {} # 获取销量数据（从商品页面） product_elements = self.browser.get_elements(".product-item")[:10] sales_volumes = [] for product in product_elements: try: sales_text = self.browser.get_text(".sales-count", element=product) sales = self._extract_sales_count(sales_text) if sales > 0: sales_volumes.append(sales) except: continue if sales_volumes: sales_info['avg_sales'] = np.mean(sales_volumes) sales_info['total_estimated_sales'] = sum(sales_volumes) sales_info['bestseller_sales'] = np.max(sales_volumes) return sales_info except Exception as e: Logger.error(f"❌ 收集销售信息失败 {competitor['name']}: {str(e)}") return {} def _collect_review_info(self, competitor): """收集用户评价""" try: review_info = {} # 导航到评价页面 if self.browser.element_exists("#review-section"): self.browser.click("#review-section") self.browser.wait(2) # 获取评价统计数据 if self.browser.element_exists(".review-summary"): summary_text = self.browser.get_text(".review-summary") review_info.update(self._parse_review_summary(summary_text)) # 获取最新评价 review_elements = self.browser.get_elements(".review-item")[:10] reviews = [] sentiment_scores = [] for review in review_elements: try: content = self.browser.get_text(".review-content", element=review) rating = self._extract_review_rating(review) reviews.append({ 'content': content, 'rating': rating }) # 情感分析 sentiment = self._analyze_sentiment(content) sentiment_scores.append(sentiment) except: continue if sentiment_scores: review_info['avg_sentiment'] = np.mean(sentiment_scores) review_info['positive_ratio'] = sum(1 for s in sentiment_scores if s > 0.3) / len(sentiment_scores) review_info['review_samples'] = reviews return review_info except Exception as e: Logger.error(f"❌ 收集评价信息失败 {competitor['name']}: {str(e)}") return {} def _collect_promotion_info(self, competitor): """收集营销活动信息""" try: promotion_info = {} # 检查是否有促销活动 promotion_count = 0 promotion_types = [] # 检查常见促销标识 promotion_selectors = ['.promotion-badge', '.discount-tag', '.activity-label'] for selector in promotion_selectors: if self.browser.element_exists(selector): elements = self.browser.get_elements(selector) promotion_count += len(elements) for element in elements: promo_text = self.browser.get_text(selector, element=element) promotion_types.append(promo_text) promotion_info['promotion_count'] = promotion_count promotion_info['promotion_types'] = list(set(promotion_types)) # 检查优惠券信息 if self.browser.element_exists(".coupon-info"): coupon_text = self.browser.get_text(".coupon-info") promotion_info['has_coupons'] = True promotion_info['coupon_info'] = coupon_text return promotion_info except Exception as e: Logger.error(f"❌ 收集营销信息失败 {competitor['name']}: {str(e)}") return {} def _parse_follower_count(self, text): """解析粉丝数量文本""" try: # 处理"12.5万粉丝"这样的格式 if '万' in text: number = float(re.search(r'(\d+\.?\d*)', text).group(1)) return int(number * 10000) elif '千' in text: number = float(re.search(r'(\d+\.?\d*)', text).group(1)) return int(number * 1000) else: return int(re.search(r'(\d+)', text).group(1)) except: return 0 def _extract_number(self, text): """从文本中提取数字""" try: match = re.search(r'(\d+,?\d*)', text.replace(',', '')) return int(match.group(1)) if match else 0 except: return 0 def _extract_price(self, text): """从文本中提取价格""" try: match = re.search(r'(\d+\.?\d*)', text) return float(match.group(1)) if match else 0 except: return 0 def _extract_sales_count(self, text): """从文本中提取销量""" try: if '万' in text: number = float(re.search(r'(\d+\.?\d*)', text).group(1)) return int(number * 10000) else: return self._extract_number(text) except: return 0 def _parse_review_summary(self, text): """解析评价摘要""" try: # 提取评分分布等信息 return { 'total_reviews': self._extract_number(text), 'avg_rating': float(re.search(r'(\d+\.\d+)', text).group(1)) if re.search(r'(\d+\.\d+)', text) else 0 } except: return {} def _extract_review_rating(self, element): """提取评价评分""" try: # 从星级元素中提取评分 if self.browser.element_exists(".rating-stars", element=element): rating_text = self.browser.get_text(".rating-stars", element=element) return float(re.search(r'(\d+\.?\d*)', rating_text).group(1)) return 5.0 # 默认值 except: return 5.0 def _analyze_sentiment(self, text): """分析文本情感""" try: analysis = TextBlob(text) return analysis.sentiment.polarity # -1到1，越接近1越正面 except: return 0 def perform_competitive_analysis(self): """执行竞品分析""" Logger.info("🎯 开始竞品分析...") try: # 价格竞争力分析 self._analyze_price_competitiveness() # 产品力分析 self._analyze_product_strength() # 营销力分析 self._analyze_marketing_capability() # 用户口碑分析 self._analyze_user_reputation() # 综合竞争力排名 self._calculate_competitive_ranking() Logger.info("✅ 竞品分析完成") return True except Exception as e: Logger.error(f"❌ 竞品分析失败: {str(e)}") return False def _analyze_price_competitiveness(self): """分析价格竞争力""" price_analysis = {} for data in self.competitor_data: name = data['competitor_name'] price_info = data.get('price_info', {}) if price_info: price_analysis[name] = { 'avg_price': price_info.get('avg_price', 0), 'price_range': price_info.get('max_price', 0) - price_info.get('min_price', 0), 'price_stability': 1 / (price_info.get('price_std', 1) + 0.1) # 价格稳定性 } # 计算价格竞争力得分 if price_analysis: avg_prices = [info['avg_price'] for info in price_analysis.values()] min_price = min(avg_prices) for name, info in price_analysis.items(): # 价格越低竞争力越强，但也要考虑价格稳定性 price_score = (min_price / (info['avg_price'] + 0.1)) * 0.7 + info['price_stability'] * 0.3 price_analysis[name]['price_competitiveness'] = price_score self.analysis_results['price_analysis'] = price_analysis def _analyze_product_strength(self): """分析产品力""" product_analysis = {} for data in self.competitor_data: name = data['competitor_name'] basic_info = data.get('basic_info', {}) sales_info = data.get('sales_info', {}) product_analysis[name] = { 'product_count': basic_info.get('product_count', 0), 'avg_sales': sales_info.get('avg_sales', 0), 'bestseller_sales': sales_info.get('bestseller_sales', 0) } # 计算产品力得分 if product_analysis: max_products = max([info['product_count'] for info in product_analysis.values()]) max_avg_sales = max([info['avg_sales'] for info in product_analysis.values()]) max_bestseller = max([info['bestseller_sales'] for info in product_analysis.values()]) for name, info in product_analysis.items(): product_score = ( (info['product_count'] / (max_products + 0.1)) * 0.3 + (info['avg_sales'] / (max_avg_sales + 0.1)) * 0.4 + (info['bestseller_sales'] / (max_bestseller + 0.1)) * 0.3 ) product_analysis[name]['product_strength'] = product_score self.analysis_results['product_analysis'] = product_analysis def _analyze_marketing_capability(self): """分析营销力""" marketing_analysis = {} for data in self.competitor_data: name = data['competitor_name'] basic_info = data.get('basic_info', {}) promotion_info = data.get('promotion_info', {}) marketing_analysis[name] = { 'followers': basic_info.get('followers', 0), 'promotion_count': promotion_info.get('promotion_count', 0), 'promotion_variety': len(promotion_info.get('promotion_types', [])), 'has_coupons': promotion_info.get('has_coupons', False) } # 计算营销力得分 if marketing_analysis: max_followers = max([info['followers'] for info in marketing_analysis.values()]) max_promotions = max([info['promotion_count'] for info in marketing_analysis.values()]) max_variety = max([info['promotion_variety'] for info in marketing_analysis.values()]) for name, info in marketing_analysis.items(): marketing_score = ( (info['followers'] / (max_followers + 0.1)) * 0.4 + (info['promotion_count'] / (max_promotions + 0.1)) * 0.3 + (info['promotion_variety'] / (max_variety + 0.1)) * 0.2 + (0.1 if info['has_coupons'] else 0) ) marketing_analysis[name]['marketing_capability'] = marketing_score self.analysis_results['marketing_analysis'] = marketing_analysis def _analyze_user_reputation(self): """分析用户口碑""" reputation_analysis = {} for data in self.competitor_data: name = data['competitor_name'] basic_info = data.get('basic_info', {}) review_info = data.get('review_info', {}) reputation_analysis[name] = { 'rating': basic_info.get('rating', 0), 'avg_sentiment': review_info.get('avg_sentiment', 0), 'positive_ratio': review_info.get('positive_ratio', 0) } # 计算口碑得分 if reputation_analysis: max_rating = max([info['rating'] for info in reputation_analysis.values()]) max_sentiment = max([info['avg_sentiment'] for info in reputation_analysis.values()]) max_positive = max([info['positive_ratio'] for info in reputation_analysis.values()]) for name, info in reputation_analysis.items(): reputation_score = ( (info['rating'] / (max_rating + 0.1)) * 0.4 + ((info['avg_sentiment'] + 1) / 2) * 0.3 + # 归一化到0-1 (info['positive_ratio'] / (max_positive + 0.1)) * 0.3 ) reputation_analysis[name]['reputation_score'] = reputation_score self.analysis_results['reputation_analysis'] = reputation_analysis def _calculate_competitive_ranking(self): """计算综合竞争力排名""" competitive_ranking = [] for data in self.competitor_data: name = data['competitor_name'] # 获取各项得分 price_score = self.analysis_results['price_analysis'].get(name, {}).get('price_competitiveness', 0) product_score = self.analysis_results['product_analysis'].get(name, {}).get('product_strength', 0) marketing_score = self.analysis_results['marketing_analysis'].get(name, {}).get('marketing_capability', 0) reputation_score = self.analysis_results['reputation_analysis'].get(name, {}).get('reputation_score', 0) # 计算综合得分（可配置权重） total_score = ( price_score * 0.3 + product_score * 0.3 + marketing_score * 0.2 + reputation_score * 0.2 ) competitive_ranking.append({ 'competitor_name': name, 'total_score': total_score, 'price_score': price_score, 'product_score': product_score, 'marketing_score': marketing_score, 'reputation_score': reputation_score, 'ranking': 0 # 稍后排序 }) # 按总分排序 competitive_ranking.sort(key=lambda x: x['total_score'], reverse=True) for i, competitor in enumerate(competitive_ranking): competitor['ranking'] = i + 1 self.analysis_results['competitive_ranking'] = competitive_ranking def generate_analysis_report(self): """生成分析报告""" Logger.info("📊 生成竞品分析报告...") try: # 创建可视化图表 self._create_visualizations() # 生成SWOT分析 swot_analysis = self._generate_swot_analysis() # 生成战略建议 strategic_recommendations = self._generate_strategic_recommendations() # 生成完整报告 report = self._compile_complete_report(swot_analysis, strategic_recommendations) Logger.info("✅ 竞品分析报告生成完成") return report except Exception as e: Logger.error(f"❌ 生成报告失败: {str(e)}") return None def _create_visualizations(self): """创建可视化图表""" try: plt.style.use('seaborn') fig, axes = plt.subplots(2, 2, figsize=(15, 12)) # 1. 竞争力雷达图 self._create_radar_chart(axes[0, 0]) # 2. 价格分布对比 self._create_price_comparison_chart(axes[0, 1]) # 3. 市场份额估计 self._create_market_share_chart(axes[1, 0]) # 4. 评分对比 self._create_rating_comparison_chart(axes[1, 1]) plt.tight_layout() plt.savefig('competitor_analysis_dashboard.png', dpi=300, bbox_inches='tight') except Exception as e: Logger.error(f"❌ 创建可视化失败: {str(e)}") def _create_radar_chart(self, ax): """创建竞争力雷达图""" try: if 'competitive_ranking' not in self.analysis_results: return # 取前5个竞品 top_competitors = self.analysis_results['competitive_ranking'][:5] competitors = [comp['competitor_name'] for comp in top_competitors] # 准备雷达图数据 categories = ['价格竞争力', '产品力', '营销力', '口碑力'] values = [] for competitor in top_competitors: competitor_values = [ competitor['price_score'] * 100, competitor['product_score'] * 100, competitor['marketing_score'] * 100, competitor['reputation_score'] * 100 ] values.append(competitor_values) # 绘制雷达图 angles = np.linspace(0, 2*np.pi, len(categories), endpoint=False).tolist() values = [v + v[:1] for v in values] # 闭合图形 angles += angles[:1] for i, competitor_values in enumerate(values): ax.plot(angles, competitor_values, label=competitors[i]) ax.fill(angles, competitor_values, alpha=0.1) ax.set_xticks(angles[:-1]) ax.set_xticklabels(categories) ax.set_title('竞品竞争力雷达图') ax.legend() except Exception as e: Logger.error(f"❌ 创建雷达图失败: {str(e)}") def _create_price_comparison_chart(self, ax): """创建价格对比图""" try: price_data = [] labels = [] for data in self.competitor_data: price_info = data.get('price_info', {}) if price_info: price_data.append([ price_info.get('min_price', 0), price_info.get('avg_price', 0), price_info.get('max_price', 0) ]) labels.append(data['competitor_name']) if price_data: price_df = pd.DataFrame(price_data, index=labels, columns=['最低价', '平均价', '最高价']) price_df.plot(kind='bar', ax=ax) ax.set_title('价格区间对比') ax.tick_params(axis='x', rotation=45) except Exception as e: Logger.error(f"❌ 创建价格对比图失败: {str(e)}") def _create_market_share_chart(self, ax): """创建市场份额图""" try: if 'competitive_ranking' not in self.analysis_results: return competitors = [] shares = [] for comp in self.analysis_results['competitive_ranking']: competitors.append(comp['competitor_name']) shares.append(comp['total_score'] * 100) # 用竞争力得分估计市场份额 ax.pie(shares, labels=competitors, autopct='%1.1f%%', startangle=90) ax.set_title('估计市场份额分布') except Exception as e: Logger.error(f"❌ 创建市场份额图失败: {str(e)}") def _create_rating_comparison_chart(self, ax): """创建评分对比图""" try: ratings = [] names = [] for data in self.competitor_data: basic_info = data.get('basic_info', {}) if basic_info.get('rating'): ratings.append(basic_info['rating']) names.append(data['competitor_name']) if ratings: ax.bar(names, ratings, color='lightgreen') ax.set_title('店铺评分对比') ax.tick_params(axis='x', rotation=45) ax.set_ylim(0, 5) except Exception as e: Logger.error(f"❌ 创建评分对比图失败: {str(e)}") def _generate_swot_analysis(self): """生成SWOT分析""" swot_analysis = {} try: # 分析我们的优势、劣势、机会、威胁 top_competitors = self.analysis_results['competitive_ranking'][:3] # 优势（相对于竞品） strengths = [] # 劣势 weaknesses = [] # 机会（市场机会） opportunities = [] # 威胁（竞品威胁） threats = [] # 基于数据分析生成SWOT if top_competitors: our_data = next((comp for comp in top_competitors if comp['competitor_name'] == 'Zozone'), None) if our_data: # 优势分析 if our_data['price_score'] > 0.7: strengths.append("价格竞争力强") if our_data['product_score'] > 0.7: strengths.append("产品线丰富") if our_data['reputation_score'] > 0.7: strengths.append("用户口碑良好") # 劣势分析 if our_data['marketing_score'] < 0.5: weaknesses.append("营销投入不足") if our_data['price_score'] < 0.5: weaknesses.append("价格优势不明显") # 机会分析 opportunities.extend([ "市场增长空间大", "新用户获取成本较低", "产品创新机会多" ]) # 威胁分析 for comp in top_competitors: if comp['competitor_name'] != 'Zozone': if comp['total_score'] > our_data['total_score']: threats.append(f"{comp['competitor_name']}综合竞争力更强") if comp['price_score'] > our_data['price_score']: threats.append(f"{comp['competitor_name']}价格更具竞争力") swot_analysis = { 'strengths': strengths, 'weaknesses': weaknesses, 'opportunities': opportunities, 'threats': threats } except Exception as e: Logger.error(f"❌ 生成SWOT分析失败: {str(e)}") return swot_analysis def _generate_strategic_recommendations(self): """生成战略建议""" recommendations = [] try: if 'competitive_ranking' not in self.analysis_results: return recommendations our_data = next((comp for comp in self.analysis_results['competitive_ranking'] if comp['competitor_name'] == 'Zozone'), None) if not our_data: return recommendations # 基于得分生成针对性建议 if our_data['price_score'] < 0.6: recommendations.append("优化价格策略，提升价格竞争力") if our_data['product_score'] < 0.6: recommendations.append("丰富产品线，加强产品创新") if our_data['marketing_score'] < 0.6: recommendations.append("加大营销投入，提升品牌曝光") if our_data['reputation_score'] < 0.6: recommendations.append("加强客户服务，提升用户满意度") # 针对最强竞品的建议 top_competitor = self.analysis_results['competitive_ranking'][0] if top_competitor['competitor_name'] != 'Zozone': recommendations.append(f"重点关注{top_competitor['competitor_name']}的动态，学习其成功经验") except Exception as e: Logger.error(f"❌ 生成战略建议失败: {str(e)}") return recommendations def _compile_complete_report(self, swot_analysis, strategic_recommendations): """编译完整报告""" try: report = [] report.append("🎯 Zozone竞品分析报告") report.append("=" * 50) report.append(f"📅 报告生成时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") report.append(f"🔍 分析竞品数量: {len(self.competitor_data)}") report.append("") # 竞争力排名 report.append("🏆 竞争力排名") report.append("-" * 30) if 'competitive_ranking' in self.analysis_results: for comp in self.analysis_results['competitive_ranking']: report.append(f"{comp['ranking']}. {comp['competitor_name']} - 综合得分: {comp['total_score']:.3f}") report.append("") # SWOT分析 report.append("📋 SWOT分析") report.append("-" * 30) report.append("优势 (Strengths):") for strength in swot_analysis.get('strengths', []): report.append(f" ✓ {strength}") report.append("劣势 (Weaknesses):") for weakness in swot_analysis.get('weaknesses', []): report.append(f" ✗ {weakness}") report.append("机会 (Opportunities):") for opportunity in sw
