实用脚本统计SofaScore综合评分？

我将为您创建一个实用的SofaScore综合评分统计脚本,这个脚本可以批量抓取球员评分数据并进行统计分析。

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
SofaScore综合评分统计脚本
用于抓取和分析球员的SofaScore评分数据
"""
import requests
import json
import time
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
from typing import Dict, List, Optional
import logging
from collections import defaultdict
# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
class SofaScoreAnalyzer:
    """SofaScore评分统计分析器"""
    def __init__(self):
        self.base_url = "https://api.sofascore.com/api/v1"
        self.headers = {
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
            'Accept': 'application/json',
            'Accept-Language': 'en-US,en;q=0.9',
        }
        self.session = requests.Session()
        self.session.headers.update(self.headers)
        self.rate_limit_delay = 1.0  # API请求间隔（秒）
    def get_player_stats(self, player_id: int, season: str = "2023") -> Optional[Dict]:
        """
        获取球员赛季统计数据
        Args:
            player_id: SofaScore球员ID
            season: 赛季年份
        Returns:
            球员统计数据字典，失败返回None
        """
        try:
            url = f"{self.base_url}/player/{player_id}/unique-tournament/17/season/{season}/statistics"
            response = self.session.get(url)
            if response.status_code == 200:
                return response.json()
            else:
                logger.warning(f"获取球员 {player_id} 数据失败，状态码: {response.status_code}")
                return None
        except Exception as e:
            logger.error(f"请求失败 (球员 {player_id}): {str(e)}")
            return None
    def get_player_matches(self, player_id: int, limit: int = 20) -> List[Dict]:
        """
        获取球员最近比赛评分
        Args:
            player_id: SofaScore球员ID
            limit: 比赛场次限制
        Returns:
            比赛数据列表
        """
        try:
            url = f"{self.base_url}/player/{player_id}/events/last/{limit}"
            response = self.session.get(url)
            if response.status_code == 200:
                events = response.json().get('events', [])
                match_data = []
                for event in events:
                    try:
                        # 获取该场比赛的具体数据
                        event_id = event['id']
                        stats_url = f"{self.base_url}/event/{event_id}/statistics"
                        stats_response = self.session.get(stats_url)
                        if stats_response.status_code == 200:
                            event_stats = stats_response.json()
                            # 提取球员评分（根据球队ID匹配）
                            home_team_id = event['homeTeam']['id']
                            away_team_id = event['awayTeam']['id']
                            player_rating = self._extract_player_rating(
                                event_stats, 
                                home_team_id, 
                                away_team_id, 
                                player_id
                            )
                            if player_rating is not None:
                                match_data.append({
                                    'match_id': event_id,
                                    'date': event.get('startTimestamp'),
                                    'opponent': self._get_opponent_name(event, player_id, home_team_id, away_team_id),
                                    'rating': player_rating,
                                    'result': self._get_match_result(event, home_team_id, away_team_id)
                                })
                        time.sleep(self.rate_limit_delay * 0.5)  # 限制请求频率
                    except Exception as e:
                        logger.error(f"处理比赛 {event.get('id', 'unknown')} 时出错: {str(e)}")
                        continue
                return match_data
        except Exception as e:
            logger.error(f"获取球员 {player_id} 比赛数据失败: {str(e)}")
            return []
    def _extract_player_rating(self, event_stats: Dict, home_team_id: int, 
                              away_team_id: int, player_id: int) -> Optional[float]:
        """
        从比赛统计数据中提取球员评分
        """
        try:
            statistics = event_stats.get('statistics', [])
            # 遍历球队统计
            for team_stats in statistics:
                team_id = team_stats.get('teamId')
                # 查找包含球员评分的组
                for group in team_stats.get('groups', []):
                    for item in group.get('statisticsItems', []):
                        if 'rating' in item.get('name', '').lower():
                            # 尝试从详细数据中获取球员评分
                            for detail in item.get('details', []):
                                if detail.get('playerId') == player_id:
                                    rating = detail.get('value')
                                    if rating and isinstance(rating, (int, float)):
                                        return float(rating)
        except Exception as e:
            logger.debug(f"提取评分失败: {str(e)}")
            return None
    def _get_opponent_name(self, event: Dict, player_id: int, 
                          home_team_id: int, away_team_id: int) -> str:
        """获取对手球队名称"""
        try:
            if home_team_id == player_id or away_team_id == player_id:
                return "Self"  # 防御性检查
            # 判断球员属于哪个队，返回对手名称
            if event['homeTeam']['id'] == home_team_id:
                # 检查球员是否在主队
                if 'players' in event.get('homeTeam', {}):
                    player_ids = [p['id'] for p in event['homeTeam']['players']]
                    if player_id in player_ids:
                        return event['awayTeam']['name']
            return event['homeTeam']['name'] if away_team_id != event['awayTeam']['id'] else event['awayTeam']['name']
        except Exception as e:
            logger.error(f"获取对手名称失败: {str(e)}")
            return "Unknown"
    def _get_match_result(self, event: Dict, home_team_id: int, away_team_id: int) -> str:
        """获取比赛结果"""
        try:
            home_score = event.get('homeScore', {}).get('current', 0)
            away_score = event.get('awayScore', {}).get('current', 0)
            if home_score > away_score:
                return f"W {home_score}-{away_score}"
            elif home_score < away_score:
                return f"L {home_score}-{away_score}"
            else:
                return f"D {home_score}-{away_score}"
        except:
            return "N/A"
    def analyze_player_ratings(self, player_ids: List[int], player_names: List[str], 
                              season: str = "2023") -> pd.DataFrame:
        """
        分析多个球员的评分数据
        Args:
            player_ids: 球员ID列表
            player_names: 球员名称列表
            season: 赛季
        Returns:
            统计结果DataFrame
        """
        all_data = []
        for player_id, player_name in zip(player_ids, player_names):
            logger.info(f"正在分析球员: {player_name} (ID: {player_id})")
            # 获取赛季统计数据
            season_stats = self.get_player_stats(player_id, season)
            # 获取最近比赛评分
            matches = self.get_player_matches(player_id, limit=20)
            if matches:
                ratings = [m['rating'] for m in matches if m['rating'] is not None]
                if ratings:
                    player_data = {
                        'player_name': player_name,
                        'player_id': player_id,
                        'matches_played': len(ratings),
                        'average_rating': np.mean(ratings),
                        'highest_rating': max(ratings),
                        'lowest_rating': min(ratings),
                        'std_deviation': np.std(ratings),
                        'last_5_avg': np.mean(ratings[-5:]) if len(ratings) >= 5 else np.mean(ratings),
                        'recent_form': self._calculate_form_rating(ratings)
                    }
                    # 添加赛季统计信息（如果可用）
                    if season_stats:
                        player_data.update(self._extract_season_stats(season_stats))
                    # 计算稳定性指标
                    player_data['consistency'] = self._calculate_consistency(ratings)
                    all_data.append(player_data)
            time.sleep(self.rate_limit_delay)
        return pd.DataFrame(all_data)
    def _extract_season_stats(self, stats: Dict) -> Dict:
        """提取赛季统计数据"""
        season_data = {}
        try:
            # 根据实际API返回结构调整
            if 'statistics' in stats:
                for stat in stats['statistics']:
                    name = stat.get('name', '')
                    value = stat.get('value', 0)
                    if 'goals' in name.lower():
                        season_data['season_goals'] = value
                    elif 'assists' in name.lower():
                        season_data['season_assists'] = value
                    elif 'yellow' in name.lower():
                        season_data['yellow_cards'] = value
                    elif 'red' in name.lower():
                        season_data['red_cards'] = value
        except Exception as e:
            logger.debug(f"提取赛季数据失败: {str(e)}")
        return season_data
    def _calculate_form_rating(self, ratings: List[float]) -> float:
        """计算近期状态（最后3场加权平均）"""
        if len(ratings) <= 3:
            return np.mean(ratings)
        recent_3 = ratings[-3:]
        weights = [0.2, 0.3, 0.5]  # 最近一场权重最高
        return sum(r * w for r, w in zip(recent_3, weights)) / sum(weights)
    def _calculate_consistency(self, ratings: List[float]) -> float:
        """计算球员稳定性（评分变异系数）"""
        mean_rating = np.mean(ratings)
        if mean_rating == 0:
            return 0
        std = np.std(ratings)
        cv = (std / mean_rating) * 100  # 变异系数百分比
        # 转换为稳定性分数（0-100分）
        consistency = max(0, 100 - cv * 5)
        return round(consistency, 1)
    def generate_report(self, df: pd.DataFrame) -> str:
        """
        生成统计报告
        Args:
            df: 球员评分统计DataFrame
        Returns:
            格式化的报告文本
        """
        if df.empty:
            return "没有可用的统计数据。"
        report_lines = []
        report_lines.append("=" * 60)
        report_lines.append("SOFASCORE球员评分统计报告")
        report_lines.append(f"生成时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
        report_lines.append("=" * 60)
        report_lines.append("")
        # 总体统计
        report_lines.append("【总体统计】")
        report_lines.append(f"球员数量: {len(df)}")
        report_lines.append(f"平均评分: {df['average_rating'].mean():.2f}")
        report_lines.append(f"最高平均评分: {df.loc[df['average_rating'].idxmax(), 'player_name']} "
                          f"({df['average_rating'].max():.2f})")
        report_lines.append(f"最低平均评分: {df.loc[df['average_rating'].idxmin(), 'player_name']} "
                          f"({df['average_rating'].min():.2f})")
        report_lines.append("")
        # 球员排名
        report_lines.append("【球员评分排名】")
        df_sorted = df.sort_values('average_rating', ascending=False)
        for idx, row in df_sorted.iterrows():
            report_lines.append(f"第{df_sorted.index.tolist().index(idx)+1}名: "
                              f"{row['player_name']} - 平均评分 {row['average_rating']:.2f} "
                              f"(最佳: {row['highest_rating']:.2f}, 最差: {row['lowest_rating']:.2f})")
        report_lines.append("")
        # 状态分析
        report_lines.append("【近期状态分析】")
        for _, row in df_sorted.iterrows():
            form_status = "状态优秀" if row['last_5_avg'] >= 7.5 else \
                         "状态良好" if row['last_5_avg'] >= 7.0 else \
                         "状态一般" if row['last_5_avg'] >= 6.5 else "状态低迷"
            report_lines.append(f"{row['player_name']}: "
                              f"近5场 {row['last_5_avg']:.2f} ({form_status}), "
                              f"稳定性 {row['consistency']:.1f}%")
        # 稳定性排名
        report_lines.append("")
        report_lines.append("【稳定性排名】")
        df_consistency = df.sort_values('consistency', ascending=False)
        for idx, row in df_consistency.iterrows():
            report_lines.append(f"{row['player_name']}: 稳定性得分 {row['consistency']:.1f}")
        return "\n".join(report_lines)
def main():
    """主函数 - 示例用法"""
    # 示例球员数据（需要替换为实际的SofaScore球员ID）
    players = [
        {"id": 882017, "name": "示例球员1"},  # 替换为实际ID
        {"id": 882018, "name": "示例球员2"},
        {"id": 882019, "name": "示例球员3"},
    ]
    analyzer = SofaScoreAnalyzer()
    try:
        # 分析球员数据
        df = analyzer.analyze_player_ratings(
            [p["id"] for p in players],
            [p["name"] for p in players],
            season="2023"
        )
        if not df.empty:
            # 输出统计报告
            report = analyzer.generate_report(df)
            print(report)
            # 保存到CSV
            df.to_csv('sofascore_ratings_analysis.csv', index=False, encoding='utf-8-sig')
            logger.info("统计结果已保存到 sofascore_ratings_analysis.csv")
            # 保存报告到文件
            with open('sofascore_report.txt', 'w', encoding='utf-8') as f:
                f.write(report)
            logger.info("报告已保存到 sofascore_report.txt")
        else:
            logger.warning("未获取到有效数据，请检查球员ID是否正确")
    except Exception as e:
        logger.error(f"程序执行错误: {str(e)}")
        logger.error("请确保网络连接正常，并且球员ID有效")
if __name__ == "__main__":
    main()

这里提供一个简化版的控制台交互脚本,方便快速使用：

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
SofaScore评分统计 - 简易命令行版本
"""
import requests
import json
from collections import defaultdict
from datetime import datetime
def get_player_ratings(player_id, days=30):
    """
    获取球员最近评分数据
    Args:
        player_id: SofaScore球员ID
        days: 统计天数
    Returns:
        dict: 包含评分统计信息
    """
    headers = {
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64)'
    }
    # 这里使用示例数据演示（实际使用时需要调用真实API）
    # 注意：实际使用需要处理API限流和数据格式
    demo_data = {
        'player_id': player_id,
        'player_name': '示例球员',
        'matches': [
            {'date': '2024-01-01', 'rating': 7.5},
            {'date': '2024-01-05', 'rating': 8.0},
            {'date': '2024-01-10', 'rating': 6.8},
            {'date': '2024-01-15', 'rating': 7.2},
            {'date': '2024-01-20', 'rating': 8.5},
            {'date': '2024-01-25', 'rating': 7.8},
        ]
    }
    # 计算统计数据
    ratings = [m['rating'] for m in demo_data['matches']]
    stats = {
        '球员ID': player_id,
        '比赛场次': len(ratings),
        '平均评分': round(sum(ratings) / len(ratings), 2),
        '最高评分': max(ratings),
        '最低评分': min(ratings),
        '最近3场平均': round(sum(ratings[-3:]) / 3, 2),
        '数据截止': datetime.now().strftime('%Y-%m-%d %H:%M')
    }
    return stats
def main():
    """简易命令行界面"""
    print("=" * 50)
    print("SofaScore球员评分统计工具")
    print("=" * 50)
    while True:
        print("\n请选择操作:")
        print("1. 查询球员评分统计")
        print("2. 退出")
        choice = input("输入选择 (1/2): ").strip()
        if choice == "1":
            player_id = input("请输入SofaScore球员ID: ").strip()
            if player_id.isdigit():
                stats = get_player_ratings(int(player_id))
                print("\n" + "=" * 50)
                print("球员评分统计结果:")
                print("=" * 50)
                for key, value in stats.items():
                    print(f"{key}: {value}")
                print("=" * 50)
            else:
                print("错误：请输入有效的球员ID")
        elif choice == "2":
            print("感谢使用，再见！")
            break
        else:
            print("无效选择，请重新输入")
if __name__ == "__main__":
    main()

使用说明

安装依赖

pip install requests pandas numpy

主要功能

1. 批量球员评分统计：统计多名球员的比赛评分数据
2. 近期状态分析：计算球员最近5场的表现趋势
3. 稳定性评估：分析球员评分的波动情况
4. 比赛数据对比：获取球员特定场次的详细评分

使用方法

1. 获取球员ID：在SofaScore网站搜索球员，URL中的数字就是球员ID

2. 修改配置：替换脚本中的示例球员数据为实际球员ID

3. 运行脚本：

   python sofascore_stats.py

4. 查看结果：

   • 控制台输出统计报告
   • CSV文件保存详细数据
   • 报告文件保存综合分析

注意事项

1. API限制：请控制请求频率，避免被封IP
2. 数据准确性：请确认为正式赛季数据
3. 合规性：建议遵守网站使用条款，不要频繁请求

如需真实数据，建议使用SofaScore官方API或考虑使用爬虫方案，这个脚本提供了一个完整的框架,您可以根据实际需要调整和扩展。