qualifiers-uefa-api

Integriert die UEFA Digital API für Team-Daten, Koeffizienten und Clashes.

Trigger

UEFA API-Aufrufe
Team-Synchronisation
Clash-Updates
"uefa api", "team sync", "clashes aktualisieren"

Kontext

Die UEFA Digital API verwendet OAuth2 Client Credentials Flow. API-Skripte befinden sich in qualifiers/uefadigitalapi/.

Regeln

Umgebungsvariablen für Credentials:

UEFA_OAUTH_TENANT_ID=<azure-tenant-id>
UEFA_OAUTH_CLIENT_ID=<client-id>
UEFA_OAUTH_CLIENT_SECRET=<client-secret>
UEFA_OAUTH_SCOPE=<api-scope>
UEFA_SUBSCRIPTION_KEY=<subscription-key>
UEFA_SUBSCRIPTION_KEY_TOKEN=<subscription-key-for-token>

Token-Abruf:

import requests
import os

def get_uefa_token():
    tenant_id = os.getenv("UEFA_OAUTH_TENANT_ID")
    response = requests.post(
        f"https://login.microsoftonline.com/{tenant_id}/oauth2/v2.0/token",
        headers={"Content-Type": "application/x-www-form-urlencoded"},
        data={
            "grant_type": "client_credentials",
            "client_id": os.getenv("UEFA_OAUTH_CLIENT_ID"),
            "client_secret": os.getenv("UEFA_OAUTH_CLIENT_SECRET"),
            "scope": os.getenv("UEFA_OAUTH_SCOPE"),
        },
    )
    return response.json()['access_token']

API-Aufruf mit Bearer Token:

token = get_uefa_token()
headers = {
    "Authorization": f"Bearer {token}",
    "Ocp-Apim-Subscription-Key": os.getenv("UEFA_SUBSCRIPTION_KEY_TOKEN"),
}

response = requests.get(
    "https://api.digital.uefa.com/comp/v2/...",
    headers=headers,
    timeout=30
)

Retry-Logik bei Fehlern:

import time

def safe_api_call(url, headers, retries=3):
    for attempt in range(retries):
        try:
            response = requests.get(url, headers=headers, timeout=30)
            if response.status_code == 429:  # Rate limit
                retry_after = int(response.headers.get('Retry-After', 60))
                time.sleep(retry_after)
                continue
            response.raise_for_status()
            return response.json()
        except requests.exceptions.RequestException as e:
            if attempt == retries - 1:
                raise
            time.sleep(2 ** attempt)  # Exponential backoff

Daten-Mapping:

UEFA Feld Django Model Feld

teamId Team external_id

teamName Team name

countryCode Team.countryObj shortname

coefficient Team coefficient

position Team position

UEFA Feld	Django Model	Feld
`teamId`	Team	`external_id`
`teamName`	Team	`name`
`countryCode`	Team.countryObj	`shortname`
`coefficient`	Team	`coefficient`
`position`	Team	`position`

Beispiel: Teams importieren

import requests
from scheduler.models import Team, Country

def import_teams(scenario, competition_id):
    token = get_uefa_token()
    headers = {
        "Authorization": f"Bearer {token}",
        "Ocp-Apim-Subscription-Key": os.getenv("UEFA_SUBSCRIPTION_KEY_TOKEN"),
    }

    url = f"https://api.digital.uefa.com/comp/v2/competitions/{competition_id}/teams"
    data = safe_api_call(url, headers)

    for team_data in data['teams']:
        country = Country.objects.get(shortname=team_data['countryCode'])

        team, created = Team.objects.update_or_create(
            season=scenario.season,
            external_id=team_data['teamId'],
            defaults={
                'name': team_data['teamName'],
                'countryObj': country,
                'coefficient': team_data.get('coefficient', 0),
                'position': team_data.get('position', 0),
                'active': True,
            }
        )

        if created:
            print(f"Neues Team: {team.name}")

Beispiel: Clashes aktualisieren

from qualifiers.models import QClash

def update_clashes(scenario, season_code):
    token = get_uefa_token()
    headers = {...}

    url = f"https://api.digital.uefa.com/comp/v2/seasons/{season_code}/clashes"
    data = safe_api_call(url, headers)

    for clash_data in data['clashes']:
        team1 = Team.objects.get(
            season=scenario.season,
            external_id=clash_data['team1Id']
        )
        team2 = Team.objects.get(
            season=scenario.season,
            external_id=clash_data['team2Id']
        )

        QClash.objects.update_or_create(
            scenario=scenario,
            team1=team1,
            team2=team2,
            defaults={
                'type': clash_data.get('type', 0),
                'active_q1': True,
                'active_q2': True,
                'active_q3': True,
                'active_po': True,
            }
        )

Caching (geplant)

from django.core.cache import cache

def get_cached_api_data(key, url, headers, timeout=3600):
    """Cached API-Aufruf."""
    cached = cache.get(key)
    if cached:
        return cached

    data = safe_api_call(url, headers)
    cache.set(key, data, timeout)
    return data

5.3 KiB Raw Blame History