引言：足球赛程管理的重要性
在现代足球世界中，球迷们面临着前所未有的信息爆炸。从英超、西甲、德甲、意甲、法甲五大联赛，到欧冠、欧联杯、世界杯、欧洲杯等国际赛事，再到各国杯赛和友谊赛，比赛场次之多令人眼花缭乱。对于真正的球迷来说，如何高效地管理这些赛程信息，精准掌握自己关注球队的比赛时间，已经成为一个实际而迫切的需求。
传统的赛程查询方式往往依赖于记忆、纸质日历或简单的手机提醒，这些方法在面对复杂多变的赛程安排时显得力不从心。特别是当比赛时间因转播需求、天气原因或其他因素而临时调整时，传统方式很容易导致球迷错过关键场次。因此，开发一套智能化的排期预测和赛程查询系统，不仅能够帮助球迷精准掌握比赛时间，还能通过数据分析预测可能的赛程变化，为球迷提供前所未有的观赛体验。
本文将深入探讨如何构建一个完整的足球赛程管理系统，包括数据获取、存储、查询、预测分析以及用户通知等核心模块，并提供详细的代码实现示例。
足球赛程数据的获取与处理
数据源的选择与API集成
要构建一个可靠的赛程查询系统，首先需要稳定、准确的数据源。目前市面上有多个提供足球数据的API服务，如：
- Football-Data.org - 提供免费的足球数据API
- API-Football - 功能全面的付费API
- Sportradar - 专业的体育数据提供商
- Opta - 高级数据分析服务
以Football-Data.org为例，我们可以通过以下Python代码获取赛程数据：
import requests
import json
from datetime import datetime, timedelta
class FootballScheduleManager:
def __init__(self, api_key):
self.api_key = api_key
self.base_url = "http://api.football-data.org/v4"
self.headers = {"X-Auth-Token": api_key}
def get_competition_schedule(self, competition_id, season):
"""
获取特定联赛的赛程
:param competition_id: 联赛ID (如: PL=英超, PD=西甲)
:param season: 赛季年份 (如: 2023)
:return: 赛程数据
"""
url = f"{self.base_url}/competitions/{competition_id}/matches"
params = {"season": season}
try:
response = requests.get(url, headers=self.headers, params=params)
response.raise_for_status()
return response.json()
except requests.exceptions.RequestException as e:
print(f"获取赛程数据失败: {e}")
return None
def get_team_schedule(self, team_id, competition_id=None, season=None):
"""
获取特定球队的赛程
:param team_id: 球队ID
:param competition_id: 可选，特定联赛
:param season: 可选，特定赛季
:return: 球队赛程数据
"""
url = f"{self.base_url}/teams/{team_id}/matches"
params = {}
if competition_id:
params["competition"] = competition_id
if season:
params["season"] = season
try:
response = requests.get(url, headers=self.headers, params=params)
response.raise_for_status()
return response.json()
except requests.exceptions.RequestException as e:
print(f"获取球队赛程失败: {e}")
return None
# 使用示例
if __name__ == "__main__":
# 替换为你的API密钥
API_KEY = "YOUR_API_KEY"
manager = FootballScheduleManager(API_KEY)
# 获取英超2023-24赛季赛程
premier_league_schedule = manager.get_competition_schedule("PL", 2023)
if premier_league_schedule:
print("成功获取英超赛程数据")
# 保存到文件
with open("premier_league_2023.json", "w") as f:
json.dump(premier_league_schedule, f, indent=2)
数据清洗与标准化
从API获取的原始数据往往包含冗余信息或格式不统一的问题，需要进行清洗和标准化处理：
import pandas as pd
from datetime import datetime
class DataCleaner:
@staticmethod
def clean_match_data(raw_data):
"""
清洗比赛数据，提取关键信息
"""
matches = []
for match in raw_data["matches"]:
match_info = {
"match_id": match["id"],
"competition": match["competition"]["name"],
"season": f"{match['season']['startDate'][:4]}-{match['season']['endDate'][:4]}",
"matchday": match.get("matchday", "N/A"),
"status": match["status"],
"home_team": match["homeTeam"]["name"],
"away_team": match["awayTeam"]["name"],
"home_score": match.get("score", {}).get("fullTime", {}).get("home"),
"away_score": match.get("score", {}).get("fullTime", {}).get("away"),
"utc_date": match["utcDate"],
"local_date": datetime.fromisoformat(match["utcDate"].replace('Z', '+00:00')).strftime("%Y-%m-%d %H:%M:%S"),
"venue": match.get("venue", "Unknown")
}
matches.append(match_info)
return pd.DataFrame(matches)
# 使用示例
cleaner = DataCleaner()
if premier_league_schedule:
df_matches = cleaner.clean_match_data(premier_league_schedule)
print(df_matches.head())
# 保存为CSV
df_matches.to_csv("premier_league_matches.csv", index=False)
赛程存储与数据库设计
数据库架构设计
为了高效存储和查询大量赛程数据，我们需要设计合理的数据库结构。以下是使用SQLite的示例：
import sqlite3
import json
from datetime import datetime
class ScheduleDatabase:
def __init__(self, db_path="football_schedule.db"):
self.db_path = db_path
self.init_database()
def init_database(self):
"""初始化数据库表结构"""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
# 创建联赛表
cursor.execute("""
CREATE TABLE IF NOT EXISTS competitions (
id INTEGER PRIMARY KEY,
name TEXT NOT NULL,
code TEXT UNIQUE,
type TEXT,
emblem TEXT
)
""")
# 创建球队表
cursor.execute("""
CREATE TABLE IF NOT EXISTS teams (
id INTEGER PRIMARY KEY,
name TEXT NOT NULL,
short_name TEXT,
tla TEXT,
crest TEXT,
UNIQUE(name)
)
""")
# 创建比赛表
cursor.execute("""
CREATE TABLE IF NOT EXISTS matches (
id INTEGER PRIMARY KEY,
competition_id INTEGER,
season TEXT,
matchday INTEGER,
status TEXT,
home_team_id INTEGER,
away_team_id INTEGER,
home_score INTEGER,
away_score INTEGER,
utc_date TEXT,
local_date TEXT,
venue TEXT,
FOREIGN KEY (competition_id) REFERENCES competitions(id),
FOREIGN KEY (home_team_id) REFERENCES teams(id),
FOREIGN KEY (away_team_id) REFERENCES teams(id)
)
""")
# 创建用户关注表
cursor.execute("""
CREATE TABLE IF NOT EXISTS user_follows (
user_id INTEGER,
team_id INTEGER,
competition_id INTEGER,
notify_before_minutes INTEGER DEFAULT 60,
PRIMARY KEY (user_id, team_id, competition_id),
FOREIGN KEY (team_id) REFERENCES teams(id),
FOREIGN KEY (competition_id) REFERENCES competitions(id)
)
""")
conn.commit()
conn.close()
def insert_competition(self, competition_data):
"""插入联赛数据"""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cursor.execute("""
INSERT OR REPLACE INTO competitions (id, name, code, type, emblem)
VALUES (?, ?, ?, ?, ?)
""", (
competition_data["id"],
competition_data["name"],
competition_data.get("code"),
competition_data.get("type"),
competition_data.get("emblem")
))
conn.commit()
conn.close()
def insert_team(self, team_data):
"""插入球队数据"""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cursor.execute("""
INSERT OR REPLACE INTO teams (id, name, short_name, tla, crest)
VALUES (?, ?, ?, ?, ?)
""", (
team_data["id"],
team_data["name"],
team_data.get("shortName"),
team_data.get("tla"),
team_data.get("crest")
))
conn.commit()
conn.close()
def insert_match(self, match_data):
"""插入比赛数据"""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cursor.execute("""
INSERT OR REPLACE INTO matches (
id, competition_id, season, matchday, status,
home_team_id, away_team_id, home_score, away_score,
utc_date, local_date, venue
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""", (
match_data["match_id"],
match_data.get("competition_id"),
match_data.get("season"),
match_data.get("matchday"),
match_data["status"],
match_data["home_team_id"],
match_data["away_team_id"],
match_data.get("home_score"),
match_data.get("away_score"),
match_data["utc_date"],
match_data["local_date"],
match_data.get("venue")
))
conn.commit()
conn.close()
def batch_insert_matches(self, matches_df):
"""批量插入比赛数据"""
conn = sqlite3.connect(self.db_path)
# 转换DataFrame为列表
matches_list = matches_df.to_dict('records')
# 批量插入
matches_df.to_sql('matches', conn, if_exists='replace', index=False)
conn.close()
# 使用示例
db = ScheduleDatabase()
# 假设我们已经有了清洗后的DataFrame df_matches
# db.batch_insert_matches(df_matches)
智能排期预测功能
基于历史数据的赛程预测
赛程预测是系统的核心功能之一。虽然官方赛程通常提前公布，但有时会因各种原因（如电视转播调整、天气、球队参加其他赛事等）而临时调整。我们可以通过分析历史数据来预测可能的调整：
import numpy as np
from sklearn.ensemble import RandomForestRegressor
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import LabelEncoder
class SchedulePredictor:
def __init__(self, db_path="football_schedule.db"):
self.db_path = db_path
self.model = RandomForestRegressor(n_estimators=100, random_state=42)
self.label_encoders = {}
def prepare_training_data(self):
"""
从数据库中准备训练数据
"""
conn = sqlite3.connect(self.db_path)
# 查询历史比赛数据，包括那些被调整过的比赛
query = """
SELECT
m.season,
m.matchday,
m.competition_id,
m.home_team_id,
m.away_team_id,
m.utc_date,
c.type as competition_type,
CASE
WHEN m.status = 'SCHEDULED' THEN 0
WHEN m.status = 'POSTPONED' THEN 1
WHEN m.status = 'RESCHEDULED' THEN 2
ELSE 0
END as schedule_change
FROM matches m
JOIN competitions c ON m.competition_id = c.id
WHERE m.status IN ('SCHEDULED', 'POSTPONED', 'RESCHEDULED', 'FINISHED')
"""
df = pd.read_sql_query(query, conn)
conn.close()
# 特征工程
df['utc_date'] = pd.to_datetime(df['utc_date'])
df['month'] = df['utc_date'].dt.month
df['day_of_week'] = df['utc_date'].dt.dayofweek
df['hour'] = df['utc_date'].dt.hour
# 编码分类变量
categorical_columns = ['season', 'competition_id', 'competition_type', 'home_team_id', 'away_team_id']
for col in categorical_columns:
if col not in self.label_encoders:
self.label_encoders[col] = LabelEncoder()
df[col] = self.label_encoders[col].fit_transform(df[col].astype(str))
# 特征和标签
feature_columns = ['season', 'matchday', 'competition_id', 'home_team_id',
'away_team_id', 'competition_type', 'month', 'day_of_week', 'hour']
X = df[feature_columns]
y = df['schedule_change']
return X, y
def train_model(self):
"""训练预测模型"""
X, y = self.prepare_training_data()
# 分割训练测试集
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)
# 训练模型
self.model.fit(X_train, y_train)
# 评估模型
train_score = self.model.score(X_train, y_train)
test_score = self.model.score(X_test, y_test)
print(f"训练集准确率: {train_score:.4f}")
print(f"测试集准确率: {test_score:.4f}")
# 特征重要性
feature_importance = pd.DataFrame({
'feature': feature_columns,
'importance': self.model.feature_importances_
}).sort_values('importance', ascending=False)
print("\n特征重要性:")
print(feature_importance)
return train_score, test_score
def predict_schedule_change(self, match_data):
"""
预测单场比赛的赛程变更概率
:param match_data: 包含比赛信息的字典
:return: 变更概率 (0-1)
"""
# 准备特征
features = {}
# 编码分类变量
for col in ['season', 'competition_id', 'competition_type', 'home_team_id', 'away_team_id']:
if col in match_data:
if col not in self.label_encoders:
# 如果没有编码器，创建临时编码器
self.label_encoders[col] = LabelEncoder()
self.label_encoders[col].fit([str(match_data[col])])
features[col] = self.label_encoders[col].transform([str(match_data[col])])[0]
# 时间特征
if 'utc_date' in match_data:
date = pd.to_datetime(match_data['utc_date'])
features['month'] = date.month
features['day_of_week'] = date.dayofweek
features['hour'] = date.hour
features['matchday'] = match_data.get('matchday', 1)
# 确保特征顺序一致
feature_columns = ['season', 'matchday', 'competition_id', 'home_team_id',
'away_team_id', 'competition_type', 'month', 'day_of_week', 'hour']
X = np.array([[features.get(col, 0) for col in feature_columns]])
# 预测
probability = self.model.predict(X)[0]
return min(probability, 1.0) # 确保不超过1
# 使用示例
predictor = SchedulePredictor()
# predictor.train_model()
# 预测示例
# new_match = {
# 'season': '2023-2024',
# 'matchday': 15,
# 'competition_id': 'PL',
# 'home_team_id': 'Arsenal',
# 'away_team_id': 'Chelsea',
# 'competition_type': 'LEAGUE',
# 'utc_date': '2024-01-15T15:00:00Z'
# }
# probability = predictor.predict_schedule_change(new_match)
# print(f"赛程变更概率: {probability:.2%}")
赛程查询与用户界面
命令行查询界面
首先，我们实现一个基础的命令行界面，用于查询赛程：
import sys
from datetime import datetime, timedelta
class ScheduleCLI:
def __init__(self, db_path="football_schedule.db"):
self.db_path = db_path
def query_upcoming_matches(self, team_name=None, competition=None, days=7):
"""
查询未来比赛
"""
conn = sqlite3.connect(self.db_path)
query = """
SELECT
m.utc_date,
m.local_date,
c.name as competition,
ht.name as home_team,
at.name as away_team,
m.status,
m.venue
FROM matches m
JOIN competitions c ON m.competition_id = c.id
JOIN teams ht ON m.home_team_id = ht.id
JOIN teams at ON m.away_team_id = at.id
WHERE m.utc_date >= ?
"""
params = [datetime.utcnow().isoformat()]
if team_name:
query += " AND (ht.name LIKE ? OR at.name LIKE ?)"
params.extend([f"%{team_name}%", f"%{team_name}%"])
if competition:
query += " AND c.name LIKE ?"
params.append(f"%{competition}%")
query += " ORDER BY m.utc_date LIMIT ?"
params.append(days * 5) # 假设每天最多5场比赛
df = pd.read_sql_query(query, conn, params=params)
conn.close()
return df
def query_team_schedule(self, team_name, competition=None, season=None):
"""
查询特定球队的完整赛程
"""
conn = sqlite3.connect(self.db_path)
query = """
SELECT
m.matchday,
m.utc_date,
m.local_date,
c.name as competition,
ht.name as home_team,
at.name as away_team,
m.status,
m.home_score,
m.away_score,
m.venue
FROM matches m
JOIN competitions c ON m.competition_id = c.id
JOIN teams ht ON m.home_team_id = ht.id
JOIN teams at ON m.away_team_id = at.id
WHERE (ht.name LIKE ? OR at.name LIKE ?)
"""
params = [f"%{team_name}%", f"%{team_name}%"]
if competition:
query += " AND c.name LIKE ?"
params.append(f"%{competition}%")
if season:
query += " AND m.season = ?"
params.append(season)
query += " ORDER BY m.utc_date"
df = pd.read_sql_query(query, conn, params=params)
conn.close()
return df
def display_matches(self, df):
"""格式化显示比赛信息"""
if df.empty:
print("未找到比赛记录")
return
print("\n" + "="*80)
print(f"{'日期时间':<20} {'联赛':<15} {'主队':<20} {'客队':<20} {'状态':<10}")
print("="*80)
for _, row in df.iterrows():
date_str = row['local_date'][:16] if row['local_date'] else 'N/A'
competition = row['competition'][:14]
home_team = row['home_team'][:19]
away_team = row['away_team'][:19]
status = row['status'][:9]
print(f"{date_str:<20} {competition:<15} {home_team:<20} {away_team:<20} {status:<10}")
print("="*80)
def interactive_query(self):
"""交互式查询界面"""
print("\n足球赛程查询系统")
print("1. 查询未来比赛")
print("2. 查询球队赛程")
print("3. 退出")
while True:
choice = input("\n请选择操作 (1-3): ").strip()
if choice == "1":
team = input("输入球队名称 (可为空): ").strip() or None
comp = input("输入联赛名称 (可为空): ").strip() or None
days = input("查询天数 (默认7): ").strip()
days = int(days) if days.isdigit() else 7
df = self.query_upcoming_matches(team, comp, days)
self.display_matches(df)
elif choice == "2":
team = input("输入球队名称: ").strip()
if not team:
print("球队名称不能为空")
continue
comp = input("输入联赛名称 (可为空): ").strip() or None
season = input("输入赛季 (如2023-2024, 可为空): ").strip() or None
df = self.query_team_schedule(team, comp, season)
self.display_matches(df)
elif choice == "3":
print("感谢使用！")
break
else:
print("无效选择，请重新输入")
# 使用示例
if __name__ == "__main__":
cli = ScheduleCLI()
cli.interactive_query()
Web界面实现（Flask）
对于更友好的用户体验，我们可以使用Flask构建Web界面：
from flask import Flask, render_template, request, jsonify
import sqlite3
import json
from datetime import datetime, timedelta
app = Flask(__name__)
DB_PATH = "football_schedule.db"
@app.route('/')
def index():
"""首页 - 显示未来7天比赛"""
conn = sqlite3.connect(DB_PATH)
query = """
SELECT
m.utc_date,
m.local_date,
c.name as competition,
ht.name as home_team,
at.name as away_team,
m.status,
m.venue,
m.home_score,
m.away_score
FROM matches m
JOIN competitions c ON m.competition_id = c.id
JOIN teams ht ON m.home_team_id = ht.id
JOIN teams at ON m.away_team_id = at.id
WHERE m.utc_date >= ?
ORDER BY m.utc_date
LIMIT 30
"""
df = pd.read_sql_query(query, conn, params=[datetime.utcnow().isoformat()])
conn.close()
# 转换为JSON格式供前端使用
matches = df.to_dict('records')
return render_template('index.html', matches=matches)
@app.route('/api/schedule')
def api_schedule():
"""API接口 - 获取赛程数据"""
team = request.args.get('team')
competition = request.args.get('competition')
days = int(request.args.get('days', 7))
conn = sqlite3.connect(DB_PATH)
query = """
SELECT
m.utc_date,
m.local_date,
c.name as competition,
ht.name as home_team,
at.name as away_team,
m.status,
m.venue,
m.home_score,
m.away_score
FROM matches m
JOIN competitions c ON m.competition_id = c.id
JOIN teams ht ON m.home_team_id = ht.id
JOIN teams at ON m.away_team_id = at.id
WHERE m.utc_date >= ?
"""
params = [datetime.utcnow().isoformat()]
if team:
query += " AND (ht.name LIKE ? OR at.name LIKE ?)"
params.extend([f"%{team}%", f"%{team}%"])
if competition:
query += " AND c.name LIKE ?"
params.append(f"%{competition}%")
query += " ORDER BY m.utc_date LIMIT ?"
params.append(days * 5)
df = pd.read_sql_query(query, conn, params=params)
conn.close()
return jsonify(df.to_dict('records'))
@app.route('/api/team/<team_name>')
def team_schedule(team_name):
"""API接口 - 获取特定球队赛程"""
conn = sqlite3.connect(DB_PATH)
query = """
SELECT
m.matchday,
m.utc_date,
m.local_date,
c.name as competition,
ht.name as home_team,
at.name as away_team,
m.status,
m.home_score,
m.away_score,
m.venue
FROM matches m
JOIN competitions c ON m.competition_id = c.id
JOIN teams ht ON m.home_team_id = ht.id
JOIN teams at ON m.away_team_id = at.id
WHERE (ht.name LIKE ? OR at.name LIKE ?)
ORDER BY m.utc_date
"""
df = pd.read_sql_query(query, conn, params=[f"%{team_name}%", f"%{team_name}%"])
conn.close()
return jsonify(df.to_dict('records'))
if __name__ == '__main__':
app.run(debug=True, host='0.0.0.0', port=5000)
对应的HTML模板（templates/index.html）：
<!DOCTYPE html>
<html lang="zh-CN">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>足球赛程查询系统</title>
<link href="https://cdn.jsdelivr.net/npm/bootstrap@5.3.0/dist/css/bootstrap.min.css" rel="stylesheet">
<style>
.match-card {
transition: transform 0.2s;
margin-bottom: 10px;
}
.match-card:hover {
transform: translateY(-2px);
box-shadow: 0 4px 8px rgba(0,0,0,0.1);
}
.status-scheduled { background-color: #e7f3ff; }
.status-live { background-color: #ffebee; font-weight: bold; }
.status-finished { background-color: #f5f5f5; opacity: 0.8; }
.status-postponed { background-color: #fff3cd; }
</style>
</head>
<body>
<div class="container mt-4">
<h1 class="text-center mb-4">⚽ 足球赛程查询系统</h1>
<!-- 搜索栏 -->
<div class="card mb-4">
<div class="card-body">
<div class="row g-3">
<div class="col-md-4">
<input type="text" id="teamSearch" class="form-control" placeholder="搜索球队...">
</div>
<div class="col-md-4">
<input type="text" id="competitionSearch" class="form-control" placeholder="搜索联赛...">
</div>
<div class="col-md-2">
<input type="number" id="daysInput" class="form-control" value="7" min="1" max="30" placeholder="天数">
</div>
<div class="col-md-2">
<button class="btn btn-primary w-100" onclick="searchMatches()">搜索</button>
</div>
</div>
</div>
</div>
<!-- 比赛列表 -->
<div id="matchesList" class="row">
<!-- 动态填充 -->
</div>
<!-- 统计信息 -->
<div class="card mt-4">
<div class="card-body">
<h5>统计信息</h5>
<div id="statsInfo">加载中...</div>
</div>
</div>
</div>
<script src="https://cdn.jsdelivr.net/npm/bootstrap@5.3.0/dist/js/bootstrap.bundle.min.js"></script>
<script>
// 页面加载时显示未来7天比赛
document.addEventListener('DOMContentLoaded', function() {
searchMatches();
});
function searchMatches() {
const team = document.getElementById('teamSearch').value;
const competition = document.getElementById('competitionSearch').value;
const days = document.getElementById('daysInput').value;
let url = `/api/schedule?days=${days}`;
if (team) url += `&team=${encodeURIComponent(team)}`;
if (competition) url += `&competition=${encodeURIComponent(competition)}`;
fetch(url)
.then(response => response.json())
.then(matches => {
displayMatches(matches);
updateStats(matches);
})
.catch(error => {
console.error('Error:', error);
document.getElementById('matchesList').innerHTML =
'<div class="alert alert-danger">加载失败，请稍后重试</div>';
});
}
function displayMatches(matches) {
const container = document.getElementById('matchesList');
if (matches.length === 0) {
container.innerHTML = '<div class="alert alert-info">未找到比赛记录</div>';
return;
}
let html = '';
matches.forEach(match => {
const date = new Date(match.utc_date);
const dateStr = date.toLocaleString('zh-CN', {
year: 'numeric', month: '2-digit', day: '2-digit',
hour: '2-digit', minute: '2-digit'
});
const statusClass = `status-${match.status.toLowerCase()}`;
const score = match.home_score !== null ?
`${match.home_score} - ${match.away_score}` : 'VS';
html += `
<div class="col-md-6 col-lg-4">
<div class="card match-card ${statusClass}">
<div class="card-body">
<div class="d-flex justify-content-between align-items-center mb-2">
<span class="badge bg-primary">${match.competition}</span>
<small class="text-muted">${dateStr}</small>
</div>
<div class="d-flex justify-content-between align-items-center mb-2">
<strong>${match.home_team}</strong>
<span class="badge bg-dark">${score}</span>
<strong>${match.away_team}</strong>
</div>
<div class="d-flex justify-content-between">
<small class="text-muted">${match.venue || '待定'}</small>
<span class="badge bg-${getStatusBadgeColor(match.status)}">${getStatusText(match.status)}</span>
</div>
</div>
</div>
</div>
`;
});
container.innerHTML = html;
}
function updateStats(matches) {
const total = matches.length;
const live = matches.filter(m => m.status === 'LIVE').length;
const postponed = matches.filter(m => m.status === 'POSTPONED').length;
const upcoming = matches.filter(m => ['SCHEDULED', 'TIMED'].includes(m.status)).length;
const statsHtml = `
<div class="row text-center">
<div class="col">
<h3>${total}</h3>
<p class="text-muted">总场次</p>
</div>
<div class="col">
<h3>${upcoming}</h3>
<p class="text-muted">待进行</p>
</div>
<div class="col">
<h3>${live}</h3>
<p class="text-muted">进行中</p>
</div>
<div class="col">
<h3>${postponed}</h3>
<p class="text-muted">延期</p>
</div>
</div>
`;
document.getElementById('statsInfo').innerHTML = statsHtml;
}
function getStatusBadgeColor(status) {
const colors = {
'SCHEDULED': 'primary',
'TIMED': 'primary',
'LIVE': 'danger',
'IN_PLAY': 'danger',
'PAUSED': 'warning',
'FINISHED': 'secondary',
'POSTPONED': 'warning',
'SUSPENDED': 'warning',
'CANCELED': 'dark'
};
return colors[status] || 'secondary';
}
function getStatusText(status) {
const texts = {
'SCHEDULED': '待定',
'TIMED': '已定时',
'LIVE': '进行中',
'IN_PLAY': '进行中',
'PAUSED': '暂停',
'FINISHED': '已结束',
'POSTPONED': '延期',
'SUSPENDED': '中止',
'CANCELED': '取消'
};
return texts[status] || status;
}
</script>
</body>
</html>
通知与提醒系统
基于时间的通知调度
为了确保球迷不会错过比赛，我们需要实现一个通知系统，在比赛开始前发送提醒：
import schedule
import time
import threading
from datetime import datetime, timedelta
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
class NotificationManager:
def __init__(self, db_path="football_schedule.db"):
self.db_path = db_path
self.running = False
self.scheduler_thread = None
def get_upcoming_matches(self, hours_ahead=24):
"""获取即将开始的比赛"""
conn = sqlite3.connect(self.db_path)
# 获取未来24小时内开始的比赛
future_time = (datetime.utcnow() + timedelta(hours=hours_ahead)).isoformat()
current_time = datetime.utcnow().isoformat()
query = """
SELECT
m.id as match_id,
m.utc_date,
m.local_date,
c.name as competition,
ht.name as home_team,
at.name as away_team,
uf.user_id,
uf.notify_before_minutes,
m.venue
FROM matches m
JOIN competitions c ON m.competition_id = c.id
JOIN teams ht ON m.home_team_id = ht.id
JOIN teams at ON m.away_team_id = at.id
JOIN user_follows uf ON (m.home_team_id = uf.team_id OR m.away_team_id = uf.team_id)
WHERE m.utc_date BETWEEN ? AND ?
AND m.status IN ('SCHEDULED', 'TIMED')
"""
df = pd.read_sql_query(query, conn, params=[current_time, future_time])
conn.close()
return df
def check_notifications(self):
"""检查需要发送的通知"""
upcoming = self.get_upcoming_matches(hours_ahead=2)
if upcoming.empty:
return
now = datetime.utcnow()
for _, row in upcoming.iterrows():
match_time = datetime.fromisoformat(row['utc_date'].replace('Z', '+00:00'))
notify_time = match_time - timedelta(minutes=row['notify_before_minutes'])
# 如果当前时间在通知时间之后，但未超过比赛时间，发送通知
if now >= notify_time and now < match_time:
self.send_notification(row)
def send_notification(self, match_info):
"""发送通知（示例：邮件通知）"""
# 这里可以扩展为多种通知方式：邮件、短信、推送通知等
subject = f"⚽ 比赛提醒: {match_info['home_team']} vs {match_info['away_team']}"
body = f"""
您关注的比赛即将开始！
比赛信息:
联赛: {match_info['competition']}
时间: {match_info['local_date']}
对阵: {match_info['home_team']} vs {match_info['away_team']}
场地: {match_info['venue']}
请不要错过精彩比赛！
"""
print(f"\n{'='*60}")
print(f"【比赛提醒】")
print(f"{'='*60}")
print(body)
print(f"{'='*60}\n")
# 实际应用中，这里会调用邮件或推送API
# self.send_email(match_info['user_id'], subject, body)
def start_scheduler(self):
"""启动定时检查器"""
self.running = True
def run_scheduler():
while self.running:
self.check_notifications()
time.sleep(60) # 每分钟检查一次
self.scheduler_thread = threading.Thread(target=run_scheduler)
self.scheduler_thread.daemon = True
self.scheduler_thread.start()
print("通知系统已启动")
def stop_scheduler(self):
"""停止定时器"""
self.running = False
if self.scheduler_thread:
self.scheduler_thread.join()
print("通知系统已停止")
# 使用示例
if __name__ == "__main__":
notifier = NotificationManager()
# 启动通知系统
notifier.start_scheduler()
try:
# 保持程序运行
while True:
time.sleep(1)
except KeyboardInterrupt:
notifier.stop_scheduler()
用户关注管理
class UserFollowManager:
def __init__(self, db_path="football_schedule.db"):
self.db_path = db_path
def follow_team(self, user_id, team_name, competition_name=None, notify_before_minutes=60):
"""
用户关注球队
"""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
# 查找球队ID
cursor.execute("SELECT id FROM teams WHERE name LIKE ?", (f"%{team_name}%",))
team_result = cursor.fetchone()
if not team_result:
print(f"未找到球队: {team_name}")
conn.close()
return False
team_id = team_result[0]
# 查找联赛ID（如果指定了）
competition_id = None
if competition_name:
cursor.execute("SELECT id FROM competitions WHERE name LIKE ?", (f"%{competition_name}%",))
comp_result = cursor.fetchone()
if comp_result:
competition_id = comp_result[0]
# 插入关注记录
try:
cursor.execute("""
INSERT OR REPLACE INTO user_follows (user_id, team_id, competition_id, notify_before_minutes)
VALUES (?, ?, ?, ?)
""", (user_id, team_id, competition_id, notify_before_minutes))
conn.commit()
print(f"用户 {user_id} 已成功关注 {team_name}")
success = True
except Exception as e:
print(f"关注失败: {e}")
success = False
conn.close()
return success
def unfollow_team(self, user_id, team_name):
"""取消关注球队"""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cursor.execute("""
DELETE FROM user_follows
WHERE user_id = ? AND team_id IN (
SELECT id FROM teams WHERE name LIKE ?
)
""", (user_id, f"%{team_name}%"))
affected = cursor.rowcount
conn.commit()
conn.close()
if affected > 0:
print(f"用户 {user_id} 已取消关注 {team_name}")
return True
else:
print(f"未找到关注记录")
return False
def get_user_follows(self, user_id):
"""获取用户关注的所有球队"""
conn = sqlite3.connect(self.db_path)
query = """
SELECT
t.name as team_name,
c.name as competition_name,
uf.notify_before_minutes
FROM user_follows uf
JOIN teams t ON uf.team_id = t.id
LEFT JOIN competitions c ON uf.competition_id = c.id
WHERE uf.user_id = ?
"""
df = pd.read_sql_query(query, conn, params=[user_id])
conn.close()
return df
# 使用示例
follow_manager = UserFollowManager()
# 用户1关注阿森纳
follow_manager.follow_team(1, "Arsenal", notify_before_minutes=120)
# 用户1关注曼城在英超的比赛
follow_manager.follow_team(1, "Manchester City", "Premier League", notify_before_minutes=90)
# 查看用户1的关注列表
follows = follow_manager.get_user_follows(1)
print("\n用户1的关注列表:")
print(follows)
高级功能：赛程变化预测与分析
基于机器学习的赛程变更预测
我们可以进一步优化预测模型，加入更多特征：
import pandas as pd
import numpy as np
from sklearn.model_selection import train_test_split, GridSearchCV
from sklearn.ensemble import GradientBoostingClassifier
from sklearn.metrics import classification_report, confusion_matrix
import joblib
class AdvancedSchedulePredictor:
def __init__(self, db_path="football_schedule.db"):
self.db_path = db_path
self.model = None
self.feature_columns = None
def engineer_features(self, df):
"""
高级特征工程
"""
# 时间特征
df['utc_date'] = pd.to_datetime(df['utc_date'])
df['month'] = df['utc_date'].dt.month
df['day_of_week'] = df['utc_date'].dt.dayofweek
df['hour'] = df['utc_date'].dt.hour
df['is_weekend'] = df['day_of_week'].isin([5, 6]).astype(int)
# 赛季特征
df['season_part'] = pd.cut(df['month'],
bins=[0, 3, 6, 9, 12],
labels=['冬', '春', '夏', '秋'])
# 比赛重要性特征（基于历史数据）
df['is_derby'] = self._detect_derby(df)
df['is_top6_clash'] = self._detect_top_clash(df)
# 联赛特征
df['competition_popularity'] = df['competition_id'].map({
'PL': 1.0, 'PD': 0.9, 'BL1': 0.85, 'SA': 0.85, 'FL1': 0.7,
'CL': 1.0, 'EL': 0.8, 'WC': 1.0
}).fillna(0.5)
# 球队特征（基于历史表现）
df['team_rank_difference'] = self._calculate_rank_difference(df)
# 转换分类变量
categorical_features = ['season', 'competition_id', 'competition_type',
'home_team_id', 'away_team_id', 'season_part']
for col in categorical_features:
if col in df.columns:
df[col] = df[col].astype('category').cat.codes
# 选择最终特征
self.feature_columns = [
'season', 'matchday', 'competition_id', 'home_team_id', 'away_team_id',
'competition_type', 'month', 'day_of_week', 'hour', 'is_weekend',
'competition_popularity', 'is_derby', 'is_top6_clash', 'team_rank_difference'
]
return df[self.feature_columns]
def _detect_derby(self, df):
"""检测德比战（简化版）"""
# 实际应用中需要球队地理位置数据
# 这里仅作示例
derby_pairs = [
('Arsenal', 'Tottenham'),
('Liverpool', 'Everton'),
('Manchester United', 'Manchester City'),
('Real Madrid', 'Atletico Madrid'),
('Barcelona', 'Espanyol')
]
def is_derby_match(row):
home = row.get('home_team_name', '')
away = row.get('away_team_name', '')
return int((home, away) in derby_pairs or (away, home) in derby_pairs)
return df.apply(is_derby_match, axis=1)
def _detect_top_clash(self, df):
"""检测顶级球队对决"""
top_teams = ['Arsenal', 'Chelsea', 'Liverpool', 'Manchester City', 'Manchester United',
'Real Madrid', 'Barcelona', 'Atletico Madrid', 'Bayern Munich', 'Borussia Dortmund']
def is_top_clash(row):
home = row.get('home_team_name', '')
away = row.get('away_team_name', '')
return int(home in top_teams and away in top_teams)
return df.apply(is_top_clash, axis=1)
def _calculate_rank_difference(self, df):
"""计算球队排名差（需要历史排名数据）"""
# 简化版本，实际需要查询历史排名
return np.random.normal(0, 3, len(df)) # 占位符
def prepare_training_data_advanced(self):
"""准备高级训练数据"""
conn = sqlite3.connect(self.db_path)
query = """
SELECT
m.*,
c.type as competition_type,
c.name as competition_name,
ht.name as home_team_name,
at.name as away_team_name
FROM matches m
JOIN competitions c ON m.competition_id = c.id
JOIN teams ht ON m.home_team_id = ht.id
JOIN teams at ON m.away_team_id = at.id
WHERE m.status IN ('SCHEDULED', 'POSTPONED', 'RESCHEDULED', 'FINISHED')
"""
df = pd.read_sql_query(query, conn)
conn.close()
# 创建目标变量：是否发生赛程变更
df['schedule_changed'] = df['status'].isin(['POSTPONED', 'RESCHEDULED']).astype(int)
# 特征工程
X = self.engineer_features(df)
y = df['schedule_changed']
return X, y
def train_advanced_model(self):
"""训练高级预测模型"""
X, y = self.prepare_training_data_advanced()
# 分割数据
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)
# 使用梯度提升树
self.model = GradientBoostingClassifier(
n_estimators=200,
learning_rate=0.1,
max_depth=5,
random_state=42
)
# 网格搜索优化参数
param_grid = {
'n_estimators': [100, 200],
'learning_rate': [0.05, 0.1],
'max_depth': [3, 5]
}
grid_search = GridSearchCV(self.model, param_grid, cv=3, scoring='f1')
grid_search.fit(X_train, y_train)
self.model = grid_search.best_estimator_
# 评估
y_pred = self.model.predict(X_test)
print("最佳参数:", grid_search.best_params_)
print("\n分类报告:")
print(classification_report(y_test, y_pred))
# 特征重要性
importance = pd.DataFrame({
'feature': self.feature_columns,
'importance': self.model.feature_importances_
}).sort_values('importance', ascending=False)
print("\n特征重要性排序:")
print(importance)
return self.model
def predict_with_confidence(self, match_data):
"""
带置信度的预测
"""
if self.model is None:
raise ValueError("模型尚未训练")
# 准备特征
df = pd.DataFrame([match_data])
X = self.engineer_features(df)
# 预测概率
probability = self.model.predict_proba(X)[0][1]
# 置信度
confidence = max(probability, 1 - probability)
return {
'change_probability': probability,
'confidence': confidence,
'recommendation': '建议关注' if probability > 0.3 else '正常进行'
}
def save_model(self, filepath="schedule_predictor.pkl"):
"""保存模型"""
if self.model is not None:
joblib.dump({
'model': self.model,
'feature_columns': self.feature_columns,
'label_encoders': getattr(self, 'label_encoders', {})
}, filepath)
print(f"模型已保存到 {filepath}")
def load_model(self, filepath="schedule_predictor.pkl"):
"""加载模型"""
data = joblib.load(filepath)
self.model = data['model']
self.feature_columns = data['feature_columns']
self.label_encoders = data.get('label_encoders', {})
print(f"模型已从 {filepath} 加载")
# 使用示例
advanced_predictor = AdvancedSchedulePredictor()
# 训练模型
# advanced_predictor.train_advanced_model()
# 预测示例
# new_match = {
# 'season': '2023-2024',
# 'matchday': 20,
# 'competition_id': 'PL',
# 'home_team_id': 'Arsenal',
# 'away_team_id': 'Chelsea',
# 'competition_type': 'LEAGUE',
# 'utc_date': '2024-02-10T15:00:00Z',
# 'home_team_name': 'Arsenal',
# 'away_team_name': 'Chelsea'
# }
# result = advanced_predictor.predict_with_confidence(new_match)
# print(f"预测结果: {result}")
系统部署与维护
Docker部署方案
为了确保系统的可移植性和稳定性，我们可以使用Docker进行部署：
# Dockerfile
FROM python:3.9-slim
WORKDIR /app
# 安装系统依赖
RUN apt-get update && apt-get install -y \
gcc \
&& rm -rf /var/lib/apt/lists/*
# 安装Python依赖
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
# 复制应用代码
COPY . .
# 创建数据目录
RUN mkdir -p /app/data
# 暴露端口
EXPOSE 5000
# 启动命令
CMD ["python", "app.py"]
requirements.txt:
requests==2.31.0
pandas==2.0.3
numpy==1.24.3
scikit-learn==1.3.0
Flask==2.3.2
schedule==1.2.0
joblib==1.3.1
docker-compose.yml:
version: '3.8'
services:
football-schedule:
build: .
container_name: football_schedule_app
ports:
- "5000:5000"
volumes:
- ./data:/app/data
- ./logs:/app/logs
environment:
- API_KEY=${FOOTBALL_DATA_API_KEY}
- DB_PATH=/app/data/football_schedule.db
restart: unless-stopped
healthcheck:
test: ["CMD", "curl", "-f", "http://localhost:5000/health"]
interval: 30s
timeout: 10s
retries: 3
数据备份与恢复脚本
import shutil
import sqlite3
from datetime import datetime
import os
class BackupManager:
def __init__(self, db_path="football_schedule.db", backup_dir="backups"):
self.db_path = db_path
self.backup_dir = backup_dir
if not os.path.exists(backup_dir):
os.makedirs(backup_dir)
def create_backup(self):
"""创建数据库备份"""
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
backup_path = os.path.join(self.backup_dir, f"backup_{timestamp}.db")
try:
# 使用SQLite的备份机制
conn = sqlite3.connect(self.db_path)
backup_conn = sqlite3.connect(backup_path)
conn.backup(backup_conn)
backup_conn.close()
conn.close()
print(f"备份创建成功: {backup_path}")
return backup_path
except Exception as e:
print(f"备份失败: {e}")
return None
def restore_backup(self, backup_filename):
"""从备份恢复"""
backup_path = os.path.join(self.backup_dir, backup_filename)
if not os.path.exists(backup_path):
print(f"备份文件不存在: {backup_path}")
return False
try:
# 创建当前数据库的备份（以防万一）
self.create_backup()
# 恢复备份
backup_conn = sqlite3.connect(backup_path)
current_conn = sqlite3.connect(self.db_path)
backup_conn.backup(current_conn)
current_conn.close()
backup_conn.close()
print(f"从 {backup_filename} 恢复成功")
return True
except Exception as e:
print(f"恢复失败: {e}")
return False
def list_backups(self):
"""列出所有备份"""
backups = []
for filename in os.listdir(self.backup_dir):
if filename.endswith('.db'):
path = os.path.join(self.backup_dir, filename)
size = os.path.getsize(path)
mtime = datetime.fromtimestamp(os.path.getmtime(path))
backups.append({
'filename': filename,
'size': size,
'modified': mtime
})
# 按修改时间排序
backups.sort(key=lambda x: x['modified'], reverse=True)
return backups
# 使用示例
backup_mgr = BackupManager()
# 创建备份
backup_mgr.create_backup()
# 列出备份
backups = backup_mgr.list_backups()
print("可用备份:")
for backup in backups:
print(f" {backup['filename']} - {backup['size']} bytes - {backup['modified']}")
总结
通过本文的详细介绍，我们构建了一个完整的足球赛程管理系统，涵盖了以下核心功能：
- 数据获取与处理：通过API获取实时赛程数据，并进行清洗和标准化
- 数据库存储：设计了高效的数据库结构，支持复杂查询
- 智能预测：使用机器学习模型预测赛程变更概率
- 用户交互：提供命令行和Web界面两种查询方式
- 通知系统：实时监控比赛时间，提前发送提醒
- 高级分析：基于历史数据的深度分析和预测
- 部署维护：Docker容器化部署和数据备份方案
这套系统能够帮助球迷精准掌握比赛时间，通过智能预测功能提前了解可能的赛程变化，避免错过关键场次。无论是个人使用还是作为商业产品，都具有很高的实用价值。
未来可以进一步扩展的功能包括：
- 集成更多数据源，提高预测准确性
- 开发移动应用，提供推送通知
- 增加社交功能，让球迷可以分享观赛体验
- 结合实时比分和赔率数据，提供更丰富的分析
通过持续优化和扩展，这个系统将成为球迷不可或缺的智能助手。