这个python案例能否提供实时比分预警功能？

Python 实时比分预警功能实现方案

可以实现！下面给你一套完整可用的方案，涵盖数据获取、预警判断、消息推送三个环节。

整体架构

数据源(API/爬虫) → 定时轮询 → 比分变化检测 → 触发预警 → 消息推送(微信/钉钉/邮件)

核心代码实现

基础版本（轮询 + 控制台预警）

import requests
import time
from datetime import datetime
# 模拟数据源（实际替换为真实API）
def fetch_score(match_id):
    # 示例：使用真实API如 football-data.org、懂球帝、雷速等
    url = f"https://api.example.com/match/{match_id}"
    resp = requests.get(url, timeout=5).json()
    return {
        "home": resp["home_team"],
        "away": resp["away_team"],
        "home_score": resp["home_score"],
        "away_score": resp["away_score"],
        "status": resp["status"],  # 进行中/结束
        "minute": resp.get("minute", 0)
    }
class ScoreMonitor:
    def __init__(self, match_id, interval=30):
        self.match_id = match_id
        self.interval = interval
        self.last_score = None
        self.last_status = None
    def check(self):
        data = fetch_score(self.match_id)
        current = (data["home_score"], data["away_score"])
        # 首次初始化
        if self.last_score is None:
            self.last_score = current
            self.last_status = data["status"]
            print(f"[初始化] {data['home']} vs {data['away']} 当前比分 {current}")
            return
        # 比分变化预警
        if current != self.last_score:
            self.alert_goal(data, self.last_score, current)
            self.last_score = current
        # 状态变化预警（如比赛结束）
        if data["status"] != self.last_status:
            self.alert_status(data)
            self.last_status = data["status"]
    def alert_goal(self, data, old, new):
        scorer = "主队" if new[0] > old[0] else "客队"
        msg = (f"⚽ 进球预警 [{datetime.now():%H:%M:%S}]\n"
               f"{data['home']} {new[0]}-{new[1]} {data['away']}\n"
               f"{scorer}进球！第{data['minute']}分钟")
        print(msg)
        self.push(msg)  # 推送到微信/钉钉
    def alert_status(self, data):
        msg = f"🏁 比赛状态变更: {data['status']}\n{data['home']} {data['home_score']}-{data['away_score']} {data['away']}"
        print(msg)
        self.push(msg)
    def push(self, msg):
        """消息推送（示例：钉钉机器人）"""
        webhook = "https://oapi.dingtalk.com/robot/send?access_token=YOUR_TOKEN"
        try:
            requests.post(webhook, json={
                "msgtype": "text",
                "text": {"content": msg}
            }, timeout=5)
        except Exception as e:
            print(f"推送失败: {e}")
    def run(self):
        print(f"开始监控比赛 {self.match_id}...")
        while True:
            try:
                self.check()
            except Exception as e:
                print(f"异常: {e}")
            time.sleep(self.interval)
if __name__ == "__main__":
    monitor = ScoreMonitor(match_id="12345", interval=15)  # 15秒轮询
    monitor.run()

进阶：多场比赛 + 条件预警

class MultiMatchMonitor:
    def __init__(self, match_ids, interval=20):
        self.monitors = {mid: ScoreMonitor(mid, interval) for mid in match_ids}
    def run(self):
        while True:
            for mid, m in self.monitors.items():
                try:
                    m.check()
                except Exception as e:
                    print(f"比赛 {mid} 异常: {e}")
            time.sleep(self.monitors[list(self.monitors)[0]].interval)

数据源推荐

推送渠道对比

微信推送（Server酱示例）

def push_wechat(msg):
    key = "YOUR_SENDKEY"
    requests.post(f"https://sctapi.ftqq.com/{key}.send",
                  data={"title": "比分预警", "desp": msg})

优化建议

1. 轮询频率：比赛中 10-15 秒，未开始 5-10 分钟即可，节省配额
2. 去重机制：用 (match_id, score, minute) 组合键避免重复推送
3. 容错重试：加 tenacity 库做指数退避重试
4. 存储历史：用 SQLite 记录每次比分，便于赛后复盘
5. 异步化：改用 asyncio + aiohttp 并发监控几十场比赛
6. 规则引擎：不止进球，可扩展"红牌"、"点球"、"大球预警(总进球≥3)"等

关键预警场景示例

def advanced_alerts(data, last):
    score = data["home_score"] + data["away_score"]
    # 大小球预警
    if score >= 3:
        push_wechat(f"🎯 总进球达{score}球！大球打出")
    # 冷门预警：弱队领先
    if data["home_score"] > data["away_score"] and data["home_rank"] > 15:
        push_wechat(f"❄️ 冷门！{data['home']} 领先")
    # 关键时间：85分钟后有球
    if data["minute"] >= 85 and score != last:
        push_wechat("⏰ 绝杀/补时进球！")

完全可行,30 行核心代码就能跑起来，难点主要在数据源稳定性和推送渠道，建议从 PushPlus + football-data.org 起步，跑通后再切换到商业 API 提升实时性。

需要我针对某个具体数据源（比如雷速/懂球帝/API-Football）或推送渠道（钉钉/微信）写完整可运行的代码吗？