import sqlite3 import os import csv from collections import defaultdict from pathlib import Path def import_match_results(): """导入比赛结果数据""" # 连接数据库 conn = sqlite3.connect('footviz.db') cursor = conn.cursor() # 定义国家映射 country_mapping = { 'uk': 1, # England 'spain': 2, # Spain 'germany': 3, # Germany 'france': 4 # France (假设France对应数据库中的China位置，需要根据实际情况调整) } base_dir = Path('data/matches') match_count = 0 # 遍历所有国家目录 for country_dir in base_dir.iterdir(): if not country_dir.is_dir(): continue country_code = country_mapping.get(country_dir.name) if not country_code: print(f"未知的国家目录: {country_dir.name}") continue print(f"处理 {country_dir.name} 的比赛数据...") # 遍历该国家下的所有球队CSV文件 for csv_file in country_dir.iterdir(): if not csv_file.is_file() or not csv_file.name.endswith('.csv'): continue team_name = csv_file.stem # 文件名即为球队名 # 查找球队ID cursor.execute("SELECT id FROM api_team WHERE name = ?", (team_name,)) team_result = cursor.fetchone() if not team_result: print(f" 警告: 数据库中未找到球队 {team_name}") continue team_id = team_result[0] # 读取比赛数据 try: with open(csv_file, 'r', encoding='utf-8') as f: reader = csv.reader(f) next(reader) # 跳过表头 for row in reader: if len(row) < 6: continue result_code = row[0] # 比赛结果代码 competition = row[1] # 联赛/杯赛名称 match_date = row[2] # 比赛日期 team_a = row[3] # 主队 score = row[4] # 比分 team_b = row[5] # 客队 # 只处理英超联赛数据 if competition != 'EPL': continue # 解析比分 if ':' in score: try: goals_a, goals_b = map(int, score.split(':')) except ValueError: continue # 无法解析比分，跳过 else: continue # 无效比分格式，跳过 # 确定主客队ID cursor.execute("SELECT id FROM api_team WHERE name = ?", (team_a,)) team_a_result = cursor.fetchone() if not team_a_result: continue team_a_id = team_a_result[0] cursor.execute("SELECT id FROM api_team WHERE name = ?", (team_b,)) team_b_result = cursor.fetchone() if not team_b_result: continue team_b_id = team_b_result[0] # 确定比赛结果 result = None if goals_a > goals_b: result = 'W' # 主队胜利 elif goals_a < goals_b: result = 'L' # 主队失败 else: result = 'D' # 平局 # 插入比赛结果 cursor.execute(""" INSERT OR IGNORE INTO api_matchresult (country_id, home_team_id, away_team_id, match_date, home_goals, away_goals, result) VALUES (?, ?, ?, ?, ?, ?, ?) """, (country_code, team_a_id, team_b_id, match_date, goals_a, goals_b, result)) match_count += 1 except Exception as e: print(f" 处理文件 {csv_file.name} 时出错: {e}") conn.commit() conn.close() print(f"成功导入 {match_count} 条比赛结果记录") return match_count def calculate_league_standings(): """根据比赛结果计算联赛积分榜""" # 连接数据库 conn = sqlite3.connect('footviz.db') cursor = conn.cursor() # 获取所有比赛结果 cursor.execute(""" SELECT mr.home_team_id, mr.away_team_id, mr.home_goals, mr.away_goals, t1.country_id FROM api_matchresult mr JOIN api_team t1 ON mr.home_team_id = t1.id WHERE mr.result IS NOT NULL ORDER BY mr.match_date """) matches = cursor.fetchall() if not matches: print("没有找到比赛结果数据，无法生成积分榜") conn.close() return # 为每个国家/联赛初始化积分榜数据 standings = defaultdict(lambda: defaultdict(lambda: { 'position': 0, 'played': 0, 'won': 0, 'drawn': 0, 'lost': 0, 'goals_for': 0, 'goals_against': 0, 'goal_difference': 0, 'points': 0, 'recent_form': [] })) # 处理每场比赛 for home_team_id, away_team_id, home_goals, away_goals, country_id in matches: if home_goals is None or away_goals is None: continue # 更新场次统计 standings[country_id][home_team_id]['played'] += 1 standings[country_id][away_team_id]['played'] += 1 # 更新进球统计 standings[country_id][home_team_id]['goals_for'] += home_goals standings[country_id][home_team_id]['goals_against'] += away_goals standings[country_id][away_team_id]['goals_for'] += away_goals standings[country_id][away_team_id]['goals_against'] += home_goals # 计算比赛结果和积分 if home_goals > away_goals: # 主队胜利 standings[country_id][home_team_id]['won'] += 1 standings[country_id][home_team_id]['points'] += 3 standings[country_id][home_team_id]['recent_form'].append('W') standings[country_id][away_team_id]['lost'] += 1 standings[country_id][away_team_id]['recent_form'].append('L') elif home_goals < away_goals: # 客队胜利 standings[country_id][away_team_id]['won'] += 1 standings[country_id][away_team_id]['points'] += 3 standings[country_id][away_team_id]['recent_form'].append('W') standings[country_id][home_team_id]['lost'] += 1 standings[country_id][home_team_id]['recent_form'].append('L') else: # 平局 standings[country_id][home_team_id]['drawn'] += 1 standings[country_id][home_team_id]['points'] += 1 standings[country_id][home_team_id]['recent_form'].append('D') standings[country_id][away_team_id]['drawn'] += 1 standings[country_id][away_team_id]['points'] += 1 standings[country_id][away_team_id]['recent_form'].append('D') # 保留最近5场比赛结果 if len(standings[country_id][home_team_id]['recent_form']) > 5: standings[country_id][home_team_id]['recent_form'].pop(0) if len(standings[country_id][away_team_id]['recent_form']) > 5: standings[country_id][away_team_id]['recent_form'].pop(0) # 计算净胜球 for country_id in standings: for team_id in standings[country_id]: standings[country_id][team_id]['goal_difference'] = ( standings[country_id][team_id]['goals_for'] - standings[country_id][team_id]['goals_against'] ) # 为每个联赛排序 sorted_standings = {} for country_id in standings: # 按积分、净胜球、进球数排序 sorted_teams = sorted( standings[country_id].items(), key=lambda x: ( x[1]['points'], x[1]['goal_difference'], x[1]['goals_for'] ), reverse=True ) # 设置排名 for position, (team_id, stats) in enumerate(sorted_teams, 1): stats['position'] = position # 将最近比赛结果转换为字符串 stats['recent_form'] = ''.join(stats['recent_form']) sorted_standings[country_id] = sorted_teams # 插入数据到数据库 cursor.execute("DELETE FROM api_leaguestanding") record_count = 0 for country_id in sorted_standings: for team_id, stats in sorted_standings[country_id]: cursor.execute(""" INSERT INTO api_leaguestanding (team_id, country_id, position, played, won, drawn, lost, goals_for, goals_against, goal_difference, points, recent_form) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( team_id, country_id, stats['position'], stats['played'], stats['won'], stats['drawn'], stats['lost'], stats['goals_for'], stats['goals_against'], stats['goal_difference'], stats['points'], stats['recent_form'] )) record_count += 1 conn.commit() conn.close() print(f"成功生成 {record_count} 条联赛积分榜记录") return sorted_standings def main(): """主函数""" print("开始导入比赛结果数据...") import_match_results() print("开始生成联赛积分榜...") calculate_league_standings() if __name__ == "__main__": main()