#!/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()