temu跨境电商商家电话批量采集分析

import pandas as pd
import matplotlib.pyplot as plt
import seaborn as sns
import numpy as np
from collections import Counter
import re
import os
from geopy.geocoders import Nominatim
from geopy.extra.rate_limiter import RateLimiter
import folium
from folium.plugins import HeatMap
class TemuDataAnalyzer:
    def __init__(self, data_file):
        """初始化分析器并加载数据"""
        self.data = self._load_data(data_file)
        self.geolocator = Nominatim(user_agent="temu_analysis")
        self.geocode = RateLimiter(self.geolocator.geocode, min_delay_seconds=1)
        
    def _load_data(self, data_file):
        """加载数据文件"""
        file_ext = os.path.splitext(data_file)[1].lower()
        
        if file_ext == '.csv':
            return pd.read_csv(data_file)
        elif file_ext == '.json':
            return pd.read_json(data_file)
        else:
            raise ValueError(f"不支持的文件格式: {file_ext}")
    
    def clean_data(self):
        """数据清洗和预处理"""
        # 去除重复记录
        self.data.drop_duplicates(subset=['seller_name', 'seller_url'], inplace=True)
        
        # 处理缺失值
        self.data['seller_rating'] = self.data['seller_rating'].fillna('0')
        
        # 提取评分数值
        self.data['rating_value'] = self.data['seller_rating'].str.extract(r'(\d+\.?\d*)').astype(float)
        
        # 提取联系方式类型
        def extract_contact_type(contacts):
            if pd.isna(contacts):
                return []
                
            if isinstance(contacts, str):
                try:
                    contacts = eval(contacts)  # 将字符串转为列表
                except:
                    return []
                    
            types = []
            for contact in contacts:
                if re.match(r'^[\w\.-]+@[\w\.-]+$', contact):
                    types.append('email')
                elif re.match(r'^\+?\d{1,3}\s?\d{3,4}\s?\d{3,4}\s?\d{3,4}$', contact):
                    types.append('phone')
                else:
                    types.append('other')
            return types
        
        self.data['contact_types'] = self.data['contact'].apply(extract_contact_type)
        
        # 提取是否有电话联系方式
        self.data['has_phone'] = self.data['contact_types'].apply(lambda x: 'phone' in x)
        
        # 提取产品类别
        def extract_category(url):
            if pd.isna(url):
                return 'unknown'
                
            categories = ['electronics', 'clothing', 'home', 'beauty', 'toys', 'sports', 'jewelry', 'shoes']
            url_lower = url.lower()
            
            for cat in categories:
                if cat in url_lower:
                    return cat
            return 'other'
            
        self.data['product_category'] = self.data['product_url'].apply(extract_category)
        
        print(f"数据清洗完成，剩余有效记录: {len(self.data)}")
        return self.data
    
    def analyze_contact_methods(self):
        """分析联系方式分布"""
        contact_counts = Counter()
        
        for contact_types in self.data['contact_types']:
            for contact_type in contact_types:
                contact_counts[contact_type] += 1
        
        # 计算有联系方式的卖家比例
        sellers_with_contact = len(self.data[self.data['contact_types'].apply(len) > 0])
        total_sellers = len(self.data)
        contact_percentage = (sellers_with_contact / total_sellers) * 100
        
        print(f"联系方式分布: {contact_counts}")
        print(f"有联系方式的卖家比例: {contact_percentage:.2f}%")
        
        # 可视化
        plt.figure(figsize=(10, 6))
        sns.barplot(x=list(contact_counts.keys()), y=list(contact_counts.values()))
        plt.title('卖家联系方式分布')
        plt.xlabel('联系方式类型')
        plt.ylabel('卖家数量')
        plt.xticks(rotation=45)
        plt.tight_layout()
        plt.savefig('contact_methods_distribution.png')
        plt.close()
        
        return {
            'contact_distribution': contact_counts,
            'contact_percentage': contact_percentage
        }
    
    def analyze_seller_ratings(self):
        """分析卖家评分分布"""
        # 过滤无效评分
        valid_ratings = self.data[self.data['rating_value'] > 0]['rating_value']
        
        if valid_ratings.empty:
            print("没有有效的评分数据")
            return
            
        # 计算统计量
        rating_stats = {
            'mean': valid_ratings.mean(),
            'median': valid_ratings.median(),
            'min': valid_ratings.min(),
            'max': valid_ratings.max(),
            'std': valid_ratings.std()
        }
        
        print(f"卖家评分统计: {rating_stats}")
        
        # 可视化
        plt.figure(figsize=(10, 6))
        sns.histplot(valid_ratings, kde=True, bins=20)
        plt.axvline(rating_stats['mean'], color='red', linestyle='dashed', linewidth=2, label='Mean')
        plt.axvline(rating_stats['median'], color='green', linestyle='dashed', linewidth=2, label='Median')
        plt.title('卖家评分分布')
        plt.xlabel('评分')
        plt.ylabel('卖家数量')
        plt.legend()
        plt.tight_layout()
        plt.savefig('seller_ratings_distribution.png')
        plt.close()
        
        return rating_stats
    
    def analyze_categories(self):
        """分析不同类别的卖家分布"""
        category_counts = self.data['product_category'].value_counts().to_dict()
        
        print(f"卖家类别分布: {category_counts}")
        
        # 可视化
        plt.figure(figsize=(12, 7))
        sns.barplot(x=list(category_counts.keys()), y=list(category_counts.values()))
        plt.title('卖家产品类别分布')
        plt.xlabel('产品类别')
        plt.ylabel('卖家数量')
        plt.xticks(rotation=45)
        plt.tight_layout()
        plt.savefig('product_categories_distribution.png')
        plt.close()
        
        # 分析不同类别卖家的联系方式比例
        category_contact = {}
        for category in category_counts.keys():
            cat_data = self.data[self.data['product_category'] == category]
            contact_ratio = len(cat_data[cat_data['has_phone']]) / len(cat_data) * 100
            category_contact[category] = contact_ratio
        
        print(f"不同类别卖家电话联系方式比例: {category_contact}")
        
        return {
            'category_distribution': category_counts,
            'category_contact_ratio': category_contact
        }
    
    def geocode_addresses(self, address_column='seller_name'):
        """地理编码卖家地址"""
        # 为简化示例，这里假设卖家名称包含地址信息
        # 实际应用中应使用真实地址数据
        
        locations = []
        for index, row in self.data.iterrows():
            address = row[address_column]
            if pd.notna(address):
                try:
                    location = self.geocode(address)
                    if location:
                        locations.append({
                            'seller_name': row['seller_name'],
                            'latitude': location.latitude,
                            'longitude': location.longitude,
                            'address': location.address
                        })
                except Exception as e:
                    print(f"地理编码出错: {e}")
        
        # 保存地理位置数据
        geo_df = pd.DataFrame(locations)
        geo_df.to_csv('seller_locations.csv', index=False)
        
        return geo_df
    
    def create_heatmap(self, locations_df):
        """创建卖家分布热力图"""
        # 创建基础地图，以平均位置为中心
        if locations_df.empty:
            print("没有位置数据，无法创建热力图")
            return
            
        avg_lat = locations_df['latitude'].mean()
        avg_lon = locations_df['longitude'].mean()
        
        m = folium.Map(location=[avg_lat, avg_lon], zoom_start=2)
        
        # 添加热力层
        heat_data = [[row['latitude'], row['longitude']] for index, row in locations_df.iterrows()]
        HeatMap(heat_data, radius=15).add_to(m)
        
        # 保存地图
        m.save('seller_heatmap.html')
        print("热力图已保存为 seller_heatmap.html")
        return m
# 使用示例
if __name__ == "__main__":
    analyzer = TemuDataAnalyzer("temu_sellers.csv")
    cleaned_data = analyzer.clean_data()
    
    # 进行各项分析
    contact_analysis = analyzer.analyze_contact_methods()
    rating_analysis = analyzer.analyze_seller_ratings()
    category_analysis = analyzer.analyze_categories()
    
    # 地理编码和热力图（示例中使用卖家名称作为地址，实际应使用真实地址）
    locations = analyzer.geocode_addresses()
    if not locations.empty:
        analyzer.create_heatmap(locations)
    
    # 保存分析结果
    analysis_results = {
        'contact_analysis': contact_analysis,
        'rating_analysis': rating_analysis,
        'category_analysis': category_analysis
    }
    
    with open('analysis_results.json', 'w') as f:
        json.dump(analysis_results, f, indent=2)
    
    print("数据分析完成，结果已保存")