dota2-match-calendar/sync_dota2_matches.py
Ching cf679b7107 fix: 防止未开始的比赛被错误标记为完成 v4.4
问题修复:
- 添加时间检查,只有已开始的比赛才能标记为完成
- 防止Liquipedia的预设比分导致未来比赛被误标为完成
- 修复了XG vs PV等未开始比赛显示完成状态的问题

技术细节:
- 在_parse_match方法中增加match_has_started检查
- 只有match_datetime <= now时才判断比赛是否完成
- 对未来比赛的比分记录debug日志但不标记为完成

🤖 Generated with Claude Code

Co-Authored-By: Claude <noreply@anthropic.com>
2025-09-14 12:27:35 +08:00

1621 lines
65 KiB
Python

#!/usr/bin/env python3
"""
Dota 2 Tournament Calendar Sync v4.0
Fetches Tier 1 Dota 2 matches from Liquipedia and syncs them to Google Calendar
Features:
- Sync upcoming matches
- Update completed match results
- Update match times if they change
- Improved code structure with dataclasses
- Enhanced error handling with retry mechanism
- Professional logging system
- Graceful degradation
"""
import requests
from bs4 import BeautifulSoup
from google.oauth2 import service_account
from googleapiclient.discovery import build
from datetime import datetime, timedelta
import pytz
import re
import hashlib
import sys
import argparse
import time
import logging
from dataclasses import dataclass, field
from typing import Optional, List, Dict, Tuple, Any
from functools import wraps
from enum import Enum
# Configure logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s',
datefmt='%Y-%m-%d %H:%M:%S'
)
logger = logging.getLogger(__name__)
class MatchFormat(Enum):
BO1 = "Bo1"
BO3 = "Bo3"
BO5 = "Bo5"
@dataclass
class Match:
"""Data class for match information
Attributes:
id: Unique identifier for the match (MD5 hash)
team1: Name of the first team
team2: Name of the second team
datetime: Match scheduled datetime (timezone aware)
tournament: Tournament name (e.g., 'The International 2025')
format: Match format (Bo1, Bo3, Bo5)
score: Current or final score (e.g., '2-1')
completed: Whether the series is completed
has_score: Whether the match has any score (including partial)
winner: Name of the winning team if completed
"""
id: str
team1: str
team2: str
datetime: datetime
tournament: Optional[str] = None
format: Optional[str] = None
score: Optional[str] = None
completed: bool = False
has_score: bool = False
winner: Optional[str] = None
def __post_init__(self):
"""Post-initialization processing
Ensures datetime is always timezone aware (UTC)
"""
if self.datetime and self.datetime.tzinfo is None:
self.datetime = self.datetime.replace(tzinfo=pytz.UTC)
def to_dict(self) -> Dict[str, Any]:
"""Convert Match object to dictionary
Returns:
Dict containing all match attributes
Used for backward compatibility with existing code
"""
return {
'id': self.id,
'team1': self.team1,
'team2': self.team2,
'datetime': self.datetime,
'tournament': self.tournament,
'format': self.format,
'score': self.score,
'completed': self.completed,
'has_score': self.has_score,
'winner': self.winner
}
def retry_on_exception(max_retries: int = 3, delay: float = 1.0, backoff: float = 2.0):
"""Decorator for retrying functions with exponential backoff
Args:
max_retries: Maximum number of retry attempts
delay: Initial delay between retries in seconds
backoff: Multiplier for delay after each retry (exponential backoff)
Returns:
Decorated function that automatically retries on exceptions
Example:
@retry_on_exception(max_retries=3, delay=2.0)
def fetch_data():
# This will retry up to 3 times with delays of 2s, 4s, 8s
"""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
retry_delay = delay
last_exception = None
for attempt in range(max_retries):
try:
return func(*args, **kwargs)
except Exception as e:
last_exception = e
if attempt < max_retries - 1:
logger.warning(f"Attempt {attempt + 1} failed for {func.__name__}: {e}. Retrying in {retry_delay}s...")
time.sleep(retry_delay)
retry_delay *= backoff
else:
logger.error(f"All {max_retries} attempts failed for {func.__name__}: {e}")
raise last_exception
return wrapper
return decorator
class Dota2CalendarSync:
"""Main class for synchronizing Dota 2 matches to Google Calendar
Handles fetching matches from Liquipedia and syncing them to Google Calendar,
including creating new events, updating scores, and managing TBD placeholders.
"""
def __init__(self, credentials_file='credentials.json', calendar_id='primary'):
"""Initialize the sync service
Args:
credentials_file: Path to Google service account credentials JSON
calendar_id: Google Calendar ID or 'primary' for main calendar
"""
self.credentials_file = credentials_file
self.calendar_id = calendar_id
self.service = self._authenticate()
def _authenticate(self):
"""Authenticate with Google Calendar using service account credentials
Returns:
Google Calendar service object
Raises:
SystemExit: If authentication fails
"""
try:
credentials = service_account.Credentials.from_service_account_file(
self.credentials_file,
scopes=['https://www.googleapis.com/auth/calendar']
)
service = build('calendar', 'v3', credentials=credentials)
logger.info("✓ Successfully authenticated with Google Calendar")
return service
except Exception as e:
logger.error(f"✗ Authentication failed: {e}")
sys.exit(1)
@retry_on_exception(max_retries=3, delay=2.0)
def fetch_all_matches(self) -> Tuple[List[Match], List[Match]]:
"""Fetch both upcoming and completed matches from Liquipedia
Scrapes the Liquipedia matches page for Tier 1 Dota 2 tournaments.
Automatically retries on network failures.
Returns:
Tuple of (upcoming_matches, completed_matches) lists
"""
url = 'https://liquipedia.net/dota2/Liquipedia:Matches'
headers = {
'User-Agent': 'Dota2CalendarSync/3.0 (https://github.com/youruser/dota2-calendar)'
}
logger.info("Fetching matches from Liquipedia...")
try:
response = requests.get(url, headers=headers, timeout=30)
response.raise_for_status()
soup = BeautifulSoup(response.text, 'lxml')
matches = []
# Find all timestamps (these contain match info)
timestamps = soup.find_all('span', {'data-timestamp': True})
for timestamp_elem in timestamps:
parent = timestamp_elem.find_parent('div')
if not parent:
continue
text_content = parent.get_text()
# Check if this is a Tier 1 match
is_tier1 = any(tier in text_content for tier in [
'TI2025', 'The International', 'Major', 'Premier',
'Tier 1', 'DreamLeague', 'ESL One', 'PGL Major'
])
if is_tier1:
match_data = self._parse_match(parent, timestamp_elem)
if match_data:
matches.append(match_data)
# Remove duplicates
matches = self._remove_duplicates(matches)
# Separate upcoming and completed matches
upcoming = [m for m in matches if not m.completed]
completed = [m for m in matches if m.completed]
logger.info(f"✓ Found {len(upcoming)} upcoming matches")
logger.info(f"✓ Found {len(completed)} completed matches with results")
# Merge duplicate matches at the same time
upcoming = self._merge_duplicate_matches(upcoming)
completed = self._merge_duplicate_matches(completed)
return upcoming, completed
except requests.RequestException as e:
logger.error(f"✗ Error fetching Liquipedia data: {e}")
return [], []
def _parse_match(self, parent, timestamp_elem) -> Optional[Match]:
"""Parse match data from an HTML element
Extracts team names, scores, format, and tournament information
from Liquipedia's HTML structure.
Args:
parent: Parent div element containing match information
timestamp_elem: Span element with data-timestamp attribute
Returns:
Match object if parsing successful, None otherwise
"""
try:
# Get timestamp
timestamp = timestamp_elem.get('data-timestamp')
if not timestamp:
return None
match_datetime = datetime.fromtimestamp(int(timestamp), tz=pytz.UTC)
# Extract team names from HTML structure
team_blocks = parent.find_all('div', class_='block-team')
team1 = 'TBD'
team2 = 'TBD'
if len(team_blocks) >= 2:
# Get team names - prefer span.name over a tag
team1_elem = team_blocks[0].find('span', class_='name')
if not team1_elem or not team1_elem.get_text().strip():
# Try finding any a tag with text
for a_tag in team_blocks[0].find_all('a'):
if a_tag.get_text().strip():
team1_elem = a_tag
break
team2_elem = team_blocks[1].find('span', class_='name')
if not team2_elem or not team2_elem.get_text().strip():
# Try finding any a tag with text
for a_tag in team_blocks[1].find_all('a'):
if a_tag.get_text().strip():
team2_elem = a_tag
break
if team1_elem and team2_elem:
team1 = self._clean_team_name(team1_elem.get_text().strip())
team2 = self._clean_team_name(team2_elem.get_text().strip())
# If team blocks not found, try fallback
if team1 == 'TBD' and team2 == 'TBD':
# Fallback to text parsing
text = parent.get_text()
vs_match = re.search(r'([A-Za-z0-9\s\.\-_]+?)\s*vs\s*([A-Za-z0-9\s\.\-_]+)', text)
if vs_match:
team1_raw = vs_match.group(1).strip()
team2_raw = vs_match.group(2).strip()
team1_raw = re.sub(r'^.*CEST?', '', team1_raw).strip()
team1 = self._clean_team_name(team1_raw)
team2 = self._clean_team_name(team2_raw)
# Extract score, format, and tournament
score = self._extract_score(parent)
format_str = self._extract_format(parent)
tournament = self._extract_tournament(parent)
# Check if completed
completed = False
has_score = False
winner = None
if score and not (team1 == 'TBD' and team2 == 'TBD'):
has_score = True
score_parts = re.match(r'(\d+)-(\d+)', score)
if score_parts:
score1 = int(score_parts.group(1))
score2 = int(score_parts.group(2))
# Check if series is completed
# Only mark as completed if the match has already started
now = datetime.now(pytz.UTC)
match_has_started = match_datetime <= now
if match_has_started:
completed = self._is_series_completed(score1, score2, format_str)
else:
# Future match with score - likely a placeholder/prediction
completed = False
logger.debug(f"Future match {team1} vs {team2} has score {score} but not marking as completed")
if completed:
winner = team1 if score1 > score2 else team2
# Generate match ID
match_id = self._generate_match_id(team1, team2, tournament, match_datetime)
return Match(
id=match_id,
team1=team1,
team2=team2,
datetime=match_datetime,
tournament=tournament,
format=format_str,
score=score,
completed=completed,
has_score=has_score,
winner=winner
)
except Exception as e:
logger.debug(f"Failed to parse match: {e}")
return None
def _extract_score(self, parent) -> Optional[str]:
"""Extract score from match element
Looks for score in structured HTML elements first,
then falls back to text pattern matching.
Args:
parent: Parent element containing match information
Returns:
Score string (e.g., '2-1') or None if no valid score found
"""
# Look for score in structured elements
score_holder = parent.find('div', class_='match-info-header-scoreholder')
if score_holder:
score_elems = score_holder.find_all('span', class_='match-info-header-scoreholder-score')
if len(score_elems) >= 2:
try:
score1 = int(score_elems[0].get_text().strip())
score2 = int(score_elems[1].get_text().strip())
if 0 <= score1 <= 5 and 0 <= score2 <= 5 and (score1 + score2) > 0:
return f"{score1}-{score2}"
except ValueError:
pass
# Try text pattern
text = parent.get_text()
score_pattern = re.search(r'(\d{1,2})-(\d{1,2})', text)
if score_pattern:
score1 = int(score_pattern.group(1))
score2 = int(score_pattern.group(2))
if 0 <= score1 <= 5 and 0 <= score2 <= 5 and (score1 + score2) > 0:
surrounding_text = text[max(0, score_pattern.start()-5):score_pattern.end()+5]
if not re.search(r'\d{4}-\d{1,2}-\d{1,2}', surrounding_text):
return f"{score1}-{score2}"
return None
def _extract_format(self, parent) -> Optional[str]:
"""Extract match format (Bo1, Bo3, Bo5)
Args:
parent: Parent element containing match information
Returns:
Format string (e.g., 'Bo3') or None if not found
"""
format_elem = parent.find('span', class_='match-info-header-scoreholder-lower')
if format_elem:
format_text = format_elem.get_text().strip()
format_match = re.search(r'(Bo\d)', format_text)
if format_match:
return format_match.group(1)
# Fallback to text search
text = parent.get_text()
format_match = re.search(r'\(?(Bo\d)\)?', text)
if format_match:
return format_match.group(1)
return None
def _extract_tournament(self, parent) -> Optional[str]:
"""Extract tournament name from match element
Special handling for TI2025 and Major tournaments.
Args:
parent: Parent element containing match information
Returns:
Tournament name or None if not found
"""
tournament_elem = parent.find('div', class_='match-info-tournament')
if tournament_elem:
tournament_text = tournament_elem.get_text().strip()
tournament_text = re.sub(r'\+ Add details.*', '', tournament_text).strip()
if 'TI2025' in tournament_text:
tournament = 'The International 2025'
round_match = re.search(r'Round\s+\d+', tournament_text)
if round_match:
tournament += f" - {round_match.group(0)}"
return tournament
return tournament_text
# Fallback
text = parent.get_text()
if 'TI2025' in text:
tournament = 'The International 2025'
round_match = re.search(r'Round\s+\d+', text)
if round_match:
tournament += f" - {round_match.group(0)}"
return tournament
elif 'Major' in text:
major_match = re.search(r'[\w\s]+Major', text)
if major_match:
return major_match.group(0).strip()
return None
def _is_series_completed(self, score1: int, score2: int, format_str: Optional[str]) -> bool:
"""Check if a series is completed based on score and format
Args:
score1: Score for team 1
score2: Score for team 2
format_str: Match format (Bo1, Bo3, Bo5)
Returns:
True if series is completed (someone has won enough games)
"""
if not format_str:
return score1 >= 2 or score2 >= 2
if 'Bo3' in format_str:
return score1 >= 2 or score2 >= 2
elif 'Bo5' in format_str:
return score1 >= 3 or score2 >= 3
elif 'Bo1' in format_str:
return True
return True
def _clean_team_name(self, name: str) -> str:
"""Clean and normalize team name
Removes extra whitespace, parenthetical notes, dates, etc.
Args:
name: Raw team name from HTML
Returns:
Cleaned team name
"""
name = re.sub(r'\s+', ' ', name).strip()
name = re.sub(r'\s*\(.*?\)\s*$', '', name)
name = re.sub(r'^\d{4}-\d{2}-\d{2}.*', '', name).strip()
name = re.sub(r'^\w+\s+\d+,\s+\d{4}.*', '', name).strip()
return name
def _merge_duplicate_matches(self, matches: List[Match]) -> List[Match]:
"""Merge duplicate matches at the same time with one common team
When multiple matches occur at the same time with one common team,
they likely represent the same match with different team name variations.
This method merges such duplicates, keeping the longer/more complete team name.
Args:
matches: List of Match objects to check for duplicates
Returns:
List of Match objects with duplicates merged
"""
if not matches:
return matches
# Group matches by time (30-minute window)
from collections import defaultdict
matches_by_time = defaultdict(list)
for match in matches:
# Create time key with 30-minute precision
time_key = match.datetime.replace(minute=(match.datetime.minute // 30) * 30, second=0, microsecond=0)
matches_by_time[time_key].append(match)
merged_matches = []
for time_key, time_matches in matches_by_time.items():
if len(time_matches) <= 1:
# No duplicates at this time
merged_matches.extend(time_matches)
continue
# Check for matches with common teams
processed = set()
for i, match1 in enumerate(time_matches):
if i in processed:
continue
merged = False
for j, match2 in enumerate(time_matches[i+1:], i+1):
if j in processed:
continue
# Check if matches share a common team
common_team = None
different_teams = []
if match1.team1 == match2.team1:
common_team = match1.team1
different_teams = [match1.team2, match2.team2]
elif match1.team1 == match2.team2:
common_team = match1.team1
different_teams = [match1.team2, match2.team1]
elif match1.team2 == match2.team1:
common_team = match1.team2
different_teams = [match1.team1, match2.team2]
elif match1.team2 == match2.team2:
common_team = match1.team2
different_teams = [match1.team1, match2.team1]
if common_team and common_team != 'TBD':
# Found matches with a common non-TBD team
# Keep the longer/more complete team name for the different team
chosen_different = max(different_teams, key=len)
# Create merged match
merged_match = Match(
id=match1.id, # Keep first match's ID
team1=common_team if match1.team1 == common_team else chosen_different,
team2=chosen_different if match1.team1 == common_team else common_team,
datetime=match1.datetime,
tournament=match1.tournament or match2.tournament,
format=match1.format or match2.format,
score=match1.score or match2.score,
completed=match1.completed or match2.completed,
has_score=match1.has_score or match2.has_score,
winner=match1.winner or match2.winner
)
merged_matches.append(merged_match)
processed.add(i)
processed.add(j)
merged = True
logger.info(f"Merged duplicate matches at {time_key}: "
f"{match1.team1} vs {match1.team2} + {match2.team1} vs {match2.team2} "
f"-> {merged_match.team1} vs {merged_match.team2}")
break
if not merged and i not in processed:
# No merge found for this match
merged_matches.append(match1)
processed.add(i)
return merged_matches
def _generate_match_id(self, team1: str, team2: str, tournament: Optional[str], match_datetime: datetime) -> str:
"""Generate a unique ID for a match
Uses MD5 hash of team names and tournament. For TBD vs TBD matches,
includes datetime to ensure uniqueness.
Args:
team1: First team name
team2: Second team name
tournament: Tournament name
match_datetime: Match scheduled time
Returns:
16-character hex string ID
"""
id_parts = []
# For TBD vs TBD matches, include datetime to make them unique
if team1 == 'TBD' and team2 == 'TBD':
id_parts.append(str(match_datetime))
if tournament:
id_parts.append(tournament)
else:
# Normal matches: use teams and tournament
id_parts.extend([team1, team2])
if tournament:
id_parts.append(tournament)
else:
id_parts.append(str(match_datetime.date()))
unique_string = '_'.join(id_parts)
return hashlib.md5(unique_string.encode()).hexdigest()[:16]
def _remove_duplicates(self, matches: List[Match]) -> List[Match]:
"""Remove duplicate matches based on ID
Args:
matches: List of Match objects
Returns:
List with duplicates removed (keeps last occurrence)
"""
unique_matches = {}
for match in matches:
unique_matches[match.id] = match
return list(unique_matches.values())
@retry_on_exception(max_retries=3, delay=1.0)
def get_existing_events(self, days_back=7, days_ahead=30) -> Dict[str, Any]:
"""Get existing Dota 2 events from Google Calendar
Fetches events within the specified time range and builds two indexes:
- By match ID (from event description)
- By match details (teams + tournament)
Args:
days_back: Number of days to look back
days_ahead: Number of days to look ahead
Returns:
Dictionary with event indexes:
- Keys are match IDs
- Special '_by_match' key contains team+tournament index
"""
try:
now = datetime.utcnow()
time_min = (now - timedelta(days=days_back)).isoformat() + 'Z'
time_max = (now + timedelta(days=days_ahead)).isoformat() + 'Z'
logger.info("Checking existing events in calendar...")
events_result = self.service.events().list(
calendarId=self.calendar_id,
timeMin=time_min,
timeMax=time_max,
maxResults=500,
singleEvents=True,
orderBy='startTime'
).execute()
events = events_result.get('items', [])
# Filter for Dota 2 events
dota_events_by_id = {}
dota_events_by_match = {}
for event in events:
summary = event.get('summary', '')
is_dota = ('Dota 2' in summary or
'The International' in summary or
'TI2025' in summary or
'[' in summary and 'vs' in summary)
if is_dota:
description = event.get('description', '')
# Extract ID from description
id_match = re.search(r'ID:\s*([a-f0-9]+)', description)
if id_match:
dota_events_by_id[id_match.group(1)] = event
# Create key based on teams and tournament
summary = event.get('summary', '')
summary = summary.replace('[COMPLETED] ', '')
summary = re.sub(r'^✓\s+\d+[-:]\d+\s+', '', summary)
summary = re.sub(r'^\d+[-:]\d+\s+', '', summary)
summary = re.sub(r'\s*\([0-9\-\?]+\)\s*$', '', summary)
# Extract teams and tournament
match = re.search(r'^(.*?)\s+vs\s+(.*?)\s*\[(.*?)\]$', summary)
if not match:
match = re.search(r'Dota 2 - (.*?):\s*(.*?)\s+vs\s+(.*?)$', summary)
if match:
tournament = match.group(1).strip()
team1 = match.group(2).strip()
team2 = match.group(3).strip()
else:
continue
else:
team1 = match.group(1).strip()
team2 = match.group(2).strip()
tournament = match.group(3).strip()
match_key = f"{team1}_{team2}_{tournament}"
dota_events_by_match[match_key] = event
logger.info(f"✓ Found {len(dota_events_by_id)} existing Dota 2 events")
combined = {}
combined.update(dota_events_by_id)
combined['_by_match'] = dota_events_by_match
return combined
except Exception as e:
logger.error(f"✗ Error fetching calendar events: {e}")
raise
def find_existing_event(self, match: Match, existing_events: Dict[str, Any]) -> Optional[Dict[str, Any]]:
"""Find existing calendar event for a match
Tries multiple matching strategies:
1. Direct ID match
2. Team + tournament match
3. Teams only (for live score updates)
4. TBD placeholder match (within 1 hour time window)
Args:
match: Match object to find
existing_events: Dictionary of existing calendar events
Returns:
Calendar event dict if found, None otherwise
"""
# Try by ID first
if match.id in existing_events:
return existing_events[match.id]
# Try by match details
if '_by_match' not in existing_events:
return None
by_match = existing_events['_by_match']
# Direct match
match_key = f"{match.team1}_{match.team2}_{match.tournament}"
if match_key in by_match:
return by_match[match_key]
# Try to find by teams only (for live updates)
for event_key, event in by_match.items():
if ((f"{match.team1}_{match.team2}" in event_key or
f"{match.team2}_{match.team1}" in event_key) and
match.tournament and match.tournament in event_key):
return event
# Special handling for TBD matches - check if this match should update a TBD placeholder
if not (match.team1 == 'TBD' and match.team2 == 'TBD'):
for event_key, event in by_match.items():
# Check if this is a TBD event that could be updated with real teams
if 'TBD' in event_key and match.tournament and match.tournament in event_key:
# Parse the event key to get teams
key_parts = event_key.split('_')
if len(key_parts) >= 3:
existing_team1 = key_parts[0]
existing_team2 = key_parts[1]
# Check if this TBD event matches the incoming match
# Match if: one team is TBD and the other matches, or both are TBD
teams_match = False
if existing_team1 == 'TBD' and existing_team2 == 'TBD':
# Both TBD - match by time
teams_match = True
elif existing_team1 == 'TBD' and existing_team2 == match.team2:
# First team is TBD, second matches
teams_match = True
elif existing_team1 == match.team1 and existing_team2 == 'TBD':
# First team matches, second is TBD
teams_match = True
elif existing_team2 == 'TBD' and existing_team1 == match.team1:
# Reverse check
teams_match = True
elif existing_team1 == 'TBD' and existing_team2 == match.team1:
# Reverse check
teams_match = True
if teams_match:
# Check if time matches (within 1 hour)
event_start = event['start'].get('dateTime', event['start'].get('date'))
event_dt = datetime.fromisoformat(event_start.replace('Z', '+00:00'))
if abs((event_dt - match.datetime).total_seconds()) < 3600:
logger.info(f"Found TBD match to update: {existing_team1} vs {existing_team2} -> {match.team1} vs {match.team2}")
return event
return None
def create_calendar_event(self, match: Match) -> Dict[str, Any]:
"""Create a Google Calendar event for a match
Args:
match: Match object with event details
Returns:
Dictionary with Google Calendar event format
"""
# Build summary
if match.tournament:
summary = f"{match.team1} vs {match.team2} [{match.tournament}]"
else:
summary = f"{match.team1} vs {match.team2}"
# Build description
description_parts = []
if match.tournament:
description_parts.append(f"Tournament: {match.tournament}")
description_parts.append(f"Match: {match.team1} vs {match.team2}")
if match.format:
description_parts.append(f"Format: {match.format}")
if match.completed:
description_parts.append(f"\n🏆 RESULT: {match.score or 'Unknown'}")
description_parts.append(f"Winner: {match.winner or 'Unknown'}")
description_parts.append(f"ID: {match.id}")
description_parts.append("\nSource: Liquipedia")
description = '\n'.join(description_parts)
# Set duration based on format
duration = 2
if match.format:
if 'Bo5' in match.format:
duration = 4
elif 'Bo3' in match.format:
duration = 3
elif 'Bo1' in match.format:
duration = 1
end_time = match.datetime + timedelta(hours=duration)
event = {
'summary': summary,
'description': description,
'start': {
'dateTime': match.datetime.isoformat(),
'timeZone': 'UTC',
},
'end': {
'dateTime': end_time.isoformat(),
'timeZone': 'UTC',
},
'reminders': {
'useDefault': False,
'overrides': [
{'method': 'popup', 'minutes': 30},
],
},
'colorId': '9', # Blue
}
return event
@retry_on_exception(max_retries=2, delay=1.0)
def update_event_time(self, event_id: str, new_datetime: datetime) -> bool:
"""Update the time of an existing event
Preserves event duration when updating start time.
Args:
event_id: Google Calendar event ID
new_datetime: New start time for the event
Returns:
True if update successful, False otherwise
"""
try:
# Get the existing event
event = self.service.events().get(
calendarId=self.calendar_id,
eventId=event_id
).execute()
# Calculate duration from existing event
start_dt = datetime.fromisoformat(event['start']['dateTime'].replace('Z', '+00:00'))
end_dt = datetime.fromisoformat(event['end']['dateTime'].replace('Z', '+00:00'))
duration = end_dt - start_dt
# Update times
event['start']['dateTime'] = new_datetime.isoformat()
event['end']['dateTime'] = (new_datetime + duration).isoformat()
# Add note about time change
description = event.get('description', '')
timestamp = datetime.now(pytz.UTC).strftime('%Y-%m-%d %H:%M UTC')
if 'Last updated:' in description:
description = re.sub(r'Last updated:.*', f"Last updated: {timestamp}", description)
else:
description += f"\nLast updated: {timestamp}"
event['description'] = description
# Update the event
self.service.events().update(
calendarId=self.calendar_id,
eventId=event_id,
body=event
).execute()
return True
except Exception as e:
logger.error(f"Error updating event time for {event_id}: {e}")
return False
@retry_on_exception(max_retries=2, delay=1.0)
def update_event_with_teams(self, event_id: str, match: Match) -> bool:
"""Update a TBD event with actual team names
Used when TBD placeholder teams are determined.
Args:
event_id: Google Calendar event ID
match: Match object with updated team names
Returns:
True if update successful, False otherwise
"""
try:
# Get the existing event
event = self.service.events().get(
calendarId=self.calendar_id,
eventId=event_id
).execute()
# Update summary
if match.tournament:
new_summary = f"{match.team1} vs {match.team2} [{match.tournament}]"
else:
new_summary = f"{match.team1} vs {match.team2}"
# Update description
description = event.get('description', '')
description = re.sub(r'Match: .*?\n', f"Match: {match.team1} vs {match.team2}\n", description)
if 'ID:' in description:
description = re.sub(r'ID: [a-f0-9]+', f"ID: {match.id}", description)
# Add update timestamp
timestamp = datetime.now(pytz.UTC).strftime('%Y-%m-%d %H:%M UTC')
if 'Teams updated:' in description:
description = re.sub(r'Teams updated:.*', f"Teams updated: {timestamp}", description)
else:
description = description.replace('\nSource:', f"\nTeams updated: {timestamp}\nSource:")
event['summary'] = new_summary
event['description'] = description
# Update the event
self.service.events().update(
calendarId=self.calendar_id,
eventId=event_id,
body=event
).execute()
return True
except Exception as e:
logger.error(f"Error updating event with teams for {event_id}: {e}")
return False
@retry_on_exception(max_retries=2, delay=1.0)
def update_event_with_score(self, event_id: str, match: Match) -> bool:
"""Update an existing calendar event with in-progress score
Shows current score for ongoing matches (not yet completed).
Args:
event_id: Google Calendar event ID
match: Match object with current score
Returns:
True if update successful, False otherwise
"""
try:
# Get the existing event
event = self.service.events().get(
calendarId=self.calendar_id,
eventId=event_id
).execute()
# Update description
description = event.get('description', '')
score_text = f"📊 CURRENT SCORE: {match.score or 'Unknown'}"
if '📊 CURRENT SCORE:' in description:
description = re.sub(r'📊 CURRENT SCORE:.*?\n', f"{score_text}\n", description)
else:
if 'ID:' in description:
description = description.replace('ID:', f"\n{score_text}\nID:")
else:
description += f"\n{score_text}"
# Update summary - remove ALL existing scores first
summary = event.get('summary', '')
# Remove all score patterns (including those after checkmark)
summary = re.sub(r'\d+-\d+\s+', '', summary)
# Clean up extra spaces
summary = ' '.join(summary.split())
# Add new score at the beginning
summary = f"{match.score} {summary}"
event['description'] = description
event['summary'] = summary
# Update the event
self.service.events().update(
calendarId=self.calendar_id,
eventId=event_id,
body=event
).execute()
return True
except Exception as e:
logger.error(f"Error updating event score for {event_id}: {e}")
return False
@retry_on_exception(max_retries=2, delay=1.0)
def update_event_with_result(self, event_id: str, match: Match) -> bool:
"""Update an existing calendar event with match results
Marks match as completed with final score and winner.
Args:
event_id: Google Calendar event ID
match: Match object with final results
Returns:
True if update successful, False otherwise
"""
try:
# Get the existing event
event = self.service.events().get(
calendarId=self.calendar_id,
eventId=event_id
).execute()
# Update description
description = event.get('description', '')
result_text = f"🏆 RESULT: {match.score or 'Unknown'}\nWinner: {match.winner or 'Unknown'}"
if '🏆 RESULT:' in description:
description = re.sub(
r'🏆 RESULT:.*?\n.*?Winner:.*?\n',
f"{result_text}\n",
description,
flags=re.DOTALL
)
else:
if 'ID:' in description:
description = description.replace('ID:', f"\n{result_text}\nID:")
else:
description += f"\n{result_text}"
# Update summary - remove ALL existing scores and checkmarks first
summary = event.get('summary', '')
# Remove all score patterns
summary = re.sub(r'\d+-\d+\s+', '', summary)
# Remove checkmark if present
summary = re.sub(r'^✓\s+', '', summary)
# Clean up extra spaces
summary = ' '.join(summary.split())
# Add checkmark and final score
summary = f"{match.score} {summary}"
event['description'] = description
event['summary'] = summary
# Update the event
self.service.events().update(
calendarId=self.calendar_id,
eventId=event_id,
body=event
).execute()
return True
except Exception as e:
logger.error(f"Error updating event result for {event_id}: {e}")
return False
def check_time_difference(self, event_datetime: Any, new_datetime: datetime) -> bool:
"""Check if there's a significant time difference (>= 5 minutes)
Args:
event_datetime: Current event datetime (string or datetime object)
new_datetime: New datetime to compare
Returns:
True if difference is 5 minutes or more
"""
# Parse event datetime
if isinstance(event_datetime, str):
if event_datetime.endswith('Z'):
event_dt = datetime.fromisoformat(event_datetime.replace('Z', '+00:00'))
else:
event_dt = datetime.fromisoformat(event_datetime)
else:
event_dt = event_datetime
# Ensure timezone aware
if event_dt.tzinfo is None:
event_dt = event_dt.replace(tzinfo=pytz.UTC)
if new_datetime.tzinfo is None:
new_datetime = new_datetime.replace(tzinfo=pytz.UTC)
# Calculate difference in minutes
diff = abs((event_dt - new_datetime).total_seconds() / 60)
return diff >= 5
@retry_on_exception(max_retries=2, delay=1.0)
def delete_calendar_event(self, event_id: str) -> bool:
"""Delete a calendar event
Args:
event_id: Google Calendar event ID to delete
Returns:
True if deletion successful, False otherwise
"""
try:
self.service.events().delete(
calendarId=self.calendar_id,
eventId=event_id
).execute()
return True
except Exception as e:
logger.error(f"Error deleting event {event_id}: {e}")
return False
def process_upcoming_matches(self, upcoming_matches: List[Match], existing_events: Dict[str, Any],
dry_run: bool, update_results: bool, update_times: bool) -> Dict[str, int]:
"""Process upcoming matches and update calendar
Handles:
- Adding new matches
- Updating TBD placeholders with team names
- Updating in-progress scores
- Updating match times
Args:
upcoming_matches: List of upcoming Match objects
existing_events: Dictionary of existing calendar events
dry_run: If True, only simulate changes
update_results: Whether to update match scores
update_times: Whether to update match times
Returns:
Dictionary with counters for added, updated, skipped matches
"""
counters = {
'added': 0,
'updated': 0,
'time_updated': 0,
'skipped': 0,
'errors': 0
}
updated_tbd_events = set()
now = datetime.now(pytz.UTC)
twelve_hours_ago = now - timedelta(hours=12)
future_matches = [m for m in upcoming_matches if m.datetime >= twelve_hours_ago]
logger.info("\nProcessing upcoming matches...")
logger.info("-" * 30)
for match in future_matches:
try:
existing_event = self.find_existing_event(match, existing_events)
if existing_event:
# Check if this is a TBD match that now has team names
summary = existing_event.get('summary', '')
is_tbd_update = 'TBD' in summary and (match.team1 != 'TBD' or match.team2 != 'TBD')
if is_tbd_update:
if dry_run:
logger.info(f"◯ Would update TBD match with teams: {match.team1} vs {match.team2}")
counters['updated'] += 1
else:
if self.update_event_with_teams(existing_event['id'], match):
logger.info(f"✓ Updated TBD match with teams: {match.team1} vs {match.team2}")
counters['updated'] += 1
updated_tbd_events.add(existing_event['id'])
time.sleep(0.2)
else:
counters['errors'] += 1
# Check if this match has a score and needs update
elif match.has_score and update_results:
# Check current event status
summary = existing_event.get('summary', '')
current_score = None
score_in_summary = re.search(r'✓?\s*(\d+[-:]\d+)', summary)
if score_in_summary:
current_score = score_in_summary.group(1).replace(':', '-')
if current_score == match.score:
logger.info(f"⊘ Score unchanged: {match.team1} vs {match.team2} ({match.score})")
counters['skipped'] += 1
else:
if match.completed:
if dry_run:
logger.info(f"◯ Would update completed result: {match.team1} vs {match.team2} - {match.score}")
counters['updated'] += 1
else:
if self.update_event_with_result(existing_event['id'], match):
logger.info(f"✓ Updated completed result: {match.team1} vs {match.team2} - {match.score}")
counters['updated'] += 1
time.sleep(0.2)
else:
counters['errors'] += 1
else:
if dry_run:
logger.info(f"◯ Would update in-progress score: {match.team1} vs {match.team2} - {match.score}")
counters['updated'] += 1
else:
if self.update_event_with_score(existing_event['id'], match):
logger.info(f"📊 Updated in-progress score: {match.team1} vs {match.team2} - {match.score}")
counters['updated'] += 1
time.sleep(0.2)
else:
counters['errors'] += 1
# Check if time has changed
elif update_times:
event_start = existing_event['start'].get('dateTime', existing_event['start'].get('date'))
if self.check_time_difference(event_start, match.datetime):
if dry_run:
old_time = datetime.fromisoformat(event_start.replace('Z', '+00:00'))
logger.info(f"◯ Would update time: {match.team1} vs {match.team2}")
logger.info(f" Old: {old_time.strftime('%Y-%m-%d %H:%M UTC')}")
logger.info(f" New: {match.datetime.strftime('%Y-%m-%d %H:%M UTC')}")
counters['time_updated'] += 1
else:
old_time = datetime.fromisoformat(event_start.replace('Z', '+00:00'))
if self.update_event_time(existing_event['id'], match.datetime):
logger.info(f"⏰ Updated time: {match.team1} vs {match.team2}")
logger.info(f" Old: {old_time.strftime('%Y-%m-%d %H:%M UTC')}")
logger.info(f" New: {match.datetime.strftime('%Y-%m-%d %H:%M UTC')}")
counters['time_updated'] += 1
time.sleep(0.2)
else:
counters['errors'] += 1
else:
logger.info(f"⊘ No change: {match.team1} vs {match.team2}")
counters['skipped'] += 1
else:
logger.info(f"⊘ Skipping (exists): {match.team1} vs {match.team2}")
counters['skipped'] += 1
else:
# New match
if dry_run:
logger.info(f"◯ Would add: {match.team1} vs {match.team2} at {match.datetime.strftime('%Y-%m-%d %H:%M UTC')}")
counters['added'] += 1
else:
try:
event = self.create_calendar_event(match)
self.service.events().insert(
calendarId=self.calendar_id,
body=event
).execute()
logger.info(f"✓ Added: {match.team1} vs {match.team2} at {match.datetime.strftime('%Y-%m-%d %H:%M UTC')}")
counters['added'] += 1
time.sleep(0.2)
except Exception as e:
logger.error(f"✗ Error adding {match.team1} vs {match.team2}: {e}")
counters['errors'] += 1
except Exception as e:
logger.error(f"Error processing match {match.team1} vs {match.team2}: {e}")
counters['errors'] += 1
continue
counters['updated_tbd_events'] = updated_tbd_events
return counters
def process_completed_matches(self, completed_matches: List[Match], existing_events: Dict[str, Any],
dry_run: bool) -> Dict[str, int]:
"""Process completed matches and update results
Updates calendar events with final scores and winners.
Args:
completed_matches: List of completed Match objects
existing_events: Dictionary of existing calendar events
dry_run: If True, only simulate changes
Returns:
Dictionary with counters for updated matches and errors
"""
counters = {
'updated': 0,
'errors': 0
}
logger.info("\nProcessing completed match results...")
logger.info("-" * 30)
for match in completed_matches:
try:
existing_event = self.find_existing_event(match, existing_events)
if existing_event:
# Check if already marked as completed
summary = existing_event.get('summary', '')
if '' in summary or '[COMPLETED]' in summary:
logger.info(f"⊘ Already updated: {match.team1} vs {match.team2} ({match.score})")
else:
if dry_run:
logger.info(f"◯ Would update result: {match.team1} vs {match.team2} - {match.score}")
counters['updated'] += 1
else:
if self.update_event_with_result(existing_event['id'], match):
logger.info(f"✓ Updated result: {match.team1} vs {match.team2} - {match.score}")
counters['updated'] += 1
time.sleep(0.2)
else:
counters['errors'] += 1
except Exception as e:
logger.error(f"Error processing completed match {match.team1} vs {match.team2}: {e}")
counters['errors'] += 1
continue
return counters
def clean_duplicate_and_expired_events(self, existing_events: Dict[str, Any],
updated_tbd_events: set, dry_run: bool) -> int:
"""Clean up duplicate and expired TBD events
Removes:
- TBD events that have ended
- Duplicate TBD vs TBD events at the same time
- TBD events superseded by confirmed matches
Args:
existing_events: Dictionary of existing calendar events
updated_tbd_events: Set of event IDs that were updated (to skip)
dry_run: If True, only simulate deletions
Returns:
Number of events deleted
"""
deleted_count = 0
now = datetime.now(pytz.UTC)
logger.info("\nChecking for expired, duplicate, and superseded TBD events to delete...")
logger.info("-" * 30)
# Group events by time
events_by_time = {}
tbd_by_time = {}
for key, event in existing_events.items():
if key == '_by_match':
continue
summary = event.get('summary', '')
event_id = event['id']
# Skip if this event was updated
if event_id in updated_tbd_events:
continue
# Get event time and end time
event_start = event['start'].get('dateTime', event['start'].get('date'))
event_dt = datetime.fromisoformat(event_start.replace('Z', '+00:00'))
# Get event end time
event_end = event.get('end', {}).get('dateTime', event.get('end', {}).get('date'))
if event_end:
event_end_dt = datetime.fromisoformat(event_end.replace('Z', '+00:00'))
else:
event_end_dt = event_dt + timedelta(hours=3)
# Use 30-minute window for "same time"
time_key = (event_dt.year, event_dt.month, event_dt.day,
event_dt.hour, event_dt.minute // 30)
if time_key not in events_by_time:
events_by_time[time_key] = {'tbd': [], 'confirmed': []}
# Categorize events
if 'vs TBD' in summary or 'TBD vs' in summary:
events_by_time[time_key]['tbd'].append(event)
# Delete if match has ended and still contains TBD
if event_end_dt < now:
if not dry_run:
if self.delete_calendar_event(event_id):
logger.info(f"🗑️ Deleted ended TBD event: {summary}")
deleted_count += 1
time.sleep(0.2)
else:
logger.info(f"◯ Would delete ended TBD event: {summary}")
deleted_count += 1
continue
# Track non-expired TBD events
if 'TBD vs TBD' in summary:
simple_time_key = event_dt.strftime('%Y-%m-%d %H:%M')
if simple_time_key not in tbd_by_time:
tbd_by_time[simple_time_key] = []
tbd_by_time[simple_time_key].append(event)
else:
events_by_time[time_key]['confirmed'].append(event)
# Delete TBD events superseded by confirmed matches
for time_key, events in events_by_time.items():
if events['confirmed'] and events['tbd']:
for tbd_event in events['tbd']:
tbd_summary = tbd_event.get('summary', '')
# Delete any TBD event that has a confirmed match at the same time
# This includes "TBD vs TBD", "TBD vs Team", and "Team vs TBD"
if events['confirmed']:
if not dry_run:
if self.delete_calendar_event(tbd_event['id']):
logger.info(f"🗑️ Deleted TBD event superseded by confirmed match: {tbd_summary}")
deleted_count += 1
time.sleep(0.2)
else:
logger.info(f"◯ Would delete TBD event superseded by confirmed match: {tbd_summary}")
deleted_count += 1
# Delete duplicate TBD vs TBD events
for time_key, events in tbd_by_time.items():
if len(events) > 1:
logger.info(f"Found {len(events)} duplicate TBD events at {time_key}")
for event in events[1:]:
if not dry_run:
if self.delete_calendar_event(event['id']):
logger.info(f"🗑️ Deleted duplicate TBD event")
deleted_count += 1
time.sleep(0.2)
else:
logger.info(f"◯ Would delete duplicate TBD event")
deleted_count += 1
return deleted_count
def sync_matches_to_calendar(self, dry_run=False, update_results=True, update_times=True, delete_old_tbd=True):
"""Main sync function with improved structure and error handling
Coordinates the entire sync process:
1. Fetches matches from Liquipedia
2. Gets existing events from Google Calendar
3. Processes upcoming and completed matches
4. Cleans up duplicates and expired TBD events
5. Displays summary
Args:
dry_run: If True, only simulate changes
update_results: Whether to update match scores/results
update_times: Whether to update changed match times
delete_old_tbd: Whether to delete expired TBD events
"""
logger.info("\n" + "="*50)
logger.info("Starting Dota 2 Calendar Sync v4.0")
logger.info("="*50 + "\n")
try:
# Fetch all matches
upcoming_matches, completed_matches = self.fetch_all_matches()
if not upcoming_matches and not completed_matches:
logger.info("No matches found to sync")
return
# Get existing events
existing_events = self.get_existing_events(days_back=7, days_ahead=30)
# Process upcoming matches
upcoming_counters = self.process_upcoming_matches(
upcoming_matches, existing_events, dry_run, update_results, update_times
)
# Process completed matches
completed_counters = {}
if update_results and completed_matches:
completed_counters = self.process_completed_matches(
completed_matches, existing_events, dry_run
)
# Clean up duplicate and expired events
deleted_count = 0
if delete_old_tbd and not dry_run:
deleted_count = self.clean_duplicate_and_expired_events(
existing_events,
upcoming_counters.get('updated_tbd_events', set()),
dry_run
)
# Summary
logger.info("\n" + "="*50)
logger.info("Sync Summary")
logger.info("="*50)
logger.info(f"✓ Added: {upcoming_counters.get('added', 0)} matches")
if upcoming_counters.get('time_updated', 0) > 0:
logger.info(f"⏰ Time updated: {upcoming_counters.get('time_updated', 0)} matches")
total_updated = upcoming_counters.get('updated', 0) + completed_counters.get('updated', 0)
if total_updated > 0:
logger.info(f"✓ Results updated: {total_updated} matches")
if deleted_count > 0:
logger.info(f"🗑️ Deleted: {deleted_count} expired TBD events")
logger.info(f"⊘ Skipped: {upcoming_counters.get('skipped', 0)} matches (no changes)")
total_errors = upcoming_counters.get('errors', 0) + completed_counters.get('errors', 0)
if total_errors > 0:
logger.warning(f"✗ Errors: {total_errors} matches")
if dry_run:
logger.info("\n⚠ DRY RUN - No actual changes were made")
logger.info("\n✓ Sync complete!")
except Exception as e:
logger.error(f"Fatal error during sync: {e}", exc_info=True)
raise
def main():
"""Main entry point for the script
Parses command line arguments and runs the sync process.
"""
parser = argparse.ArgumentParser(
description='Sync Dota 2 Tier 1 matches from Liquipedia to Google Calendar v4.0'
)
parser.add_argument(
'--calendar-id',
default='primary',
help='Google Calendar ID (default: primary).'
)
parser.add_argument(
'--dry-run',
action='store_true',
help='Perform a dry run without actually creating/updating events'
)
parser.add_argument(
'--no-results',
action='store_true',
help='Skip updating completed match results'
)
parser.add_argument(
'--no-time-updates',
action='store_true',
help='Skip updating match times'
)
parser.add_argument(
'--credentials',
default='credentials.json',
help='Path to Google service account credentials JSON file'
)
parser.add_argument(
'--log-level',
default='INFO',
choices=['DEBUG', 'INFO', 'WARNING', 'ERROR'],
help='Set the logging level'
)
args = parser.parse_args()
# Set log level
logger.setLevel(getattr(logging, args.log_level))
# Notice
logger.info("\n" + "!"*60)
logger.info("Dota 2 Calendar Sync v4.0")
logger.info("Features: Dataclasses, retry mechanism, professional logging")
logger.info("Service Account: calendar-bot@tunpok.iam.gserviceaccount.com")
logger.info("!"*60 + "\n")
# Initialize and run sync
try:
sync = Dota2CalendarSync(
credentials_file=args.credentials,
calendar_id=args.calendar_id
)
sync.sync_matches_to_calendar(
dry_run=args.dry_run,
update_results=not args.no_results,
update_times=not args.no_time_updates,
delete_old_tbd=True
)
except KeyboardInterrupt:
logger.info("\n\nSync cancelled by user")
sys.exit(0)
except Exception as e:
logger.error(f"\n✗ Fatal error: {e}")
sys.exit(1)
if __name__ == "__main__":
main()