import sqlite3 import csv import os from pathlib import Path def main(): # 定义路径 - 使用相对路径 BASE_DIR = Path(__file__).resolve().parent.parent.parent.parent data_dir = BASE_DIR / 'footviz' / 'data' countries_dir = data_dir / 'countries' db_path = data_dir / 'football.db' # 创建数据库连接 conn = sqlite3.connect(db_path) cursor = conn.cursor() # 创建表结构 try: with open(Path(r'C:\Users\jerry\PyCharmProjects\footviz\footviz\schema.sql'), 'r', encoding='utf-8') as f: schema = f.read() # 移除 DROP TABLE 语句，只保留 CREATE TABLE 语句 create_statements = [] for line in schema.split('\n'): if line.startswith('--') or line.strip() == '': continue if not line.startswith('DROP TABLE'): create_statements.append(line) create_script = '\n'.join(create_statements) cursor.executescript(create_script) except sqlite3.OperationalError as e: if "already exists" in str(e): print("表结构已存在，跳过创建") # 如果表已存在，先删除再重新创建 cursor.executescript(''' DROP TABLE IF EXISTS LeagueStandings; DROP TABLE IF EXISTS MatchResults; DROP TABLE IF EXISTS PlayerStats; DROP TABLE IF EXISTS Players; DROP TABLE IF EXISTS Teams; DROP TABLE IF EXISTS Countries; ''') # 重新执行完整的schema with open(Path(r'C:\Users\jerry\PyCharmProjects\footviz\footviz\schema.sql'), 'r', encoding='utf-8') as f: schema = f.read() # 移除 DROP TABLE 语句，只保留 CREATE TABLE 语句 create_statements = [] for line in schema.split('\n'): if line.startswith('--') or line.strip() == '': continue if not line.startswith('DROP TABLE'): create_statements.append(line) create_script = '\n'.join(create_statements) cursor.executescript(create_script) else: raise e print("数据库表结构创建完成") # 国家代码映射 country_codes = { "England": "ENG", "France": "FRA", "Germany": "GER", "Spain": "ESP", "China": "CHN" } # 插入国家数据 print("插入国家数据...") for country_name, country_code in country_codes.items(): # 获取联赛名 league_name = None mapping_file = data_dir / 'league_country_mapping.csv' if mapping_file.exists(): with open(mapping_file, 'r', encoding='utf-8') as f: reader = csv.DictReader(f) for row in reader: if row['country_name'] == country_name: league_name = row['league_name'] break cursor.execute( "INSERT OR IGNORE INTO Countries (name, code, league_name) VALUES (?, ?, ?)", (country_name, country_code, league_name) ) conn.commit() print("国家数据插入完成") # 收集所有球队名称 all_teams = set() # 遍历所有国家目录，收集球队名称 if countries_dir.exists(): for country_dir in countries_dir.iterdir(): if country_dir.is_dir(): country_name = country_dir.name 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)} 支球队") # 插入球队数据 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 cursor.execute( "INSERT OR IGNORE INTO Teams (name, country_code) VALUES (?, ?)", (team_name, country_code) ) conn.commit() print("球队数据插入完成") # 插入联赛积分数据 print("插入联赛积分数据...") if countries_dir.exists(): 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(): 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)) conn.commit() print("联赛积分数据插入完成") # 插入比赛数据 print("插入比赛数据...") if countries_dir.exists(): 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] 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: # 解析数据 competition = row[0] match_date = row[1] team_a = row[2] team_b = row[3] goals_a = int(row[4]) if row[4].isdigit() else None goals_b = int(row[5]) if row[5].isdigit() else None # 确定主客队ID cursor.execute("SELECT id FROM Teams WHERE name = ?", (team_a,)) team_a_result = cursor.fetchone() if team_a_result is None: print(f"警告: 无法找到球队 {team_a}，跳过该记录") continue team_a_id = team_a_result[0] cursor.execute("SELECT id FROM Teams WHERE name = ?", (team_b,)) team_b_result = cursor.fetchone() if team_b_result is None: print(f"警告: 无法找到球队 {team_b}，跳过该记录") continue team_b_id = team_b_result[0] # 确定比赛结果 result = None if goals_a is not None and goals_b is not None: if goals_a > goals_b: result = 'W' # 主队胜利 elif goals_a < goals_b: result = 'L' # 主队失败 else: result = 'D' # 平局 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, team_a_id, team_b_id, match_date, goals_a, goals_b, result)) conn.commit() print("比赛数据插入完成") # 插入球员数据 print("插入球员数据...") if countries_dir.exists(): 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 players_file = team_dir / 'players.csv' if players_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] with open(players_file, 'r', encoding='utf-8') as f: reader = csv.reader(f) header = next(reader) # 跳过表头 for row in reader: if len(row) >= 8: # 解析数据 player_name = row[0] age = int(row[1]) if row[1].isdigit() else None date_of_birth = row[2] if row[2] else None height = int(row[3]) if row[3].isdigit() else None nationality = row[4] if row[4] else None positions = row[5] if row[5] else None image_url = row[6] if row[6] else None # 插入球员数据 cursor.execute(""" INSERT OR IGNORE INTO Players (name, age, date_of_birth, height, nationality, positions, image_url) VALUES (?, ?, ?, ?, ?, ?, ?) """, (player_name, age, date_of_birth, height, nationality, positions, image_url)) # 获取球员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] # 插入球员统计数据 stats_data = row[7] if row[7] else None if stats_data: # 解析统计数据（假设格式为JSON或简单的键值对） # 这里简化处理，实际应根据具体格式解析 cursor.execute(""" INSERT OR IGNORE INTO PlayerStats (player_id, team_id, tournament, appearances, minutes_played, goals, assists, tackles, interceptions, fouls_committed, pass_success_rate, shots_per_game, dribbles_per_game, rating) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, (player_id, team_id, "League", stats_data, None, None, None, None, None, None, None, None, None, None)) conn.commit() print("球员数据插入完成") # 关闭数据库连接 conn.close() print("所有数据导入完成！") if __name__ == "__main__": main()