#!/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""" 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): """Ensure datetime is timezone aware""" 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 to dictionary for compatibility""" 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""" 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: def __init__(self, credentials_file='credentials.json', calendar_id='primary'): 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""" 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""" 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") 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 element using HTML structure""" 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 completed = self._is_series_completed(score1, score2, format_str) 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""" # 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""" 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""" 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""" 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""" 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 _generate_match_id(self, team1: str, team2: str, tournament: Optional[str], match_datetime: datetime) -> str: """Generate a unique ID for a match""" 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""" 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""" 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""" # 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 if not (match.team1 == 'TBD' and match.team2 == 'TBD'): for event_key, event in by_match.items(): if 'TBD_TBD' in event_key and match.tournament and match.tournament in event_key: # 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: {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""" # 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""" 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""" 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""" 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 summary = event.get('summary', '') summary = re.sub(r'^(\d+[-:]\d+\s+)+', '', summary) 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""" 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 summary = event.get('summary', '') summary = re.sub(r'^(\d+[-:]\d+\s+)+', '', summary) summary = re.sub(r'^✓\s+', '', summary) 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)""" # 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""" 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""" 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""" 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""" 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', '') if 'TBD vs TBD' in tbd_summary and events['confirmed']: if not dry_run: if self.delete_calendar_event(tbd_event['id']): logger.info(f"🗑️ Deleted TBD vs TBD event at same time as confirmed match") deleted_count += 1 time.sleep(0.2) else: logger.info(f"◯ Would delete TBD vs TBD event: {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""" 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(): 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()