""" 重新创建和初始化SQLite数据库的脚本 此脚本将删除现有的数据库并重新创建所有表结构，然后导入数据 """ import sqlite3 import csv import os from pathlib import Path # 定义路径 BASE_DIR = Path(__file__).resolve().parent.parent.parent.parent data_dir = BASE_DIR / 'footviz' / 'data' db_path = data_dir / 'football.db' def drop_and_create_tables(conn): """删除现有表并重新创建表结构""" cursor = conn.cursor() # 删除现有表 tables = ['LeagueStandings', 'MatchResults', 'PlayerStats', 'Players', 'Teams', 'Countries'] for table in tables: try: cursor.execute(f"DROP TABLE IF EXISTS {table}") print(f"已删除表 {table}") except sqlite3.Error as e: print(f"删除表 {table} 时出错: {e}") # 创建新表 create_tables_sql = """ -- 国家表 CREATE TABLE Countries ( id INTEGER PRIMARY KEY, name TEXT NOT NULL, code TEXT NOT NULL, league_name TEXT, UNIQUE(name, code) ); -- 球队表 CREATE TABLE Teams ( id INTEGER PRIMARY KEY, name TEXT UNIQUE NOT NULL, country_code TEXT, FOREIGN KEY (country_code) REFERENCES Countries(code) ); -- 球员表 CREATE TABLE Players ( id INTEGER PRIMARY KEY, name TEXT UNIQUE NOT NULL, age INTEGER, date_of_birth TEXT, height INTEGER, nationality TEXT, positions TEXT, image_url TEXT, FOREIGN KEY (nationality) REFERENCES Countries(code) ); -- 球员数据表 CREATE TABLE PlayerStats ( id INTEGER PRIMARY KEY, player_id INTEGER, team_id INTEGER, tournament TEXT, -- 添加tournament字段 appearances TEXT, minutes_played INTEGER, goals INTEGER, assists INTEGER, tackles REAL, interceptions REAL, fouls_committed REAL, pass_success_rate REAL, shots_per_game REAL, dribbles_per_game REAL, rating REAL, FOREIGN KEY (player_id) REFERENCES Players(id), FOREIGN KEY (team_id) REFERENCES Teams(id) ); -- 比赛结果表（避免 Matches 关键字） CREATE TABLE MatchResults ( id INTEGER PRIMARY KEY, country_code TEXT, home_team_id INTEGER, away_team_id INTEGER, match_date TEXT, home_goals INTEGER, away_goals INTEGER, result TEXT, FOREIGN KEY (country_code) REFERENCES Countries(code), FOREIGN KEY (home_team_id) REFERENCES Teams(id), FOREIGN KEY (away_team_id) REFERENCES Teams(id) ); -- 联赛积分表 CREATE TABLE LeagueStandings ( id INTEGER PRIMARY KEY, team_id INTEGER, country_code TEXT, position INTEGER, played INTEGER, won INTEGER, drawn INTEGER, lost INTEGER, goals_for INTEGER, goals_against INTEGER, goal_difference INTEGER, points INTEGER, recent_form TEXT, FOREIGN KEY (team_id) REFERENCES Teams(id), FOREIGN KEY (country_code) REFERENCES Countries(code) ); """ try: cursor.executescript(create_tables_sql) print("所有表已成功创建") except sqlite3.Error as e: print(f"创建表时出错: {e}") conn.commit() def insert_countries_data(conn): """插入国家数据""" cursor = conn.cursor() # 国家代码映射 country_codes = { "England": "ENG", "France": "FRA", "Germany": "GER", "Spain": "ESP", "China": "CHN" } print("插入国家数据...") for country_name, country_code in country_codes.items(): try: cursor.execute( "INSERT OR IGNORE INTO Countries (name, code) VALUES (?, ?)", (country_name, country_code) ) except sqlite3.Error as e: print(f"插入国家 {country_name} 时出错: {e}") conn.commit() print("国家数据插入完成") def insert_teams_data(conn): """插入球队数据""" cursor = conn.cursor() countries_dir = data_dir / 'countries' if not countries_dir.exists(): print(f"目录 {countries_dir} 不存在") return print("收集所有球队名称...") all_teams = set() # 遍历所有国家目录，收集球队名称 for country_dir in countries_dir.iterdir(): if country_dir.is_dir(): for team_dir in country_dir.iterdir(): if team_dir.is_dir(): team_name = team_dir.name all_teams.add(team_name) print(f"总共发现 {len(all_teams)} 支球队") # 国家代码映射 country_codes = { "England": "ENG", "France": "FRA", "Germany": "GER", "Spain": "ESP", "China": "CHN" } print("插入球队数据...") for team_name in all_teams: # 确定球队所属国家 country_code = None for country_dir in countries_dir.iterdir(): if country_dir.is_dir(): for team_dir in country_dir.iterdir(): if team_dir.is_dir() and team_dir.name == team_name: country_name = country_dir.name country_code = country_codes[country_name] break if country_code: break if country_code is None: print(f"警告: 无法确定球队 {team_name} 所属的国家，跳过") continue try: cursor.execute( "INSERT OR IGNORE INTO Teams (name, country_code) VALUES (?, ?)", (team_name, country_code) ) except sqlite3.Error as e: print(f"插入球队 {team_name} 时出错: {e}") conn.commit() print("球队数据插入完成") def insert_league_standings_data(conn): """插入联赛积分数据""" cursor = conn.cursor() countries_dir = data_dir / 'countries' # 国家代码映射 country_codes = { "England": "ENG", "France": "FRA", "Germany": "GER", "Spain": "ESP", "China": "CHN" } print("插入联赛积分数据...") for country_dir in countries_dir.iterdir(): if country_dir.is_dir(): country_name = country_dir.name country_code = country_codes[country_name] scores_file = country_dir / 'scores.csv' if scores_file.exists(): try: with open(scores_file, 'r', encoding='utf-8') as f: reader = csv.reader(f) header = next(reader) # 跳过表头 for row in reader: if len(row) >= 10: # 解析球队名称（去掉排名数字） team_name = ''.join([c for c in row[0] if not c.isdigit()]) # 查找球队ID cursor.execute("SELECT id FROM Teams WHERE name = ?", (team_name,)) team_result = cursor.fetchone() if team_result is None: print(f"警告: 无法在数据库中找到球队 {team_name}，跳过该记录") continue team_id = team_result[0] # 解析数据 position = int(''.join([c for c in row[0] if c.isdigit()])) if any(c.isdigit() for c in row[0]) else None played = int(row[1]) if row[1].isdigit() else None won = int(row[2]) if row[2].isdigit() else None drawn = int(row[3]) if row[3].isdigit() else None lost = int(row[4]) if row[4].isdigit() else None goals_for = int(row[5]) if row[5].isdigit() else None goals_against = int(row[6]) if row[6].isdigit() else None goal_difference = int(row[7]) if row[7].lstrip('-').isdigit() else None points = int(row[8]) if row[8].isdigit() else None recent_form = row[9] if len(row) > 9 else None cursor.execute(""" INSERT OR IGNORE INTO LeagueStandings (team_id, country_code, position, played, won, drawn, lost, goals_for, goals_against, goal_difference, points, recent_form) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, (team_id, country_code, position, played, won, drawn, lost, goals_for, goals_against, goal_difference, points, recent_form)) except Exception as e: print(f"处理 {scores_file} 时出错: {e}") conn.commit() print("联赛积分数据插入完成") def insert_match_results_data(conn): """插入比赛数据""" cursor = conn.cursor() countries_dir = data_dir / 'countries' # 国家代码映射 country_codes = { "England": "ENG", "France": "FRA", "Germany": "GER", "Spain": "ESP", "China": "CHN" } print("插入比赛数据...") for country_dir in countries_dir.iterdir(): if country_dir.is_dir(): country_name = country_dir.name country_code = country_codes[country_name] # 遍历所有球队目录 for team_dir in country_dir.iterdir(): if team_dir.is_dir(): team_name = team_dir.name matches_file = team_dir / 'matches.csv' if matches_file.exists(): # 查找球队ID cursor.execute("SELECT id FROM Teams WHERE name = ?", (team_name,)) team_result = cursor.fetchone() if team_result is None: print(f"警告: 无法在数据库中找到球队 {team_name}，跳过") continue team_id = team_result[0] try: with open(matches_file, 'r', encoding='utf-8') as f: reader = csv.reader(f) header = next(reader) # 跳过表头 for row in reader: if len(row) >= 6: result = row[0] if row[0] else None league = row[1] if len(row) > 1 else None match_date = row[2] if len(row) > 2 else None team_a = row[3] if len(row) > 3 else None goals = row[4] if len(row) > 4 else None team_b = row[5] if len(row) > 5 else None # 解析比分 home_goals = away_goals = None if goals and ':' in goals: try: home_goals, away_goals = map(int, goals.split(':')) except ValueError: pass # 无法解析比分，保持为None # 确定主队和客队 home_team_name = team_a away_team_name = team_b # 清理球队名称（移除数字前缀） home_team_name = ''.join([c for c in home_team_name if not c.isdigit()]) away_team_name = ''.join([c for c in away_team_name if not c.isdigit()]) # 如果当前球队是team_b，则交换主客队 if team_name == team_b: home_team_name, away_team_name = away_team_name, home_team_name if home_goals is not None and away_goals is not None: home_goals, away_goals = away_goals, home_goals # 查找主队ID home_team_id = None if home_team_name == team_name: home_team_id = team_id else: cursor.execute("SELECT id FROM Teams WHERE name = ?", (home_team_name,)) home_team_result = cursor.fetchone() if home_team_result is None: print(f"警告: 无法在数据库中找到球队 {home_team_name}，跳过该记录") continue home_team_id = home_team_result[0] # 查找客队ID away_team_id = None if away_team_name == team_name: away_team_id = team_id else: cursor.execute("SELECT id FROM Teams WHERE name = ?", (away_team_name,)) away_team_result = cursor.fetchone() if away_team_result is None: print(f"警告: 无法在数据库中找到球队 {away_team_name}，跳过该记录") continue away_team_id = away_team_result[0] cursor.execute(""" INSERT OR IGNORE INTO MatchResults (country_code, home_team_id, away_team_id, match_date, home_goals, away_goals, result) VALUES (?, ?, ?, ?, ?, ?, ?) """, (country_code, home_team_id, away_team_id, match_date, home_goals, away_goals, result)) except Exception as e: print(f"处理 {matches_file} 时出错: {e}") conn.commit() print("比赛数据插入完成") def insert_player_data(conn): """插入球员数据""" cursor = conn.cursor() countries_dir = data_dir / 'countries' # 国家代码映射 country_codes = { "England": "ENG", "France": "FRA", "Germany": "GER", "Spain": "ESP", "China": "CHN" } print("插入球员数据...") for country_dir in countries_dir.iterdir(): if country_dir.is_dir(): country_name = country_dir.name country_code = country_codes[country_name] # 遍历所有球队目录 for team_dir in country_dir.iterdir(): if team_dir.is_dir(): team_name = team_dir.name # 查找球队ID cursor.execute("SELECT id FROM Teams WHERE name = ?", (team_name,)) team_result = cursor.fetchone() if team_result is None: print(f"警告: 无法在数据库中找到球队 {team_name}，跳过") continue team_id = team_result[0] # 处理球员数据 players_dir = team_dir / 'players' if players_dir.exists(): for player_file in players_dir.iterdir(): if player_file.suffix == '.csv': try: with open(player_file, 'r', encoding='utf-8') as f: reader = csv.reader(f) header = next(reader) # 读取表头 # 解析球员基本信息（第一行） first_row = next(reader) if len(first_row) >= 7: player_name = first_row[0] current_team = first_row[1] if len(first_row) > 1 else None shirt_number = None try: shirt_number = int(first_row[2]) if first_row[2] else None except ValueError: pass # 解析年龄 age = None if len(first_row) > 3 and 'years old' in first_row[3]: try: age = int(first_row[3].split()[0]) except ValueError: pass # 解析身高 height = None if len(first_row) > 4 and 'cm' in first_row[4]: try: height = int(first_row[4].replace('cm', '')) except ValueError: pass nationality = first_row[5] if len(first_row) > 5 else None positions = first_row[6] if len(first_row) > 6 else None # 插入球员基本信息 cursor.execute(""" INSERT OR IGNORE INTO Players (name, age, height, nationality, positions) VALUES (?, ?, ?, ?, ?) """, (player_name, age, height, nationality, positions)) # 获取球员ID cursor.execute("SELECT id FROM Players WHERE name = ?", (player_name,)) player_result = cursor.fetchone() if player_result is None: print(f"警告: 无法在数据库中找到球员 {player_name}，跳过") continue player_id = player_result[0] # 处理球员统计数据（从第二行开始） for stat_row in reader: if len(stat_row) >= 19: tournament = stat_row[7] if len(stat_row) > 7 else None appearances = stat_row[8] if len(stat_row) > 8 else None # 解析分钟数 minutes_played = None try: minutes_played = int(stat_row[9]) if stat_row[9] and stat_row[9] != '-' else None except ValueError: pass # 解析进球数 goals = None try: goals = int(stat_row[10]) if stat_row[10] and stat_row[10] != '-' else None except ValueError: pass # 解析助攻数 assists = None try: assists = int(stat_row[11]) if stat_row[11] and stat_row[11] != '-' else None except ValueError: pass # 解析场均射门 shots_per_game = None try: shots_per_game = float(stat_row[14]) if stat_row[14] and stat_row[14] != '-' else None except ValueError: pass # 解析传球成功率 pass_success_rate = None try: pass_success_rate = float(stat_row[15].replace('%', '')) if stat_row[15] and stat_row[15] != '-' else None except ValueError: pass # 解析评分 rating = None try: rating = float(stat_row[19]) if len(stat_row) > 19 and stat_row[19] and stat_row[19] != '-' else None except ValueError: pass # 插入球员统计数据 cursor.execute(""" INSERT OR IGNORE INTO PlayerStats (player_id, team_id, tournament, appearances, minutes_played, goals, assists, shots_per_game, pass_success_rate, rating) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, (player_id, team_id, tournament, appearances, minutes_played, goals, assists, shots_per_game, pass_success_rate, rating)) except Exception as e: print(f"处理 {player_file} 时出错: {e}") conn.commit() print("球员数据插入完成") def main(): """主函数""" print("开始重新创建数据库...") # 确保数据目录存在 if not data_dir.exists(): print(f"错误: 数据目录 {data_dir} 不存在") return # 删除现有的数据库文件 if db_path.exists(): try: db_path.unlink() print(f"已删除现有数据库文件: {db_path}") except Exception as e: print(f"删除现有数据库文件时出错: {e}") return # 创建数据库连接 try: conn = sqlite3.connect(db_path) print(f"已创建新数据库: {db_path}") except Exception as e: print(f"创建数据库连接时出错: {e}") return try: # 删除并重新创建表 drop_and_create_tables(conn) # 插入数据 insert_countries_data(conn) insert_teams_data(conn) insert_league_standings_data(conn) insert_match_results_data(conn) insert_player_data(conn) print("数据库重建完成!") except Exception as e: print(f"重建数据库时出错: {e}") finally: conn.close() if __name__ == "__main__": main()